mp_audio_manage/app/Console/Commands/ProcessBatchAnimeEpisodeGenerationCommand.php

<?php

namespace App\Console\Commands;

use Illuminate\Console\Command;
use Illuminate\Support\Facades\DB;
use App\Services\DeepSeek\DeepSeekService;

/**
 * 批量生成分集子进程 worker
 * 由 batch:generate-episodes 调度，每个进程通过 MySQL 命名锁独占一个 anime_id，
 * 串行生成该 anime 的 pending 剧集（按集数顺序），直到没有待处理任务为止。
 */
class ProcessBatchAnimeEpisodeGenerationCommand extends Command
{
    /**
     * anime 处理锁前缀（配合 MySQL GET_LOCK 使用，同一 anime 同一时刻只允许一个进程处理）
     */
    public const LOCK_PREFIX = 'batch_episode_generation_anime_';

    /**
     * 单集最大重试次数，超过后停止重试并标记为失败
     */
    private const MAX_RETRY_COUNT = 5;

    /**
     * The name and signature of the console command.
     *
     * @var string
     */
    protected $signature = 'batch:generate-episodes:anime {anime_id : 动漫ID}';

    /**
     * The console command description.
     *
     * @var string
     */
    protected $description = '批量生成分集子进程：独占处理指定anime_id的待生成剧集';

    protected $deepSeekService;

    /**
     * Create a new command instance.
     *
     * @return void
     */
    public function __construct(DeepSeekService $deepSeekService)
    {
        parent::__construct();
        $this->deepSeekService = $deepSeekService;
    }

    /**
     * Execute the console command.
     *
     * @return int
     */
    public function handle()
    {
        set_time_limit(0);

        $animeId = (int) $this->argument('anime_id');
        $lockName = self::LOCK_PREFIX . $animeId;

        // 非阻塞抢占该 anime 的处理锁；抢不到说明已有其他进程正在处理
        $acquired = DB::selectOne('SELECT GET_LOCK(?, 0) AS acquired', [$lockName]);
        if (!$acquired || (int) $acquired->acquired !== 1) {
            dLog('command')->info("anime_id: {$animeId} 正在被其他进程处理，本次直接退出");
            return 0;
        }

        dLog('command')->info("anime_id: {$animeId} 抢占处理锁成功，开始处理...");

        try {
            $this->processAnimeEpisodes($animeId);
        } catch (\Throwable $e) {
            dLog('command')->error("anime_id: {$animeId} 处理异常: " . $e->getMessage());
            logDB('batch_episode_generation', 'error', "anime_id: {$animeId} 批量生成处理异常", [
                'anime_id' => $animeId,
                'error' => $e->getMessage(),
                'trace' => $e->getTraceAsString()
            ]);
        } finally {
            // 更新任务中心：该 anime 的批量生成任务若已全部结束，同步状态和结果
            try {
                $this->deepSeekService->finishBatchGenerateEpisodesTaskCenters($animeId);
            } catch (\Throwable $e) {
                dLog('command')->error('批量生成分集任务中心同步异常: ' . $e->getMessage());
            }

            // 显式释放处理锁；即使进程异常退出，连接断开后 MySQL 也会自动释放
            DB::statement('SELECT RELEASE_LOCK(?)', [$lockName]);
            dLog('command')->info("anime_id: {$animeId} 处理结束，已释放处理锁");
        }

        return 0;
    }

    /**
     * 串行处理指定 anime 的所有待生成剧集
     *
     * @param int $animeId
     * @return void
     */
    private function processAnimeEpisodes($animeId)
    {
        while (true) {
            // 取该 anime 集数最小的 pending 任务
            $task = DB::table('mp_batch_episode_generation_details')
                ->where('anime_id', $animeId)
                ->where('status', 'pending')
                ->orderBy('episode_number')
                ->first();

            if (!$task) {
                dLog('command')->info('[' . date('Y-m-d H:i:s') . "] anime_id: {$animeId} 没有待处理的任务");
                return;
            }

            try {
                // 返回 false 表示该 anime 暂时无法继续（如前一集未完成），退出等待下一轮调度
                $keepGoing = $this->processEpisode($task);
                if (!$keepGoing) {
                    dLog('command')->info("anime_id: {$animeId} 暂时无法继续生成，退出本次处理");
                    return;
                }
            } catch (\Exception $e) {
                $animeLogInfo = isset($task) ? ' (anime_id: ' . $task->anime_id . ', 第' . $task->episode_number . '集)' : '';
                dLog('command')->error('[' . date('Y-m-d H:i:s') . '] 生成失败: ' . $e->getMessage() . $animeLogInfo);

                if (isset($task)) {
                    // 更新重试次数
                    $retryCount = $task->retry_count + 1;

                    // 未超过最大重试次数则标记为待处理继续重试，超过后标记为失败并通知
                    $this->markTaskRetryOrFail($task, $retryCount, $e->getMessage());

                    // 记录错误日志
                    logDB('batch_episode_generation', 'error', "anime_id: {$task->anime_id} 第{$task->episode_number}集生成失败", [
                        'anime_id' => $task->anime_id,
                        'episode_number' => $task->episode_number,
                        'error' => $e->getMessage(),
                        'trace' => $e->getTraceAsString()
                    ]);
                }
            }

            // 与原有逻辑保持一致：每5秒处理一轮
            sleep(5);
        }
    }

    /**
     * 处理一个待生成的剧集（沿用原有校验与生成逻辑）
     *
     * @param object $task
     * @return bool true-可继续处理下一个任务，false-该anime暂时无法继续
     */
    private function processEpisode($task)
    {
        dLog('command')->info('[' . date('Y-m-d H:i:s') . '] 找到待处理任务 - Anime ID: ' . $task->anime_id . ', Episode: ' . $task->episode_number);

        // 检查前一集是否已完成（如果不是第1集）
        if ($task->episode_number > 1) {
            $prevEpisodeNumber = $task->episode_number - 1;

            // 优先判断前一集是否已在系统中生成（查询 mp_anime_episodes）。
            // 用户手动创建的前一集也视为已存在，直接跳过批量任务状态判断，
            // 避免批量任务前一集失败后卡住整个批量任务。
            $prevEpisodeExists = DB::table('mp_anime_episodes')
                ->where('anime_id', $task->anime_id)
                ->where('episode_number', $prevEpisodeNumber)
                ->where('is_default', 1)
                ->exists();

            if (!$prevEpisodeExists) {
                // 系统中不存在前一集时，才进行批量任务状态判断
                $prevTaskInBatch = DB::table('mp_batch_episode_generation_details')
                    ->where('anime_id', $task->anime_id)
                    ->where('episode_number', $prevEpisodeNumber)
                    ->first();

                if ($prevTaskInBatch && $prevTaskInBatch->status !== 'completed') {
                    dLog('command')->info('[' . date('Y-m-d H:i:s') . '] 前一集(第' . $prevEpisodeNumber . '集)尚未完成，跳过当前任务 (anime_id: ' . $task->anime_id . ')');
                    return false;
                }

                dLog('command')->error('[' . date('Y-m-d H:i:s') . '] 前一集(第' . $prevEpisodeNumber . '集)不存在，无法继续生成 (anime_id: ' . $task->anime_id . ')');

                // 标记为失败
                DB::table('mp_batch_episode_generation_details')
                    ->where('id', $task->id)
                    ->update([
                        'status' => 'failed',
                        'error_message' => "前一集(第{$prevEpisodeNumber}集)不存在，无法继续生成",
                        'updated_at' => now()
                    ]);

                return false;
            }
        }

        // 检查当前集是否已经存在（避免重复生成）
        $currentEpisodeExists = DB::table('mp_anime_episodes')
            ->where('anime_id', $task->anime_id)
            ->where('episode_number', $task->episode_number)
            ->where('is_default', 1)
            ->exists();

        if ($currentEpisodeExists) {
            dLog('command')->info('[' . date('Y-m-d H:i:s') . '] 第' . $task->episode_number . '集已存在，跳过生成 (anime_id: ' . $task->anime_id . ')');

            // 标记为已完成
            DB::table('mp_batch_episode_generation_details')
                ->where('id', $task->id)
                ->update([
                    'status' => 'completed',
                    'completed_at' => now(),
                    // 'updated_at' => now()
                ]);

            return true;
        }

        // 标记为处理中
        DB::table('mp_batch_episode_generation_details')
            ->where('id', $task->id)
            ->update([
                'status' => 'processing',
                'updated_at' => now()
            ]);

        // 设置用户上下文（绑定到容器）
        app()->instance('siteData', [
            'uid' => $task->uid,
            'cpid' => $task->cpid
        ]);

        // 解析请求数据
        $requestData = json_decode($task->request_data, true);
        $requestData['episode_number'] = $task->episode_number;

        // 使用"继续策划下一集"逻辑
        $requestData['prompt'] = '继续策划下一集';

        dLog('command')->info('[' . date('Y-m-d H:i:s') . '] 开始生成第' . $task->episode_number . '集... (anime_id: ' . $task->anime_id . ')');

        // 调用非流式生成方法，并记录实际执行耗时
        $chatStartTime = microtime(true);
        try {
            $result = $this->deepSeekService->chatForAceNonStream($requestData);
        } finally {
            $chatDuration = round(microtime(true) - $chatStartTime, 2);
            dLog('command')->info('[' . date('Y-m-d H:i:s') . '] chatForAceNonStream 执行完成，耗时: ' . $chatDuration . ' 秒 (anime_id: ' . $task->anime_id . ', 第' . $task->episode_number . '集)');
        }

        // 检查是否有错误
        if (isset($result['error']) && $result['error']) {
            dLog('command')->error('[' . date('Y-m-d H:i:s') . '] 生成失败: ' . $result['error'] . ' (anime_id: ' . $task->anime_id . ', 第' . $task->episode_number . '集)');

            // 更新重试次数
            $retryCount = $task->retry_count + 1;

            // 未超过最大重试次数则标记为待处理继续重试，超过后标记为失败并通知
            $this->markTaskRetryOrFail($task, $retryCount, $result['error']);

            // 记录错误日志
            logDB('batch_episode_generation', 'error', "anime_id: {$task->anime_id} 第{$task->episode_number}集生成失败", [
                'anime_id' => $task->anime_id,
                'episode_number' => $task->episode_number,
                'error' => $result['error']
            ]);

            return true;
        }

        // 标记为完成
        DB::table('mp_batch_episode_generation_details')
            ->where('id', $task->id)
            ->update([
                'status' => 'completed',
                'result_data' => json_encode($result, JSON_UNESCAPED_UNICODE),
                'completed_at' => now(),
                // 'updated_at' => now()
            ]);

        dLog('command')->info('[' . date('Y-m-d H:i:s') . '] 第' . $task->episode_number . '集生成成功 (anime_id: ' . $task->anime_id . ')');

        return true;
    }

    /**
     * 任务失败处理：未超过最大重试次数则标记为待处理继续重试，
     * 超过后直接标记为失败并通过 sendNotice 发送报错通知。
     *
     * @param object $task
     * @param int $retryCount
     * @param string $error
     * @return void
     */
    private function markTaskRetryOrFail($task, $retryCount, $error)
    {
        if ($retryCount > self::MAX_RETRY_COUNT) {
            // 超过最大重试次数，标记为失败
            DB::table('mp_batch_episode_generation_details')
                ->where('id', $task->id)
                ->update([
                    'status' => 'failed',
                    'error_message' => '重试超过' . self::MAX_RETRY_COUNT . '次，已停止重试: ' . $error,
                    'retry_count' => $retryCount,
                    'updated_at' => now()
                ]);

            dLog('command')->error('[' . date('Y-m-d H:i:s') . '] 第' . $task->episode_number . '集重试超过' . self::MAX_RETRY_COUNT . '次，已标记为失败 (anime_id: ' . $task->anime_id . ')');

            // 发送报错通知（通知失败不影响任务状态）
            try {
                $notice = "批量分集生成失败（重试超过" . self::MAX_RETRY_COUNT . "次，已停止重试）\n"
                    . "anime_id: {$task->anime_id}\n"
                    . "集数: 第{$task->episode_number}集\n"
                    . "错误: {$error}\n"
                    . "时间: " . date('Y-m-d H:i:s');
                sendNotice($notice);
            } catch (\Throwable $e) {
                dLog('command')->error('[' . date('Y-m-d H:i:s') . '] 发送失败通知异常 (anime_id: ' . $task->anime_id . ', 第' . $task->episode_number . '集): ' . $e->getMessage());
            }

            return;
        }

        // 未超过最大重试次数，标记为待处理，下一轮重试
        DB::table('mp_batch_episode_generation_details')
            ->where('id', $task->id)
            ->update([
                'status' => 'pending',
                'error_message' => $error,
                'retry_count' => $retryCount,
                'updated_at' => now()
            ]);
    }
}