<?php

declare(strict_types=1);

namespace Drupal\flowdrop_node_processor\Plugin\FlowDropNodeProcessor;

use Drupal\flowdrop\Attribute\FlowDropNodeProcessor;
use Drupal\flowdrop\DTO\ParameterBagInterface;
use Drupal\flowdrop\DTO\ValidationResult;
use Drupal\flowdrop\Plugin\FlowDropNodeProcessor\AbstractFlowDropNodeProcessor;
use Symfony\Component\DependencyInjection\ContainerInterface;

/**
 * Executor for Dataframe Operations nodes.
 */
#[FlowDropNodeProcessor(
  id: "dataframe_operations",
  label: new \Drupal\Core\StringTranslation\TranslatableMarkup("Dataframe Operations"),
  description: "Operations on dataframe data",
  version: "1.0.0"
)]
class DataframeOperations extends AbstractFlowDropNodeProcessor {

  /**
   * {@inheritdoc}
   */
  public static function create(
    ContainerInterface $container,
    array $configuration,
    $plugin_id,
    $plugin_definition,
  ): static {
    return new static($configuration, $plugin_id, $plugin_definition);
  }

  /**
   * {@inheritdoc}
   */
  public function validateParams(array $params): ValidationResult {
    // Dataframe operations nodes can accept any inputs or none.
    return ValidationResult::success();
  }

  /**
   * {@inheritdoc}
   */
  public function process(ParameterBagInterface $params): array {
    $operation = $params->get("operation", "head");
    $columns = $params->get("columns", []);
    $rows = $params->get("rows", 5);
    $condition = $params->get("condition", "");
    $dataframe = $params->get("dataframe") ?: [];

    $result = [];
    switch ($operation) {
      case "head":
        $result = $this->head($dataframe, $rows);
        break;

      case "tail":
        $result = $this->tail($dataframe, $rows);
        break;

      case "select":
        $result = $this->select($dataframe, $columns);
        break;

      case "filter":
        $result = $this->filter($dataframe, $condition);
        break;

      case "sort":
        $result = $this->sort($dataframe, $columns);
        break;

      case "group":
        $result = $this->group($dataframe, $columns);
        break;

      case "aggregate":
        $result = $this->aggregate($dataframe, $columns);
        break;

      case "merge":
        $result = $this->merge($dataframe);
        break;
    }

    return [
      "result" => $result,
      "operation" => $operation,
      "input_rows" => count($dataframe["data"] ?? []),
      "output_rows" => count($result["data"] ?? []),
      "columns" => $columns,
      "rows" => $rows,
    ];
  }

  /**
   * Get first n rows.
   *
   * @param array<string, mixed> $dataframe
   *   The dataframe.
   * @param int $rows
   *   Number of rows.
   *
   * @return array<string, mixed>
   *   The result.
   */
  private function head(array $dataframe, int $rows): array {
    $data = $dataframe["data"] ?? [];
    return [
      "data" => array_slice($data, 0, $rows),
      "columns" => $dataframe["columns"] ?? [],
      "index" => array_slice($dataframe["index"] ?? [], 0, $rows),
    ];
  }

  /**
   * Get last n rows.
   *
   * @param array<string, mixed> $dataframe
   *   The dataframe.
   * @param int $rows
   *   Number of rows.
   *
   * @return array<string, mixed>
   *   The result.
   */
  private function tail(array $dataframe, int $rows): array {
    $data = $dataframe["data"] ?? [];
    return [
      "data" => array_slice($data, -$rows),
      "columns" => $dataframe["columns"] ?? [],
      "index" => array_slice($dataframe["index"] ?? [], -$rows),
    ];
  }

  /**
   * Select specific columns.
   *
   * @param array<string, mixed> $dataframe
   *   The dataframe.
   * @param array<int, string> $columns
   *   The columns to select.
   *
   * @return array<string, mixed>
   *   The result.
   */
  private function select(array $dataframe, array $columns): array {
    $data = $dataframe["data"] ?? [];
    if (empty($columns)) {
      return $dataframe;
    }

    $filteredData = [];
    foreach ($data as $row) {
      $filteredRow = [];
      foreach ($columns as $column) {
        if (isset($row[$column])) {
          $filteredRow[$column] = $row[$column];
        }
      }
      $filteredData[] = $filteredRow;
    }

    return [
      "data" => $filteredData,
      "columns" => $columns,
      "index" => $dataframe["index"] ?? [],
    ];
  }

  /**
   * Filter rows based on condition.
   *
   * @param array<string, mixed> $dataframe
   *   The dataframe.
   * @param string $condition
   *   The condition.
   *
   * @return array<string, mixed>
   *   The result.
   */
  private function filter(array $dataframe, string $condition): array {
    $data = $dataframe["data"] ?? [];
    if (empty($condition)) {
      return $dataframe;
    }

    $filteredData = array_filter($data, function ($row) use ($condition) {
      switch ($condition) {
        case "empty":
          return empty($row);

        case "not empty":
          return !empty($row);
      }
      return TRUE;
    });

    return [
      "data" => array_values($filteredData),
      "columns" => $dataframe["columns"] ?? [],
      "index" => array_slice($dataframe["index"] ?? [], 0, count($filteredData)),
    ];
  }

  /**
   * Sort by columns.
   *
   * @param array<string, mixed> $dataframe
   *   The dataframe.
   * @param array<int, string> $columns
   *   The columns to sort by.
   *
   * @return array<string, mixed>
   *   The result.
   */
  private function sort(array $dataframe, array $columns): array {
    $data = $dataframe["data"] ?? [];
    if (empty($columns)) {
      return $dataframe;
    }

    usort($data, function ($a, $b) use ($columns) {
      foreach ($columns as $column) {
        $aVal = $a[$column] ?? "";
        $bVal = $b[$column] ?? "";
        $comparison = $aVal <=> $bVal;
        if ($comparison !== 0) {
          return $comparison;
        }
      }
      return 0;
    });

    return [
      "data" => $data,
      "columns" => $dataframe["columns"] ?? [],
      "index" => $dataframe["index"] ?? [],
    ];
  }

  /**
   * Group by columns.
   *
   * @param array<string, mixed> $dataframe
   *   The dataframe.
   * @param array<int, string> $columns
   *   The columns to group by.
   *
   * @return array<string, mixed>
   *   The result.
   */
  private function group(array $dataframe, array $columns): array {
    $data = $dataframe["data"] ?? [];
    if (empty($columns)) {
      return $dataframe;
    }

    $groups = [];
    foreach ($data as $row) {
      $groupKey = [];
      foreach ($columns as $column) {
        $groupKey[] = $row[$column] ?? "";
      }
      $key = implode("|", $groupKey);

      if (!isset($groups[$key])) {
        $groups[$key] = [];
      }
      $groups[$key][] = $row;
    }

    return [
      "data" => $groups,
      "columns" => $columns,
      "index" => array_keys($groups),
    ];
  }

  /**
   * Aggregate data.
   *
   * @param array<string, mixed> $dataframe
   *   The dataframe.
   * @param array<int, string> $columns
   *   The columns to aggregate.
   *
   * @return array<string, mixed>
   *   The result.
   */
  private function aggregate(array $dataframe, array $columns): array {
    $data = $dataframe["data"] ?? [];
    if (empty($columns)) {
      return $dataframe;
    }

    $aggregates = [];
    foreach ($columns as $column) {
      $values = array_column($data, $column);
      $aggregates[$column] = [
        "count" => count($values),
        "sum" => array_sum($values),
        "avg" => count($values) > 0 ? array_sum($values) / count($values) : 0,
        "min" => min($values),
        "max" => max($values),
      ];
    }

    return [
      "data" => $aggregates,
      "columns" => $columns,
      "index" => array_keys($aggregates),
    ];
  }

  /**
   * Merge dataframes.
   *
   * @param array<string, mixed> $dataframe
   *   The dataframe.
   *
   * @return array<string, mixed>
   *   The result.
   */
  private function merge(array $dataframe): array {
    // Placeholder for merge operation.
    return $dataframe;
  }

  /**
   * {@inheritdoc}
   */
  public function getParameterSchema(): array {
    return [
      "type" => "object",
      "properties" => [
        "dataframe" => [
          "type" => "object",
          "title" => "Dataframe",
          "description" => "The dataframe to operate on",
          "required" => FALSE,
        ],
        "operation" => [
          "type" => "string",
          "title" => "Operation",
          "description" => "Dataframe operation to perform",
          "default" => "head",
          "enum" => ["head", "tail", "select", "filter", "sort", "group", "aggregate", "merge"],
        ],
        "columns" => [
          "type" => "array",
          "title" => "Columns",
          "description" => "Columns to operate on",
          "default" => [],
        ],
        "rows" => [
          "type" => "integer",
          "title" => "Rows",
          "description" => "Number of rows for head/tail operations",
          "default" => 5,
        ],
        "condition" => [
          "type" => "string",
          "title" => "Condition",
          "description" => "Filter condition",
          "default" => "",
        ],
      ],
    ];
  }

  /**
   * {@inheritdoc}
   */
  public function getOutputSchema(): array {
    return [
      "type" => "object",
      "properties" => [
        "result" => [
          "type" => "object",
          "description" => "The processed dataframe",
        ],
        "operation" => [
          "type" => "string",
          "description" => "The operation performed",
        ],
        "input_rows" => [
          "type" => "integer",
          "description" => "Number of input rows",
        ],
        "output_rows" => [
          "type" => "integer",
          "description" => "Number of output rows",
        ],
        "columns" => [
          "type" => "array",
          "description" => "The columns used",
        ],
        "rows" => [
          "type" => "integer",
          "description" => "The number of rows",
        ],
      ],
    ];
  }

}
