From 0ee498d8073d4023712b968c667b8c1dcd85dd81 Mon Sep 17 00:00:00 2001 From: "Azmat@qq.com" Date: Wed, 26 Aug 2026 15:18:18 +0800 Subject: [PATCH] =?UTF-8?q?=E6=B5=8B=E8=AF=95=E6=9E=81=E9=80=9F=E6=88=90?= =?UTF-8?q?=E7=89=87?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- core/backend/.env | 4 +- core/backend/.env.example | 8 +- core/backend/README.md | 2 +- core/backend/airshelf/settings/base.py | 9 +- .../apps/ai/providers/openai_compatible.py | 24 +- core/backend/apps/ai/providers/yunqi.py | 3 +- core/backend/apps/ai/services.py | 130 ++- .../apps/ai/test_openai_compatible_image.py | 34 + .../apps/ai/test_project_triview_routing.py | 52 ++ core/backend/apps/ai/test_routing_policy.py | 4 +- .../apps/ai/test_storyboard_routing.py | 10 + core/backend/apps/ai/tests.py | 51 +- core/backend/apps/projects/serializers.py | 52 +- .../apps/projects/services/quick_create.py | 834 ++++++++++++++---- core/backend/apps/projects/tasks.py | 49 +- .../apps/projects/test_quick_create.py | 446 +++++++++- core/backend/apps/projects/tests.py | 17 +- core/backend/apps/projects/views.py | 64 +- core/frontend/src/App.tsx | 46 +- core/frontend/src/api.ts | 1 + core/frontend/src/pipeline-page.css | 23 +- core/frontend/src/projects-page.css | 4 +- core/frontend/src/quick-create-lock.ts | 46 + core/frontend/src/routes/dashboard.tsx | 5 +- core/frontend/src/routes/pipeline.tsx | 159 +++- core/frontend/src/routes/products.tsx | 2 + core/frontend/src/routes/projects.tsx | 2 + core/frontend/src/routes/quick-create.tsx | 238 +++-- core/frontend/src/routes/route-config.ts | 4 + core/frontend/src/types.ts | 7 + 30 files changed, 1963 insertions(+), 367 deletions(-) create mode 100644 core/backend/apps/ai/test_openai_compatible_image.py create mode 100644 core/frontend/src/quick-create-lock.ts diff --git a/core/backend/.env b/core/backend/.env index 4f2be41..01c58de 100644 --- a/core/backend/.env +++ b/core/backend/.env @@ -61,9 +61,9 @@ MODEL_ROUTING_TEXT_RETRY_DELAYS=1,3 # 文本：普通非流式请求单次超时秒数。 MODEL_ROUTING_TEXT_REQUEST_TIMEOUT=120 # 文本：脚本等流式请求单次超时秒数。 -MODEL_ROUTING_TEXT_STREAM_TIMEOUT=300 +MODEL_ROUTING_TEXT_STREAM_TIMEOUT=1740 # 文本：一次逻辑任务从首次请求起的总时限秒数。 -MODEL_ROUTING_TEXT_TOTAL_TIMEOUT=480 +MODEL_ROUTING_TEXT_TOTAL_TIMEOUT=1800 # 文本：429 Retry-After 允许等待的最长秒数。 MODEL_ROUTING_TEXT_RETRY_AFTER_CAP=15 diff --git a/core/backend/.env.example b/core/backend/.env.example index 8b74dba..680c2b7 100644 --- a/core/backend/.env.example +++ b/core/backend/.env.example @@ -52,10 +52,10 @@ MODEL_ROUTING_JITTER_RATIO=0.20 MODEL_ROUTING_TEXT_RETRY_DELAYS=1,3 # 普通文本单次调用超时；调小可能截断正常生成，调大会延长故障等待。 MODEL_ROUTING_TEXT_REQUEST_TIMEOUT=120 -# 流式文本单次调用超时；脚本 SSE 通常比普通文本耗时更长。 -MODEL_ROUTING_TEXT_STREAM_TIMEOUT=300 -# 单个文本逻辑任务总时限，默认 480 秒（8 分钟）。 -MODEL_ROUTING_TEXT_TOTAL_TIMEOUT=480 +# 流式文本单次调用超时。豆包 2.1 Pro 思考出脚本常超过 4 分钟，短了会掐成 stream aborted。 +MODEL_ROUTING_TEXT_STREAM_TIMEOUT=1740 +# 单个文本逻辑任务总时限，默认 900 秒（15 分钟）。 +MODEL_ROUTING_TEXT_TOTAL_TIMEOUT=1800 # 文本遇到 429 时接受 Retry-After 的最长等待时间。 MODEL_ROUTING_TEXT_RETRY_AFTER_CAP=15 diff --git a/core/backend/README.md b/core/backend/README.md index 4b36644..d99606f 100644 --- a/core/backend/README.md +++ b/core/backend/README.md @@ -21,7 +21,7 @@ Start workers in separate terminals: ```bash cd /Users/maidong/Desktop/zyc/qiyuan_gitea/AirShelf/core/backend source .venv/bin/activate -celery -A airshelf worker -l info -P threads -c 4 -Q celery,airshelf.quick # 必须带 airshelf.quick，否则极速成片会一直停在「等待开始」 +celery -A airshelf worker -l info -P threads -c 4 -Q celery,airshelf.quick # 建议带 airshelf.quick；没带时编排会回退到 celery，避免一直停在「等待开始」 ``` `ffmpeg` must be available on `PATH` for Stage5 export jobs. diff --git a/core/backend/airshelf/settings/base.py b/core/backend/airshelf/settings/base.py index 1b5fe9a..7f2169a 100644 --- a/core/backend/airshelf/settings/base.py +++ b/core/backend/airshelf/settings/base.py @@ -292,10 +292,11 @@ MODEL_ROUTING_POLICY = { "retry_delays": env_int_list("MODEL_ROUTING_TEXT_RETRY_DELAYS", (1, 3)), # 普通非流式文本请求的单次最长等待时间。 "request_timeout": env_int("MODEL_ROUTING_TEXT_REQUEST_TIMEOUT", 120), - # 流式文本请求的单次最长等待时间；脚本 SSE 正常生成可能明显慢于普通文本。 - "stream_timeout": env_int("MODEL_ROUTING_TEXT_STREAM_TIMEOUT", 300), - # 单个文本逻辑任务从首次调用 Provider 起允许的总执行时间，默认 8 分钟。 - "total_timeout": env_int("MODEL_ROUTING_TEXT_TOTAL_TIMEOUT", 480), + # 流式文本请求的单次最长等待时间。给 29 分钟，略短于任务软上限， + # 确保 30 分钟内可正常收尾，不会在长思考中被 HTTP 客户端提前掐断。 + "stream_timeout": env_int("MODEL_ROUTING_TEXT_STREAM_TIMEOUT", 1740), + # 单个文本逻辑任务从首次调用 Provider 起允许的总执行时间，默认 30 分钟。 + "total_timeout": env_int("MODEL_ROUTING_TEXT_TOTAL_TIMEOUT", 1800), # 429 响应 Retry-After 的最长接受时间；超过该值时不继续长时间等待，转入 Fallback 或最终失败。 "retry_after_cap": env_int("MODEL_ROUTING_TEXT_RETRY_AFTER_CAP", 15), }, diff --git a/core/backend/apps/ai/providers/openai_compatible.py b/core/backend/apps/ai/providers/openai_compatible.py index edb305e..3a485b7 100644 --- a/core/backend/apps/ai/providers/openai_compatible.py +++ b/core/backend/apps/ai/providers/openai_compatible.py @@ -5,6 +5,26 @@ import requests from .volcano import VolcanoArkProvider +# gpt-image / New API 网关只认这三档。业务层为了「真 16:9」会传 1536x864， +# 不在这里收成合法横图的话，立绘能成、三视图 image_edit 直接 400。 +_OPENAI_IMAGE_SIZES = {"1024x1024", "1024x1536", "1536x1024"} + + +def openai_image_size(size: str, default: str = "1024x1536") -> str: + """把任意宽高收成 gpt-image 接受的三档：方 / 竖 / 横。""" + raw = (size or "").strip() + if raw in _OPENAI_IMAGE_SIZES: + return raw + try: + width, height = (int(part) for part in raw.lower().replace("×", "x").split("x", 1)) + except (TypeError, ValueError): + return default + if width <= 0 or height <= 0: + return default + if width == height: + return "1024x1024" + return "1536x1024" if width > height else "1024x1536" + def _downscale_ref_for_edit(data: bytes, content_type: str, max_edge: int = 2048) -> tuple[bytes, str]: """图生图参考图过大时按需降采样,避免中转站(yunqi/gpt-image edits)拒收大图报 400 @@ -91,7 +111,7 @@ class OpenAICompatibleProvider(VolcanoArkProvider): """文生图(可选单图参考 base64)。多图参考请用 image_edit。返回体含 url 或 b64_json。""" if not self.api_key: raise ValueError("中转站 api_key 未配置") - body: dict[str, Any] = {"model": model, "prompt": prompt, "size": size, "n": 1} + body: dict[str, Any] = {"model": model, "prompt": prompt, "size": openai_image_size(size), "n": 1} if image: body["image"] = image # 实测中转站生图延迟可达 75s+,超时给到 300s @@ -135,7 +155,7 @@ class OpenAICompatibleProvider(VolcanoArkProvider): files.append(("image[]", (f"ref{idx + 1}.{ext}", img_bytes, content_type))) if not files: raise ValueError("image_edit 至少需要一张参考图") - data = {"model": model, "prompt": prompt, "size": size, "n": "1"} + data = {"model": model, "prompt": prompt, "size": openai_image_size(size), "n": "1"} response = requests.post( self._endpoint_url(endpoint), headers={"Authorization": f"Bearer {self.api_key}"}, # multipart 不要手设 Content-Type diff --git a/core/backend/apps/ai/providers/yunqi.py b/core/backend/apps/ai/providers/yunqi.py index 76cbe7b..d6c544b 100644 --- a/core/backend/apps/ai/providers/yunqi.py +++ b/core/backend/apps/ai/providers/yunqi.py @@ -3,6 +3,7 @@ from typing import Any import requests from django.conf import settings +from .openai_compatible import openai_image_size from .volcano import VolcanoArkProvider @@ -32,7 +33,7 @@ class YunqiProvider(VolcanoArkProvider): raise ValueError("YUNQI_API_KEY is not configured") if image: raise ValueError("gpt-image-2 via YunQi only supports text-to-image; reference image is not supported") - body: dict[str, Any] = {"model": model, "prompt": prompt, "size": size, "n": 1} + body: dict[str, Any] = {"model": model, "prompt": prompt, "size": openai_image_size(size), "n": 1} # 实测该网关生图平均延迟 75s+,超时须显著高于火山的 180s response = requests.post( f"{self.base_url.rstrip('/')}/{endpoint.lstrip('/')}", diff --git a/core/backend/apps/ai/services.py b/core/backend/apps/ai/services.py index 9590ea3..eda804c 100644 --- a/core/backend/apps/ai/services.py +++ b/core/backend/apps/ai/services.py @@ -1411,6 +1411,7 @@ def _ratio_to_image_size(ratio: str) -> str: "9:16": "1024x1536", # 近似竖图(网关无精确 9:16) "4:3": "1536x1024", "16:9": "1536x864", # 真 16:9(原来误用 1536x1024 = 3:2) + "21:9": "1536x1024", # 超宽近似横图(网关无精确 21:9) } normalized = (ratio or "").strip() if normalized in known: @@ -1429,6 +1430,56 @@ def _ratio_to_image_size(ratio: str) -> str: return "1024x1024" +def project_output_spec(project) -> dict: + """专业创作 / 极速成片共用的成片规格：画幅、分辨率、视频模型。缺省 9:16 + 720p。""" + wizard = dict((project.metadata or {}).get("wizard") or {}) + return { + "aspect_ratio": str(wizard.get("aspect_ratio") or "9:16").strip() or "9:16", + "resolution": str(wizard.get("resolution") or "720p").strip().lower() or "720p", + "video_model_config_id": str(wizard.get("video_model_config_id") or "").strip() or None, + } + + +def _storyboard_canvas_phrase(ratio: str) -> str: + r = (ratio or "9:16").strip() or "9:16" + if r in {"9:16", "3:4"}: + return f"电商竖屏 {r}" + if r == "1:1": + return f"电商方形 {r}" + return f"电商横屏 {r}" + + +def _apply_storyboard_output_ratio(text: str, project) -> str: + phrase = _storyboard_canvas_phrase(project_output_spec(project)["aspect_ratio"]) + return (text or "").replace("电商竖屏 9:16", phrase) + + +def _sync_timeline_output_spec(project, *, aspect_ratio: str, resolution: str) -> None: + from apps.ai.video_pricing import get_resolution + + try: + width, height = get_resolution(aspect_ratio, resolution) + except Exception: # noqa: BLE001 — 规格不合法时不挡提交，时间线保持原值 + return + pixels = f"{width}x{height}" + timeline, created = Timeline.objects.get_or_create( + project=project, + defaults={ + "name": f"{project.name} Timeline", + "duration_seconds": 60, + "aspect_ratio": aspect_ratio, + "resolution": pixels, + }, + ) + if created: + return + if timeline.aspect_ratio == aspect_ratio and timeline.resolution == pixels: + return + timeline.aspect_ratio = aspect_ratio + timeline.resolution = pixels + timeline.save(update_fields=["aspect_ratio", "resolution", "updated_at"]) + + def _ratio_to_volcano_size(ratio: str) -> str: """前端比例 → 火山 Seedream 尺寸(~2K 面积,各边夹在 [1024,4096] 且取 16 的倍数)。 预设比例直接给好尺寸;自定义 W:H 按 2K 面积换算;解析不到回落 '2K'。""" @@ -2060,9 +2111,9 @@ def generate_base_asset(*, project, user, kind: str, prompt: str, label: str = " "model": model_config.name, "endpoint": model_config.endpoint, "prompt": gen_prompt, "kind": kind, "label": label or "", "group_id": str(group_id) if group_id else "", "use_edit": use_edit, "reference_image": ref_url, "model_routing_v1": True, - # 角色立绘不再自动接力三视图；三视图只通过角色详情里的显式按钮生成。 - # 保留字段为 False，兼容旧前端/旧任务读取，但不允许再触发自动链路。 - "auto_triview": False, + # 角色从这里生成时，立绘落库后由 worker 自动接力生成绑定它的三视图。 + # 非角色一律忽略该参数，避免商品/场景误入人物三视图链路。 + "auto_triview": bool(auto_triview and kind == BaseAssetGroup.Kind.PERSON), } task = create_ai_task( project=project, @@ -2189,6 +2240,21 @@ def run_base_asset_task(*, task_id: str) -> None: from apps.assets.review import submit_asset_for_review transaction.on_commit(lambda a=asset: submit_asset_for_review(a)) + # 三视图是专业创作可选的增强资产。只有调用方明确要求时才接力， + # 极速成片只需人物立绘即可进入故事板，不能为三视图额外等待或失败。 + if kind == BaseAssetGroup.Kind.PERSON and payload.get("auto_triview"): + def _kickoff_person_triview(portrait=asset): + try: + generate_person_triview( + project=project, user=user, portrait_asset=portrait + ) + except Exception: + logger.exception( + "auto person triview kickoff failed for portrait %s", + getattr(portrait, "id", ""), + ) + + transaction.on_commit(_kickoff_person_triview) except Exception as exc: # noqa: BLE001 — 失败要退费并把错误记进 AITask 供前端轮询读取;不向上抛(避免 celery 重试二次扣费) task.status = AITask.Status.FAILED task.error_message = str(exc) @@ -2201,6 +2267,16 @@ def run_base_asset_task(*, task_id: str) -> None: ) +def _triview_reference_url(*, asset_id: str, fallback: str = "") -> str: + """三视图 worker 现取立绘可访问 URL。提交时写进 payload 的 TOS 签名链接会过期， + 立绘落库瞬间也可能还没签出 URL；执行时再取一次，避免「立绘成了、三视图没参考图」。""" + if not asset_id: + return str(fallback or "") + portrait = Asset.objects.filter(id=asset_id, is_deleted=False).first() + live = _asset_preview_url(portrait) + return live or str(fallback or "") + + def generate_person_triview(*, project, user, portrait_asset) -> AITask: """流程步骤4 · 据「某一版立绘资产」生成它配套的三视图(**异步**:image_edit 慢,交给 worker)。 Web 请求只建 RESERVED 任务 + 预留额度后秒回;worker 内跑 image_edit 并把三视图归组(run_triview_task)。 @@ -2214,7 +2290,7 @@ def generate_person_triview(*, project, user, portrait_asset) -> AITask: model_config = get_default_model(ModelConfig.Capability.IMAGE) if model_config is None: raise ValueError("no active image model configured") - ref_url = _asset_preview_url(portrait_asset) + ref_url = _triview_reference_url(asset_id=asset_key) # 人物三视图提示词:正文可在 admin「提示词」页改(无占位符) tri_prompt = render_prompt("person_triview", THREE_VIEW_PROMPT) portrait_label = "" @@ -2222,7 +2298,11 @@ def generate_person_triview(*, project, user, portrait_asset) -> AITask: meta = group.metadata or {} if meta.get("triview_of"): continue - candidates = [str(value) for value in (group.candidate_assets or [])] + # candidate_assets 是 Django 的 ManyRelatedManager，不能直接遍历； + # 资产生成完成后这里会立即触发人物三视图，直接遍历会抛 + # “ManyRelatedManager object is not iterable”，进而让极速成片 + # 在所有基础资产已成功时被错误终止。 + candidates = [str(value) for value in group.candidate_assets.all()] if str(group.adopted_asset_id or "") == asset_key or asset_key in candidates: portrait_label = str(meta.get("label") or "") break @@ -2343,7 +2423,10 @@ def run_model_triview_task(*, task_id: str) -> None: payload = task.request_payload or {} model_id = str(payload.get("model_id") or "") portrait_asset_id = str(payload.get("portrait_asset_id") or "") - ref_url = str(payload.get("reference_image") or "") + ref_url = _triview_reference_url( + asset_id=portrait_asset_id, + fallback=str(payload.get("reference_image") or ""), + ) prompt = str(payload.get("prompt") or "") use_model_routing = bool(payload.get("model_routing_v1")) provider = None if use_model_routing else get_image_provider(task.model_config) @@ -2465,13 +2548,19 @@ def run_triview_task(*, task_id: str) -> None: user = task.created_by payload = task.request_payload or {} asset_key = str(payload.get("triview_of") or "") - ref_url = str(payload.get("reference_image") or "") + ref_url = _triview_reference_url( + asset_id=asset_key, + fallback=str(payload.get("reference_image") or ""), + ) prompt = str(payload.get("prompt") or THREE_VIEW_PROMPT) model_config = task.model_config use_model_routing = bool(payload.get("model_routing_v1")) provider = None if use_model_routing else get_image_provider(model_config) reservation = task.credit_reservation try: + if not asset_key or not ref_url: + raise ValueError("立绘参考图不可用，无法生成三视图") + # 构图意图仍是 16:9；gpt-image 不认 1536x864，OpenAICompatibleProvider 会收成 1536x1024。 tri_size = prompt_ratio_size("person_triview", "1536x864") def _make(): @@ -2603,7 +2692,8 @@ def build_storyboard_frame_prompt(project, segment, extra_prompt: str = "") -> s 脚本=(_segment_script_text(segment) or f"第 {segment.sort_order + 1} 镜"), 补充=(("\n" + extra_prompt.strip()) if extra_prompt else ""), ) - return "\n".join(line for line in rendered.split("\n") if line.strip()) + cleaned = "\n".join(line for line in rendered.split("\n") if line.strip()) + return _apply_storyboard_output_ratio(cleaned, project) def build_video_segment_prompt(project, video_segment, scene, refs, user_prompt: str = "") -> str: @@ -2813,7 +2903,8 @@ def build_storyboard_frame_prompt_refs(project, segment, refs: list[dict], extra 脚本=(_segment_script_text(segment) or f"第 {segment.sort_order + 1} 镜"), 补充=(("\n" + extra_prompt.strip()) if extra_prompt else ""), ) - return "\n".join(line for line in rendered.split("\n") if line.strip()) + cleaned = "\n".join(line for line in rendered.split("\n") if line.strip()) + return _apply_storyboard_output_ratio(cleaned, project) def _is_transient_error(exc: Exception) -> bool: @@ -2968,6 +3059,9 @@ def _storyboard_shot_worker(task_id, shot_id, user_id) -> None: model_config = task.model_config reservation = task.credit_reservation extra_prompt = (project.metadata or {}).get("storyboard_prompt", "") or "" + spec = project_output_spec(project) + frame_ratio = spec["aspect_ratio"] + frame_size = _ratio_to_image_size(frame_ratio) task.status = AITask.Status.SUBMITTED task.save(update_fields=["status", "updated_at"]) try: @@ -2990,9 +3084,9 @@ def _storyboard_shot_worker(task_id, shot_id, user_id) -> None: primary_model=model_config, prompt=frame_prompt, reference_images=ref_urls, - aspect_ratio="9:16", - edit_size="1024x1536", - direct_size="1024x1536", + aspect_ratio=frame_ratio, + edit_size=frame_size, + direct_size=frame_size, request_summary={ "storyboard_shot": str(shot.id), "storyboard_sort_order": shot.sort_order, @@ -3006,7 +3100,7 @@ def _storyboard_shot_worker(task_id, shot_id, user_id) -> None: model=model_config.name, prompt=frame_prompt, images=ref_urls, - size="1024x1536", + size=frame_size, ) ) else: @@ -3270,9 +3364,14 @@ def submit_video_segment( user, prompt: str, model_config_id=None, - aspect_ratio: str = "9:16", - resolution: str = "720p", + aspect_ratio: str | None = None, + resolution: str | None = None, ) -> VideoSegmentVersion | None: + spec = project_output_spec(video_segment.project) + aspect_ratio = str(aspect_ratio or spec["aspect_ratio"] or "9:16") + resolution = str(resolution or spec["resolution"] or "720p").lower() + if not model_config_id: + model_config_id = spec["video_model_config_id"] model_config = None if model_config_id: model_config = ( @@ -3392,6 +3491,7 @@ def submit_video_segment( ) video_segment.status = VideoSegment.Status.RUNNING video_segment.save(update_fields=["status", "updated_at"]) + _sync_timeline_output_spec(project, aspect_ratio=aspect_ratio, resolution=resolution) return None except Exception as exc: public_error = classify_generation_error( diff --git a/core/backend/apps/ai/test_openai_compatible_image.py b/core/backend/apps/ai/test_openai_compatible_image.py new file mode 100644 index 0000000..88e93c3 --- /dev/null +++ b/core/backend/apps/ai/test_openai_compatible_image.py @@ -0,0 +1,34 @@ +from io import BytesIO +from unittest.mock import Mock, patch + +from django.test import SimpleTestCase + +from apps.ai.providers.openai_compatible import OpenAICompatibleProvider, openai_image_size + + +class OpenAICompatibleImageSizeTests(SimpleTestCase): + def test_coerces_true_16_9_to_supported_landscape(self): + self.assertEqual(openai_image_size("1536x864"), "1536x1024") + self.assertEqual(openai_image_size("1024x1536"), "1024x1536") + self.assertEqual(openai_image_size("1536x1024"), "1536x1024") + self.assertEqual(openai_image_size("1024x1024"), "1024x1024") + self.assertEqual(openai_image_size("512x768"), "1024x1536") + + @patch("apps.ai.providers.openai_compatible.requests.post") + @patch.object(OpenAICompatibleProvider, "media_to_bytes") + def test_image_edit_sends_gateway_supported_size(self, media_to_bytes, post): + media_to_bytes.return_value = (BytesIO(b"img"), "image/png") + response = Mock() + response.ok = True + response.json.return_value = {"data": [{"b64_json": "x"}]} + post.return_value = response + provider = OpenAICompatibleProvider(base_url="https://img.example/v1", api_key="secret") + + provider.image_edit( + model="gpt-image-2", + prompt="三视图", + images=["http://example.test/portrait.png"], + size="1536x864", + ) + + self.assertEqual(post.call_args.kwargs["data"]["size"], "1536x1024") diff --git a/core/backend/apps/ai/test_project_triview_routing.py b/core/backend/apps/ai/test_project_triview_routing.py index dc0d6f0..0c3acb1 100644 --- a/core/backend/apps/ai/test_project_triview_routing.py +++ b/core/backend/apps/ai/test_project_triview_routing.py @@ -150,6 +150,24 @@ class ProjectPersonTriviewRoutingTests(TestCase): self.assertEqual(self.ledger_count(task, CreditLedger.Type.RESERVE), 1) self.assertEqual(self.ledger_count(task, CreditLedger.Type.CHARGE), 1) + @patch("apps.ai.services.create_ai_task") + def test_submit_reads_candidate_assets_manager_to_keep_portrait_label(self, create_task): + """基础资产落库后自动接力三视图时，关联集合必须可正常读取。""" + self.model(self.provider("project-tri-label", 20), "project-tri-label") + group = BaseAssetGroup.objects.create( + project=self.project, + kind=BaseAssetGroup.Kind.PERSON, + adopted_asset=self.portrait, + metadata={"label": "专业测评师"}, + ) + group.candidate_assets.add(self.portrait) + create_task.return_value = Mock(id="triview-task") + + task = self.submit() + + self.assertEqual(create_task.call_args.kwargs["request_payload"]["label"], "专业测评师") + self.assertEqual(task, create_task.return_value) + def test_retry_then_dynamic_candidate_success_charges_once(self): primary = self.model( self.provider("project-tri-fallback-primary", 100), @@ -232,3 +250,37 @@ class ProjectPersonTriviewRoutingTests(TestCase): self.assertEqual(call.kwargs["image"], ["http://example.test/portrait.png"]) self.assertEqual(call.kwargs["size"], "1536x864") self.assertEqual(task.model_attempts.get().public_model_name, primary.display_name) + + def test_worker_refreshes_stale_reference_url_from_portrait(self): + """立绘成功后立刻出三视图：worker 必须用当前立绘 URL，不能沿用已过期的 payload 链接。""" + self.model(self.provider("project-tri-refresh", 20), "project-tri-refresh") + task = self.submit() + payload = dict(task.request_payload) + payload["reference_image"] = "http://expired.test/old-portrait.png" + task.request_payload = payload + task.save(update_fields=["request_payload", "updated_at"]) + + run_triview_task(task_id=str(task.id)) + + task.refresh_from_db() + self.assertEqual(task.status, AITask.Status.SUCCEEDED) + primary = next(iter(self.provider_mocks.values())) + self.assertEqual( + primary.image_edit.call_args.kwargs["images"], + ["http://example.test/portrait.png"], + ) + + def test_worker_fails_clearly_when_portrait_reference_is_missing(self): + self.model(self.provider("project-tri-missing-ref", 20), "project-tri-missing-ref") + self.portrait.files.all().delete() + task = self.submit() + payload = dict(task.request_payload) + payload["reference_image"] = "" + task.request_payload = payload + task.save(update_fields=["request_payload", "updated_at"]) + + run_triview_task(task_id=str(task.id)) + + task.refresh_from_db() + self.assertEqual(task.status, AITask.Status.FAILED) + self.assertIn("立绘参考图不可用", task.error_message) diff --git a/core/backend/apps/ai/test_routing_policy.py b/core/backend/apps/ai/test_routing_policy.py index 0dc71a3..a1613a8 100644 --- a/core/backend/apps/ai/test_routing_policy.py +++ b/core/backend/apps/ai/test_routing_policy.py @@ -24,8 +24,8 @@ class ModelRoutingPolicyTests(SimpleTestCase): self.assertEqual(policy.jitter_ratio, 0.20) self.assertEqual(policy.text.retry_delays, (1.0, 3.0)) self.assertEqual(policy.text.request_timeout, 120.0) - self.assertEqual(policy.text.stream_timeout, 300.0) - self.assertEqual(policy.text.total_timeout, 480.0) + self.assertEqual(policy.text.stream_timeout, 1740.0) + self.assertEqual(policy.text.total_timeout, 1800.0) self.assertEqual(policy.image.retry_delays, (3.0,)) self.assertEqual(policy.image.request_timeout, 300.0) self.assertEqual(policy.image.total_timeout, 900.0) diff --git a/core/backend/apps/ai/test_storyboard_routing.py b/core/backend/apps/ai/test_storyboard_routing.py index 10d0782..81e7c08 100644 --- a/core/backend/apps/ai/test_storyboard_routing.py +++ b/core/backend/apps/ai/test_storyboard_routing.py @@ -185,6 +185,16 @@ class StoryboardRoutingTests(TestCase): self.assertEqual(self.ledger_count(task, CreditLedger.Type.RESERVE), 1) self.assertEqual(self.ledger_count(task, CreditLedger.Type.CHARGE), 1) + def test_wizard_aspect_ratio_drives_storyboard_size(self): + self.project.metadata = {"wizard": {"aspect_ratio": "16:9", "resolution": "480p"}} + self.project.save(update_fields=["metadata"]) + primary = self.model(self.provider("storyboard-wide", 20), "storyboard-wide") + task = self.enqueue() + attempt = task.model_attempts.get() + self.assertEqual(attempt.request_summary["aspect_ratio"], "16:9") + call = self.provider_mocks[primary.id].image_edit.call_args + self.assertEqual(call.kwargs["size"], "1536x864") + def test_retry_then_dynamic_candidate_success_charges_once(self): primary = self.model( self.provider("storyboard-fallback-primary", 100), diff --git a/core/backend/apps/ai/tests.py b/core/backend/apps/ai/tests.py index 3cab6cd..6852301 100644 --- a/core/backend/apps/ai/tests.py +++ b/core/backend/apps/ai/tests.py @@ -1,5 +1,6 @@ import json from pathlib import Path +from types import SimpleNamespace from django.conf import settings from django.test import SimpleTestCase @@ -1544,6 +1545,7 @@ class StandaloneCategoryTests(TestCase): ) +@override_settings(CACHES={"default": {"BACKEND": "django.core.cache.backends.locmem.LocMemCache"}}) class TriviewModelDecouplingTests(TestCase): """项目角色三视图只归项目，不自动创建或更新团队模特。""" @@ -1606,9 +1608,9 @@ class TriviewModelDecouplingTests(TestCase): model.refresh_from_db() self.assertEqual(model.triview_asset_id, existing_tri.id) - @patch("apps.ai.services.generate_person_triview") - def test_person_base_asset_ignores_legacy_auto_triview_flag(self, kickoff_triview): - """即使旧前端/旧任务传 auto_triview=True，角色立绘完成后也不能自动接力三视图。""" + @patch("apps.ai.tasks.generate_base_asset_task.delay") + def test_person_base_asset_marks_auto_triview_for_worker(self, _enqueue_base_asset): + """角色立绘任务需标记自动接力三视图，实际接力只在立绘成功落库后发生。""" from apps.ai.services import generate_base_asset from apps.projects.models import BaseAssetGroup @@ -1633,8 +1635,47 @@ class TriviewModelDecouplingTests(TestCase): ) task.refresh_from_db() - self.assertFalse(task.request_payload.get("auto_triview")) - kickoff_triview.assert_not_called() + self.assertTrue(task.request_payload.get("auto_triview")) + + @patch("apps.assets.review.submit_asset_for_review") + @patch("apps.ai.services.generate_person_triview") + @patch("apps.ai.services._store_generated_media") + @patch("apps.ai.services.execute_routed_image_request") + @patch("apps.ai.tasks.generate_base_asset_task.delay") + def test_completed_auto_person_generation_queues_triview( + self, _enqueue_base_asset, routed, store_media, kickoff_triview, _review + ): + """角色立绘成功落库后，worker 必须自动创建绑定该立绘的三视图任务。""" + from apps.ai.services import generate_base_asset, run_base_asset_task + from apps.projects.models import BaseAssetGroup + + ModelConfig.objects.create( + provider=ModelProvider.objects.create(name="auto-tri-success-provider", display_name="Auto Tri Success Provider"), + name="auto-tri-success-img", + display_name="Auto Tri Success Img", + capability=ModelConfig.Capability.IMAGE, + endpoint="images/generations", + unit_price="1.0000", + is_default=True, + ) + routed.return_value = SimpleNamespace(value=({"data": [{"url": "http://x/person.png"}]}, "http://x/person.png")) + store_media.return_value = self.portrait + task = generate_base_asset( + project=self.project, + user=self.user, + kind=BaseAssetGroup.Kind.PERSON, + prompt="项目角色立绘", + label="女主", + # 专业模式显式要求时才接力三视图；极速成片传 False 会跳过。 + auto_triview=True, + ) + + with self.captureOnCommitCallbacks(execute=True): + run_base_asset_task(task_id=str(task.id)) + + kickoff_triview.assert_called_once_with( + project=self.project, user=self.user, portrait_asset=self.portrait + ) class ModelLibraryTriviewTaskTests(TestCase): diff --git a/core/backend/apps/projects/serializers.py b/core/backend/apps/projects/serializers.py index e6315fd..253f85d 100644 --- a/core/backend/apps/projects/serializers.py +++ b/core/backend/apps/projects/serializers.py @@ -1,5 +1,6 @@ import uuid +from django.core.exceptions import ObjectDoesNotExist from rest_framework import serializers from apps.assets.serializers import AssetFileSerializer @@ -341,6 +342,7 @@ class QuickCreateJobSerializer(serializers.ModelSerializer): phase_index = serializers.SerializerMethodField() settings = serializers.SerializerMethodField() result = serializers.SerializerMethodField() + error_message = serializers.SerializerMethodField() class Meta: model = QuickCreateJob @@ -365,6 +367,15 @@ class QuickCreateJobSerializer(serializers.ModelSerializer): ] read_only_fields = fields + def get_error_message(self, obj) -> str: + from apps.projects.services.quick_create import _public_error + + stored = obj.error_message or "" + if obj.status != QuickCreateJob.Status.FAILED: + return stored + hidden = str((obj.metadata or {}).get("internal_error") or "") + return _public_error(" ".join(part for part in (stored, hidden) if part)) or stored + def get_product_images(self, obj) -> list[dict]: product = getattr(obj.project, "product", None) if product is None: @@ -403,7 +414,8 @@ class QuickCreateJobSerializer(serializers.ModelSerializer): def get_result(self, obj) -> dict | None: project = obj.project settings = self.get_settings(obj) - video_url = _final_video_url(project) + final_video_url = _final_video_url(project) + video_url = final_video_url segments = list(project.video_segments.all()) if not video_url: for segment in sorted(segments, key=lambda item: item.sort_order): @@ -446,6 +458,7 @@ class QuickCreateJobSerializer(serializers.ModelSerializer): ) return { "video_url": video_url, + "final_video_url": final_video_url, "poster_url": poster_url, "duration_seconds": duration or 15, "aspect_ratio": settings["aspect_ratio"], @@ -479,6 +492,23 @@ class ScriptVersionSerializer(serializers.ModelSerializer): read_only_fields = fields +def _quick_create_job(obj: Project): + try: + return obj.quick_create_job + except ObjectDoesNotExist: + return None + + +def _quick_create_status(obj: Project) -> str: + job = _quick_create_job(obj) + return job.status if job else "" + + +def _quick_create_job_id(obj: Project) -> str: + job = _quick_create_job(obj) + return str(job.id) if job else "" + + class ProjectListSerializer(serializers.ModelSerializer): """列表/仪表盘/侧栏用的轻量项目序列化:不嵌套 阶段/片段/故事板/时间线(那些只详情页要)。 脚本数/镜数走 annotate 计数(见 ProjectViewSet.get_queryset),避免逐项目拉全套关联(原列表 2-3s)。""" @@ -490,13 +520,15 @@ class ProjectListSerializer(serializers.ModelSerializer): # 合成成片地址:项目列表的播放按钮据此直接播成片(没合成过为空 → 退回进流水线) final_video_url = serializers.SerializerMethodField() quick_create = serializers.SerializerMethodField() + quick_create_status = serializers.SerializerMethodField() + quick_create_job_id = serializers.SerializerMethodField() class Meta: model = Project fields = [ "id", "name", "product", "product_title", "cover_preview_url", "status", "current_stage", "script_version_count", "video_segment_count", - "final_video_url", "quick_create", + "final_video_url", "quick_create", "quick_create_status", "quick_create_job_id", "is_deleted", "purged_at", "created_at", "updated_at", ] @@ -509,6 +541,12 @@ class ProjectListSerializer(serializers.ModelSerializer): def get_quick_create(self, obj) -> bool: return bool((obj.metadata or {}).get("quick_create")) + def get_quick_create_status(self, obj) -> str: + return _quick_create_status(obj) + + def get_quick_create_job_id(self, obj) -> str: + return _quick_create_job_id(obj) + class ProjectSerializer(serializers.ModelSerializer): stages = ProjectStageSerializer(many=True, read_only=True) @@ -520,6 +558,8 @@ class ProjectSerializer(serializers.ModelSerializer): timeline = TimelineSerializer(read_only=True) # 合成成片地址(最新一次成功拼接):视频阶段的「播放成片 / 下载成片」直接用它 final_video_url = serializers.SerializerMethodField() + quick_create_status = serializers.SerializerMethodField() + quick_create_job_id = serializers.SerializerMethodField() class Meta: model = Project @@ -542,6 +582,8 @@ class ProjectSerializer(serializers.ModelSerializer): "video_segments", "timeline", "final_video_url", + "quick_create_status", + "quick_create_job_id", "created_at", "updated_at", ] @@ -550,6 +592,12 @@ class ProjectSerializer(serializers.ModelSerializer): def get_final_video_url(self, obj) -> str: return _final_video_url(obj) + def get_quick_create_status(self, obj) -> str: + return _quick_create_status(obj) + + def get_quick_create_job_id(self, obj) -> str: + return _quick_create_job_id(obj) + class ScriptTemplateSerializer(serializers.ModelSerializer): """套路模板 · 列表与详情共用。写入只开放 name(其余字段由存模板端点从脚本抽)。""" diff --git a/core/backend/apps/projects/services/quick_create.py b/core/backend/apps/projects/services/quick_create.py index c8b4f29..3b974a8 100644 --- a/core/backend/apps/projects/services/quick_create.py +++ b/core/backend/apps/projects/services/quick_create.py @@ -52,15 +52,21 @@ logger = logging.getLogger(__name__) POLL_DELAY_SECONDS = 10 QUICK_SCRIPT_MODEL_NAME = "doubao-seed-2-1-pro-260628" SCRIPT_POLL_SECONDS = 5 -SCRIPT_TIMEOUT = timedelta(minutes=4) +SCRIPT_TIMEOUT = timedelta(minutes=30) SCRIPT_STOLEN_AFTER = timedelta(seconds=25) +SCRIPT_CONSUME_STALE = timedelta(minutes=30) STALE_AFTER = timedelta(seconds=12) +ASSET_FAIL_RETRY_LIMIT = 2 +TRIVIEW_FAIL_RETRY_LIMIT = 2 +DEFAULT_WORKER_QUEUE = "celery" +QUICK_CREATE_QUEUE = "airshelf.quick" _ACTIVE_TASK_STATUSES = { AITask.Status.CREATED, AITask.Status.RESERVED, AITask.Status.SUBMITTED, AITask.Status.POLLING, AITask.Status.POSTPROCESSING, + AITask.Status.COMPENSATING, } _FINISHED_STATUSES = { QuickCreateJob.Status.SUCCEEDED, @@ -130,21 +136,69 @@ def _save_job(job: QuickCreateJob, **changes) -> None: TRANSIENT_RETRY_LIMIT = 8 +REVIEW_FAIL_MESSAGE = "图片或生成画面未通过审核。请更换商品图，或进入专业模式调整素材后再生成" +_REVIEW_ERROR_TOKENS = ( + "审核", + "review", + "moderation", + "safety_violation", + "safety system", + "content_policy", + "content_rejected", + "sensitivecontent", + "sensitive content", + "inputimagesensitive", + "outputimagesensitive", + "policyviolation", + "敏感内容", +) + + +def _looks_like_review_error(raw: str) -> bool: + text = raw or "" + lower = text.lower() + return any(token in text or token in lower for token in _REVIEW_ERROR_TOKENS) + + +def _public_error(raw: str) -> str: + text = (raw or "").strip() + lower = text.lower() + if QUICK_SCRIPT_MODEL_NAME in lower: + return "极速成片需要的豆包 Seed 2.1 Pro 模型未启用，请联系管理员配置" + if "insufficient credit" in lower or "额度不足" in text: + return "可用积分不足，极速成片已暂停" + if "no active" in lower or "not configured" in lower or "没有可用" in text: + return "当前缺少可用的生成模型，请联系管理员配置" + if _looks_like_review_error(text): + return REVIEW_FAIL_MESSAGE + return "极速成片暂未完成，请稍后重试或进入专业模式查看" + def _safe_error(exc: Exception) -> str: raw = str(exc or "").strip() - lower = raw.lower() - if QUICK_SCRIPT_MODEL_NAME in lower: - return "极速成片需要的豆包 Seed 2.1 Pro 模型未启用，请联系管理员配置" - if "insufficient credit" in lower or "额度不足" in raw: - return "可用积分不足，极速成片已暂停" - if "no active" in lower or "not configured" in lower or "没有可用" in raw: - return "当前缺少可用的生成模型，请联系管理员配置" - if "review" in lower or "审核" in raw: - return "生成素材未通过审核，请进入专业模式调整后重试" - if _is_retryable_exc(exc): + if _is_retryable_exc(exc) and not _looks_like_review_error(raw): return "脚本已保留，后续步骤遇到网络波动。点重试会从上次进度继续" - return "极速成片暂未完成，请稍后重试或进入专业模式查看" + return _public_error(raw) + + +def _is_hard_failure(job: QuickCreateJob) -> bool: + text = f"{job.error_message or ''} {(job.metadata or {}).get('internal_error') or ''}" + lower = text.lower() + return any( + token in text or token in lower + for token in ( + "积分不足", + "insufficient credit", + "额度不足", + "未通过审核", + "缺少可用的生成模型", + "未启用", + "no active", + "not configured", + "moderation", + "sensitivecontent", + ) + ) def _is_retryable_exc(exc: Exception) -> bool: @@ -166,12 +220,21 @@ def _is_retryable_exc(exc: Exception) -> bool: "client disconnected", "lost connection", "operationalerror", + "softtimelimit", + "timelimitexceeded", ) ) def _transient_internal(job: QuickCreateJob) -> bool: - return _is_retryable_exc(Exception(str((job.metadata or {}).get("internal_error") or ""))) + if _is_retryable_exc(Exception(str((job.metadata or {}).get("internal_error") or ""))): + return True + if job.phase != QuickCreateJob.Phase.SCRIPT: + return False + latest = _latest_script_task(job.project) + if latest is None or latest.status != AITask.Status.FAILED: + return False + return _is_retryable_exc(Exception(latest.error_message or "")) def _videos_have_started(job: QuickCreateJob) -> bool: @@ -198,6 +261,8 @@ def restore_false_failed_quick_creates(team) -> None: team=team, status=QuickCreateJob.Status.FAILED, phase=QuickCreateJob.Phase.PRODUCTION, + project__is_deleted=False, + project__purged_at__isnull=True, )[:20] ) for job in jobs: @@ -352,25 +417,27 @@ def _consume_script_agent(job: QuickCreateJob) -> None: def consume_quick_script(job_id: str) -> None: """Celery 专用：跑完专业创作同一套脚本 agent，再把编排推去资产阶段。""" - from apps.projects.tasks import advance_quick_create_task - job = QuickCreateJob.objects.select_related("project__product", "created_by", "team").get(id=job_id) if _is_finished(job): return - adopted = job.project.script_versions.filter(is_adopted=True).first() - if adopted is not None and adopted.segments.exists(): - advance_quick_create_task.apply_async(args=[str(job.id)], queue="airshelf.quick") + if _adopted_script(job.project) is not None: + _enqueue_advance(job) + return + if not _claim_script_consume(job): return try: - _consume_script_agent(job) - except Exception as exc: # noqa: BLE001 — 脚本失败必须落终态 - logger.exception("quick create script failed for job %s", job.id) - fail_quick_create(job, _safe_error(exc), internal_error=str(exc)) - return - job.refresh_from_db(fields=["status"]) - if _is_finished(job): - return - advance_quick_create_task.apply_async(args=[str(job.id)], queue="airshelf.quick") + try: + _consume_script_agent(job) + except Exception as exc: # noqa: BLE001 — 脚本失败必须落终态 + logger.exception("quick create script failed for job %s", job.id) + fail_quick_create(job, _safe_error(exc), internal_error=str(exc)) + return + job.refresh_from_db(fields=["status"]) + if _is_finished(job): + return + _enqueue_advance(job) + finally: + _release_script_consume(job_id) def _adopted_script(project: Project): @@ -419,27 +486,27 @@ def _advance_script(job: QuickCreateJob) -> int | None: _save_job(job, phase=QuickCreateJob.Phase.ASSETS, progress=48, message="脚本已生成，正在匹配模特与场景") return 1 - metadata = dict(job.metadata or {}) - if not metadata.get("script_started"): - from apps.projects.tasks import run_quick_script_task + with transaction.atomic(): + job = QuickCreateJob.objects.select_for_update().select_related("project").get(id=job.id) + if _is_finished(job): + return None + if _adopted_script(job.project) is not None: + _save_job(job, phase=QuickCreateJob.Phase.ASSETS, progress=48, message="脚本已生成，正在匹配模特与场景") + return 1 + metadata = dict(job.metadata or {}) + should_start = not metadata.get("script_started") + if should_start: + metadata["script_started"] = True + metadata["script_started_at"] = timezone.now().isoformat() + _save_job( + job, + metadata=metadata, + progress=28, + message="正在根据商品名称与图片生成带货脚本", + ) - from apps.common.celery_health import worker_consumes_queue - - metadata["script_started"] = True - metadata["script_started_at"] = timezone.now().isoformat() - listens = worker_consumes_queue("airshelf.quick") - if listens is False: - metadata["script_local"] = True - _save_job( - job, - metadata=metadata, - progress=28, - message="正在根据商品名称与图片生成带货脚本", - ) - if listens is False: - _run_quick_script_in_thread(str(job.id)) - else: - run_quick_script_task.apply_async(args=[str(job.id)], queue="airshelf.quick") + if should_start: + _enqueue_script(str(job.id)) return SCRIPT_POLL_SECONDS latest = _latest_script_task(job.project) @@ -449,6 +516,13 @@ def _advance_script(job: QuickCreateJob) -> int | None: _save_job(job, progress=min(44, 28 + elapsed // 8), message="正在生成分镜脚本…") return SCRIPT_POLL_SECONDS if latest is not None and latest.status == AITask.Status.FAILED: + if str(metadata.get("script_retry_of") or "") == str(latest.id): + started_at = _parse_iso(metadata.get("script_started_at")) + elapsed = int((timezone.now() - started_at).total_seconds()) if started_at else 0 + _save_job(job, progress=min(44, 28 + elapsed // 8), message="正在生成分镜脚本…") + return SCRIPT_POLL_SECONDS + if _restart_failed_script(job, latest): + return SCRIPT_POLL_SECONDS fail_quick_create(job, _task_public_error(latest), internal_error=latest.error_message) return None if latest is not None and latest.status == AITask.Status.SUCCEEDED: @@ -521,6 +595,84 @@ def _ensure_fallback_entities(project: Project) -> list[dict]: return entities +def _entities_of_type(entities: list[dict], type_name: str) -> list[dict]: + seen: set[str] = set() + out: list[dict] = [] + for item in entities: + if str(item.get("type") or "") != type_name: + continue + name = str(item.get("name") or item.get("id") or "").strip() + if not name or name in seen: + continue + seen.add(name) + out.append(item) + return out + + +def _asset_key(kind: str, label: str) -> tuple[str, str]: + return (str(kind or ""), str(label or "").strip()) + + +def _specs_from_entities(project: Project, entities: list[dict]) -> list[tuple[str, str, str]]: + return [ + (BaseAssetGroup.Kind.PRODUCT, project.product.title, project.product.title), + *[ + ( + BaseAssetGroup.Kind.PERSON, + str(item.get("visual_prompt") or item.get("name") or "电商模特"), + str(item.get("name") or "推荐模特"), + ) + for item in _entities_of_type(entities, "character") + ], + *[ + ( + BaseAssetGroup.Kind.SCENE, + str(item.get("visual_prompt") or item.get("name") or "电商场景"), + str(item.get("name") or "推荐场景"), + ) + for item in _entities_of_type(entities, "scene") + ], + ] + + +def _script_asset_specs(project: Project) -> list[tuple[str, str, str]]: + entities = [dict(item) for item in ((project.metadata or {}).get("script_entities") or []) if isinstance(item, dict)] + if not entities: + return [] + return _specs_from_entities(project, entities) + + +def _assets_status_message(specs: list[tuple[str, str, str]], done: int, total: int) -> str: + n_person = sum(1 for kind, _, _ in specs if kind == BaseAssetGroup.Kind.PERSON) + n_scene = sum(1 for kind, _, _ in specs if kind == BaseAssetGroup.Kind.SCENE) + if total: + return f"正在生成 {n_person} 个角色图、{n_scene} 个场景图（{done}/{total}）" + return f"正在生成 {n_person} 个角色图、{n_scene} 个场景图" + + +def _known_asset_keys(project: Project, task_ids: list[str]) -> set[tuple[str, str]]: + keys: set[tuple[str, str]] = set() + for group in project.base_asset_groups.all(): + if (group.metadata or {}).get("triview_of"): + continue + keys.add(_asset_key(group.kind, str((group.metadata or {}).get("label") or ""))) + if task_ids: + for task in AITask.objects.filter(id__in=task_ids): + payload = task.request_payload or {} + if payload.get("triview_of"): + continue + kind = str(payload.get("kind") or "") + label = str(payload.get("label") or "") + if kind or label: + keys.add(_asset_key(kind, label)) + return keys + + +def _missing_asset_specs(project: Project, task_ids: list[str]) -> list[tuple[str, str, str]]: + known = _known_asset_keys(project, task_ids) + return [spec for spec in _script_asset_specs(project) if _asset_key(spec[0], spec[2]) not in known] + + def _start_base_assets(job: QuickCreateJob) -> None: project = job.project user = job.created_by or project.created_by @@ -529,96 +681,88 @@ def _start_base_assets(job: QuickCreateJob) -> None: with transaction.atomic(): job = QuickCreateJob.objects.select_for_update().select_related("project").get(id=job.id) metadata = dict(job.metadata or {}) - if metadata.get("base_asset_task_ids"): + if metadata.get("assets_starting"): return + metadata["assets_starting"] = True + job.metadata = metadata + job.save(update_fields=["metadata", "updated_at"]) entities = _ensure_fallback_entities(job.project) - specs = [ - (BaseAssetGroup.Kind.PRODUCT, job.project.product.title, job.project.product.title), - *[ - (BaseAssetGroup.Kind.PERSON, str(item.get("visual_prompt") or item.get("name") or "电商模特"), str(item.get("name") or "推荐模特")) - for item in entities - if item.get("type") == "character" - ], - *[ - (BaseAssetGroup.Kind.SCENE, str(item.get("visual_prompt") or item.get("name") or "电商场景"), str(item.get("name") or "推荐场景")) - for item in entities - if item.get("type") == "scene" - ], - ] - task_ids = [] - for kind, prompt, label in specs: - task = generate_base_asset(project=job.project, user=user, kind=kind, prompt=prompt, label=label) - task_ids.append(str(task.id)) + specs = _specs_from_entities(job.project, entities) + created_ids: list[str] = [] + failed: Exception | None = None + try: + task_ids = [str(item) for item in (dict(job.metadata or {}).get("base_asset_task_ids") or [])] + known = _known_asset_keys(job.project, task_ids) + for kind, prompt, label in specs: + if _asset_key(kind, label) in known: + continue + task = generate_base_asset( + project=job.project, + user=user, + kind=kind, + prompt=prompt, + label=label, + # 极速成片只用角色立绘。三视图属于专业模式增强项，不能阻塞一键出片。 + auto_triview=False, + ) + created_ids.append(str(task.id)) + known.add(_asset_key(kind, label)) + except Exception as exc: # noqa: BLE001 — 部分任务已提交时要留下 ID，下一轮只补缺的 + failed = exc + job.refresh_from_db() metadata = dict(job.metadata or {}) - metadata["base_asset_task_ids"] = task_ids - metadata["assets_started"] = True - _save_job(job, metadata=metadata, message="正在生成商品、模特与场景资产", progress=52) - stage, _ = ProjectStage.objects.get_or_create(project=job.project, stage=ProjectStage.Stage.BASE_ASSETS) - stage.status = ProjectStage.Status.RUNNING - stage.save(update_fields=["status", "updated_at"]) + metadata.pop("assets_starting", None) + existing_ids = [str(item) for item in (metadata.get("base_asset_task_ids") or [])] + if created_ids: + metadata["base_asset_task_ids"] = [*existing_ids, *created_ids] + metadata["assets_started"] = True + _save_job( + job, + metadata=metadata, + message=_assets_status_message(specs, 0, len(metadata["base_asset_task_ids"])), + progress=52, + ) + stage, _ = ProjectStage.objects.get_or_create(project=job.project, stage=ProjectStage.Stage.BASE_ASSETS) + stage.status = ProjectStage.Status.RUNNING + stage.save(update_fields=["status", "updated_at"]) + else: + if failed is not None and not existing_ids: + metadata.pop("assets_started", None) + else: + metadata["assets_started"] = True + _save_job(job, metadata=metadata) + if failed is not None and not created_ids and not existing_ids: + raise failed + if failed is not None: + logger.warning("quick create job %s started some assets then hit %s; continuing with submitted tasks", job.id, failed) -def _task_failure(tasks) -> AITask | None: - return next( - ( - task - for task in tasks - if task.status == AITask.Status.FAILED - or task.status not in _ACTIVE_TASK_STATUSES | {AITask.Status.SUCCEEDED} - ), - None, - ) +def _failed_tasks(tasks) -> list[AITask]: + return [task for task in tasks if task.status in {AITask.Status.FAILED, AITask.Status.CANCELLED}] def _task_public_error(task: AITask) -> str: from apps.ai.generation_errors import public_error_for_task public_error = public_error_for_task(task) - return public_error.fallback_message if public_error else "生成任务失败，请稍后重试" - - -def _advance_assets(job: QuickCreateJob) -> int | None: - metadata = dict(job.metadata or {}) - base_ids = metadata.get("base_asset_task_ids") or [] - if not base_ids: - _start_base_assets(job) - return POLL_DELAY_SECONDS - tasks = list(AITask.objects.filter(project=job.project, id__in=base_ids)) - failed = _task_failure(tasks) - if failed is not None: - fail_quick_create(job, _task_public_error(failed), internal_error=failed.error_message) - return None - if len(tasks) < len(base_ids) or any(task.status in _ACTIVE_TASK_STATUSES for task in tasks): - done = sum(task.status == AITask.Status.SUCCEEDED for task in tasks) - _save_job(job, progress=min(64, 52 + done * 4), message="正在匹配并生成模特与场景") - return POLL_DELAY_SECONDS - - triview_ids = metadata.get("triview_task_ids") - if triview_ids is None: - person_groups = list( - job.project.base_asset_groups.filter( - kind=BaseAssetGroup.Kind.PERSON, - task_id__in=base_ids, - adopted_asset__isnull=False, - ).select_related("adopted_asset") + combined = " ".join( + part + for part in ( + public_error.fallback_message if public_error else "", + getattr(public_error, "code", "") or "", + task.error_message or "", ) - triview_ids = [ - str(generate_person_triview(project=job.project, user=job.created_by or job.project.created_by, portrait_asset=group.adopted_asset).id) - for group in person_groups - ] - metadata["triview_task_ids"] = triview_ids - _save_job(job, metadata=metadata, progress=65, message="正在为推荐模特生成三视图") - return POLL_DELAY_SECONDS + if part + ) + return _public_error(combined or "生成任务失败，请稍后重试") - triview_tasks = list(AITask.objects.filter(project=job.project, id__in=triview_ids)) - failed = _task_failure(triview_tasks) - if failed is not None: - fail_quick_create(job, _task_public_error(failed), internal_error=failed.error_message) - return None - if len(triview_tasks) < len(triview_ids) or any(task.status in _ACTIVE_TASK_STATUSES for task in triview_tasks): - _save_job(job, progress=68, message="正在为推荐模特生成三视图") - return POLL_DELAY_SECONDS +def _all_portraits_have_triview(project: Project) -> bool: + portraits = _portrait_groups(project) + return all(_portrait_has_triview(project, group.adopted_asset_id) for group in portraits) + + +def _finish_assets_and_enter_storyboard(job: QuickCreateJob) -> int: for group in job.project.base_asset_groups.filter(adopted_asset__isnull=False): if group.kind != BaseAssetGroup.Kind.PRODUCT: group_meta = dict(group.metadata or {}) @@ -642,6 +786,55 @@ def _advance_assets(job: QuickCreateJob) -> int | None: return 1 +def _advance_assets(job: QuickCreateJob) -> int | None: + metadata = dict(job.metadata or {}) + base_ids = [str(item) for item in (metadata.get("base_asset_task_ids") or [])] + if not base_ids: + if (metadata.get("assets_started") or metadata.get("assets_starting")) and timezone.now() - job.updated_at > STALE_AFTER: + metadata.pop("assets_started", None) + metadata.pop("assets_starting", None) + _save_job(job, metadata=metadata) + _start_base_assets(job) + return POLL_DELAY_SECONDS + missing = _missing_asset_specs(job.project, base_ids) + if missing: + _start_base_assets(job) + job.refresh_from_db() + base_ids = [str(item) for item in ((job.metadata or {}).get("base_asset_task_ids") or [])] + tasks = list(AITask.objects.filter(project=job.project, id__in=base_ids)) + failed_list = [ + task for task in _failed_tasks(tasks) + if not (task.request_payload or {}).get("triview_of") + ] + if failed_list: + if _retry_failed_base_assets(job, failed_list, base_ids): + return POLL_DELAY_SECONDS + fail_quick_create(job, _task_public_error(failed_list[0]), internal_error=failed_list[0].error_message) + return None + specs = _script_asset_specs(job.project) + if len(tasks) < len(base_ids) or any(task.status in _ACTIVE_TASK_STATUSES for task in tasks): + done = sum(task.status == AITask.Status.SUCCEEDED for task in tasks) + _save_job(job, progress=min(64, 52 + done * 4), message=_assets_status_message(specs, done, len(base_ids))) + return POLL_DELAY_SECONDS + + expected_people = _entities_of_type( + [dict(item) for item in ((job.project.metadata or {}).get("script_entities") or []) if isinstance(item, dict)], + "character", + ) + portraits = _portrait_groups(job.project) + if expected_people and len(portraits) < len(expected_people): + _save_job(job, progress=64, message="角色图已提交，正在写入资产…") + return POLL_DELAY_SECONDS + + # 极速模式不创建、不等待、也不重试人物三视图；已有的历史三视图任务同样不影响后续。 + metadata = dict(job.metadata or {}) + metadata["triview_skipped"] = True + metadata.pop("triview_task_ids", None) + metadata.pop("triview_ready", None) + _save_job(job, metadata=metadata) + return _finish_assets_and_enter_storyboard(job) + + def _reviews_ready(job: QuickCreateJob) -> bool | None: """True=可出视频，False=继续等，None=审核失败且任务已终止。""" metadata = dict(job.metadata or {}) @@ -667,7 +860,7 @@ def _reviews_ready(job: QuickCreateJob) -> bool | None: if not blockers: return True if any(item.get("review_status") == "failed" for item in blockers): - fail_quick_create(job, "生成素材未通过审核，请进入专业模式调整后重试") + fail_quick_create(job, REVIEW_FAIL_MESSAGE) return None wait_started = metadata.get("review_wait_started") if not wait_started: @@ -816,7 +1009,14 @@ def _videos_ready(job: QuickCreateJob) -> bool: def _can_complete(job: QuickCreateJob) -> bool: - return _videos_ready(job) + # 极速成片和专业模式共用同一个 Project。用户从极速页进入专业模式后， + # 专业模式可能已经将所有镜头收口并把项目标记为 completed，但极速编排 + # 自己最后一次轮询曾因网络/worker 波动落成 failed。此时项目的完成态是 + # 更可靠的事实来源，不能继续让极速页显示「成片尚未完成」。 + # + # 正常的 Project.completed 只能由 finish_video_stage() 在全部片段有采用版 + # 且均成功时写入；因此这里不会把仅有一段预览的半成品误报为成片。 + return job.project.status == Project.Status.COMPLETED or _videos_ready(job) def _complete(job: QuickCreateJob) -> None: @@ -834,19 +1034,57 @@ def _complete(job: QuickCreateJob) -> None: def _advance_production(job: QuickCreateJob) -> int | None: metadata = dict(job.metadata or {}) - if not metadata.get("storyboard_started"): - submit_storyboard(project=job.project, user=job.created_by or job.project.created_by, prompt="") - stage, _ = ProjectStage.objects.get_or_create(project=job.project, stage=ProjectStage.Stage.STORYBOARD) - stage.status = ProjectStage.Status.RUNNING - stage.save(update_fields=["status", "updated_at"]) - metadata["storyboard_started"] = True - _save_job(job, metadata=metadata, progress=74, message="正在生成故事板与镜头画面") + if ( + metadata.get("storyboard_started") + and job.project.current_stage != ProjectStage.Stage.VIDEO + and not job.project.storyboard_shots.exists() + ): + metadata["storyboard_started"] = False + _save_job(job, metadata=metadata, message="正在生成故事板与镜头画面") + if not (job.metadata or {}).get("storyboard_started"): + with transaction.atomic(): + job = QuickCreateJob.objects.select_for_update().select_related("project").get(id=job.id) + metadata = dict(job.metadata or {}) + if not metadata.get("storyboard_started"): + metadata["storyboard_started"] = True + _save_job(job, metadata=metadata, progress=74, message="正在生成故事板与镜头画面") + should_submit = True + else: + should_submit = False + if should_submit: + try: + submit_storyboard(project=job.project, user=job.created_by or job.project.created_by, prompt="") + stage, _ = ProjectStage.objects.get_or_create(project=job.project, stage=ProjectStage.Stage.STORYBOARD) + stage.status = ProjectStage.Status.RUNNING + stage.save(update_fields=["status", "updated_at"]) + except Exception as exc: + metadata = dict(job.metadata or {}) + metadata.pop("storyboard_started", None) + metadata["internal_error"] = str(exc)[:2000] + hard = any( + token in str(exc).lower() + for token in ("insufficient credit", "no active", "not configured") + ) + retries = int(metadata.get("storyboard_submit_retries") or 0) + if not hard and retries < 3: + metadata["storyboard_submit_retries"] = retries + 1 + _save_job(job, metadata=metadata, message="正在继续生成故事板…") + return POLL_DELAY_SECONDS + _save_job(job, metadata=metadata) + raise if job.project.current_stage != ProjectStage.Stage.VIDEO: result = poll_storyboard(project=job.project, user=job.created_by or job.project.created_by) if result.get("status") == "failed": - fail_quick_create(job, str(result.get("error") or "故事板生成失败")) - return None + retries = int((job.metadata or {}).get("storyboard_fail_retries") or 0) + if retries >= 2: + fail_quick_create(job, _public_error(str(result.get("error") or "故事板生成失败"))) + return None + metadata = dict(job.metadata or {}) + metadata["storyboard_fail_retries"] = retries + 1 + metadata["storyboard_started"] = False + _save_job(job, metadata=metadata, message="故事板未成功，正在重试…") + return 1 if result.get("status") != "succeeded": total = max(1, int(result.get("total") or 1)) done = int(result.get("done") or 0) @@ -870,7 +1108,7 @@ def _advance_production(job: QuickCreateJob) -> int | None: if failed: retries = int((job.metadata or {}).get("video_fail_retries") or 0) if retries >= 2: - fail_quick_create(job, failed[0].error_message or f"第{failed[0].sort_order + 1}段视频生成失败") + fail_quick_create(job, _public_error(failed[0].error_message or f"第{failed[0].sort_order + 1}段视频生成失败")) return None metadata = dict(job.metadata or {}) metadata["video_fail_retries"] = retries + 1 @@ -915,8 +1153,220 @@ def _advance_production(job: QuickCreateJob) -> int | None: _local_script_jobs: set[str] = set() -def _script_task_started(job: QuickCreateJob) -> bool: - return job.project.ai_tasks.filter(task_type=AITask.Type.SCRIPT_GENERATION).exists() +def _target_queue() -> str: + """airshelf.quick 明确没人听时回退 celery，避免任务进黑洞后一直停在「等待开始」。""" + from apps.common.celery_health import worker_consumes_queue + + if worker_consumes_queue(QUICK_CREATE_QUEUE) is True: + return QUICK_CREATE_QUEUE + return DEFAULT_WORKER_QUEUE + + +def _script_consume_active(job: QuickCreateJob) -> bool: + metadata = dict(job.metadata or {}) + started_at = _parse_iso(metadata.get("script_consuming_at")) + if not metadata.get("script_consuming") or started_at is None: + return False + return timezone.now() - started_at < SCRIPT_CONSUME_STALE + + +def _claim_script_consume(job: QuickCreateJob) -> bool: + with transaction.atomic(): + locked = QuickCreateJob.objects.select_for_update().select_related("project").get(id=job.id) + if _is_finished(locked) or _adopted_script(locked.project) is not None: + return False + if _script_consume_active(locked) or _script_generation_inflight(locked.project): + return False + metadata = dict(locked.metadata or {}) + metadata["script_consuming"] = True + metadata["script_consuming_at"] = timezone.now().isoformat() + locked.metadata = metadata + locked.save(update_fields=["metadata", "updated_at"]) + return True + + +def _release_script_consume(job_id: str) -> None: + job = QuickCreateJob.objects.filter(id=job_id).first() + if job is None: + return + metadata = dict(job.metadata or {}) + if "script_consuming" not in metadata and "script_consuming_at" not in metadata: + return + metadata.pop("script_consuming", None) + metadata.pop("script_consuming_at", None) + _save_job(job, metadata=metadata) + + +def _restart_failed_script(job: QuickCreateJob, latest: AITask) -> bool: + if not _is_retryable_exc(Exception(latest.error_message or "")): + return False + metadata = dict(job.metadata or {}) + retries = int(metadata.get("transient_retries") or 0) + if retries >= TRANSIENT_RETRY_LIMIT: + return False + metadata["transient_retries"] = retries + 1 + metadata["internal_error"] = (latest.error_message or "")[:2000] + metadata["script_started"] = True + metadata["script_started_at"] = timezone.now().isoformat() + metadata["script_retry_of"] = str(latest.id) + metadata.pop("script_local", None) + metadata.pop("script_consuming", None) + metadata.pop("script_consuming_at", None) + _save_job(job, metadata=metadata, message="脚本生成中断，正在重新生成…") + _enqueue_script(str(job.id)) + return True + + +def _retry_failed_base_assets(job: QuickCreateJob, failed_tasks: list[AITask], base_ids: list) -> bool: + metadata = dict(job.metadata or {}) + retries_map = { + str(key): int(value) + for key, value in dict(metadata.get("asset_fail_retries_by_key") or {}).items() + } + old_retries = int(metadata.get("asset_fail_retries") or 0) + user = job.created_by or job.project.created_by + if user is None: + return False + new_ids = [str(task_id) for task_id in base_ids] + retried = False + for failed in failed_tasks: + payload = failed.request_payload or {} + if payload.get("triview_of"): + continue + group = job.project.base_asset_groups.filter(task_id=failed.id).first() + kind = str(payload.get("kind") or (group.kind if group is not None else "")) + if not kind: + return False + label = str( + payload.get("label") + or ((group.metadata or {}).get("label") if group is not None else "") + or kind + ) + key = f"{kind}:{label}" + retries = retries_map.get(key, old_retries) + if retries >= ASSET_FAIL_RETRY_LIMIT: + return False + prompt = str( + payload.get("prompt") + or (group.prompt if group is not None else "") + or job.project.product.title + ) + new_task = generate_base_asset( + project=job.project, + user=user, + kind=kind, + prompt=prompt, + label=label, + group_id=str(group.id) if group is not None else None, + auto_triview=False, + ) + retries_map[key] = retries + 1 + new_ids = [ + str(new_task.id) if str(task_id) == str(failed.id) else str(task_id) for task_id in new_ids + ] + retried = True + if not retried: + return False + metadata["asset_fail_retries_by_key"] = retries_map + metadata["base_asset_task_ids"] = new_ids + _save_job(job, metadata=metadata, message="有资产未成功，正在重试…") + return True + + +def _portrait_groups(project: Project) -> list[BaseAssetGroup]: + return [ + group + for group in project.base_asset_groups.filter( + kind=BaseAssetGroup.Kind.PERSON, + adopted_asset__isnull=False, + ).select_related("adopted_asset") + if not (group.metadata or {}).get("triview_of") + ] + + +def _portrait_has_triview(project: Project, portrait_id) -> bool: + key = str(portrait_id) + return any( + (group.metadata or {}).get("triview_of") == key and group.adopted_asset_id + for group in project.base_asset_groups.filter(kind=BaseAssetGroup.Kind.PERSON) + ) + + +def _latest_triview_task(project: Project, portrait_id) -> AITask | None: + key = str(portrait_id) + return ( + AITask.objects.filter(project=project, request_payload__triview_of=key) + .order_by("-created_at") + .first() + ) + + +def _ensure_triview_tasks(job: QuickCreateJob) -> list[str]: + """每个角色立绘都必须有三视图。优先复用 worker 已发起的任务，避免重复扣费。""" + task_ids: list[str] = [] + user = job.created_by or job.project.created_by + for group in _portrait_groups(job.project): + portrait_id = group.adopted_asset_id + if _portrait_has_triview(job.project, portrait_id): + continue + existing = _latest_triview_task(job.project, portrait_id) + if existing is not None and existing.status in _ACTIVE_TASK_STATUSES | { + AITask.Status.FAILED, + AITask.Status.CANCELLED, + }: + task_ids.append(str(existing.id)) + continue + adopted = group.adopted_asset + if user is None or adopted is None: + continue + try: + task = generate_person_triview(project=job.project, user=user, portrait_asset=adopted) + except Exception as exc: # noqa: BLE001 — 单个角色三视图提交失败时继续处理其余角色 + logger.warning("quick create job %s skipped triview for %s: %s", job.id, portrait_id, exc) + continue + task_ids.append(str(task.id)) + return task_ids + + +def _retry_failed_triviews(job: QuickCreateJob, failed_tasks: list[AITask], triview_ids: list) -> bool: + metadata = dict(job.metadata or {}) + retries_map = { + str(key): int(value) + for key, value in dict(metadata.get("triview_fail_retries_by_portrait") or {}).items() + } + old_retries = int(metadata.get("triview_fail_retries") or 0) + user = job.created_by or job.project.created_by + if user is None: + return False + new_ids = [str(task_id) for task_id in triview_ids] + retried = False + for failed in failed_tasks: + portrait_id = str((failed.request_payload or {}).get("triview_of") or "") + retries = retries_map.get(portrait_id, old_retries) + if retries >= TRIVIEW_FAIL_RETRY_LIMIT: + return False + portrait = ( + job.project.base_asset_groups.filter( + kind=BaseAssetGroup.Kind.PERSON, + adopted_asset_id=portrait_id, + ) + .select_related("adopted_asset") + .first() + ) + if portrait is None or portrait.adopted_asset is None: + return False + new_task = generate_person_triview(project=job.project, user=user, portrait_asset=portrait.adopted_asset) + retries_map[portrait_id] = retries + 1 + new_ids = [ + str(new_task.id) if str(task_id) == str(failed.id) else str(task_id) for task_id in new_ids + ] + retried = True + if not retried: + return False + metadata["triview_fail_retries_by_portrait"] = retries_map + metadata["triview_task_ids"] = new_ids + _save_job(job, metadata=metadata, message="角色三视图未成功，正在重试…") + return True def _claim_next_advance(job_id: str, delay_seconds: int) -> bool: @@ -934,7 +1384,7 @@ def _claim_next_advance(job_id: str, delay_seconds: int) -> bool: def _run_quick_script_in_thread(job_id: str) -> None: - """共享 Redis 上的旧 worker 会把未注册任务直接丢掉；本机线程兜底，避免页面永久转圈。""" + """broker 都发不出去时的最后兜底；正常路径都走 Celery，不占用 API 进程。""" if job_id in _local_script_jobs: return _local_script_jobs.add(job_id) @@ -953,22 +1403,32 @@ def _run_quick_script_in_thread(job_id: str) -> None: threading.Thread(target=_worker, daemon=True, name=f"quick-script-{job_id[:8]}").start() -def _enqueue_advance(job: QuickCreateJob) -> None: - from apps.common.celery_health import worker_consumes_queue +def _enqueue_script(job_id: str) -> None: + from apps.projects.tasks import run_quick_script_task + + try: + run_quick_script_task.apply_async(args=[job_id], queue=_target_queue()) + except Exception: # noqa: BLE001 — 队列不可用时才退回本机线程 + logger.exception("quick create enqueue script failed for job %s", job_id) + _run_quick_script_in_thread(job_id) + + +def _enqueue_advance(job: QuickCreateJob, *, countdown: int = 0) -> None: from apps.projects.tasks import advance_quick_create_task - job_id = str(job.id) - if worker_consumes_queue("airshelf.quick") is False: - _advance_without_quick_queue(job_id) - return + options = {"args": [str(job.id)], "queue": _target_queue()} + if countdown: + options["countdown"] = countdown try: - advance_quick_create_task.apply_async(args=[job_id], queue="airshelf.quick") - except Exception: # noqa: BLE001 — 队列不可用时就地推进一步 - _advance_without_quick_queue(job_id) + advance_quick_create_task.apply_async(**options) + except Exception: # noqa: BLE001 — 队列不可用时就地推进一步，倒计时失败交给 recover + logger.exception("quick create enqueue advance failed for job %s", job.id) + if not countdown: + _advance_without_quick_queue(str(job.id)) def _advance_without_quick_queue(job_id: str) -> None: - """部署后的 worker 若只听 celery 队列，airshelf.quick 里的任务会永远没人拿。""" + """队列暂时发不出去时，先把商品识别和脚本启动做掉，避免页面一直停在等待开始。""" delay = advance_quick_create(job_id) job = QuickCreateJob.objects.filter(id=job_id).first() if job is None or delay is None: @@ -995,6 +1455,16 @@ def resume_quick_create(job: QuickCreateJob) -> QuickCreateJob: metadata.pop("transient_retries", None) metadata.pop("review_poll_retries", None) metadata.pop("next_advance_at", None) + metadata.pop("video_fail_retries", None) + metadata.pop("asset_fail_retries", None) + metadata.pop("asset_fail_retries_by_key", None) + metadata.pop("storyboard_fail_retries", None) + metadata.pop("triview_fail_retries", None) + metadata.pop("triview_fail_retries_by_portrait", None) + metadata.pop("storyboard_submit_retries", None) + if metadata.pop("triview_skipped", None): + metadata.pop("triview_task_ids", None) + metadata.pop("triview_ready", None) _save_job( job, status=QuickCreateJob.Status.RUNNING, @@ -1002,13 +1472,42 @@ def resume_quick_create(job: QuickCreateJob) -> QuickCreateJob: message="正在从上次进度继续生成…", metadata=metadata, ) - _enqueue_advance(job) + # 重试请求里先推进一步：只入队时页面会显示“继续了”，实际可能没人消费。 + delay = advance_quick_create(str(job.id)) + job.refresh_from_db() + if job.status == QuickCreateJob.Status.FAILED and not _is_hard_failure(job): + metadata = dict(job.metadata or {}) + metadata.pop("next_advance_at", None) + _save_job( + job, + status=QuickCreateJob.Status.RUNNING, + error_message="", + message="正在从上次进度继续生成…", + metadata=metadata, + ) + delay = POLL_DELAY_SECONDS + if not _is_finished(job): + _enqueue_advance(job, countdown=max(1, int(delay or 1))) job.refresh_from_db() return job +def _resume_transient_failure(job: QuickCreateJob, message: str) -> None: + metadata = dict(job.metadata or {}) + retries = int(metadata.get("transient_retries") or 0) + metadata["transient_retries"] = retries + 1 + _save_job( + job, + status=QuickCreateJob.Status.RUNNING, + error_message="", + message=message, + metadata=metadata, + ) + _enqueue_advance(job) + + def recover_quick_create(job: QuickCreateJob) -> None: - """前端轮询时把卡住的编排拉起来：超时落失败，被旧 worker 丢掉的脚本改走本机线程。""" + """前端轮询时把卡住的编排拉起来：超时落失败，丢进没人听的队列的任务改走默认 celery。""" job.refresh_from_db() if job.status == QuickCreateJob.Status.FAILED and _can_complete(job): _complete(job) @@ -1017,28 +1516,28 @@ def recover_quick_create(job: QuickCreateJob) -> None: _restore_project_after_orchestrator_timeout(job) job.refresh_from_db() retries = int((job.metadata or {}).get("transient_retries") or 0) - if ( - job.status == QuickCreateJob.Status.FAILED - and job.phase in {QuickCreateJob.Phase.SCRIPT, QuickCreateJob.Phase.ASSETS} - and _adopted_script(job.project) is not None - and _transient_internal(job) - and retries < TRANSIENT_RETRY_LIMIT - ): - _save_job(job, status=QuickCreateJob.Status.RUNNING, error_message="", message="网络波动，正在继续生成…") - _enqueue_advance(job) + if job.status != QuickCreateJob.Status.FAILED or retries >= TRANSIENT_RETRY_LIMIT: return if ( - job.status == QuickCreateJob.Status.FAILED - and job.phase == QuickCreateJob.Phase.SCRIPT + job.phase in {QuickCreateJob.Phase.SCRIPT, QuickCreateJob.Phase.ASSETS} + and _adopted_script(job.project) is not None + and _transient_internal(job) + ): + _resume_transient_failure(job, "网络波动，正在继续生成…") + return + if ( + job.phase == QuickCreateJob.Phase.SCRIPT and _adopted_script(job.project) is None and _transient_internal(job) - and retries < TRANSIENT_RETRY_LIMIT ): metadata = dict(job.metadata or {}) metadata["transient_retries"] = retries + 1 metadata.pop("script_started", None) metadata.pop("script_started_at", None) metadata.pop("script_local", None) + metadata.pop("script_consuming", None) + metadata.pop("script_consuming_at", None) + metadata.pop("script_retry_of", None) _save_job( job, status=QuickCreateJob.Status.RUNNING, @@ -1047,13 +1546,16 @@ def recover_quick_create(job: QuickCreateJob) -> None: metadata=metadata, ) _enqueue_advance(job) + return + if job.phase == QuickCreateJob.Phase.PRODUCTION and _transient_internal(job): + _resume_transient_failure(job, "网络波动，正在继续生成…") return if _is_finished(job): return if job.status == QuickCreateJob.Status.QUEUED or job.phase == QuickCreateJob.Phase.PRODUCT: if timezone.now() - job.updated_at > STALE_AFTER: try: - _advance_without_quick_queue(str(job.id)) + _enqueue_advance(job) except Exception: # noqa: BLE001 — 恢复失败不能把进度接口打成 500 logger.exception("quick create recover advance failed for job %s", job.id) return @@ -1070,12 +1572,14 @@ def recover_quick_create(job: QuickCreateJob) -> None: and adopted is None and not local_running and not _script_generation_inflight(job.project) + and not _script_consume_active(job) ) if stolen: - metadata["script_local"] = True metadata["script_started_at"] = timezone.now().isoformat() + metadata.pop("script_consuming", None) + metadata.pop("script_consuming_at", None) _save_job(job, metadata=metadata, message="正在继续生成分镜脚本…") - _run_quick_script_in_thread(job_id) + _enqueue_script(job_id) return if timed_out and adopted is None and not local_running and not _script_generation_inflight(job.project): fail_quick_create(job, "脚本生成超时，请稍后重试或进入专业模式查看") diff --git a/core/backend/apps/projects/tasks.py b/core/backend/apps/projects/tasks.py index c5a6f3e..67d27b5 100644 --- a/core/backend/apps/projects/tasks.py +++ b/core/backend/apps/projects/tasks.py @@ -47,9 +47,15 @@ def run_export_job_task(self, export_job_id: str) -> str: QUICK_CREATE_QUEUE = "airshelf.quick" -@app.task(bind=True, max_retries=0, soft_time_limit=240, time_limit=270, queue=QUICK_CREATE_QUEUE) +# 豆包长思考脚本允许完整跑 30 分钟；硬上限额外留 60 秒让 soft timeout 的 +# 收尾/落库完成，避免 15 分钟时仍在正常输出却被 worker 强制中断。 +@app.task(bind=True, max_retries=0, soft_time_limit=1800, time_limit=1860, queue=QUICK_CREATE_QUEUE) def run_quick_script_task(self, quick_job_id: str) -> str: - """脚本生成单独跑，避免把整条极速成片编排堵在一次 SSE 消费里。""" + """脚本生成单独跑，避免把整条极速成片编排堵在一次 SSE 消费里。 + + 软超时必须长于豆包思考流（允许最长 30 分钟）。短于 HTTP 流超时会 SIGUSR1 掐连接， + 任务监视器就记成 stream aborted (client disconnected)。 + """ from celery.exceptions import SoftTimeLimitExceeded from apps.projects.models import QuickCreateJob @@ -57,41 +63,36 @@ def run_quick_script_task(self, quick_job_id: str) -> str: try: consume_quick_script(quick_job_id) - except SoftTimeLimitExceeded: + except SoftTimeLimitExceeded as exc: job = QuickCreateJob.objects.select_related("project").filter(id=quick_job_id).first() if job is not None and job.status not in { QuickCreateJob.Status.SUCCEEDED, QuickCreateJob.Status.FAILED, QuickCreateJob.Status.CANCELLED, }: - fail_quick_create(job, "脚本生成超时，请稍后重试或进入专业模式查看") - raise + fail_quick_create( + job, + "脚本生成时间较长，系统会自动重试", + internal_error=f"SoftTimeLimitExceeded: stream aborted (client disconnected); {exc}", + ) + return quick_job_id return quick_job_id -@app.task(bind=True, max_retries=0, queue=QUICK_CREATE_QUEUE) +@app.task(bind=True, max_retries=0) def advance_quick_create_task(self, quick_job_id: str) -> str: - """一次只推进一个可重入状态，等待型阶段通过重新入队轮询，不占 worker 睡眠。""" - from apps.projects.services.quick_create import advance_quick_create + """一次只推进一个可重入状态，等待型阶段通过重新入队轮询，不占 worker 睡眠。 + + 编排走默认 celery 队列：worker 即使没听 airshelf.quick，也不会停在「等待开始」。 + """ + from apps.projects.models import QuickCreateJob + from apps.projects.services.quick_create import _claim_next_advance, _enqueue_advance, advance_quick_create next_delay = advance_quick_create(quick_job_id) if next_delay is not None: - from apps.projects.services.quick_create import _claim_next_advance - if not _claim_next_advance(quick_job_id, int(next_delay)): return quick_job_id - try: - advance_quick_create_task.apply_async( - args=[quick_job_id], - countdown=max(1, int(next_delay)), - queue=QUICK_CREATE_QUEUE, - ) - except Exception as exc: # noqa: BLE001 — 重排失败不能留下永久“生成中” - from apps.projects.models import QuickCreateJob - from apps.projects.services.quick_create import fail_quick_create - - job = QuickCreateJob.objects.select_related("project").filter(id=quick_job_id).first() - if job is not None: - fail_quick_create(job, "生成队列暂时中断，请稍后重试", internal_error=str(exc)) - raise + job = QuickCreateJob.objects.filter(id=quick_job_id).first() + if job is not None: + _enqueue_advance(job, countdown=max(1, int(next_delay))) return quick_job_id diff --git a/core/backend/apps/projects/test_quick_create.py b/core/backend/apps/projects/test_quick_create.py index bf353e0..357d974 100644 --- a/core/backend/apps/projects/test_quick_create.py +++ b/core/backend/apps/projects/test_quick_create.py @@ -12,12 +12,14 @@ from apps.accounts.models import Team, TeamMember, User from apps.ai.models import AITask, ModelConfig, ModelProvider from apps.assets.models import Asset, AssetFile from apps.products.models import Product, ProductImage -from apps.projects.models import Project, ProjectStage, QuickCreateJob, ScriptSegment, ScriptVersion, VideoSegment, VideoSegmentVersion +from apps.projects.models import BaseAssetGroup, Project, ProjectStage, QuickCreateJob, ScriptSegment, ScriptVersion, VideoSegment, VideoSegmentVersion from apps.projects.serializers import ProjectListSerializer, QuickCreateJobSerializer from apps.projects.services.pipeline import initialize_project_pipeline from apps.projects.services.quick_create import ( + REVIEW_FAIL_MESSAGE, _consume_script_agent, _reviews_ready, + _safe_error, _start_videos, advance_quick_create, cancel_quick_create, @@ -94,7 +96,7 @@ class QuickCreateApiTests(TestCase): self.assertEqual(str(job.id), response.data["id"]) enqueue.assert_called_once() self.assertEqual(enqueue.call_args.kwargs["args"], [str(job.id)]) - self.assertEqual(enqueue.call_args.kwargs["queue"], "airshelf.quick") + self.assertIn(enqueue.call_args.kwargs["queue"], {"celery", "airshelf.quick"}) require_worker_task.assert_called_once_with("apps.projects.tasks.advance_quick_create_task") self.assertEqual(get_model.call_count, 2) get_quick_model.assert_called_once() @@ -290,7 +292,7 @@ class QuickCreateApiTests(TestCase): self.assertEqual(job.status, QuickCreateJob.Status.RUNNING) enqueue.assert_called_once() - def test_history_lists_team_jobs_including_failed(self): + def test_history_lists_completed_jobs_only(self): product = Product.objects.create(team=self.team, created_by=self.user, title="历史商品") project = Project.objects.create(team=self.team, created_by=self.user, product=product, name="历史商品 · 极速成片") QuickCreateJob.objects.create( @@ -334,12 +336,45 @@ class QuickCreateApiTests(TestCase): response = self.client.get("/api/projects/quick-create-history/") self.assertEqual(response.status_code, 200) - self.assertEqual(response.data["count"], 2) + self.assertEqual(response.data["count"], 1) titles = [item["title"] for item in response.data["results"]] self.assertIn("历史商品 · 极速成片", titles) - self.assertIn("失败商品 · 极速成片", titles) + self.assertNotIn("失败商品 · 极速成片", titles) self.assertNotIn("进行中商品 · 极速成片", titles) + def test_history_hides_deleted_projects(self): + product = Product.objects.create(team=self.team, created_by=self.user, title="还在的商品") + alive = Project.objects.create(team=self.team, created_by=self.user, product=product, name="还在的商品 · 极速成片") + QuickCreateJob.objects.create( + team=self.team, + created_by=self.user, + project=alive, + status=QuickCreateJob.Status.SUCCEEDED, + phase=QuickCreateJob.Phase.COMPLETE, + ) + deleted_product = Product.objects.create(team=self.team, created_by=self.user, title="已删商品") + deleted = Project.objects.create( + team=self.team, + created_by=self.user, + product=deleted_product, + name="已删商品 · 极速成片", + is_deleted=True, + ) + QuickCreateJob.objects.create( + team=self.team, + created_by=self.user, + project=deleted, + status=QuickCreateJob.Status.CANCELLED, + phase=QuickCreateJob.Phase.SCRIPT, + ) + + response = self.client.get("/api/projects/quick-create-history/") + self.assertEqual(response.status_code, 200) + titles = [item["title"] for item in response.data["results"]] + self.assertEqual(response.data["count"], 1) + self.assertIn("还在的商品 · 极速成片", titles) + self.assertNotIn("已删商品 · 极速成片", titles) + def test_list_serializer_flags_quick_create_projects(self): product = Product.objects.create(team=self.team, created_by=self.user, title="列表商品") quick = Project.objects.create( @@ -357,6 +392,44 @@ class QuickCreateApiTests(TestCase): ) self.assertTrue(ProjectListSerializer(quick).data["quick_create"]) self.assertFalse(ProjectListSerializer(normal).data["quick_create"]) + self.assertEqual(ProjectListSerializer(quick).data["quick_create_status"], "") + self.assertEqual(ProjectListSerializer(quick).data["quick_create_job_id"], "") + + QuickCreateJob.objects.create(team=self.team, created_by=self.user, project=quick, status=QuickCreateJob.Status.RUNNING) + self.assertEqual(ProjectListSerializer(quick).data["quick_create_status"], "running") + self.assertTrue(ProjectListSerializer(quick).data["quick_create_job_id"]) + + def test_running_quick_create_blocks_professional_edits(self): + product = Product.objects.create(team=self.team, created_by=self.user, title="锁单商品") + project = Project.objects.create( + team=self.team, + created_by=self.user, + product=product, + name="锁单商品 · 极速成片", + metadata={"quick_create": True}, + ) + job = QuickCreateJob.objects.create( + team=self.team, + created_by=self.user, + project=project, + status=QuickCreateJob.Status.RUNNING, + ) + + blocked = self.client.patch(f"/api/projects/{project.id}/", {"name": "不该改"}, format="json") + self.assertEqual(blocked.status_code, 409) + self.assertIn("极速成片", str(blocked.data)) + + allowed = self.client.get(f"/api/projects/{project.id}/") + self.assertEqual(allowed.status_code, 200) + self.assertEqual(allowed.data["quick_create_status"], "running") + self.assertEqual(allowed.data["quick_create_job_id"], str(job.id)) + + job.status = QuickCreateJob.Status.FAILED + job.save(update_fields=["status", "updated_at"]) + resumed = self.client.patch(f"/api/projects/{project.id}/", {"name": "专业模式可改"}, format="json") + self.assertEqual(resumed.status_code, 200) + project.refresh_from_db() + self.assertEqual(project.name, "专业模式可改") class QuickCreateCoordinatorTests(TestCase): @@ -410,7 +483,7 @@ class QuickCreateCoordinatorTests(TestCase): self.assertTrue(self.job.metadata.get("script_started")) start_script.assert_called_once() self.assertEqual(start_script.call_args.kwargs["args"], [str(self.job.id)]) - self.assertEqual(start_script.call_args.kwargs["queue"], "airshelf.quick") + self.assertIn(start_script.call_args.kwargs["queue"], {"celery", "airshelf.quick"}) def test_adopted_script_moves_to_assets_without_rerunning(self): script = ScriptVersion.objects.create(project=self.project, title="脚本", content="口播", is_adopted=True) @@ -431,8 +504,8 @@ class QuickCreateCoordinatorTests(TestCase): self.assertIsNone(delay) self.assertEqual(self.job.status, QuickCreateJob.Status.CANCELLED) - @patch("apps.projects.services.quick_create._run_quick_script_in_thread") - def test_recover_reruns_script_locally_when_queue_drops_it(self, run_local): + @patch("apps.projects.services.quick_create._enqueue_script") + def test_recover_requeues_script_when_queue_drops_it(self, enqueue_script): self.job.status = QuickCreateJob.Status.RUNNING self.job.phase = QuickCreateJob.Phase.SCRIPT self.job.metadata = { @@ -443,9 +516,8 @@ class QuickCreateCoordinatorTests(TestCase): recover_quick_create(self.job) self.job.refresh_from_db() - self.assertTrue(self.job.metadata.get("script_local")) self.assertEqual(self.job.status, QuickCreateJob.Status.RUNNING) - run_local.assert_called_once_with(str(self.job.id)) + enqueue_script.assert_called_once_with(str(self.job.id)) def test_recover_starts_stale_queued_job_without_waiting_for_quick_queue(self): QuickCreateJob.objects.filter(id=self.job.id).update( @@ -455,13 +527,13 @@ class QuickCreateCoordinatorTests(TestCase): updated_at=timezone.now() - timedelta(seconds=30), ) self.job.refresh_from_db() - with patch("apps.projects.services.quick_create._advance_without_quick_queue") as advance: + with patch("apps.projects.services.quick_create._enqueue_advance") as enqueue: recover_quick_create(self.job) - advance.assert_called_once_with(str(self.job.id)) + enqueue.assert_called_once() @patch("apps.common.celery_health.worker_consumes_queue", return_value=False) - @patch("apps.projects.services.quick_create._run_quick_script_in_thread") - def test_advance_script_runs_locally_when_quick_queue_has_no_consumer(self, run_local, _listens): + @patch("apps.projects.tasks.run_quick_script_task.apply_async") + def test_advance_script_falls_back_to_celery_when_quick_queue_has_no_consumer(self, start_script, _listens): self.job.status = QuickCreateJob.Status.RUNNING self.job.phase = QuickCreateJob.Phase.SCRIPT self.job.save(update_fields=["status", "phase", "updated_at"]) @@ -470,11 +542,11 @@ class QuickCreateCoordinatorTests(TestCase): self.job.refresh_from_db() self.assertEqual(delay, 5) self.assertTrue(self.job.metadata.get("script_started")) - self.assertTrue(self.job.metadata.get("script_local")) - run_local.assert_called_once_with(str(self.job.id)) + start_script.assert_called_once() + self.assertEqual(start_script.call_args.kwargs["queue"], "celery") - @patch("apps.projects.services.quick_create._run_quick_script_in_thread") - def test_recover_does_not_steal_script_when_model_call_is_inflight(self, run_local): + @patch("apps.projects.services.quick_create._enqueue_script") + def test_recover_does_not_steal_script_when_model_call_is_inflight(self, enqueue_script): self._script_task(AITask.Status.SUBMITTED, key="script-inflight-1") self.job.status = QuickCreateJob.Status.RUNNING self.job.phase = QuickCreateJob.Phase.SCRIPT @@ -486,9 +558,8 @@ class QuickCreateCoordinatorTests(TestCase): recover_quick_create(self.job) self.job.refresh_from_db() - self.assertFalse(self.job.metadata.get("script_local")) self.assertEqual(self.job.status, QuickCreateJob.Status.RUNNING) - run_local.assert_not_called() + enqueue_script.assert_not_called() def test_advance_script_ignores_old_failed_task_while_latest_is_running(self): self._script_task(AITask.Status.FAILED, key="script-old-failed", error_message="stream aborted (client disconnected)") @@ -506,6 +577,25 @@ class QuickCreateCoordinatorTests(TestCase): self.assertEqual(delay, 5) self.assertEqual(self.job.status, QuickCreateJob.Status.RUNNING) + @patch("apps.projects.services.quick_create._enqueue_script") + def test_advance_script_retries_retryable_failure_without_failing_job(self, enqueue_script): + self._script_task(AITask.Status.FAILED, key="script-aborted", error_message="stream aborted (client disconnected)") + self.job.status = QuickCreateJob.Status.RUNNING + self.job.phase = QuickCreateJob.Phase.SCRIPT + self.job.metadata = { + "script_started": True, + "script_started_at": timezone.now().isoformat(), + } + self.job.save(update_fields=["status", "phase", "metadata", "updated_at"]) + + delay = advance_quick_create(str(self.job.id)) + self.job.refresh_from_db() + self.assertEqual(delay, 5) + self.assertEqual(self.job.status, QuickCreateJob.Status.RUNNING) + self.assertTrue(self.job.metadata.get("script_started")) + self.assertEqual(self.job.metadata.get("transient_retries"), 1) + enqueue_script.assert_called_once_with(str(self.job.id)) + @patch("apps.projects.tasks.advance_quick_create_task.apply_async") def test_recover_restarts_script_after_stream_abort(self, enqueue): self.job.status = QuickCreateJob.Status.FAILED @@ -525,6 +615,42 @@ class QuickCreateCoordinatorTests(TestCase): self.assertEqual(self.job.metadata.get("transient_retries"), 1) enqueue.assert_called_once() + @patch("apps.projects.tasks.advance_quick_create_task.apply_async") + def test_recover_retries_soft_time_limit_as_transient(self, enqueue): + self.job.status = QuickCreateJob.Status.FAILED + self.job.phase = QuickCreateJob.Phase.SCRIPT + self.job.error_message = "脚本生成时间较长，系统会自动重试" + self.job.metadata = { + "script_started": True, + "script_started_at": timezone.now().isoformat(), + "internal_error": "SoftTimeLimitExceeded: stream aborted (client disconnected)", + } + self.job.save(update_fields=["status", "phase", "error_message", "metadata", "updated_at"]) + + recover_quick_create(self.job) + self.job.refresh_from_db() + self.assertEqual(self.job.status, QuickCreateJob.Status.RUNNING) + self.assertFalse(self.job.metadata.get("script_started")) + enqueue.assert_called_once() + + @patch("apps.projects.tasks.advance_quick_create_task.apply_async") + def test_recover_retries_failed_script_task_without_internal_error(self, enqueue): + self._script_task( + AITask.Status.FAILED, + key="script-soft-limit", + error_message="SoftTimeLimitExceeded()", + ) + self.job.status = QuickCreateJob.Status.FAILED + self.job.phase = QuickCreateJob.Phase.SCRIPT + self.job.error_message = "脚本生成超时，请稍后重试或进入专业模式查看" + self.job.metadata = {"script_started": True} + self.job.save(update_fields=["status", "phase", "error_message", "metadata", "updated_at"]) + + recover_quick_create(self.job) + self.job.refresh_from_db() + self.assertEqual(self.job.status, QuickCreateJob.Status.RUNNING) + enqueue.assert_called_once() + @patch("apps.projects.services.quick_create.get_quick_script_model") @patch("apps.projects.services.quick_create.stream_script_agent") def test_consume_script_does_not_hit_db_on_every_sse_frame(self, stream_fn, get_model): @@ -581,12 +707,29 @@ class QuickCreateCoordinatorTests(TestCase): self.assertEqual(self.project.status, Project.Status.VIDEOING) self.assertEqual(self.project.failure_reason, "") + @patch("apps.projects.tasks.advance_quick_create_task.apply_async") + def test_recover_resumes_production_after_orchestrator_timeout(self, enqueue): + self.project.status = Project.Status.VIDEOING + self.project.current_stage = ProjectStage.Stage.VIDEO + self.project.save(update_fields=["status", "current_stage", "updated_at"]) + self.job.status = QuickCreateJob.Status.FAILED + self.job.phase = QuickCreateJob.Phase.PRODUCTION + self.job.error_message = "脚本已保留，后续步骤遇到网络波动。点重试会从上次进度继续" + self.job.metadata = {"storyboard_started": True, "internal_error": "Timeout reading from socket"} + self.job.save(update_fields=["status", "phase", "error_message", "metadata", "updated_at"]) + + recover_quick_create(self.job) + self.job.refresh_from_db() + self.assertEqual(self.job.status, QuickCreateJob.Status.RUNNING) + self.assertEqual(self.job.metadata.get("transient_retries"), 1) + enqueue.assert_called_once() + @patch("apps.projects.tasks.advance_quick_create_task.apply_async") def test_resume_failed_production_job_keeps_progress(self, enqueue): self.job.status = QuickCreateJob.Status.FAILED self.job.phase = QuickCreateJob.Phase.PRODUCTION self.job.error_message = "极速成片暂未完成，请稍后重试或进入专业模式查看" - self.job.metadata = {"storyboard_started": True, "transient_retries": 8} + self.job.metadata = {"storyboard_started": True, "transient_retries": 8, "video_fail_retries": 2} self.job.save(update_fields=["status", "phase", "error_message", "metadata", "updated_at"]) resume_quick_create(self.job) @@ -594,6 +737,7 @@ class QuickCreateCoordinatorTests(TestCase): self.assertEqual(self.job.status, QuickCreateJob.Status.RUNNING) self.assertEqual(self.job.error_message, "") self.assertIsNone(self.job.metadata.get("transient_retries")) + self.assertIsNone(self.job.metadata.get("video_fail_retries")) enqueue.assert_called_once() @patch("apps.projects.services.quick_create.assets_client.is_enabled", return_value=True) @@ -610,6 +754,40 @@ class QuickCreateCoordinatorTests(TestCase): self.job.refresh_from_db() self.assertTrue(self.job.metadata.get("reviews_skipped")) + def test_safe_error_maps_image_moderation_to_review_copy(self): + self.assertEqual( + _safe_error(ValueError("400 moderation_blocked safety_violations=[sexual]")), + REVIEW_FAIL_MESSAGE, + ) + self.assertEqual( + _safe_error(RuntimeError("InputImageSensitiveContentDetected")), + REVIEW_FAIL_MESSAGE, + ) + + @patch("apps.projects.services.quick_create.assets_client.is_enabled", return_value=True) + @patch("apps.projects.services.quick_create.poll_team_reviews", return_value={}) + @patch( + "apps.projects.services.quick_create.collect_video_review_blockers", + return_value=[{"review_status": "failed", "name": "女主立绘"}], + ) + def test_failed_asset_review_stops_job_with_clear_message(self, _blockers, _poll, _enabled): + self.job.status = QuickCreateJob.Status.RUNNING + self.job.phase = QuickCreateJob.Phase.PRODUCTION + self.job.save(update_fields=["status", "phase", "updated_at"]) + self.assertIsNone(_reviews_ready(self.job)) + self.job.refresh_from_db() + self.assertEqual(self.job.status, QuickCreateJob.Status.FAILED) + self.assertEqual(self.job.error_message, REVIEW_FAIL_MESSAGE) + + def test_failed_job_serializer_rewrites_hidden_moderation_error(self): + self.job.status = QuickCreateJob.Status.FAILED + self.job.phase = QuickCreateJob.Phase.PRODUCTION + self.job.error_message = "极速成片暂未完成，请稍后重试或进入专业模式查看" + self.job.metadata = {"internal_error": "400 moderation_blocked safety_violations=[sexual]"} + self.job.save(update_fields=["status", "phase", "error_message", "metadata", "updated_at"]) + data = QuickCreateJobSerializer(self.job).data + self.assertEqual(data["error_message"], REVIEW_FAIL_MESSAGE) + def test_recover_marks_success_when_video_already_finished(self): self.project.video_segments.exclude(sort_order=0).delete() segment = self.project.video_segments.get(sort_order=0) @@ -635,6 +813,24 @@ class QuickCreateCoordinatorTests(TestCase): self.assertEqual(self.job.status, QuickCreateJob.Status.SUCCEEDED) self.assertEqual(self.job.phase, QuickCreateJob.Phase.COMPLETE) + def test_recover_marks_success_when_professional_mode_already_completed(self): + """专业模式完成后，极速任务的旧失败状态必须自动被回收。""" + self.job.status = QuickCreateJob.Status.FAILED + self.job.phase = QuickCreateJob.Phase.PRODUCTION + self.job.error_message = "极速成片暂未完成，请稍后重试或进入专业模式查看" + self.job.save(update_fields=["status", "phase", "error_message", "updated_at"]) + self.project.status = Project.Status.COMPLETED + self.project.current_stage = ProjectStage.Stage.VIDEO + self.project.failure_reason = "" + self.project.save(update_fields=["status", "current_stage", "failure_reason", "updated_at"]) + + recover_quick_create(self.job) + + self.job.refresh_from_db() + self.assertEqual(self.job.status, QuickCreateJob.Status.SUCCEEDED) + self.assertEqual(self.job.phase, QuickCreateJob.Phase.COMPLETE) + self.assertEqual(self.job.progress, 100) + @patch("apps.projects.services.quick_create.generate_base_asset", side_effect=TimeoutError("Timeout reading from socket")) def test_asset_start_timeout_retries_without_locking_the_job(self, _generate): script = ScriptVersion.objects.create(project=self.project, title="脚本", content="口播", is_adopted=True) @@ -680,6 +876,213 @@ class QuickCreateCoordinatorTests(TestCase): self.assertEqual(self.job.status, QuickCreateJob.Status.FAILED) self.assertNotEqual(self.project.status, Project.Status.FAILED) + def _ready_asset_tasks(self): + script = ScriptVersion.objects.create(project=self.project, title="脚本", content="口播", is_adopted=True) + ScriptSegment.objects.create(script_version=script, sort_order=0, narration="开场") + ids = [] + for kind, key in ( + (AITask.Type.PRODUCT_IMAGE, "asset-product"), + (AITask.Type.PERSON_IMAGE, "asset-person"), + (AITask.Type.SCENE_IMAGE, "asset-scene"), + ): + task = self._script_task(AITask.Status.SUCCEEDED, key=key) + task.task_type = kind + task.save(update_fields=["task_type"]) + ids.append(str(task.id)) + return ids + + def _portrait_group(self, *, task=None, name="推荐模特"): + portrait = Asset.objects.create( + team=self.team, + created_by=self.user, + name=name, + asset_type=Asset.Type.IMAGE, + source=Asset.Source.AI_GENERATED, + category=Asset.Category.PERSON, + ) + return BaseAssetGroup.objects.create( + project=self.project, + kind=BaseAssetGroup.Kind.PERSON, + task=task, + adopted_asset=portrait, + metadata={"label": name}, + ), portrait + + @patch("apps.projects.services.quick_create.generate_person_triview") + def test_assets_skip_inflight_person_triview(self, start_triview): + base_ids = self._ready_asset_tasks() + _group, portrait = self._portrait_group() + inflight = self._script_task(AITask.Status.SUBMITTED, key="auto-triview") + inflight.request_payload = {"triview_of": str(portrait.id)} + inflight.save(update_fields=["request_payload"]) + self.job.status = QuickCreateJob.Status.RUNNING + self.job.phase = QuickCreateJob.Phase.ASSETS + self.job.metadata = {"base_asset_task_ids": base_ids, "assets_started": True} + self.job.save(update_fields=["status", "phase", "metadata", "updated_at"]) + + delay = advance_quick_create(str(self.job.id)) + self.job.refresh_from_db() + start_triview.assert_not_called() + self.assertEqual(delay, 1) + self.assertIsNone(self.job.metadata.get("triview_task_ids")) + self.assertTrue(self.job.metadata.get("triview_skipped")) + self.assertEqual(self.job.status, QuickCreateJob.Status.RUNNING) + self.assertEqual(self.job.phase, QuickCreateJob.Phase.PRODUCTION) + + @patch("apps.projects.services.quick_create.generate_person_triview") + def test_assets_never_starts_triview(self, start_triview): + new_id = uuid.uuid4() + start_triview.return_value = SimpleNamespace(id=new_id) + base_ids = self._ready_asset_tasks() + group, portrait = self._portrait_group() + self.job.status = QuickCreateJob.Status.RUNNING + self.job.phase = QuickCreateJob.Phase.ASSETS + self.job.metadata = {"base_asset_task_ids": base_ids, "assets_started": True} + self.job.save(update_fields=["status", "phase", "metadata", "updated_at"]) + + delay = advance_quick_create(str(self.job.id)) + self.job.refresh_from_db() + start_triview.assert_not_called() + self.assertEqual(delay, 1) + self.assertIsNone(self.job.metadata.get("triview_task_ids")) + self.assertTrue(self.job.metadata.get("triview_skipped")) + self.assertEqual(self.job.phase, QuickCreateJob.Phase.PRODUCTION) + self.assertIsNone(group.task_id) + + def test_failed_triview_does_not_block_storyboard(self): + base_ids = self._ready_asset_tasks() + _group, portrait = self._portrait_group() + failed = self._script_task(AITask.Status.FAILED, key="triview-failed", error_message="image_edit timeout") + failed.request_payload = {"triview_of": str(portrait.id)} + failed.save(update_fields=["request_payload"]) + self.job.status = QuickCreateJob.Status.RUNNING + self.job.phase = QuickCreateJob.Phase.ASSETS + self.job.metadata = { + "base_asset_task_ids": base_ids, + "assets_started": True, + "triview_task_ids": [str(failed.id)], + "triview_fail_retries": 2, + } + self.job.save(update_fields=["status", "phase", "metadata", "updated_at"]) + + delay = advance_quick_create(str(self.job.id)) + self.job.refresh_from_db() + self.assertEqual(delay, 1) + self.assertEqual(self.job.phase, QuickCreateJob.Phase.PRODUCTION) + self.assertTrue(self.job.metadata.get("triview_skipped")) + self.assertEqual(self.job.status, QuickCreateJob.Status.RUNNING) + + @patch("apps.projects.services.quick_create.generate_base_asset") + def test_assets_start_all_characters_and_scenes(self, generate): + generate.side_effect = lambda **kwargs: SimpleNamespace(id=uuid.uuid4()) + ScriptVersion.objects.create(project=self.project, title="脚本", content="口播", is_adopted=True) + self.project.metadata = { + "script_entities": [ + {"id": "c1", "type": "character", "name": "女主", "visual_prompt": "都市女性"}, + {"id": "c2", "type": "character", "name": "闺蜜", "visual_prompt": "活泼女生"}, + {"id": "s1", "type": "scene", "name": "客厅", "visual_prompt": "暖光客厅"}, + {"id": "s2", "type": "scene", "name": "咖啡馆", "visual_prompt": "街边咖啡馆"}, + ] + } + self.project.save(update_fields=["metadata", "updated_at"]) + self.job.status = QuickCreateJob.Status.RUNNING + self.job.phase = QuickCreateJob.Phase.ASSETS + self.job.save(update_fields=["status", "phase", "updated_at"]) + + delay = advance_quick_create(str(self.job.id)) + self.job.refresh_from_db() + self.assertEqual(delay, 10) + self.assertEqual(generate.call_count, 5) + labels = [(call.kwargs["kind"], call.kwargs["label"]) for call in generate.call_args_list] + self.assertEqual( + labels, + [ + (BaseAssetGroup.Kind.PRODUCT, "测试精华"), + (BaseAssetGroup.Kind.PERSON, "女主"), + (BaseAssetGroup.Kind.PERSON, "闺蜜"), + (BaseAssetGroup.Kind.SCENE, "客厅"), + (BaseAssetGroup.Kind.SCENE, "咖啡馆"), + ], + ) + for call in generate.call_args_list: + if call.kwargs["kind"] == BaseAssetGroup.Kind.PERSON: + self.assertFalse(call.kwargs["auto_triview"]) + + @patch("apps.projects.services.quick_create.generate_person_triview") + def test_assets_skip_triview_for_every_character(self, start_triview): + start_triview.side_effect = lambda **kwargs: SimpleNamespace(id=uuid.uuid4()) + base_ids = self._ready_asset_tasks() + _g1, portrait_a = self._portrait_group(name="女主") + _g2, portrait_b = self._portrait_group(name="闺蜜") + self.job.status = QuickCreateJob.Status.RUNNING + self.job.phase = QuickCreateJob.Phase.ASSETS + self.job.metadata = {"base_asset_task_ids": base_ids, "assets_started": True} + self.job.save(update_fields=["status", "phase", "metadata", "updated_at"]) + + delay = advance_quick_create(str(self.job.id)) + self.job.refresh_from_db() + self.assertEqual(delay, 1) + start_triview.assert_not_called() + self.assertIsNone(self.job.metadata.get("triview_task_ids")) + self.assertTrue(self.job.metadata.get("triview_skipped")) + self.assertEqual(self.job.phase, QuickCreateJob.Phase.PRODUCTION) + self.assertEqual(self.job.status, QuickCreateJob.Status.RUNNING) + + @patch("apps.projects.services.quick_create.generate_person_triview", side_effect=ValueError("no active image model configured")) + def test_triview_submit_error_does_not_fail_job(self, _start_triview): + base_ids = self._ready_asset_tasks() + self._portrait_group() + self.job.status = QuickCreateJob.Status.RUNNING + self.job.phase = QuickCreateJob.Phase.ASSETS + self.job.metadata = {"base_asset_task_ids": base_ids, "assets_started": True} + self.job.save(update_fields=["status", "phase", "metadata", "updated_at"]) + + delay = advance_quick_create(str(self.job.id)) + self.job.refresh_from_db() + self.assertEqual(delay, 1) + self.assertEqual(self.job.status, QuickCreateJob.Status.RUNNING) + self.assertEqual(self.job.phase, QuickCreateJob.Phase.PRODUCTION) + self.assertTrue(self.job.metadata.get("triview_skipped")) + + def test_failed_triview_task_in_base_ids_does_not_fail_assets(self): + base_ids = self._ready_asset_tasks() + _group, portrait = self._portrait_group() + failed = self._script_task(AITask.Status.FAILED, key="triview-in-base", error_message="image_edit timeout") + failed.request_payload = {"kind": "person", "label": "推荐模特", "triview_of": str(portrait.id)} + failed.save(update_fields=["request_payload"]) + self.job.status = QuickCreateJob.Status.RUNNING + self.job.phase = QuickCreateJob.Phase.ASSETS + self.job.metadata = {"base_asset_task_ids": [*base_ids, str(failed.id)], "assets_started": True} + self.job.save(update_fields=["status", "phase", "metadata", "updated_at"]) + + delay = advance_quick_create(str(self.job.id)) + self.job.refresh_from_db() + self.assertEqual(delay, 1) + self.assertEqual(self.job.status, QuickCreateJob.Status.RUNNING) + self.assertEqual(self.job.phase, QuickCreateJob.Phase.PRODUCTION) + + @patch("apps.projects.tasks.advance_quick_create_task.apply_async") + def test_resume_failed_assets_continues_into_storyboard(self, enqueue): + base_ids = self._ready_asset_tasks() + self.job.status = QuickCreateJob.Status.FAILED + self.job.phase = QuickCreateJob.Phase.ASSETS + self.job.error_message = "极速成片暂未完成，请稍后重试或进入专业模式查看" + self.job.metadata = { + "base_asset_task_ids": base_ids, + "assets_started": True, + "triview_task_ids": [], + "triview_ready": True, + "triview_skipped": True, + } + self.job.save(update_fields=["status", "phase", "error_message", "metadata", "updated_at"]) + + resume_quick_create(self.job) + self.job.refresh_from_db() + self.assertEqual(self.job.status, QuickCreateJob.Status.RUNNING) + self.assertEqual(self.job.phase, QuickCreateJob.Phase.PRODUCTION) + self.assertEqual(self.job.error_message, "") + enqueue.assert_called_once() + @patch("apps.projects.services.quick_create._reviews_ready", return_value=True) def test_production_counts_ready_videos_without_adding_version_ids(self, _reviews): self.project.video_segments.exclude(sort_order=0).delete() @@ -739,6 +1142,7 @@ class QuickCreateCoordinatorTests(TestCase): data = QuickCreateJobSerializer(self.job).data self.assertEqual(data["phase_index"], 3) self.assertEqual(data["result"]["video_url"], "https://cdn.example/quick.mp4") + self.assertEqual(data["result"]["final_video_url"], "") self.assertEqual(data["result"]["duration_seconds"], 15) def test_phase_index_matches_four_step_ui(self): diff --git a/core/backend/apps/projects/tests.py b/core/backend/apps/projects/tests.py index f91b1d5..12ef1f7 100644 --- a/core/backend/apps/projects/tests.py +++ b/core/backend/apps/projects/tests.py @@ -1,4 +1,4 @@ -from django.test import TestCase +from django.test import TestCase, override_settings from unittest.mock import patch from rest_framework.test import APIClient @@ -351,9 +351,11 @@ class ProjectApiTests(TestCase): group = BaseAssetGroup.objects.get(project=project, kind=BaseAssetGroup.Kind.PERSON) self.assertEqual(group.metadata.get("label"), "女主") + @override_settings(CACHES={"default": {"BACKEND": "django.core.cache.backends.locmem.LocMemCache"}}) + @patch("apps.ai.tasks.generate_base_asset_task.delay") @patch("apps.ai.services._store_generated_media") @patch("apps.ai.services.get_image_provider") - def test_generate_base_asset_ignores_auto_triview_request(self, get_provider, store_media): + def test_generate_person_base_asset_always_enables_auto_triview(self, get_provider, store_media, _enqueue_base_asset): ModelConfig.objects.create( provider=self.provider, name="img-model-auto-tri", display_name="Img Auto Tri", capability=ModelConfig.Capability.IMAGE, endpoint="images/generations", unit_price="1.0000", @@ -370,13 +372,13 @@ class ProjectApiTests(TestCase): response = self.client.post( f"/api/projects/{project.id}/generate-base-asset/", - {"kind": "person", "prompt": "portrait", "label": "hero", "auto_triview": True}, + {"kind": "person", "prompt": "portrait", "label": "hero"}, format="json", ) self.assertEqual(response.status_code, 202) task = AITask.objects.get(id=response.data["task"]["id"]) - self.assertFalse(task.request_payload.get("auto_triview")) + self.assertTrue(task.request_payload.get("auto_triview")) @patch("apps.ai.services._store_generated_media") @patch("apps.ai.services.get_image_provider") @@ -1348,6 +1350,13 @@ class VideoSegmentTrueUpTests(TestCase): self.assertEqual(task.credit_reservation.amount, video_reserve_amount(quote.points)) self.assertEqual(task.request_payload["estimated_tokens"], tokens) + def test_submit_reads_wizard_output_spec(self): + self.project.metadata = {"wizard": {"aspect_ratio": "16:9", "resolution": "480p"}} + self.project.save(update_fields=["metadata"]) + task = self._submit() + self.assertEqual(task.request_payload["ratio"], "16:9") + self.assertEqual(task.request_payload["resolution"], "480p") + @patch("apps.ai.services._store_generated_media") def test_poll_settles_by_actual_usage_tokens(self, store): from decimal import Decimal diff --git a/core/backend/apps/projects/views.py b/core/backend/apps/projects/views.py index a081f29..a6bc52f 100644 --- a/core/backend/apps/projects/views.py +++ b/core/backend/apps/projects/views.py @@ -8,7 +8,7 @@ from django.http import HttpResponse, JsonResponse, StreamingHttpResponse from django.utils import timezone from rest_framework import status from rest_framework.decorators import action -from rest_framework.exceptions import ValidationError +from rest_framework.exceptions import APIException, ValidationError from rest_framework.parsers import FormParser, MultiPartParser from rest_framework.renderers import BaseRenderer from rest_framework.response import Response @@ -95,6 +95,12 @@ from .tasks import poll_video_segment_task logger = logging.getLogger(__name__) +class QuickCreateInProgress(APIException): + status_code = status.HTTP_409_CONFLICT + default_detail = "该项目正在极速成片中，请到极速成片页查看进度" + default_code = "quick_create_running" + + class ServerSentEventRenderer(BaseRenderer): """让 DRF 内容协商接受 Accept: text/event-stream(否则流式端点直接 406)。 实际响应由视图返回 StreamingHttpResponse 直接下发,这个 renderer 只用于通过协商。""" @@ -235,7 +241,7 @@ def settle_video_completion(project: Project) -> bool: class ProjectViewSet(TeamScopedViewSetMixin, ModelViewSet): - queryset = Project.objects.select_related("product", "timeline").prefetch_related( + queryset = Project.objects.select_related("product", "timeline", "quick_create_job").prefetch_related( "stages", "video_segments", "video_segments__adopted_version__asset__files", @@ -339,7 +345,7 @@ class ProjectViewSet(TeamScopedViewSetMixin, ModelViewSet): # ——原列表把每个项目的 阶段/片段/故事板/时间线/资产文件全拉出,20 个项目实测 ~2s。 if self.action == "list": qs = ( - Project.objects.select_related("product", "product__cover_asset", "timeline") + Project.objects.select_related("product", "product__cover_asset", "timeline", "quick_create_job") .prefetch_related( "product__cover_asset__files", # 成片地址(final_video_url)只需要「成功的导出任务」,预取到位后列表不再逐项目查库 @@ -363,6 +369,21 @@ class ProjectViewSet(TeamScopedViewSetMixin, ModelViewSet): ).order_by("-updated_at") return super().get_queryset().filter(is_deleted=False, purged_at__isnull=True) + def initial(self, request, *args, **kwargs): + super().initial(request, *args, **kwargs) + if request.method in ("GET", "HEAD", "OPTIONS"): + return + if self.action in {"create", "destroy"}: + return + pk = kwargs.get("pk") + if not pk: + return + if QuickCreateJob.objects.filter( + project_id=pk, + status__in=[QuickCreateJob.Status.QUEUED, QuickCreateJob.Status.RUNNING], + ).exists(): + raise QuickCreateInProgress() + def perform_destroy(self, instance): instance.is_deleted = True instance.save(update_fields=["is_deleted", "updated_at"]) @@ -730,30 +751,19 @@ class ProjectViewSet(TeamScopedViewSetMixin, ModelViewSet): job = self._quick_job_queryset().get(id=job.id) return Response(QuickCreateJobSerializer(job).data) - @action(detail=False, methods=["post"], url_path=r"quick-create-retry/(?P[^/.]+)") - def quick_create_retry(self, request, job_id=None): - job = self._quick_job_queryset().filter(id=job_id).first() - if job is None: - return Response({"detail": "极速成片任务不存在"}, status=status.HTTP_404_NOT_FOUND) - if job.status == QuickCreateJob.Status.SUCCEEDED: - return Response({"detail": "任务已经完成"}, status=status.HTTP_400_BAD_REQUEST) - if job.status == QuickCreateJob.Status.CANCELLED: - return Response({"detail": "已取消的任务请重新开始"}, status=status.HTTP_400_BAD_REQUEST) - from .services.quick_create import resume_quick_create - - resume_quick_create(job) - job = self._quick_job_queryset().get(id=job.id) - return Response(QuickCreateJobSerializer(job).data) - @action(detail=False, methods=["get"], url_path="quick-create-history") def quick_create_history(self, request): from .services.quick_create import restore_false_failed_quick_creates restore_false_failed_quick_creates(self.get_team()) - # 进行中的任务看上方状态卡；列表要能找回失败后去专业模式继续的项目。 + # 进行中 / 未完成的任务回填上方表单；过往列表只放已完成成片。 jobs = ( self._quick_job_queryset() - .exclude(status__in=[QuickCreateJob.Status.QUEUED, QuickCreateJob.Status.RUNNING]) + .filter( + project__is_deleted=False, + project__purged_at__isnull=True, + status=QuickCreateJob.Status.SUCCEEDED, + ) .order_by("-created_at") ) return Response({ @@ -975,8 +985,9 @@ class ProjectViewSet(TeamScopedViewSetMixin, ModelViewSet): prompt=request.data.get("prompt", ""), label=request.data.get("label", ""), reference_asset_id=request.data.get("reference_asset_id") or None, - # 角色立绘不再自动接力三视图；三视图只由角色详情里的显式按钮生成。 - auto_triview=False, + # 用户点击角色 AI 生成 = 生成立绘并在完成后自动接力三视图。 + # 三视图任务由 worker 创建，页面刷新或离开也不会漏掉。 + auto_triview=kind == BaseAssetGroup.Kind.PERSON, ) except ValueError as exc: # 无可用模型 / 余额不足等,立即反馈 internal_kind = "user_credit_insufficient" if str(exc).strip().lower() == "insufficient credit" else "" @@ -1453,7 +1464,14 @@ class ProjectViewSet(TeamScopedViewSetMixin, ModelViewSet): # 火山/中转报错(如人脸需走素材库的 InputImageSensitiveContentDetected)→ 返回真实报错的 JSON, # 任务仍保留原始错误供排障，但普通用户仅收到安全错误对象，不让 500 HTML 导致前端白屏。 try: - submit_video_segment(video_segment=segment, user=request.user, prompt=request.data.get("prompt", "")) + submit_video_segment( + video_segment=segment, + user=request.user, + prompt=request.data.get("prompt", ""), + model_config_id=request.data.get("model_config_id") or None, + aspect_ratio=request.data.get("aspect_ratio") or None, + resolution=request.data.get("resolution") or None, + ) except Exception as exc: # noqa: BLE001 public_error = classify_generation_error(exc, operation="video_generate") return Response( diff --git a/core/frontend/src/App.tsx b/core/frontend/src/App.tsx index 933f7b0..9b28beb 100644 --- a/core/frontend/src/App.tsx +++ b/core/frontend/src/App.tsx @@ -21,6 +21,7 @@ import type { } from "./types"; import { publicModelDisplayName } from "./model-display"; import { generationErrorText } from "./generation-error"; +import { isQuickCreateBusy, lockedQuickCreateProject, rememberQuickCreateJob, withQuickCreateStatus } from "./quick-create-lock"; import { AccountMenu, CornerMarks, Decorations, ModeTabs, openCommandPalette, Sidebar, ToastLike, topModuleForPage } from "./components/app-shell"; import { SystemLoading } from "./components/loading"; import { @@ -415,6 +416,17 @@ export function App() { }; }, [authed, page, activeProjectId, detailRetry]); + useEffect(() => { + if (!authed || page !== "pipeline" || !activeProjectId) return; + const listed = projects.find((item) => item.id === activeProjectId); + const detailed = projectDetail && projectDetail.id === activeProjectId ? projectDetail : null; + const locked = lockedQuickCreateProject(listed, detailed); + if (!locked) return; + rememberQuickCreateJob(locked.quick_create_job_id); + setNotice({ type: "info", text: "该项目正在极速成片中，请在极速成片页查看进度" }); + navigate("quickCreate", { productId: locked.product, replace: true }); + }, [authed, page, activeProjectId, projects, projectDetail]); + // 静默轮询运行中的视频段(本机无 Celery worker,由前端驱动 poll-video-segment),实时刷新管线进度,不弹 toast。 // 资源账:旧实现每轮「GET 项目 → 逐段串行 POST → 再 GET 项目」,4 段在途时一轮 = 2 个 26KB GET + 4 个串行 // ARK 轮询(总耗时随段数线性涨)。现用内存态定位在途段(省前置 GET),段间 Promise.all 并行,一轮只回读一次。 @@ -516,12 +528,36 @@ export function App() { // eslint-disable-next-line react-hooks/exhaustive-deps }, [activeProjectId, refreshExport]); + function applyQuickCreateStatus(projectId: string, status: string) { + setProjects((items) => withQuickCreateStatus(items, projectId, status)); + setProjectDetail((current) => ( + current && current.id === projectId ? { ...current, quick_create_status: status } : current + )); + } + function navigate(next: Page, options: NavigateOptions = {}) { // 已知为子账号时在发起导航前拦截；与直接访问 URL 的 layout guard 共用同一规则。 if (role && !isOwner && isOwnerOnlyPage(next)) { setNotice({ type: "info", text: "当前账号暂无访问权限" }); return; } + if (next === "pipeline") { + const targetId = options.projectId ?? activeProjectId; + const listed = projects.find((item) => item.id === targetId); + const detailed = projectDetail && projectDetail.id === targetId ? projectDetail : null; + const released = options.quickCreateStatus || "failed"; + const canForce = Boolean(options.forcePipeline && targetId && !isQuickCreateBusy({ quick_create_status: released })); + if (canForce && targetId) { + applyQuickCreateStatus(targetId, released); + } + const locked = canForce ? null : lockedQuickCreateProject(listed, detailed); + if (locked) { + rememberQuickCreateJob(locked.quick_create_job_id); + setNotice({ type: "info", text: "该项目正在极速成片中，请在极速成片页查看进度" }); + next = "quickCreate"; + options = { ...options, productId: locked.product, replace: options.replace }; + } + } // 图片创作 / 极速成片只有显式入口才携带商品；从工作台或视频创作进入时不能继承全局当前商品。 const productId = next === "imageOptimize" || next === "quickCreate" ? options.productId : (options.productId ?? activeProductId); const projectId = options.projectId ?? activeProjectId; @@ -1064,6 +1100,7 @@ export function App() { modelConfigs={modelConfigs} onNotify={(type, text) => setNotice({ type, text })} onProjectCreated={() => { void loadData(); }} + onQuickCreateStatus={applyQuickCreateStatus} /> ); case "videoRemix": @@ -1151,19 +1188,14 @@ export function App() { onAdoptVideoVersion={(segmentId, versionId) => action(() => api.adoptVideoVersion(pipelineProject.id, { video_segment_id: segmentId, version_id: versionId }), "已采用该版本")} onGenerateVoiceover={(payload) => action(() => api.generateVoiceover(pipelineProject.id, payload), "配音已生成")} onGenerateBaseAsset={async (kind, prompt, label, referenceAssetId) => { - // 异步:提交→轮询出图→刷新。角色立绘成功后立刻据该立绘出三视图,用户不必再进详情点一次。 + // 异步:提交→轮询出图→刷新。角色三视图由后端在立绘落库后自动接力， + // 避免页面刷新/离开时漏掉，也避免这里重复创建第二个三视图任务。 // referenceAssetId:角色重跑时传当前立绘 → 后端走 image_edit 参考它,保持人物一致 const assetId = await submitAndPollAsset( () => api.generateBaseAsset(pipelineProject.id, { kind, prompt, label, reference_asset_id: referenceAssetId }), kind === "person" ? "" : "基础资产已生成", ); if (!assetId) return null; - if (kind === "person") { - await submitAndPollAsset( - () => api.generateTriview(pipelineProject.id, { portrait_asset_id: assetId }), - "角色立绘与三视图已生成", - ); - } return { adopted_asset: assetId }; }} onGenerateStoryboard={(prompt) => diff --git a/core/frontend/src/api.ts b/core/frontend/src/api.ts index a1842de..b33860f 100644 --- a/core/frontend/src/api.ts +++ b/core/frontend/src/api.ts @@ -83,6 +83,7 @@ export type QuickCreateJob = { }; result: null | { video_url: string; + final_video_url?: string; poster_url: string; duration_seconds: number; aspect_ratio: string; diff --git a/core/frontend/src/pipeline-page.css b/core/frontend/src/pipeline-page.css index 2ecfc81..b575d32 100644 --- a/core/frontend/src/pipeline-page.css +++ b/core/frontend/src/pipeline-page.css @@ -1034,7 +1034,28 @@ line-height: 1.4; } .as-action-bar > span svg { width: 14px; height: 14px; flex-shrink: 0; } - .as-action-bar > div { display: flex; gap: 9px; flex-shrink: 0; } + .as-action-bar > div { display: flex; align-items: center; gap: 9px; flex-shrink: 0; flex-wrap: wrap; justify-content: flex-end; } + .as-spec-fields { + display: flex; + align-items: flex-end; + gap: 8px; + } + .as-spec-field { + display: flex; + flex-direction: column; + gap: 3px; + min-width: 0; + } + .as-spec-field:nth-child(1), + .as-spec-field:nth-child(2) { width: 118px; } + .as-spec-field:nth-child(3) { width: 156px; } + .as-spec-field > span { + font-size: 10px; + line-height: 1; + color: var(--pl-muted); + white-space: nowrap; + } + .stage-foot .as-spec-fields { margin-right: 2px; } .as-action-bar .pl-ghost, .as-action-bar .pl-next { height: 42px; min-height: 42px; } .as-ai-btn, diff --git a/core/frontend/src/projects-page.css b/core/frontend/src/projects-page.css index 2a078dd..bac85a3 100644 --- a/core/frontend/src/projects-page.css +++ b/core/frontend/src/projects-page.css @@ -433,7 +433,9 @@ } .projects-page .vc-grid.list .vc-thumb { width: 184px; - height: 132px; + height: 100%; + min-height: 132px; + align-self: stretch; aspect-ratio: auto; } .projects-page .vc-grid.list .vc-play { diff --git a/core/frontend/src/quick-create-lock.ts b/core/frontend/src/quick-create-lock.ts new file mode 100644 index 0000000..7973b7f --- /dev/null +++ b/core/frontend/src/quick-create-lock.ts @@ -0,0 +1,46 @@ +const QUICK_JOB_KEY = "airshelf:quick-create-job"; + +export function isQuickCreateBusy(project?: { quick_create_status?: string } | null) { + return project?.quick_create_status === "queued" || project?.quick_create_status === "running"; +} + +export function lockedQuickCreateProject( + listed?: T | null, + detailed?: T | null, +): T | null { + const source = detailed || listed || null; + return source && isQuickCreateBusy(source) ? source : null; +} + +export function withQuickCreateStatus( + items: T[], + projectId: string, + status: string, +): T[] { + return items.map((item) => (item.id === projectId ? { ...item, quick_create_status: status } : item)); +} + +export function readQuickCreateJobId() { + try { + return localStorage.getItem(QUICK_JOB_KEY) || ""; + } catch { + return ""; + } +} + +export function rememberQuickCreateJob(jobId?: string) { + if (!jobId) return; + try { + localStorage.setItem(QUICK_JOB_KEY, jobId); + } catch { + /* ignore */ + } +} + +export function forgetQuickCreateJob() { + try { + localStorage.removeItem(QUICK_JOB_KEY); + } catch { + /* ignore */ + } +} diff --git a/core/frontend/src/routes/dashboard.tsx b/core/frontend/src/routes/dashboard.tsx index 64e3b86..1cfb0c5 100644 --- a/core/frontend/src/routes/dashboard.tsx +++ b/core/frontend/src/routes/dashboard.tsx @@ -11,6 +11,7 @@ import { import type { BillingSummary, Product, Project } from "../types"; import type { NavigateFn, Page } from "./route-config"; import { ConfirmModal } from "../components/overlays"; +import { isQuickCreateBusy } from "../quick-create-lock"; type DashTab = "all" | "wip" | "done"; type EntryTone = "primary" | "subtle"; @@ -82,7 +83,7 @@ function isQuickCreateProject(project: Project) { function dashCardMeta(project: Project, productTitle: string): string { const shots = project.video_segment_count ?? project.video_segments?.length ?? 0; - const mode = isQuickCreateProject(project) ? "极速成片" : "专业创作"; + const mode = isQuickCreateBusy(project) ? "极速成片生成中" : isQuickCreateProject(project) ? "极速成片" : "专业创作"; return [mode, productTitle, shots ? `${shots} 镜` : null].filter(Boolean).join(" / "); } @@ -273,7 +274,7 @@ export function Dashboard({ type="button" onClick={(event) => { event.stopPropagation(); openProject(); }} > - {project.status === "completed" ? "查看" : "继续"} + {isQuickCreateBusy(project) ? "查看进度" : project.status === "completed" ? "查看" : "继续"} diff --git a/core/frontend/src/routes/pipeline.tsx b/core/frontend/src/routes/pipeline.tsx index 685f248..0faa549 100644 --- a/core/frontend/src/routes/pipeline.tsx +++ b/core/frontend/src/routes/pipeline.tsx @@ -9,7 +9,7 @@ import { isPublicGenerationError, presentGenerationError } from "../generation-e import type { Notice, Page } from "./route-config"; import { stageOrder, statusPill } from "./stage-config"; import { ConfirmModal, MediaLightbox, TeamModal, useBodyScrollLock } from "../components/overlays"; -import { DEFAULT_BILLING_RATES, estimateCost } from "../components/free-create/constants"; +import { DEFAULT_BILLING_RATES, estimateCost, FC_MODELS, modelLabel } from "../components/free-create/constants"; import { ModelLibrary } from "../components/model-library"; import { ReviewBadge, type ReviewStatus } from "../components/review-badge"; import { @@ -286,6 +286,29 @@ const PIPELINE_RAIL = [ { n: "04", title: "故事板", desc: "生成并确认视频分镜画面" }, { n: "05", title: "视频生成", desc: "按故事板生成视频片段" }, ]; +const OUTPUT_RATIOS = [ + { value: "9:16", label: "9:16 竖屏" }, + { value: "16:9", label: "16:9 横屏" }, + { value: "1:1", label: "1:1 方形" }, + { value: "3:4", label: "3:4 竖版" }, + { value: "4:3", label: "4:3 横版" }, + { value: "21:9", label: "21:9 超宽" }, +]; +const OUTPUT_RESOLUTIONS = [ + { value: "480p", label: "480p 流畅" }, + { value: "720p", label: "720p 高清" }, + { value: "1080p", label: "1080p 超清" }, + { value: "4k", label: "4K 超清" }, +]; +const DEFAULT_VIDEO_MODEL_NAME = "doubao-seedance-2-0-260128"; + +function modelResolutions(config: ModelConfig | undefined) { + const capabilities = (config?.metadata?.capabilities || {}) as Record; + const nested = Array.isArray(capabilities.resolutions) ? capabilities.resolutions : []; + const legacy = Array.isArray(config?.metadata?.resolutions) ? config.metadata.resolutions : []; + return (nested.length ? nested : legacy).map(String); +} + const PIPELINE_HEAD: Record = { 1: { title: "脚本创建", desc: "围绕商品卖点组织镜头脚本，确认后进入下一步内容生产", status: "镜头脚本" }, 2: { title: "资产选择", desc: "准备故事板所需的商品、角色与场景资产，确保后续画面保持一致", status: "资产选择" }, @@ -1155,13 +1178,111 @@ export function PipelinePage(props: { const [chargeConfirm, setChargeConfirm] = useState<"storyboard" | "video" | null>(null); const sbChargeShots = shots.length || sbExpectedShots; const sbChargePoints = sbChargeShots * pts(20); - const defaultVideoModel = (videoModels ?? []).find((m) => m.status === "active") || (videoModels ?? [])[0]; + const videoConfigs = videoModels ?? []; + const defaultVideoModel = videoConfigs.find((m) => m.status === "active" && m.name === DEFAULT_VIDEO_MODEL_NAME) + || videoConfigs.find((m) => m.status === "active") + || videoConfigs[0]; + const [outputAspect, setOutputAspect] = useState(() => project.metadata?.wizard?.aspect_ratio || "9:16"); + const [outputResolution, setOutputResolution] = useState(() => String(project.metadata?.wizard?.resolution || "720p").toLowerCase()); + const [outputModelId, setOutputModelId] = useState(() => project.metadata?.wizard?.video_model_config_id || defaultVideoModel?.id || ""); + const specRef = useRef({ aspect: outputAspect, resolution: outputResolution, modelId: outputModelId }); + useEffect(() => { + const wiz = project.metadata?.wizard; + const next = { + aspect: wiz?.aspect_ratio || "9:16", + resolution: String(wiz?.resolution || "720p").toLowerCase(), + modelId: wiz?.video_model_config_id || defaultVideoModel?.id || "", + }; + specRef.current = next; + setOutputAspect(next.aspect); + setOutputResolution(next.resolution); + setOutputModelId(next.modelId); + }, [project.id]); // eslint-disable-line react-hooks/exhaustive-deps + useEffect(() => { + if (outputModelId || !defaultVideoModel?.id) return; + specRef.current = { ...specRef.current, modelId: defaultVideoModel.id }; + setOutputModelId(defaultVideoModel.id); + }, [defaultVideoModel?.id, outputModelId]); + const outputModel = videoConfigs.find((model) => model.id === outputModelId) || defaultVideoModel; + const supportedResolutions = modelResolutions(outputModel); + useEffect(() => { + if (!supportedResolutions.length || supportedResolutions.includes(outputResolution)) return; + const nextResolution = supportedResolutions.includes("720p") ? "720p" : supportedResolutions[0]; + specRef.current = { ...specRef.current, resolution: nextResolution }; + setOutputResolution(nextResolution); + }, [outputResolution, supportedResolutions]); + function specWizardPatch(aspect: string, resolution: string, modelId: string) { + const model = videoConfigs.find((item) => item.id === modelId) || defaultVideoModel; + return { + aspect_ratio: aspect, + resolution, + video_model_config_id: modelId, + video_model_name: model?.name || "", + video_model_label: FC_MODELS.find((item) => item.name === model?.name)?.label || model?.display_name || modelLabel(model?.name || ""), + }; + } + async function persistOutputSpec() { + const { aspect, resolution, modelId } = specRef.current; + const wizard = { ...(project.metadata?.wizard ?? {}), ...specWizardPatch(aspect, resolution, modelId) }; + await api.updateProject(project.id, { metadata: { ...(project.metadata ?? {}), wizard } }); + } + function changeOutputSpec(patch: { aspect_ratio?: string; resolution?: string; video_model_config_id?: string }) { + let nextAspect = patch.aspect_ratio ?? specRef.current.aspect; + let nextResolution = patch.resolution ?? specRef.current.resolution; + const nextModelId = patch.video_model_config_id ?? specRef.current.modelId; + const nextModel = videoConfigs.find((model) => model.id === nextModelId) || outputModel; + if (patch.video_model_config_id) { + const supported = modelResolutions(nextModel); + if (supported.length && !supported.includes(nextResolution)) { + nextResolution = supported.includes("720p") ? "720p" : supported[0]; + } + } + specRef.current = { aspect: nextAspect, resolution: nextResolution, modelId: nextModelId }; + setOutputAspect(nextAspect); + setOutputResolution(nextResolution); + setOutputModelId(nextModelId); + const wizard = { ...(project.metadata?.wizard ?? {}), ...specWizardPatch(nextAspect, nextResolution, nextModelId) }; + void api.updateProject(project.id, { metadata: { ...(project.metadata ?? {}), wizard } }); + } + const outputSpecSummary = `${FC_MODELS.find((item) => item.name === outputModel?.name)?.label || outputModel?.display_name || modelLabel(outputModel?.name || "") || "视频模型"} · ${outputAspect} · ${outputResolution}`; + const videoPrompt = outputAspect === "1:1" + ? "方形电商短视频,镜头稳定,商品露出清晰,节奏有转化感" + : (outputAspect === "9:16" || outputAspect === "3:4") + ? "竖屏电商短视频,镜头稳定,商品露出清晰,节奏有转化感" + : "横屏电商短视频,镜头稳定,商品露出清晰,节奏有转化感"; + function renderOutputSpecFields() { + return ( +
+ + + +
+ ); + } const videoChargeDurations = segments.length ? segments.map((s) => s.target_duration_seconds || 15) : shots.map((s) => shotSeconds(s)); const videoChargeShots = videoChargeDurations.length; const videoChargePoints = videoChargeDurations.reduce( - (sum, duration) => sum + estimateCost(defaultVideoModel, { ratio: "9:16", resolution: "720p", duration, refs: [] }, billingRates).points, + (sum, duration) => sum + estimateCost(outputModel, { ratio: outputAspect, resolution: outputResolution, duration, refs: [] }, billingRates).points, 0, ); const sbNextLabel = sbAnyImage || sbAnyGenerating @@ -1192,18 +1313,22 @@ export function PipelinePage(props: { setRerunPending((s) => { if (!s.has(segId)) return s; const n = new Set(s); n.delete(segId); return n; }); function submitVideoOptimistic(segId: string, prompt: string) { setRerunPending((s) => new Set(s).add(segId)); - Promise.resolve(onSubmitVideo(segId, prompt)) - .then((res) => { if (res == null) clearRerunPending(segId); }) // 失败(action 返回 null)→ 立即解禁可重试;成功交给下方 effect + void persistOutputSpec() + .catch(() => undefined) + .then(() => Promise.resolve(onSubmitVideo(segId, prompt))) + .then((res) => { if (res == null) clearRerunPending(segId); }) .catch(() => clearRerunPending(segId)); - window.setTimeout(() => clearRerunPending(segId), 20000); // 兜底:异常下也别永久禁用 + window.setTimeout(() => clearRerunPending(segId), 20000); } // 「全部重跑」乐观态:把将被提交的段(非在途)全部立刻标 pending → 每张卡马上转圈,不必等状态回传 function submitAllVideosOptimistic() { const ids = segments.filter((s) => !["running", "queued"].includes(s.status)).map((s) => s.id); if (ids.length === 0) return; setRerunPending((s) => { const n = new Set(s); ids.forEach((id) => n.add(id)); return n; }); - Promise.resolve(onSubmitAllVideos(videoPrompt)) - .then((res) => { if (res == null) ids.forEach(clearRerunPending); }) // 失败 → 解禁;成功交给上面的 effect 逐个摘除 + void persistOutputSpec() + .catch(() => undefined) + .then(() => Promise.resolve(onSubmitAllVideos(videoPrompt))) + .then((res) => { if (res == null) ids.forEach(clearRerunPending); }) .catch(() => ids.forEach(clearRerunPending)); ids.forEach((id) => window.setTimeout(() => clearRerunPending(id), 20000)); } @@ -1950,9 +2075,11 @@ export function PipelinePage(props: { const [storyboardPrompt, setStoryboardPrompt] = useState(sbSavedPrompt || SB_PROMPT_DEFAULT); const startStoryboardGeneration = () => { setSbGenerating(true); - void Promise.resolve(onGenerateStoryboard(storyboardPrompt || SB_PROMPT_DEFAULT)).finally(() => setSbGenerating(false)); + void persistOutputSpec() + .catch(() => undefined) + .then(() => onGenerateStoryboard(storyboardPrompt || SB_PROMPT_DEFAULT)) + .finally(() => setSbGenerating(false)); }; - const videoPrompt = "竖屏电商短视频,镜头稳定,商品露出清晰,节奏有转化感"; const canExport = project.video_segments.length > 0 && project.video_segments.every((segment) => Boolean(segment.adopted_version)); // ── Stage 5 · 真实视频播放器:时间轴 clips 当作播放列表,逐段播真实视频文件 ── @@ -3708,6 +3835,11 @@ export function PipelinePage(props: {