mes-ai-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)
    {
        // 将 $data 强制转换为数组（兼容对象/字符串等类型）
        $data = (array)$data;

        if (empty($data['status_val']) || $data['status_val'] == '文生图') {
            Queue::push('app\job\TextToImageJob', $data, 'txttoimg');
            echo date('Y-m-d H:i:s') . " 队列已启动\n";
        }else if (empty($data['status_val']) || $data['status_val'] == '图生文') {
            Queue::push('app\job\ImageJob', $data, 'imgtotxt');
            echo date('Y-m-d H:i:s') . " 队列已启动\n";
        } else {

            $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));
    }
}