mp_audio_manage/app/Services/TaskCenterService.php

<?php

namespace App\Services;

use App\Facade\Site;
use App\Models\MpTaskCenter;
use App\Services\Anime\AnimeService;
use Illuminate\Support\Facades\DB;

class TaskCenterService
{
    /**
     * 创建任务中心记录
     *
     * @param string $type 任务类型：text/image/video
     * @param array  $data 任务数据（title/ref_task_id/prompt/params等）
     * @return MpTaskCenter
     */
    public function createTask(string $type, array $data = [])
    {
        $uid = 0;
        $cpid = 0;
        try {
            $uid = (int)Site::getUid();
            $cpid = (int)Site::getCpid();
        } catch (\Throwable $e) {
            // 非请求上下文（如命令行）下可能无法获取用户ID，置为0
        }

        return MpTaskCenter::create([
            'uid'          => $data['uid'] ?? $uid,
            'cpid'           => $data['cpid'] ?? $cpid,
            'task_type'    => $type,
            'title'        => $data['title'] ?? '',
            'ref_task_id'  => $data['ref_task_id'] ?? 0,
            'status'       => $data['status'] ?? MpTaskCenter::STATUS_PROCESSING,
            'result'       => $data['result'] ?? null,
            'error_message'=> $data['error_message'] ?? null,
            'prompt'       => $data['prompt'] ?? null,
            'params'       => $data['params'] ?? null,
            'params_md5'   => $data['params_md5'] ?? null,
        ]);
    }

    /**
     * 更新任务中心记录
     *
     * @param int   $taskId
     * @param array $data
     * @return bool
     */
    public function updateTask(int $taskId, array $data = []): bool
    {
        return (bool)MpTaskCenter::where('id', $taskId)->update($data);
    }

    /**
     * 查询任务详情
     *
     * @param int $taskId
     * @return MpTaskCenter|null
     */
    public function getTaskDetail(int $taskId)
    {
        $query = MpTaskCenter::where('id', $taskId);

        // 请求上下文下按当前用户过滤
        $uid = 0;
        try {
            $uid = (int)Site::getUid();
        } catch (\Throwable $e) {
        }
        if ($uid > 0) {
            $query->where('uid', $uid);
        }

        return $query->first();
    }

    /**
     * 分页查询任务列表
     *
     * @param array $params
     * @return \Illuminate\Contracts\Pagination\LengthAwarePaginator
     */
    public function getTaskList(array $params = [])
    {
        $query = MpTaskCenter::query();

        // 请求上下文下按当前用户过滤
        $uid = 0;
        try {
            $uid = (int)Site::getUid();
        } catch (\Throwable $e) {
        }
        if ($uid > 0) {
            $query->where('uid', $uid);
        }

        // 按任务ID筛选
        if (!empty($params['task_id'])) {
            $query->where('id', (int)$params['task_id']);
        }

        // 按任务状态筛选（支持逗号分隔多状态）
        if (!empty($params['status'])) {
            $statuses = is_array($params['status'])
                ? $params['status']
                : array_filter(array_map('trim', explode(',', (string)$params['status'])));
            if (!empty($statuses)) {
                $query->whereIn('status', $statuses);
            }
        }

        // 按任务类型筛选
        if (!empty($params['task_type'])) {
            $query->where('task_type', $params['task_type']);
        }

        $pageSize = (int)($params['page_size'] ?? 20);
        if ($pageSize <= 0 || $pageSize > 100) {
            $pageSize = 20;
        }

        return $query->orderBy('created_at', 'desc')->orderBy('id', 'desc')->paginate($pageSize);
    }

    /**
     * 定时同步任务中心状态和结果
     *
     * 图片任务关联 mp_generate_pic_tasks，视频任务关联 mp_generate_video_tasks，
     * 将底层任务的最新状态、结果和错误信息同步到任务中心。
     *
     * @return int 同步更新的记录数
     */
    public function syncTaskStatus(): int
    {
        $updated = 0;

        // 只同步尚未结束的任务（避免重复扫描已完成记录）
        $tasks = MpTaskCenter::whereIn('status', [
            MpTaskCenter::STATUS_PENDING,
            MpTaskCenter::STATUS_PROCESSING,
        ])
            ->where('ref_task_id', '>', 0)
            ->orderBy('id', 'desc')
            ->limit(500)
            ->get();

        foreach ($tasks as $task) {
            try {
                if ($task->task_type === MpTaskCenter::TYPE_IMAGE) {
                    $updated += $this->syncImageTask($task);
                } elseif ($task->task_type === MpTaskCenter::TYPE_VIDEO
                    || $task->task_type === MpTaskCenter::TYPE_ACT_VIDEO) {
                    // 文生视频与片段转视频的单任务结果格式一致，共用同步逻辑
                    $updated += $this->syncVideoTask($task);
                } elseif ($task->task_type === MpTaskCenter::TYPE_BATCH_ACT_VIDEOS) {
                    $updated += $this->syncBatchActVideosTask($task);
                }
            } catch (\Exception $e) {
                dLog('command')->error('任务中心同步失败: ' . $e->getMessage(), [
                    'task_id'     => $task->id,
                    'ref_task_id' => $task->ref_task_id,
                    'task_type'   => $task->task_type,
                ]);
            }
        }

        return $updated;
    }

    /**
     * 同步图片任务状态到任务中心
     *
     * @param MpTaskCenter $task
     * @return int
     */
    private function syncImageTask(MpTaskCenter $task): int
    {
        $ref = DB::table('mp_generate_pic_tasks')->where('id', $task->ref_task_id)->first();
        if (!$ref) {
            return 0;
        }

        $result = null;
        if (!empty($ref->result_url)) {
            $urls = $this->normalizeResultUrls($ref->result_url);
            $urls = array_values(array_filter($urls));

            // 与文生图 completed 返回格式保持一致
            $resultData = [
                'msg'            => '',
                'code'           => 0,
                'data'           => $urls[0] ?? '',
                'task_center_id' => $task->id,
            ];
            if (count($urls) > 1) {
                $resultData['image_urls'] = $urls;
            }
            $result = json_encode($resultData, JSON_UNESCAPED_UNICODE);
        }

        return $this->applyRefStatus($task, $ref->status, $result, $ref->error_message);
    }

    /**
     * 规范化图片结果URL（兼容字符串/数组/JSON字符串/双重编码）
     *
     * @param mixed $resultUrl
     * @return array
     */
    private function normalizeResultUrls($resultUrl): array
    {
        if (is_array($resultUrl)) {
            return array_values(array_filter($resultUrl));
        }

        $value = (string)$resultUrl;
        // 最多解析两层 JSON（防御双重编码）
        for ($i = 0; $i < 2; $i++) {
            if (!is_string($value) || !is_json($value)) {
                break;
            }
            $decoded = json_decode($value, true);
            if (!is_array($decoded)) {
                // 解码结果是字符串且仍是JSON，继续解析下一层
                if (is_string($decoded) && is_json($decoded)) {
                    $value = $decoded;
                    continue;
                }
                $value = $decoded;
                break;
            }
            $value = $decoded;
        }

        if (is_array($value)) {
            return array_values(array_filter($value));
        }
        if (is_string($value) && $value !== '') {
            return [$value];
        }
        return [];
    }

    /**
     * 同步视频任务状态到任务中心
     *
     * @param MpTaskCenter $task
     * @return int
     */
    private function syncVideoTask(MpTaskCenter $task): int
    {
        $ref = DB::table('mp_generate_video_tasks')->where('id', $task->ref_task_id)->first();
        if (!$ref) {
            return 0;
        }

        $result = null;
        if (!empty($ref->result_url) || !empty($ref->compressed_url) || !empty($ref->last_frame_url)) {
            // 与文生视频 completed 返回格式保持一致
            $result = json_encode([
                'task_id'          => $ref->id,
                'status'           => $ref->status,
                'video_url'        => $ref->compressed_url ?: $ref->result_url,
                'origin_video_url' => $ref->result_url,
                'last_frame_url'   => $ref->last_frame_url,
                'error_message'    => $ref->error_message ? mapErrorMessage($ref->error_message) : '',
            ], JSON_UNESCAPED_UNICODE);
        }

        return $this->applyRefStatus($task, $ref->status, $result, $ref->error_message);
    }

    /**
     * 同步片段一键转视频（批量）任务状态到任务中心
     *
     * ref_task_id 关联首个视频任务，完整任务列表存放在 params.task_ids 中；
     * 结果字段与 batchSetActVideos 接口 completed 事件保持一致
     * （total_tasks_count / completed_tasks_count / success_tasks_count / episode）。
     *
     * @param MpTaskCenter $task
     * @return int
     */
    private function syncBatchActVideosTask(MpTaskCenter $task): int
    {
        $params = json_decode((string)$task->params, true) ?: [];
        $animeId = (int)getProp($params, 'anime_id', 0);
        $episodeId = (int)getProp($params, 'episode_id', 0);

        $taskIds = array_values(array_filter(array_map('intval', (array)getProp($params, 'task_ids', [])), function ($id) {
            return $id > 0;
        }));
        if (!$taskIds) {
            return 0;
        }

        $refs = DB::table('mp_generate_video_tasks')->whereIn('id', $taskIds)->get();
        if ($refs->isEmpty()) {
            return 0;
        }

        $totalCount = count($taskIds);
        $completedCount = 0;
        $successCount = 0;
        $failedCount = 0;
        $errorMessage = '';

        foreach ($refs as $ref) {
            if (!in_array($ref->status, ['success', 'failed'], true)) {
                continue;
            }
            $completedCount++;
            if ($ref->status === 'success') {
                $successCount++;
                continue;
            }
            $failedCount++;
            if ($errorMessage === '' && !empty($ref->error_message)) {
                $errorMessage = mapErrorMessage($ref->error_message);
            }
        }

        // 未全部结束时仅推进整体状态，等待所有视频任务落终态后再写结果
        if ($completedCount < $totalCount) {
            $newStatus = $completedCount > 0 ? MpTaskCenter::STATUS_PROCESSING : $task->status;
            return $this->updateTask((int)$task->id, ['status' => $newStatus]) ? 1 : 0;
        }

        $result = [
            'total_tasks_count'     => $totalCount,
            'completed_tasks_count' => $completedCount,
            'success_tasks_count'   => $successCount,
        ];

        // 与接口 completed 事件一致，附带剧集完整信息
        if ($animeId > 0 && $episodeId > 0) {
            try {
                $result['episode'] = app(AnimeService::class)->episodeInfoForAce([
                    'anime_id'   => $animeId,
                    'episode_id' => $episodeId,
                ]);
            } catch (\Throwable $e) {
                dLog('command')->error('任务中心同步片段批量视频结果失败: ' . $e->getMessage(), [
                    'task_id'    => $task->id,
                    'anime_id'   => $animeId,
                    'episode_id' => $episodeId,
                ]);
            }
        }

        $updateData = [
            'status' => $failedCount > 0 ? MpTaskCenter::STATUS_FAILED : MpTaskCenter::STATUS_SUCCESS,
            'result' => json_encode($result, JSON_UNESCAPED_UNICODE),
        ];
        if ($failedCount > 0) {
            $updateData['error_message'] = $errorMessage !== '' ? $errorMessage : ('有 ' . $failedCount . ' 个视频生成失败');
        }

        return $this->updateTask((int)$task->id, $updateData) ? 1 : 0;
    }

    /**
     * 将底层任务状态应用到任务中心记录
     *
     * @param MpTaskCenter $task
     * @param string       $refStatus
     * @param string|null  $result
     * @param string|null  $errorMessage
     * @return int
     */
    private function applyRefStatus(MpTaskCenter $task, string $refStatus, $result, $errorMessage): int
    {
        $statusMap = [
            'pending'    => MpTaskCenter::STATUS_PENDING,
            'processing' => MpTaskCenter::STATUS_PROCESSING,
            'success'    => MpTaskCenter::STATUS_SUCCESS,
            'failed'     => MpTaskCenter::STATUS_FAILED,
        ];

        $newStatus = $statusMap[$refStatus] ?? $task->status;
        $updateData = ['status' => $newStatus];

        if ($result !== null) {
            $updateData['result'] = $result;
        }
        if ($errorMessage !== null) {
            $updateData['error_message'] = $errorMessage;
        }

        return $this->updateTask((int)$task->id, $updateData) ? 1 : 0;
    }
}