608 lines
26 KiB
Python
608 lines
26 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Any
|
|
|
|
from fastapi import HTTPException
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.config import settings
|
|
from app.enums.audio_reference import (
|
|
AUDIO_MAX_COUNT_LIMIT,
|
|
AUDIO_MAX_DURATION_SECONDS,
|
|
AUDIO_MAX_TOTAL_DURATION_SECONDS,
|
|
AUDIO_MIN_DURATION_SECONDS,
|
|
)
|
|
from app.enums.generation_task import CHAT_TOP_LEVEL_MODES, GenerationMode, GenerationType
|
|
from app.models.chat_generation_task import ChatGenerationTask
|
|
from app.models.user import User
|
|
from app.schemas.generation_ai import GenerationAITaskCreate
|
|
from app.services.generation.ai.engine_service import (
|
|
IMAGE_DEFAULT_PROPORTION,
|
|
IMAGE_DEFAULT_PX,
|
|
IMAGE_DEFAULT_SIZE,
|
|
VIDEO_DEFAULT_DURATION,
|
|
VIDEO_DEFAULT_RATIO,
|
|
VIDEO_DEFAULT_RESOLUTION,
|
|
build_image_snapshot,
|
|
build_video_snapshot,
|
|
get_image_engine,
|
|
get_video_engine,
|
|
image_supported_sizes,
|
|
normalize_generation_count,
|
|
normalize_px,
|
|
parse_json_list,
|
|
)
|
|
from app.services.generation.billing_service import OWNER_CHAT_GENERATION_TASK, charge_generation_media_by_params
|
|
from app.services.operation_log_service import log_operation_event
|
|
from app.services.private_portrait.reference_resolver import resolve_private_portrait_references
|
|
from app.services.resource_capacity_service import assert_user_resource_capacity_available
|
|
from app.services.video_upscale.snapshot_service import build_video_upscale_snapshot
|
|
from app.utils.id_gen import generate_id
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class GenerationTaskCreateResult:
|
|
top_level_task_id: str
|
|
enqueue_task_ids: list[str] = field(default_factory=list)
|
|
child_task_ids: list[str] = field(default_factory=list)
|
|
generation_count: int = 1
|
|
gen_type: str = GenerationType.IMAGE.value
|
|
created: bool = True
|
|
|
|
|
|
def _json(data: Any) -> str | None:
|
|
if data is None:
|
|
return None
|
|
return json.dumps(data, ensure_ascii=False, default=str)
|
|
|
|
|
|
async def find_existing_top_level_task(
|
|
db: AsyncSession,
|
|
*,
|
|
user_id: str,
|
|
idempotency_key: str | None,
|
|
) -> ChatGenerationTask | None:
|
|
if not idempotency_key:
|
|
return None
|
|
result = await db.execute(
|
|
select(ChatGenerationTask)
|
|
.where(
|
|
ChatGenerationTask.user_id == user_id,
|
|
ChatGenerationTask.idempotency_key == idempotency_key,
|
|
ChatGenerationTask.generation_mode.in_(list(CHAT_TOP_LEVEL_MODES)),
|
|
ChatGenerationTask.deleted_at.is_(None),
|
|
)
|
|
.order_by(ChatGenerationTask.created_at.desc())
|
|
.limit(1)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
def _validate_video_references(refs: list[dict], *, max_audio_count: int) -> float:
|
|
input_video_duration = 0.0
|
|
for ref in refs:
|
|
if (ref.get("type") or "").lower() != GenerationType.VIDEO.value:
|
|
continue
|
|
try:
|
|
ref_duration = float(ref.get("duration") or 0)
|
|
except (TypeError, ValueError):
|
|
ref_duration = 0.0
|
|
if ref_duration < 2:
|
|
raise HTTPException(status_code=400, detail="视频素材最短不能少于 2 秒")
|
|
input_video_duration += ref_duration
|
|
if input_video_duration > 15:
|
|
raise HTTPException(status_code=400, detail=f"所有视频素材总时长不能超过 15 秒，当前 {input_video_duration:.1f} 秒")
|
|
|
|
audio_refs = [ref for ref in refs if (ref.get("type") or "").lower() == "audio"]
|
|
if audio_refs:
|
|
allowed_count = min(AUDIO_MAX_COUNT_LIMIT, max(0, int(max_audio_count or 0)))
|
|
if allowed_count <= 0:
|
|
raise HTTPException(status_code=400, detail="当前视频引擎不支持音频参考素材")
|
|
if len(audio_refs) > allowed_count:
|
|
raise HTTPException(status_code=400, detail=f"参考音频最多可传 {allowed_count} 段，当前 {len(audio_refs)} 段")
|
|
|
|
input_audio_duration = 0.0
|
|
for ref in audio_refs:
|
|
try:
|
|
ref_duration = float(ref.get("duration") or 0)
|
|
except (TypeError, ValueError):
|
|
ref_duration = 0.0
|
|
if ref_duration < AUDIO_MIN_DURATION_SECONDS or ref_duration > AUDIO_MAX_DURATION_SECONDS:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"单段参考音频时长必须在 {AUDIO_MIN_DURATION_SECONDS}-{AUDIO_MAX_DURATION_SECONDS} 秒之间",
|
|
)
|
|
input_audio_duration += ref_duration
|
|
if input_audio_duration > AUDIO_MAX_TOTAL_DURATION_SECONDS:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"所有参考音频总时长不能超过 {AUDIO_MAX_TOTAL_DURATION_SECONDS} 秒，当前 {input_audio_duration:.1f} 秒",
|
|
)
|
|
return input_video_duration
|
|
|
|
|
|
def _base_task_kwargs(
|
|
*,
|
|
task_id: str,
|
|
user_id: str,
|
|
req: GenerationAITaskCreate,
|
|
gen_type: str,
|
|
generation_mode: str,
|
|
generation_count: int,
|
|
engine_id: str,
|
|
engine_snapshot_json: str,
|
|
media_references_json: str | None,
|
|
deadline_at: datetime,
|
|
parent_task_id: str | None = None,
|
|
generation_index: int | None = None,
|
|
credits_cost: float = 0.0,
|
|
idempotency_key: str | None = None,
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"id": task_id,
|
|
"user_id": user_id,
|
|
"original_prompt": req.original_prompt,
|
|
"gen_type": gen_type,
|
|
"status": "generating",
|
|
"generation_mode": generation_mode,
|
|
"pipeline_stage": "queued",
|
|
"parent_task_id": parent_task_id,
|
|
"generation_count": generation_count,
|
|
"generation_index": generation_index,
|
|
"engine_id": engine_id,
|
|
"engine_snapshot_json": engine_snapshot_json,
|
|
"media_references": media_references_json,
|
|
"credits_cost": round(float(credits_cost or 0), 2),
|
|
"idempotency_key": idempotency_key,
|
|
"deadline_at": deadline_at,
|
|
}
|
|
|
|
|
|
async def create_generation_task_group(
|
|
db: AsyncSession,
|
|
current_user: User,
|
|
req: GenerationAITaskCreate,
|
|
) -> GenerationTaskCreateResult:
|
|
"""创建单份 chatapi_async 或多份 chatapi_main/chatapi_child 任务组。
|
|
|
|
本函数只 flush，不主动 commit。调用方提交成功后才能投递 Celery。
|
|
"""
|
|
gen_type = (req.gen_type or "").lower().strip()
|
|
if gen_type not in (GenerationType.IMAGE.value, GenerationType.VIDEO.value):
|
|
raise HTTPException(status_code=400, detail="gen_type 仅支持 image 或 video")
|
|
|
|
existing = await find_existing_top_level_task(
|
|
db,
|
|
user_id=current_user.id,
|
|
idempotency_key=req.idempotency_key,
|
|
)
|
|
if existing:
|
|
return GenerationTaskCreateResult(
|
|
top_level_task_id=existing.id,
|
|
generation_count=int(existing.generation_count or 1),
|
|
gen_type=existing.gen_type,
|
|
created=False,
|
|
)
|
|
|
|
refs = [item.model_dump(exclude_none=True) for item in (req.media_references or [])]
|
|
refs = await resolve_private_portrait_references(
|
|
db,
|
|
user_id=current_user.id,
|
|
media_references=refs,
|
|
gen_type=gen_type,
|
|
)
|
|
media_references_json = _json(refs) if refs else None
|
|
await assert_user_resource_capacity_available(db, current_user.id)
|
|
|
|
now = datetime.now(timezone.utc)
|
|
main_id = generate_id()
|
|
child_ids: list[str] = []
|
|
enqueue_ids: list[str] = []
|
|
total_billed_credits = 0.0
|
|
|
|
log_operation_event(
|
|
domain="generation_ai_batch",
|
|
event_type="BATCH_CREATE_START",
|
|
event_status="started",
|
|
source="service",
|
|
user_id=current_user.id,
|
|
group_id=main_id,
|
|
detail={
|
|
"gen_type": gen_type,
|
|
"requested_generation_count": normalize_generation_count(req.generation_count),
|
|
"idempotency_key_present": bool(req.idempotency_key),
|
|
},
|
|
)
|
|
|
|
if gen_type == GenerationType.IMAGE.value:
|
|
if any((ref.get("type") or "").lower() == "audio" for ref in refs):
|
|
raise HTTPException(status_code=400, detail="图片生成不支持音频参考素材")
|
|
|
|
engine = await get_image_engine(db, req.engine_id)
|
|
generation_count = normalize_generation_count(req.generation_count)
|
|
multi_generation_enabled = bool(getattr(engine, "multi_generation_enabled", False))
|
|
max_generation_count = normalize_generation_count(getattr(engine, "max_generation_count", 1))
|
|
if generation_count > 1 and not multi_generation_enabled:
|
|
raise HTTPException(status_code=400, detail="当前图片引擎未开启多份生成，本次生成数量只能为 1")
|
|
if generation_count > max_generation_count:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"当前图片引擎本次最多允许生成 {max_generation_count} 份",
|
|
)
|
|
|
|
reference_image_count = sum(
|
|
1 for ref in refs if (ref.get("type") or "").lower() == GenerationType.IMAGE.value
|
|
)
|
|
max_reference_count = max(0, int(getattr(engine, "max_reference_image_count", 14) or 0))
|
|
multi_image_max_images = max(1, int(getattr(engine, "multi_image_max_images", 15) or 15))
|
|
if reference_image_count > max_reference_count:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"当前图片引擎最多支持 {max_reference_count} 张参考图，当前 {reference_image_count} 张",
|
|
)
|
|
if generation_count > 1 and reference_image_count + generation_count > multi_image_max_images:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=(
|
|
f"参考图数量与生成数量合计不能超过 {multi_image_max_images} 张，"
|
|
f"当前参考图 {reference_image_count} 张、生成 {generation_count} 张"
|
|
),
|
|
)
|
|
|
|
sizes = image_supported_sizes(engine)
|
|
size = req.image_size or engine.default_size or IMAGE_DEFAULT_SIZE
|
|
proportion = req.image_proportion or IMAGE_DEFAULT_PROPORTION
|
|
px = normalize_px(req.image_px)
|
|
if sizes:
|
|
if size not in sizes:
|
|
raise HTTPException(status_code=400, detail=f"图片分辨率档位不支持: {size}")
|
|
if proportion not in sizes.get(size, {}):
|
|
raise HTTPException(status_code=400, detail=f"图片比例不支持: {proportion}")
|
|
px = px or normalize_px((sizes.get(size) or {}).get(proportion))
|
|
px = px or IMAGE_DEFAULT_PX
|
|
|
|
mode = GenerationMode.CHATAPI_ASYNC.value if generation_count == 1 else GenerationMode.CHATAPI_MAIN.value
|
|
billing = await charge_generation_media_by_params(
|
|
db,
|
|
user_id=current_user.id,
|
|
record_id=main_id,
|
|
gen_type=GenerationType.IMAGE.value,
|
|
image_size=size,
|
|
engine_id=engine.id,
|
|
project_name="AI生成任务",
|
|
description_prefix="AI创作-",
|
|
owner_type=OWNER_CHAT_GENERATION_TASK,
|
|
attempt_no=1,
|
|
quantity=generation_count,
|
|
)
|
|
image_snapshot = build_image_snapshot(engine, size, proportion, px)
|
|
image_snapshot["generation_count"] = generation_count
|
|
snapshot_json = _json(image_snapshot) or "{}"
|
|
total_billed_credits = round(float(billing.total_charged or 0), 2)
|
|
task = ChatGenerationTask(
|
|
**_base_task_kwargs(
|
|
task_id=main_id,
|
|
user_id=current_user.id,
|
|
req=req,
|
|
gen_type=GenerationType.IMAGE.value,
|
|
generation_mode=mode,
|
|
generation_count=generation_count,
|
|
engine_id=engine.id,
|
|
engine_snapshot_json=snapshot_json,
|
|
media_references_json=media_references_json,
|
|
deadline_at=now + timedelta(minutes=settings.CHATAPI_ASYNC_IMAGE_DEADLINE_MINUTES),
|
|
credits_cost=billing.total_charged,
|
|
idempotency_key=req.idempotency_key,
|
|
),
|
|
image_size=size,
|
|
image_proportion=proportion,
|
|
image_px=px,
|
|
)
|
|
db.add(task)
|
|
enqueue_ids.append(task.id)
|
|
else:
|
|
engine = await get_video_engine(db, req.engine_id)
|
|
generation_count = normalize_generation_count(req.generation_count)
|
|
multi_generation_enabled = bool(getattr(engine, "multi_generation_enabled", False))
|
|
max_generation_count = normalize_generation_count(getattr(engine, "max_generation_count", 1))
|
|
if generation_count > 1 and not multi_generation_enabled:
|
|
raise HTTPException(status_code=400, detail="当前视频引擎未开启多份生成，本次生成数量只能为 1")
|
|
if generation_count > max_generation_count:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"当前视频引擎本次最多允许生成 {max_generation_count} 份",
|
|
)
|
|
ratio = req.aspect_ratio or VIDEO_DEFAULT_RATIO
|
|
resolution = req.resolution or VIDEO_DEFAULT_RESOLUTION
|
|
duration = req.duration or VIDEO_DEFAULT_DURATION
|
|
ratios = parse_json_list(engine.supported_ratios, [])
|
|
resolutions = parse_json_list(engine.supported_resolutions, [])
|
|
durations = parse_json_list(engine.supported_durations, [])
|
|
if ratios and ratio not in ratios:
|
|
raise HTTPException(status_code=400, detail=f"视频比例不支持: {ratio}")
|
|
if resolutions and resolution not in resolutions:
|
|
raise HTTPException(status_code=400, detail=f"视频分辨率不支持: {resolution}")
|
|
if durations and duration not in durations:
|
|
raise HTTPException(status_code=400, detail=f"视频时长不支持: {duration}")
|
|
if engine.max_duration and duration > engine.max_duration:
|
|
raise HTTPException(status_code=400, detail=f"视频时长不能超过 {engine.max_duration} 秒")
|
|
|
|
input_video_duration = _validate_video_references(refs, max_audio_count=engine.max_audio_count)
|
|
provider_generation_resolution, upscale_enabled_snapshot, upscale_snapshot_json = await build_video_upscale_snapshot(
|
|
db,
|
|
target_resolution=resolution,
|
|
aspect_ratio=ratio,
|
|
supported_provider_resolutions=resolutions,
|
|
)
|
|
video_snapshot = build_video_snapshot(engine, ratio, resolution, duration)
|
|
video_snapshot["provider_generation_resolution"] = provider_generation_resolution
|
|
video_snapshot["video_upscale_enabled_snapshot"] = upscale_enabled_snapshot
|
|
video_snapshot["generation_count"] = generation_count
|
|
snapshot_json = _json(video_snapshot) or "{}"
|
|
deadline_at = now + timedelta(hours=settings.CHATAPI_ASYNC_VIDEO_FINAL_DEADLINE_HOURS)
|
|
|
|
if generation_count == 1:
|
|
billing = await charge_generation_media_by_params(
|
|
db,
|
|
user_id=current_user.id,
|
|
record_id=main_id,
|
|
gen_type=GenerationType.VIDEO.value,
|
|
duration=duration,
|
|
resolution=resolution,
|
|
engine_id=engine.id,
|
|
input_video_duration=input_video_duration if input_video_duration > 0 else None,
|
|
project_name="AI生成任务",
|
|
description_prefix="AI创作-",
|
|
owner_type=OWNER_CHAT_GENERATION_TASK,
|
|
attempt_no=1,
|
|
)
|
|
total_billed_credits = round(float(billing.total_charged or 0), 2)
|
|
task = ChatGenerationTask(
|
|
**_base_task_kwargs(
|
|
task_id=main_id,
|
|
user_id=current_user.id,
|
|
req=req,
|
|
gen_type=GenerationType.VIDEO.value,
|
|
generation_mode=GenerationMode.CHATAPI_ASYNC.value,
|
|
generation_count=1,
|
|
engine_id=engine.id,
|
|
engine_snapshot_json=snapshot_json,
|
|
media_references_json=media_references_json,
|
|
deadline_at=deadline_at,
|
|
credits_cost=billing.total_charged,
|
|
idempotency_key=req.idempotency_key,
|
|
),
|
|
duration=duration,
|
|
aspect_ratio=ratio,
|
|
resolution=resolution,
|
|
provider_generation_resolution=provider_generation_resolution,
|
|
video_upscale_enabled_snapshot=upscale_enabled_snapshot,
|
|
video_upscale_snapshot_json=upscale_snapshot_json,
|
|
image_size=req.image_size or IMAGE_DEFAULT_SIZE,
|
|
image_proportion=req.image_proportion or IMAGE_DEFAULT_PROPORTION,
|
|
image_px=normalize_px(req.image_px) or IMAGE_DEFAULT_PX,
|
|
)
|
|
db.add(task)
|
|
enqueue_ids.append(task.id)
|
|
else:
|
|
main_task = ChatGenerationTask(
|
|
**_base_task_kwargs(
|
|
task_id=main_id,
|
|
user_id=current_user.id,
|
|
req=req,
|
|
gen_type=GenerationType.VIDEO.value,
|
|
generation_mode=GenerationMode.CHATAPI_MAIN.value,
|
|
generation_count=generation_count,
|
|
engine_id=engine.id,
|
|
engine_snapshot_json=snapshot_json,
|
|
media_references_json=media_references_json,
|
|
deadline_at=deadline_at,
|
|
idempotency_key=req.idempotency_key,
|
|
),
|
|
duration=duration,
|
|
aspect_ratio=ratio,
|
|
resolution=resolution,
|
|
provider_generation_resolution=provider_generation_resolution,
|
|
video_upscale_enabled_snapshot=upscale_enabled_snapshot,
|
|
video_upscale_snapshot_json=upscale_snapshot_json,
|
|
image_size=req.image_size or IMAGE_DEFAULT_SIZE,
|
|
image_proportion=req.image_proportion or IMAGE_DEFAULT_PROPORTION,
|
|
image_px=normalize_px(req.image_px) or IMAGE_DEFAULT_PX,
|
|
)
|
|
db.add(main_task)
|
|
await db.flush()
|
|
|
|
total_credits = 0.0
|
|
children: list[ChatGenerationTask] = []
|
|
for generation_index in range(1, generation_count + 1):
|
|
child_id = generate_id()
|
|
billing = await charge_generation_media_by_params(
|
|
db,
|
|
user_id=current_user.id,
|
|
record_id=child_id,
|
|
gen_type=GenerationType.VIDEO.value,
|
|
duration=duration,
|
|
resolution=resolution,
|
|
engine_id=engine.id,
|
|
input_video_duration=input_video_duration if input_video_duration > 0 else None,
|
|
project_name="AI生成任务",
|
|
description_prefix=f"AI创作-第{generation_index}份-",
|
|
owner_type=OWNER_CHAT_GENERATION_TASK,
|
|
attempt_no=1,
|
|
)
|
|
child = ChatGenerationTask(
|
|
**_base_task_kwargs(
|
|
task_id=child_id,
|
|
user_id=current_user.id,
|
|
req=req,
|
|
gen_type=GenerationType.VIDEO.value,
|
|
generation_mode=GenerationMode.CHATAPI_CHILD.value,
|
|
generation_count=generation_count,
|
|
generation_index=generation_index,
|
|
parent_task_id=main_id,
|
|
engine_id=engine.id,
|
|
engine_snapshot_json=snapshot_json,
|
|
media_references_json=media_references_json,
|
|
deadline_at=deadline_at,
|
|
credits_cost=billing.total_charged,
|
|
),
|
|
duration=duration,
|
|
aspect_ratio=ratio,
|
|
resolution=resolution,
|
|
provider_generation_resolution=provider_generation_resolution,
|
|
video_upscale_enabled_snapshot=upscale_enabled_snapshot,
|
|
video_upscale_snapshot_json=upscale_snapshot_json,
|
|
image_size=req.image_size or IMAGE_DEFAULT_SIZE,
|
|
image_proportion=req.image_proportion or IMAGE_DEFAULT_PROPORTION,
|
|
image_px=normalize_px(req.image_px) or IMAGE_DEFAULT_PX,
|
|
)
|
|
children.append(child)
|
|
child_ids.append(child_id)
|
|
enqueue_ids.append(child_id)
|
|
total_credits = round(total_credits + billing.total_charged, 2)
|
|
db.add_all(children)
|
|
main_task.credits_cost = total_credits
|
|
total_billed_credits = total_credits
|
|
|
|
await db.flush()
|
|
log_operation_event(
|
|
domain="generation_ai_batch",
|
|
event_type="BATCH_BILLING_SUCCESS",
|
|
event_status="success",
|
|
source="service",
|
|
user_id=current_user.id,
|
|
group_id=main_id,
|
|
detail={
|
|
"gen_type": gen_type,
|
|
"generation_count": generation_count,
|
|
"total_billed_credits": total_billed_credits,
|
|
},
|
|
)
|
|
log_operation_event(
|
|
domain="generation_ai_batch",
|
|
event_type="BATCH_CHILDREN_CREATED" if child_ids else "BATCH_MAIN_CREATED",
|
|
event_status="success",
|
|
source="service",
|
|
user_id=current_user.id,
|
|
group_id=main_id,
|
|
detail={
|
|
"gen_type": gen_type,
|
|
"generation_count": generation_count,
|
|
"child_task_ids": child_ids,
|
|
"enqueue_task_ids": enqueue_ids,
|
|
"video_upscale_enabled_snapshot": bool(locals().get("upscale_enabled_snapshot", False)),
|
|
"provider_generation_resolution": locals().get("provider_generation_resolution"),
|
|
},
|
|
)
|
|
return GenerationTaskCreateResult(
|
|
top_level_task_id=main_id,
|
|
enqueue_task_ids=enqueue_ids,
|
|
child_task_ids=child_ids,
|
|
generation_count=generation_count,
|
|
gen_type=gen_type,
|
|
created=True,
|
|
)
|
|
|
|
|
|
async def enqueue_created_generation_tasks(
|
|
db: AsyncSession,
|
|
*,
|
|
task_ids: list[str],
|
|
) -> list[str]:
|
|
"""在业务事务提交后投递任务；返回投递失败的任务ID。
|
|
|
|
投递失败会在补偿事务中将对应任务置为失败并幂等退款，视频子任务
|
|
同时触发主任务状态汇总。调用方不应在初始事务提交前调用本函数。
|
|
"""
|
|
from app.services.generation.ai.task_group_service import aggregate_parent_for_child
|
|
from app.services.generation.log_service import log_task_event
|
|
from app.services.generation.refund_service import mark_chat_generation_task_failed_and_refund_once
|
|
from app.tasks.generation_create_tasks import chatapi_create_generation_task
|
|
|
|
normalized_ids = list(dict.fromkeys(str(item) for item in task_ids if item))
|
|
meta_result = await db.execute(
|
|
select(
|
|
ChatGenerationTask.id,
|
|
ChatGenerationTask.user_id,
|
|
ChatGenerationTask.parent_task_id,
|
|
ChatGenerationTask.generation_index,
|
|
).where(ChatGenerationTask.id.in_(normalized_ids))
|
|
) if normalized_ids else None
|
|
task_meta = {
|
|
str(row.id): {
|
|
"user_id": str(row.user_id),
|
|
"parent_task_id": str(row.parent_task_id) if row.parent_task_id else None,
|
|
"generation_index": row.generation_index,
|
|
}
|
|
for row in (meta_result.all() if meta_result is not None else [])
|
|
}
|
|
|
|
failed_ids: list[str] = []
|
|
for task_id in normalized_ids:
|
|
meta = task_meta.get(task_id, {})
|
|
log_operation_event(
|
|
domain="generation_ai_batch",
|
|
event_type="CHILD_ENQUEUE_START",
|
|
event_status="started",
|
|
source="api",
|
|
user_id=meta.get("user_id"),
|
|
group_id=meta.get("parent_task_id") or task_id,
|
|
task_id=task_id,
|
|
detail={"generation_index": meta.get("generation_index")},
|
|
)
|
|
try:
|
|
chatapi_create_generation_task.delay(task_id)
|
|
await log_task_event(
|
|
task_id=task_id,
|
|
event_type="CHILD_ENQUEUE_SUCCESS",
|
|
to_status="generating",
|
|
to_stage="queued",
|
|
detail={"task_id": task_id},
|
|
)
|
|
log_operation_event(
|
|
domain="generation_ai_batch",
|
|
event_type="CHILD_ENQUEUE_SUCCESS",
|
|
event_status="success",
|
|
source="api",
|
|
user_id=meta.get("user_id"),
|
|
group_id=meta.get("parent_task_id") or task_id,
|
|
task_id=task_id,
|
|
detail={"generation_index": meta.get("generation_index")},
|
|
)
|
|
except Exception as exc:
|
|
failed_ids.append(task_id)
|
|
await db.rollback()
|
|
failed_task = await mark_chat_generation_task_failed_and_refund_once(
|
|
db,
|
|
task_id=task_id,
|
|
error_message=f"任务队列投递失败: {exc}",
|
|
pipeline_stage="failed",
|
|
)
|
|
await aggregate_parent_for_child(db, failed_task)
|
|
await db.commit()
|
|
await log_task_event(
|
|
task_id=task_id,
|
|
event_type="CHILD_ENQUEUE_FAILED",
|
|
to_status="failed",
|
|
to_stage="failed",
|
|
message=str(exc),
|
|
detail={"task_id": task_id},
|
|
)
|
|
log_operation_event(
|
|
domain="generation_ai_batch",
|
|
event_type="CHILD_ENQUEUE_FAILED",
|
|
event_status="failed",
|
|
source="api",
|
|
user_id=getattr(failed_task, "user_id", None),
|
|
group_id=getattr(failed_task, "parent_task_id", None) or task_id,
|
|
task_id=task_id,
|
|
message=str(exc),
|
|
detail={"physical_files_deleted": False},
|
|
error=str(exc),
|
|
)
|
|
return failed_ids
|