<?php

namespace XZD\AI\V5\Tools;

use XZD\AI\V5\AI\Client;
use XZD\AI\V5\AI\Types;
use XZD\AI\V5\Core\Logger;

/**
 * 并行工具执行引擎
 *
 * 核心能力: 依赖分析、curl_multi 并发、故障隔离、进度回调
 */
class ParallelExecutor
{
    private int $maxConcurrency = 5;
    private int $timeout = 30;
    private ?\Closure $onProgress = null;

    public function __construct(int $maxConcurrency = 5, int $timeout = 30)
    {
        $this->maxConcurrency = $maxConcurrency;
        $this->timeout = $timeout;
    }

    public function onProgress(?\Closure $callback): self
    {
        $this->onProgress = $callback;
        return $this;
    }

    /**
     * 分析工具依赖并分组执行
     *
     * @param array $toolCalls  [['tool_id'=>'...','params'=>[...]], ...]
     * @param int   $userId
     * @return array  [['tool_id','success','data','message','duration_ms'], ...]
     */
    public function execute(array $toolCalls, int $userId): array
    {
        if (empty($toolCalls)) {
            return [];
        }

        $groups = $this->analyzeGroups($toolCalls);
        $results = [];
        $totalGroups = count($groups);

        foreach ($groups as $gi => $group) {
            if (count($group) === 1) {
                $results[] = $this->executeOne($group[0], $userId);
            } else {
                $groupResults = $this->executeGroup($group, $userId);
                foreach ($groupResults as $r) {
                    $results[] = $r;
                }
            }

            if ($this->onProgress) {
                ($this->onProgress)($gi + 1, $totalGroups, $results);
            }
        }

        return $results;
    }

    /**
     * 分析依赖关系，返回可并行的执行分组
     *
     * 规则: 只读工具归入同一并行组，写入工具各自独立成组，操作同一 post_id 的写入工具串行排列
     */
    public function analyzeGroups(array $toolCalls): array
    {
        $registry = ToolRegistry::instance();
        $reads = [];
        $writes = [];

        foreach ($toolCalls as $call) {
            $toolId = $call['tool_id'] ?? $call['id'] ?? '';
            $tool = $registry->get($toolId);
            $danger = $tool ? $tool->danger : 'low';

            if (in_array($danger, ['medium', 'high'], true)) {
                $writes[] = $call;
            } else {
                $reads[] = $call;
            }
        }

        $groups = [];
        if (!empty($reads)) {
            $groups[] = $reads;
        }
        // 写入工具按目标分组（同 post_id 冲突的串行）
        if (!empty($writes)) {
            $byTarget = [];
            foreach ($writes as $call) {
                $params = $call['params'] ?? [];
                $targetKey = ($params['post_id'] ?? '') . '_' . ($params['user_id'] ?? '') . '_' . ($call['tool_id'] ?? '');
                $byTarget[$targetKey][] = $call;
            }
            foreach ($byTarget as $wg) {
                $groups[] = $wg;
            }
        }

        return $groups;
    }

    /**
     * 并行 AI 调用 - 使用 curl_multi 并发请求多个 AI API
     *
     * 适用于同时请求多个独立的内容分析（标签、SEO、摘要、评分等）
     *
     * @param array $requests  [['messages'=>[...],'options'=>[...]], ...]
     * @return array 按输入顺序排列的结果数组
     */
    public function executeParallelAI(array $requests): array
    {
        if (empty($requests)) {
            return [];
        }

        $provider = $this->getActiveProvider();
        if (!$provider) {
            return array_map(fn() => [
                'success' => false,
                'message' => '没有可用的 AI 连接',
                'duration_ms' => 0,
            ], $requests);
        }

        $config = $provider->getConfig();
        $baseUrl = rtrim($config['base_url'] ?: '', '/');
        $apiFormat = $config['api_format'] ?? 'openai';

        $results = [];
        // 按 maxConcurrency 分批执行，防止内存/连接耗尽
        $batches = array_chunk($requests, $this->maxConcurrency, true);

        foreach ($batches as $batchReqs) {
            $batchResults = $this->executeBatch(
                $provider, $config, $baseUrl, $apiFormat, $batchReqs
            );
            foreach ($batchResults as $idx => $res) {
                $results[$idx] = $res;
            }
        }

        // 按原始请求顺序排列
        ksort($results);
        return array_values($results);
    }

    /** 执行单批并发请求（不超过 maxConcurrency 个） */
    private function executeBatch($provider, array $config, string $baseUrl, string $apiFormat, array $requests): array
    {
        $multiHandle = curl_multi_init();
        $handles = [];
        $callIndex = [];

        foreach ($requests as $i => $req) {
            $messages = $req['messages'] ?? [];
            $options = array_merge(
                ['disable_thinking' => true],
                $req['options'] ?? []
            );

            $url = $baseUrl . '/chat/completions';
            $body = [
                'model'    => $config['model'] ?? '',
                'messages' => $messages,
                'stream'   => false,
            ];
            if (isset($options['temperature'])) {
                $body['temperature'] = $options['temperature'];
            }
            if (isset($options['max_tokens'])) {
                $body['max_tokens'] = $options['max_tokens'];
            }
            if (isset($options['tools'])) {
                $body['tools'] = $options['tools'];
            }
            if (isset($options['tool_choice'])) {
                $body['tool_choice'] = $options['tool_choice'];
            }
            if ($options['disable_thinking'] ?? false) {
                $body['thinking'] = ['type' => 'disabled'];
            }

            // 按 API 格式构建请求头
            if ($apiFormat === 'claude') {
                $url = $baseUrl . '/messages';
                $headers = [
                    'x-api-key: ' . $config['api_key'],
                    'anthropic-version: 2023-06-01',
                    'Content-Type: application/json',
                ];
                $body = [
                    'model'      => $config['model'] ?? '',
                    'max_tokens' => $options['max_tokens'] ?? 2048,
                    'messages'   => array_values(array_filter($messages, fn($m) => in_array($m['role'] ?? '', ['user', 'assistant']))),
                    'system'     => $this->extractSystemMessage($messages),
                ];
            } elseif ($apiFormat === 'google') {
                $url = $baseUrl . '/models/' . ($config['model'] ?? '') . ':generateContent';
                $apiKey = $config['api_key'] ?? '';
                $url .= (str_contains($url, '?') ? '&' : '?') . 'key=' . urlencode($apiKey);
                $headers = ['Content-Type: application/json'];
                $body = $this->toGeminiFormat($messages, $options);
            } else {
                // OpenAI 兼容格式
                $headers = [
                    'Authorization: Bearer ' . $config['api_key'],
                    'Content-Type: application/json',
                ];
            }

            $ch = curl_init($url);
            curl_setopt_array($ch, [
                CURLOPT_RETURNTRANSFER => true,
                CURLOPT_POST           => true,
                CURLOPT_POSTFIELDS     => json_encode($body),
                CURLOPT_HTTPHEADER     => $headers,
                CURLOPT_TIMEOUT        => $this->timeout,
                CURLOPT_CONNECTTIMEOUT => 10,
            ]);

            curl_multi_add_handle($multiHandle, $ch);
            $handles[] = $ch;
            $callIndex[(int)$ch] = ['index' => $i, 'start' => microtime(true)];
        }

        // 执行并发请求
        $active = null;
        do {
            $status = curl_multi_exec($multiHandle, $active);
            if ($active > 0) {
                curl_multi_select($multiHandle, 0.1);
            }
        } while ($active > 0 && $status === CURLM_OK);

        // 收集结果
        $results = [];
        foreach ($handles as $ch) {
            $idx = $callIndex[(int)$ch]['index'];
            $start = $callIndex[(int)$ch]['start'];
            $response = curl_multi_getcontent($ch);
            $httpCode = curl_getinfo($ch, CURLINFO_HTTP_CODE);
            $error = curl_error($ch);
            $duration = (int)((microtime(true) - $start) * 1000);

            if ($error) {
                $results[$idx] = [
                    'success'     => false,
                    'message'     => "API 连接失败: {$error}",
                    'duration_ms' => $duration,
                ];
            } elseif ($httpCode >= 400) {
                $body = json_decode($response, true);
                $results[$idx] = [
                    'success'     => false,
                    'message'     => $body['error']['message'] ?? "HTTP {$httpCode}",
                    'duration_ms' => $duration,
                ];
            } else {
                $data = json_decode($response, true);
                $content = '';
                if (isset($data['choices'][0]['message']['content'])) {
                    $content = $data['choices'][0]['message']['content'];
                } elseif (isset($data['content'][0]['text'])) {
                    $content = $data['content'][0]['text'];
                }
                $results[$idx] = [
                    'success'      => true,
                    'data'         => $data,
                    'content'      => $content,
                    'input_tokens'  => $data['usage']['prompt_tokens'] ?? 0,
                    'output_tokens' => $data['usage']['completion_tokens'] ?? 0,
                    'duration_ms'  => $duration,
                ];
            }

            curl_multi_remove_handle($multiHandle, $ch);
            curl_close($ch);
        }

        curl_multi_close($multiHandle);
        // 保留原始请求索引作为 key，供 executeParallelAI 合并多批结果
        return $results;
    }

    /** 执行一组工具（可并行组内的工具） */
    private function executeGroup(array $group, int $userId): array
    {
        $results = [];

        foreach ($group as $call) {
            $results[] = $this->executeOne($call, $userId);
        }

        return $results;
    }

    /** 执行单个工具调用 */
    private function executeOne(array $call, int $userId): array
    {
        $toolId = $call['tool_id'] ?? $call['id'] ?? '';
        $params = $call['params'] ?? [];
        $start = microtime(true);

        try {
            $result = ToolRegistry::instance()->execute($toolId, $params, $userId);
            return [
                'tool_id'     => $toolId,
                'success'     => $result->success,
                'data'        => $result->data,
                'message'     => $result->message,
                'duration_ms' => (int)((microtime(true) - $start) * 1000),
            ];
        } catch (\Throwable $e) {
            Logger::error('parallel_exec', "{$toolId}: {$e->getMessage()}");
            return [
                'tool_id'     => $toolId,
                'success'     => false,
                'message'     => $e->getMessage(),
                'duration_ms' => (int)((microtime(true) - $start) * 1000),
            ];
        }
    }

    /** 获取当前活跃 Provider */
    private function getActiveProvider(): ?\XZD\AI\V5\AI\ProviderInterface
    {
        try {
            $ref = new \ReflectionClass(Client::instance());
            $method = $ref->getMethod('selectProvider');
            return $method->invoke(Client::instance(), 'text');
        } catch (\Throwable $e) {
            Logger::warning('parallel_executor', 'selectProvider 反射失败: ' . $e->getMessage());
            return null;
        }
    }

    private function extractSystemMessage(array $messages): string
    {
        foreach ($messages as $m) {
            if (($m['role'] ?? '') === 'system') {
                return $m['content'] ?? '';
            }
        }
        return '';
    }

    private function toGeminiFormat(array $messages, array $options): array
    {
        $contents = [];
        $systemText = '';

        foreach ($messages as $m) {
            $role = $m['role'] ?? 'user';
            if ($role === 'system') {
                $systemText .= ($m['content'] ?? '') . "\n";
                continue;
            }
            $geminiRole = $role === 'assistant' ? 'model' : 'user';
            $contents[] = [
                'role'  => $geminiRole,
                'parts' => [['text' => $m['content'] ?? '']],
            ];
        }

        $body = ['contents' => $contents];
        if (!empty($systemText)) {
            $body['system_instruction'] = ['parts' => [['text' => trim($systemText)]]];
        }
        if (!empty($options['max_tokens'])) {
            $body['generation_config']['maxOutputTokens'] = $options['max_tokens'];
        }

        return $body;
    }
}
