mes-ai-server-api/application/job/SeedreamImageGenerationJob.php

<?php

namespace app\job;

use OSS\OssClient;
use think\Config;
use think\Db;
use think\Log;
use think\queue\Job;

/**
 * 豆包 Seedream 组图生成队列（独立于 txttoimg / imgtoimg / arrimage）
 *
 * 启动消费者：
 * php think queue:work --queue seedreamimage --timeout 900 --sleep 3 --tries 1
 */
class SeedreamImageGenerationJob
{
    const DEFAULT_MODEL = 'doubao-seedream-5-0-260128';
    const DEFAULT_API_URL = 'https://ark.cn-beijing.volces.com/api/v3/images/generations';
    const REDIS_KEY_PREFIX = 'seedream_image_task:';
    const QUEUE_NAME = 'seedreamimage';
    const TASK_TTL = 1800;

    /**
     * 队列入口
     */
    public function fire(Job $job, $data)
    {
        $taskId = trim((string)($data['task_id'] ?? ''));
        if ($taskId === '') {
            $job->delete();
            return;
        }

        $redisKey = self::REDIS_KEY_PREFIX . $taskId;
        $redis = getTaskRedis();

        echo "\n" . date('Y-m-d H:i:s') . " [SeedreamImageGenerationJob] 开始 task_id={$taskId}\n";

        try {
            $existing = $redis->get($redisKey);
            if ($existing) {
                $info = json_decode($existing, true);
                if (is_array($info) && ($info['status'] ?? '') === 'completed') {
                    echo "任务 {$taskId} 已完成，跳过\n";
                    $job->delete();
                    return;
                }
            }

            $redis->set($redisKey, json_encode([
                'status' => 'processing',
                'task_id' => $taskId,
                'started_at' => date('Y-m-d H:i:s'),
            ], JSON_UNESCAPED_UNICODE), ['EX' => self::TASK_TTL]);

            set_time_limit(900);
            ini_set('max_execution_time', '900');

            $payload = self::execute(is_array($data) ? $data : []);

            $redis->set($redisKey, json_encode(array_merge($payload, [
                'status' => 'completed',
                'task_id' => $taskId,
                'completed_at' => date('Y-m-d H:i:s'),
            ]), JSON_UNESCAPED_UNICODE), ['EX' => self::TASK_TTL]);

            echo date('Y-m-d H:i:s') . " [SeedreamImageGenerationJob] 完成 task_id={$taskId}\n";
            $job->delete();
        } catch (\Exception $e) {
            echo date('Y-m-d H:i:s') . " [SeedreamImageGenerationJob] 失败: " . $e->getMessage() . "\n";
            Log::write('[SeedreamImageGenerationJob] ' . $e->getMessage(), 'error');

            $redis->set($redisKey, json_encode([
                'status' => 'failed',
                'task_id' => $taskId,
                'error' => $e->getMessage(),
                'completed_at' => date('Y-m-d H:i:s'),
            ], JSON_UNESCAPED_UNICODE), ['EX' => self::TASK_TTL]);

            $job->delete();
        }
    }

    /**
     * 执行 Seedream 生图（组图/单图均可）
     */
    public static function execute(array $data): array
    {
        $prompt = trim((string)($data['prompt'] ?? ''));
        $model = trim((string)($data['model'] ?? self::DEFAULT_MODEL));
        $size = trim((string)($data['size'] ?? '2K'));
        $stream = filter_var($data['stream'] ?? false, FILTER_VALIDATE_BOOLEAN);
        $referenceImages = is_array($data['reference_images'] ?? null) ? $data['reference_images'] : [];
        $genConfig = is_array($data['gen_config'] ?? null) ? $data['gen_config'] : [];

        $requestBody = is_array($data['request_body'] ?? null) ? $data['request_body'] : [];
        if (empty($requestBody)) {
            $referenceCount = count($referenceImages);
            $maxImages = max(1, min(15, (int)($genConfig['max_images'] ?? 1)));
            $sequentialMode = trim((string)($genConfig['sequential_image_generation'] ?? ''));
            if ($sequentialMode === '') {
                $sequentialMode = $maxImages > 1 ? 'auto' : 'disabled';
            }

            $requestBody = [
                'model' => $model,
                'prompt' => $prompt,
                'sequential_image_generation' => $sequentialMode,
                'response_format' => 'url',
                'size' => $size,
                'stream' => $stream,
                'watermark' => filter_var($data['watermark'] ?? false, FILTER_VALIDATE_BOOLEAN),
            ];
            if ($sequentialMode !== 'disabled') {
                $requestBody['sequential_image_generation_options'] = [
                    'max_images' => $maxImages,
                ];
            }
            if ($referenceCount > 0) {
                $requestBody['image'] = $referenceCount === 1 ? $referenceImages[0] : $referenceImages;
            }

            $genConfig['max_images'] = $maxImages;
            $genConfig['sequential_image_generation'] = $sequentialMode;
            $genConfig['mode'] = self::detectGenerationMode($referenceCount, $maxImages);
        }

        $result = self::callSeedreamApi($requestBody, $model, $stream);
        $images = self::normalizeImageResult($result);
        $imgId = self::extractApiImageId($result);
        $taskId = trim((string)($data['task_id'] ?? ''));

        $stored = self::persistGeneratedImagesToOss($images, $imgId, $referenceImages, $taskId);
        $images = $stored['images'];
        $generateImgUrl = implode(',', $stored['urls']);

        $payload = [
            'img_id' => $imgId,
            'mode' => $genConfig['mode'] ?? '',
            'reference_count' => count($referenceImages),
            'reference_images' => $referenceImages,
            'generate_images' => $stored['urls'],
            'max_images' => (int)($genConfig['max_images'] ?? count($images)),
            'images' => $images,
            'model' => $result['model'] ?? $model,
            'usage' => $result['usage'] ?? null,
            'created' => $result['created'] ?? null,
        ];

        try {
            $payload['storage_id'] = self::savePictureStorage([
                'img_id' => $imgId,
                'prompt' => $prompt,
                'reference_img_url' => implode(',', $referenceImages),
                'generate_img_url' => $generateImgUrl,
                'model' => $result['model'] ?? $model,
                'img_size' => $size,
            ]);
        } catch (\Exception $e) {
            $payload['db_save_error'] = $e->getMessage();
            Log::write('[SeedreamImageGenerationJob] 数据库保存失败: ' . $e->getMessage(), 'error');
        }

        return $payload;
    }

    private static function detectGenerationMode(int $referenceCount, int $maxImages): string
    {
        if ($referenceCount === 0) {
            return $maxImages > 1 ? 'text_to_images' : 'text_to_image';
        }
        if ($referenceCount === 1) {
            return $maxImages > 1 ? 'image_to_images' : 'image_to_image';
        }
        return $maxImages > 1 ? 'images_to_images' : 'images_to_image';
    }

    private static function callSeedreamApi(array $requestBody, string $model, bool $stream = false): array
    {
        $configs = self::resolveModelApiConfigs($model);
        if (empty($configs)) {
            throw new \Exception("模型配置不存在: {$model}");
        }

        $postData = json_encode($requestBody, JSON_UNESCAPED_UNICODE);
        if (json_last_error() !== JSON_ERROR_NONE) {
            throw new \Exception('请求数据 JSON 编码失败: ' . json_last_error_msg());
        }

        $errors = [];
        foreach ($configs as $index => $config) {
            $apiUrl = self::sanitizeApiUrl((string)($config['api_url'] ?? ''));
            $apiKey = trim((string)($config['api_key'] ?? ''));
            if ($apiUrl === '' || $apiKey === '') {
                $errors[] = '第' . ($index + 1) . '个接口配置无效（URL/Key为空）';
                continue;
            }

            $headers = [
                'Content-Type: application/json',
                'Authorization: Bearer ' . $apiKey,
            ];
            if ($stream) {
                $headers[] = 'Accept: text/event-stream';
            }

            $ch = curl_init();
            curl_setopt_array($ch, [
                CURLOPT_URL => $apiUrl,
                CURLOPT_POST => true,
                CURLOPT_POSTFIELDS => $postData,
                CURLOPT_HTTPHEADER => $headers,
                CURLOPT_RETURNTRANSFER => true,
                CURLOPT_TIMEOUT => 900,
                CURLOPT_CONNECTTIMEOUT => 30,
                CURLOPT_SSL_VERIFYPEER => false,
                CURLOPT_SSL_VERIFYHOST => 0,
            ]);

            $response = curl_exec($ch);
            $httpCode = (int)curl_getinfo($ch, CURLINFO_HTTP_CODE);
            $curlErrno = curl_errno($ch);
            $curlError = curl_error($ch);
            curl_close($ch);

            if ($response === false) {
                $errors[] = '第' . ($index + 1) . '个接口请求失败 [curl:' . $curlErrno . '] '
                    . ($curlError !== '' ? $curlError : '请求超时或网络异常');
                continue;
            }

            if ($httpCode !== 200) {
                $decoded = json_decode($response, true);
                $msg = is_array($decoded)
                    ? ($decoded['error']['message'] ?? json_encode($decoded, JSON_UNESCAPED_UNICODE))
                    : self::truncateText((string)$response, 300);
                $errors[] = '第' . ($index + 1) . '个接口 HTTP ' . $httpCode . ': ' . $msg;
                continue;
            }

            try {
                return $stream ? self::parseStreamResponse($response) : self::parseJsonResponse($response);
            } catch (\Exception $e) {
                $errors[] = '第' . ($index + 1) . '个接口响应解析失败: ' . $e->getMessage();
            }
        }

        throw new \Exception(($stream ? '流式' : '') . '生成失败: ' . implode('; ', $errors));
    }

    private static function resolveModelApiConfigs(string $model): array
    {
        $rows = Db::name('ai_model')
            ->where('model_name', $model)
            ->where('status', '1')
            ->order('sort ASC')
            ->select();

        $configs = [];
        $seenUrls = [];
        $fallbackKey = '';

        foreach ($rows as $row) {
            $apiUrl = self::sanitizeApiUrl((string)($row['api_url'] ?? ''));
            $apiKey = trim((string)($row['api_key'] ?? ''));
            if ($apiKey !== '' && $fallbackKey === '') {
                $fallbackKey = $apiKey;
            }
            if ($apiUrl === '' || $apiKey === '') {
                continue;
            }
            $urlKey = strtolower(rtrim($apiUrl, '/'));
            if (isset($seenUrls[$urlKey])) {
                continue;
            }
            $seenUrls[$urlKey] = true;
            $configs[] = ['api_url' => $apiUrl, 'api_key' => $apiKey];
        }

        $defaultUrl = rtrim(self::DEFAULT_API_URL, '/');
        if ($fallbackKey !== '' && !isset($seenUrls[strtolower($defaultUrl)])) {
            $configs[] = ['api_url' => self::DEFAULT_API_URL, 'api_key' => $fallbackKey];
        }

        return $configs;
    }

    private static function sanitizeApiUrl(string $url): string
    {
        $url = trim(str_replace(["\r\n", "\r", "\n", "\t"], ' ', $url));
        if ($url === '') {
            return '';
        }
        foreach (preg_split('/\s+/', $url) as $part) {
            $part = trim($part);
            if ($part !== '' && preg_match('/^https?:\/\//i', $part)) {
                return $part;
            }
        }
        return '';
    }

    private static function parseJsonResponse(string $raw): array
    {
        $result = json_decode($raw, true);
        if (!is_array($result)) {
            throw new \Exception('API 响应解析失败');
        }
        if (isset($result['error'])) {
            throw new \Exception($result['error']['message'] ?? 'API 返回错误');
        }
        if (empty($result['data']) || !is_array($result['data'])) {
            throw new \Exception('未解析到有效图片结果');
        }
        return $result;
    }

    private static function parseStreamResponse(string $raw): array
    {
        $images = [];
        $usage = null;
        $model = '';
        $created = null;
        $apiId = '';

        foreach (preg_split('/\r?\n/', $raw) as $line) {
            $line = trim($line);
            if ($line === '' || $line === 'data: [DONE]' || stripos($line, 'data:') !== 0) {
                continue;
            }
            $event = json_decode(trim(substr($line, 5)), true);
            if (!is_array($event)) {
                continue;
            }
            if (isset($event['error'])) {
                throw new \Exception($event['error']['message'] ?? 'API 返回错误');
            }
            if (!empty($event['model'])) {
                $model = (string)$event['model'];
            }
            if (!empty($event['id'])) {
                $apiId = (string)$event['id'];
            }
            if (isset($event['created'])) {
                $created = $event['created'];
            }
            if (isset($event['usage'])) {
                $usage = $event['usage'];
            }
            $type = (string)($event['type'] ?? '');
            if ($type === 'image_generation.partial_succeeded' && !empty($event['url'])) {
                $images[] = [
                    'url' => $event['url'],
                    'size' => $event['size'] ?? '',
                    'image_index' => $event['image_index'] ?? count($images),
                ];
            }
            if ($type === 'image_generation.completed' && !empty($event['data']) && is_array($event['data'])) {
                foreach ($event['data'] as $item) {
                    if (!empty($item['url'])) {
                        $images[] = $item;
                    }
                }
            }
        }

        if (empty($images)) {
            $decoded = json_decode($raw, true);
            if (is_array($decoded) && !empty($decoded['data'])) {
                return $decoded;
            }
            throw new \Exception('未解析到有效图片结果');
        }

        return [
            'id' => $apiId,
            'model' => $model,
            'created' => $created,
            'data' => $images,
            'usage' => $usage,
        ];
    }

    private static function extractApiImageId(array $result): string
    {
        foreach (['id', 'request_id', 'task_id'] as $key) {
            if (!empty($result[$key])) {
                return substr((string)$result[$key], 0, 50);
            }
        }
        return substr(str_replace('.', '', uniqid('img', true)), 0, 50);
    }

    private static function savePictureStorage(array $data): int
    {
        $insert = [
            'img_id' => substr((string)($data['img_id'] ?? ''), 0, 50),
            'prompt' => (string)($data['prompt'] ?? ''),
            'reference_img_url' => (string)($data['reference_img_url'] ?? ''),
            'generate_img_url' => (string)($data['generate_img_url'] ?? ''),
            'model' => substr((string)($data['model'] ?? ''), 0, 255),
            'img_size' => substr((string)($data['img_size'] ?? ''), 0, 255),
            'sys_rq' => date('Y-m-d H:i:s'),
        ];
        if ($insert['img_id'] === '' || $insert['model'] === '') {
            throw new \Exception('picture_storage 入库参数不完整');
        }
        $storageId = Db::name('picture_storage')->insertGetId($insert);
        if (!$storageId) {
            throw new \Exception('picture_storage 入库失败');
        }
        return (int)$storageId;
    }

    private static function normalizeImageResult(array $result): array
    {
        $items = $result['data'] ?? [];
        if (!is_array($items)) {
            return [];
        }
        $images = [];
        foreach ($items as $item) {
            if (!is_array($item) || empty($item['url'])) {
                continue;
            }
            $images[] = [
                'url' => $item['url'],
                'size' => $item['size'] ?? '',
                'image_index' => $item['image_index'] ?? null,
            ];
        }
        return $images;
    }

    /**
     * 生成图落盘并上传 OSS（目录/命名规则与参考图一致：uploads/imagegen/{日期}/）
     * @return array{urls:string[],images:array}
     */
    private static function persistGeneratedImagesToOss(
        array $images,
        string $imgId,
        array $referenceImages,
        string $taskId = ''
    ): array {
        if (empty($images)) {
            throw new \Exception('没有可保存的生成图');
        }
        if (!self::isOssEnabled()) {
            throw new \Exception('OSS 未配置，无法保存生成图');
        }

        $defaultExt = self::resolveOutputExtFromReference($referenceImages);
        $dateDir = date('Ymd');
        $saveDir = ROOT_PATH . 'public' . DIRECTORY_SEPARATOR . 'uploads' . DIRECTORY_SEPARATOR
            . 'imagegen' . DIRECTORY_SEPARATOR . $dateDir . DIRECTORY_SEPARATOR;
        if (!is_dir($saveDir)) {
            mkdir($saveDir, 0755, true);
        }

        $safeId = preg_replace('/[^a-zA-Z0-9_]/', '_', $imgId);
        $safeTask = $taskId !== '' ? preg_replace('/[^a-zA-Z0-9_]/', '_', $taskId) : '';
        $ossUrls = [];
        $storedImages = [];

        foreach ($images as $index => $image) {
            $sourceUrl = trim((string)($image['url'] ?? ''));
            if ($sourceUrl === '') {
                continue;
            }

            $binary = self::downloadRemoteImage($sourceUrl);
            $ext = self::detectImageExt($binary, $defaultExt);
            $suffix = $safeTask !== '' ? $safeTask . '_' . $index : $safeId . '_' . $index;
            $fileName = 'gen_' . $suffix . '.' . $ext;
            $localFullPath = $saveDir . $fileName;

            if (file_put_contents($localFullPath, $binary) === false) {
                throw new \Exception('生成图本地保存失败');
            }

            $objectKey = 'uploads/imagegen/' . $dateDir . '/' . $fileName;
            if (!self::putFileToOss($localFullPath, $objectKey)) {
                throw new \Exception('生成图上传 OSS 失败: ' . $objectKey);
            }

            $ossUrl = self::buildOssPublicUrl($objectKey);
            if (stripos($ossUrl, 'http://') !== 0 && stripos($ossUrl, 'https://') !== 0) {
                throw new \Exception('无法获取生成图 OSS 访问地址');
            }

            $ossUrls[] = $ossUrl;
            $storedImages[] = array_merge($image, [
                'url' => $ossUrl,
                'oss_url' => $ossUrl,
                'object_key' => $objectKey,
                'source_url' => $sourceUrl,
            ]);
        }

        if (empty($ossUrls)) {
            throw new \Exception('生成图 OSS 保存结果为空');
        }

        return ['urls' => $ossUrls, 'images' => $storedImages];
    }

    /**
     * 从参考图推断输出扩展名（无参考图时默认 jpg）
     */
    private static function resolveOutputExtFromReference(array $referenceImages): string
    {
        foreach ($referenceImages as $url) {
            $url = trim((string)$url);
            if ($url === '') {
                continue;
            }
            $path = (string)(parse_url($url, PHP_URL_PATH) ?: '');
            $ext = strtolower(pathinfo($path, PATHINFO_EXTENSION));
            if ($ext === 'jpeg') {
                $ext = 'jpg';
            }
            if (in_array($ext, ['jpg', 'png', 'gif', 'webp', 'bmp'], true)) {
                return $ext;
            }
        }
        return 'jpg';
    }

    private static function detectImageExt(string $binary, string $fallback = 'jpg'): string
    {
        $finfo = finfo_open(FILEINFO_MIME_TYPE);
        $mime = finfo_buffer($finfo, $binary);
        finfo_close($finfo);

        $map = [
            'image/jpeg' => 'jpg',
            'image/png' => 'png',
            'image/gif' => 'gif',
            'image/webp' => 'webp',
            'image/bmp' => 'bmp',
        ];

        if ($mime && isset($map[$mime])) {
            return $map[$mime];
        }

        $fallback = strtolower($fallback);
        return in_array($fallback, ['jpg', 'png', 'gif', 'webp', 'bmp'], true) ? $fallback : 'jpg';
    }

    private static function downloadRemoteImage(string $url): string
    {
        $ch = curl_init();
        curl_setopt_array($ch, [
            CURLOPT_URL => $url,
            CURLOPT_RETURNTRANSFER => true,
            CURLOPT_FOLLOWLOCATION => true,
            CURLOPT_TIMEOUT => 120,
            CURLOPT_CONNECTTIMEOUT => 15,
            CURLOPT_SSL_VERIFYPEER => false,
            CURLOPT_SSL_VERIFYHOST => 0,
        ]);
        $binary = curl_exec($ch);
        $httpCode = (int)curl_getinfo($ch, CURLINFO_HTTP_CODE);
        $curlError = curl_error($ch);
        curl_close($ch);

        if ($binary === false || $binary === '') {
            throw new \Exception('生成图下载失败: ' . ($curlError ?: $url));
        }
        if ($httpCode !== 200) {
            throw new \Exception('生成图下载失败 HTTP ' . $httpCode);
        }
        if (strlen($binary) < 100) {
            throw new \Exception('生成图内容无效');
        }

        return $binary;
    }

    private static function getOssConfig(): array
    {
        $config = Config::get('oss');
        return is_array($config) ? $config : [];
    }

    private static function isOssEnabled(): bool
    {
        $config = self::getOssConfig();
        return !empty($config['accessKeyId'])
            && !empty($config['accessKeySecret'])
            && !empty($config['endpoint'])
            && !empty($config['bucket']);
    }

    private static function normalizeOssObjectKey(string $objectKey): string
    {
        return ltrim(str_replace('\\', '/', trim($objectKey)), '/');
    }

    private static function putFileToOss(string $localFullPath, string $objectKey): bool
    {
        if (!self::isOssEnabled() || !is_file($localFullPath)) {
            return false;
        }

        $config = self::getOssConfig();
        $objectKey = self::normalizeOssObjectKey($objectKey);
        if ($objectKey === '') {
            return false;
        }

        try {
            $ossClient = new OssClient(
                $config['accessKeyId'],
                $config['accessKeySecret'],
                $config['endpoint']
            );
            $ossClient->uploadFile($config['bucket'], $objectKey, $localFullPath);
            return true;
        } catch (\Throwable $e) {
            Log::write('[SeedreamImageGenerationJob/OSS] ' . $e->getMessage() . ' | objectKey=' . $objectKey, 'error');
            return false;
        }
    }

    private static function buildOssPublicUrl(string $path): string
    {
        $path = trim($path);
        if ($path === '' || stripos($path, 'http://') === 0 || stripos($path, 'https://') === 0) {
            return $path;
        }

        $config = self::getOssConfig();
        $host = trim((string)($config['host'] ?? ''));
        if ($host === '') {
            return $path;
        }
        if (stripos($host, 'http://') !== 0 && stripos($host, 'https://') !== 0) {
            $host = 'https://' . $host;
        }

        return rtrim($host, '/') . '/' . ltrim($path, '/');
    }

    private static function truncateText(string $text, int $maxLen): string
    {
        $text = trim(preg_replace('/\s+/', ' ', $text));
        if ($text === '' || strlen($text) <= $maxLen) {
            return $text;
        }
        return substr($text, 0, $maxLen) . '...';
    }
}