|
|
@@ -0,0 +1,332 @@
|
|
|
+<?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 {
|
|
|
+ // 显式释放处理锁;即使进程异常退出,连接断开后 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;
|
|
|
+
|
|
|
+ // 检查前一集是否在批量任务中
|
|
|
+ $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;
|
|
|
+ }
|
|
|
+
|
|
|
+ // 检查前一集是否已在系统中生成(查询 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) {
|
|
|
+ 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()
|
|
|
+ ]);
|
|
|
+ }
|
|
|
+}
|