mes-sdai-server-api/application/job/ImageArrJob.php

<?php
namespace app\job;

use think\Db;
use think\queue\Job;
use think\Queue;

/**
 * ImageArrJob 图像任务初始化队列
 * 用于将批量图像任务按类型分发至下游子队列（图生文/文生文/文生图）
 */
class ImageArrJob
{
    /**
     * 队列任务执行函数
     *
     * @param Job $job 当前队列任务对象
     * @param array $data 传入的数据，包含 task_id 和图像列表 data
     * task_id：队列id
     * data：队列数据
     */
    public function fire(Job $job, $data)
    {

        $task_id = $data['task_id'];
        $images = $data['data'];

        foreach ($images as $value) {
            $value['task_id'] = $task_id;
            // 清理路径：去除末尾 Preview 目录
            $sourceDir = rtrim($value["sourceDir"], '/\\');
            $dirParts = explode('/', str_replace('\\', '/', $sourceDir));
            if (end($dirParts) === 'Preview') {
                array_pop($dirParts);
            }
            $cleanedSourceDir = implode('/', $dirParts);

            // 获取文件名并拼接最终路径
            $fileName = basename($value['file_name']);
            $fullPath = $cleanedSourceDir . '/' . $fileName;

            // 写入 image_task_log 队列明细日志记录
            $log_id = Db::name('image_task_log')->insertGetId([
                'task_id' => $task_id,
                'file_name' => $fullPath,
                'model_name' => $value['type'] ?? '',
                'status' => 0,
                'log' => '队列中',
                'create_time' => date('Y-m-d H:i:s')
            ]);

            $value['log_id'] = $log_id;
            //若有链式任务，传递下去
            $chain_next = $value['chain_next'] ?? [];

            switch (trim($value['type'])) {
                case '图生文':
                    Queue::push('app\job\ImageJob', array_merge($value, ['chain_next' => $chain_next]), 'imgtotxt');
                    break;
                case '文生文':
                    Queue::push('app\job\TextToTextJob', array_merge($value, ['chain_next' => $chain_next]), 'txttotxt');
                    break;
                case '文生图':
                    Queue::push('app\job\TextToImageJob', array_merge($value, ['chain_next' => $chain_next]), 'txttoimg');
                    break;
                case '图生图':
                    Queue::push('app\job\ImageToImageJob', array_merge($value, ['chain_next' => $chain_next]), 'imgtoimg');
                    break;
                case '高清放大':
                    Queue::push('app\job\ImageToSingleJob', array_merge($value, ['chain_next' => $chain_next]), 'single');
                    break;
                default:
                    \think\Log::warning("未识别的任务类型：" . json_encode($value, JSON_UNESCAPED_UNICODE));
                    break;
            }

        }

        //更新任务状态为已启动
        Db::name('queue_logs')->where('id', $task_id)->update(['status' => '已启动队列']);
        echo date('Y-m-d H:i:s') . " 队列已启动\n";

        //删除当前队列任务
        $job->delete();
    }

    /**
     * 任务执行失败回调
     *
     * @param array $data 原始任务数据
     */
    public function failed($data)
    {
        \think\Log::error("ImageArrJob 任务失败：" . json_encode($data, JSON_UNESCAPED_UNICODE));
    }
}