From c8cb6d3d51b13f8ac93b88353529fdce3f5c570f Mon Sep 17 00:00:00 2001 From: seaislee1209 Date: Mon, 22 Jun 2026 23:15:11 +0800 Subject: [PATCH] =?UTF-8?q?fix(extract):=20=E5=AE=9E=E4=BD=93=E6=8F=90?= =?UTF-8?q?=E5=8F=96=E6=94=B9=E5=BC=82=E6=AD=A5(worker+=E8=BD=AE=E8=AF=A2)?= =?UTF-8?q?,=E6=A0=B9=E6=B2=BB=20502=20+=20=E5=88=B7=E6=96=B0=E4=B8=A2=20l?= =?UTF-8?q?oading=20+=20=E9=87=8D=E5=A4=8D=E6=89=A3=E8=B4=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 问题:进资产页点"提取人物/场景"——同步死等豆包思考模型(数十秒)→ 撞网关超时 502;"提取中"loading 是纯前端内存态,刷新即丢;且无幂等,刷新后重点会重复扣费。 改动(对齐全站异步三件套:出图/三视图/故事板同款): - 后端 submit_extract_entities:Web 只建 RESERVED 任务 + 预留额度秒回(不再 502); 已有在途提取则复用,绝不二次预扣(防刷新后重点 / 并发重复扣费)。 - 后端 run_extract_entities_task(Celery worker):流式调豆包 + 解析 + 落库 + 扣费; 失败退费并把可读错误记进 task.error_message。 - 新增 AITask.Type.entity_extraction(迁移 0011)+ extract-status 端点(轮询进度/成败)。 - 前端 runExtract 改提交+轮询;进资产页自动认领在途提取,刷新后 loading 自己回来。 验收:完整 Django 套件 149/149 全绿(含新增 4 测:异步落库+计费一次 / content空 回退 reasoning / 在途防重复扣费 / 状态端点成败透出);tsc + vite build 全绿; makemigrations --check 无遗漏。 Co-Authored-By: Claude Opus 4.8 (1M context) --- .../0011_aitask_entity_extraction_type.py | 31 ++++ core/backend/apps/ai/models.py | 1 + core/backend/apps/ai/services.py | 157 ++++++++++++------ core/backend/apps/ai/tasks.py | 10 ++ core/backend/apps/projects/tests.py | 49 +++++- core/backend/apps/projects/views.py | 37 ++++- core/frontend/src/api.ts | 17 +- core/frontend/src/routes/pipeline.tsx | 88 ++++++++-- 8 files changed, 304 insertions(+), 86 deletions(-) create mode 100644 core/backend/apps/ai/migrations/0011_aitask_entity_extraction_type.py diff --git a/core/backend/apps/ai/migrations/0011_aitask_entity_extraction_type.py b/core/backend/apps/ai/migrations/0011_aitask_entity_extraction_type.py new file mode 100644 index 0000000..b3f0a69 --- /dev/null +++ b/core/backend/apps/ai/migrations/0011_aitask_entity_extraction_type.py @@ -0,0 +1,31 @@ +# 实体提取改异步:新增 AITask 任务类型 entity_extraction(供在途查询/计费区分,与 script_optimization 分开)。 +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ("ai", "0010_modelconfig_is_default"), + ] + + operations = [ + migrations.AlterField( + model_name="aitask", + name="task_type", + field=models.CharField( + choices=[ + ("script_generation", "Script Generation"), + ("script_optimization", "Script Optimization"), + ("entity_extraction", "Entity Extraction"), + ("product_image", "Product Image"), + ("person_image", "Person Image"), + ("scene_image", "Scene Image"), + ("storyboard", "Storyboard"), + ("video_segment", "Video Segment"), + ("voiceover", "Voiceover"), + ("export", "Export"), + ], + max_length=48, + ), + ), + ] diff --git a/core/backend/apps/ai/models.py b/core/backend/apps/ai/models.py index 9fe4f5c..4d6a0ee 100644 --- a/core/backend/apps/ai/models.py +++ b/core/backend/apps/ai/models.py @@ -57,6 +57,7 @@ class AITask(TeamOwnedModel): class Type(models.TextChoices): SCRIPT_GENERATION = "script_generation", "Script Generation" SCRIPT_OPTIMIZATION = "script_optimization", "Script Optimization" + ENTITY_EXTRACTION = "entity_extraction", "Entity Extraction" PRODUCT_IMAGE = "product_image", "Product Image" PERSON_IMAGE = "person_image", "Person Image" SCENE_IMAGE = "scene_image", "Scene Image" diff --git a/core/backend/apps/ai/services.py b/core/backend/apps/ai/services.py index dd57759..f05fca9 100644 --- a/core/backend/apps/ai/services.py +++ b/core/backend/apps/ai/services.py @@ -344,16 +344,37 @@ def _collect_extract_text(provider, model_config, messages) -> tuple[str, dict]: return text, payload -def extract_entities_for_project(*, project, user) -> dict: - """独立实体提取步(剧本定稿后、进资产阶段时跑)。读已定稿(或最新)脚本 → 调 ecommerce-entity-extract - skill(一个文本模型)→ 产出角色 / 场景 entities + 每镜 entity_refs。**不提商品**(商品由参考图层用真实 - 主图 / 三视图注入)。落库:覆盖 project.metadata(cast/cast_prompts/scenes/scene_prompts/script_entities) - + 回填每条 ScriptSegment.entity_refs。提取是实体的唯一权威来源,覆盖脚本生成期可能产出的旧实体。 +# 在途状态:据此判「已有提取在跑」(提交侧防重复扣费 + 前端刷新后重建 loading) +_EXTRACT_INFLIGHT = ( + AITask.Status.CREATED, + AITask.Status.RESERVED, + AITask.Status.SUBMITTED, + AITask.Status.POLLING, +) - 走完整 AITask + 计费闭环(reserve→charge/release,任务类型 script_optimization)。失败抛 ValueError, - 由端点转成用户可读错误(与脚本生成的 best-effort 不同:这是用户主动点的步骤,要让他知道成没成)。 - 返回 {entities, segments, cast, cast_prompts, scenes, scene_prompts}。""" + +def get_inflight_extraction(project): + """本项目正在跑的实体提取任务(最近一条优先);无则 None。""" + return ( + AITask.objects.filter( + project=project, task_type=AITask.Type.ENTITY_EXTRACTION, status__in=_EXTRACT_INFLIGHT + ) + .order_by("-created_at") + .first() + ) + + +def submit_extract_entities(*, project, user) -> AITask: + """提交独立实体提取步(**异步**)。读已定稿(或最新)脚本 → 校验 + 建 RESERVED 任务 + 预留额度(秒级), + 慢活(豆包思考模型流式抽取,可达数十秒)交给 Celery worker(run_extract_entities_task)跑。 + + 这样 Web 层(gunicorn/nginx)不被数十秒的模型请求占住 → 不再 502。 + **防重复扣费**:本项目已有在途提取任务时直接复用它,不新建/不二次预扣 + (用户刷新后手痒重点、或并发点击都安全 —— 提取是实体唯一权威来源,本就只需跑一次)。 + 校验失败(无脚本/无分镜/无模型/额度不足)抛 ValueError 由端点转 400;运行期失败由 worker 记进 + task.error_message,前端轮询 extract-status 读取。返回 AITask。""" from apps.projects.models import ScriptVersion + from apps.ai.tasks import extract_entities_task script = ( ScriptVersion.objects.filter(project=project, is_adopted=True).order_by("-created_at").first() @@ -364,6 +385,11 @@ def extract_entities_for_project(*, project, user) -> dict: segments = list(script.segments.order_by("sort_order")) if not segments: raise ValueError("脚本没有分镜,无法提取") + + inflight = get_inflight_extraction(project) + if inflight is not None: + return inflight # 已有提取在跑:复用,绝不二次预扣 / 重复出活 + model_config = _resolve_extract_model_config() if model_config is None: raise ValueError("没有可用的文本模型") @@ -388,20 +414,67 @@ def extract_entities_for_project(*, project, user) -> dict: task = create_ai_task( project=project, user=user, - task_type=AITask.Type.SCRIPT_OPTIMIZATION, + task_type=AITask.Type.ENTITY_EXTRACTION, model_config=model_config, - request_payload={"model": model_config.name, "endpoint": model_config.endpoint, "messages": messages}, + request_payload={ + "model": model_config.name, + "endpoint": model_config.endpoint, + "messages": messages, + "script_id": str(script.id), + }, ) except Exception as exc: # 余额不足等预扣失败 raise ValueError("额度不足,无法提取(请先充值)") from exc - reservation = task.credit_reservation + extract_entities_task.delay(str(task.id)) + return task + + +def run_extract_entities_task(*, task_id: str) -> None: + """Celery worker 内执行实体提取慢活:流式调模型 → 解析 JSON → 落库(覆盖 project.metadata 的 + cast/scenes/*_prompts/script_entities + 回填每镜 entity_refs + entities_extracted 标记)并扣费; + 失败退费并把可读错误记进 task.error_message(前端轮询 extract-status 读取)。 + 幂等:只处理 RESERVED 任务,重复投递不会二次出活、二次扣费。""" + from apps.projects.models import ScriptVersion + from apps.ai.script_agent import _map_entities_to_project_metadata + + task = AITask.objects.select_related("team", "created_by", "project", "model_config").filter(id=task_id).first() + if task is None or task.status != AITask.Status.RESERVED: + return + project = task.project + payload = task.request_payload or {} + messages = payload.get("messages") or [] + model_config = task.model_config + reservation = task.credit_reservation try: task.status = AITask.Status.SUBMITTED task.submitted_at = timezone.now() task.save(update_fields=["status", "submitted_at", "updated_at"]) + provider = build_provider(model_config) text, response = _collect_extract_text(provider, model_config, messages) + + match = re.search(r"\{.*\}", text, re.DOTALL) + if not match: + raise ValueError("提取结果解析失败(模型未返回有效 JSON),请重试") + try: + data = json.loads(match.group(0)) + except json.JSONDecodeError as exc: # 抓出来给用户可读话术,别把裸解析报错塞进 error_message + raise ValueError("提取结果解析失败,请重试") from exc + entities = _normalize_extracted_entities(data.get("entities")) + if not entities: + raise ValueError("没有从脚本里识别到角色 / 场景,可调整脚本后重试") + seg_refs = _normalize_extracted_segment_refs(data.get("segments"), {e["id"] for e in entities}) + + script_id = payload.get("script_id") + script = ScriptVersion.objects.filter(id=script_id).first() if script_id else None + if script is None: # 极端兜底:payload 里 script 没了就回落项目当前定稿稿 + script = ( + ScriptVersion.objects.filter(project=project, is_adopted=True).order_by("-created_at").first() + or ScriptVersion.objects.filter(project=project).order_by("-created_at").first() + ) + segments = list(script.segments.order_by("sort_order")) if script is not None else [] + with transaction.atomic(): task.status = AITask.Status.SUCCEEDED task.response_payload = response @@ -409,53 +482,29 @@ def extract_entities_for_project(*, project, user) -> dict: task.completed_at = timezone.now() task.save(update_fields=["status", "response_payload", "actual_cost", "completed_at", "updated_at"]) charge_reserved_credit(reservation=reservation, actual_amount=task.actual_cost) - except Exception as exc: + + # 落库:覆盖 project.metadata + 回填每镜 entity_refs(提取为实体唯一权威来源)。 + # entities_extracted 标记:前端资产页提取闸门据此显隐(没走过 → 盖蒙版露按钮;走过 → 露卡片)。 + _map_entities_to_project_metadata(project, entities) + md_flag = dict(project.metadata or {}) + md_flag["entities_extracted"] = True + project.metadata = md_flag + project.save(update_fields=["metadata", "updated_at"]) + refs_by_index = {r["index"]: r["entity_refs"] for r in seg_refs} + for i, seg in enumerate(segments): + new_refs = refs_by_index.get(i, []) + if (seg.entity_refs or []) != new_refs: + seg.entity_refs = new_refs + seg.save(update_fields=["entity_refs", "updated_at"]) + except Exception as exc: # noqa: BLE001 — 失败退费并把错误记进 AITask 供前端轮询;不向上抛(避免 celery 重试二次扣费) + # ValueError 是我们给用户写好的可读话术(解析失败 / 没识别到角色等);其余(网络/模型异常)给通用话术。 + msg = str(exc) if isinstance(exc, ValueError) else "提取调用失败,请重试" with transaction.atomic(): task.status = AITask.Status.FAILED - task.error_message = str(exc) + task.error_message = msg task.completed_at = timezone.now() task.save(update_fields=["status", "error_message", "completed_at", "updated_at"]) - release_credit(reservation=reservation, reason=str(exc)) - raise ValueError("提取调用失败,请重试") from exc - - match = re.search(r"\{.*\}", text, re.DOTALL) - if not match: - raise ValueError("提取结果解析失败(模型未返回有效 JSON),请重试") - try: - data = json.loads(match.group(0)) - except Exception as exc: - raise ValueError("提取结果解析失败,请重试") from exc - entities = _normalize_extracted_entities(data.get("entities")) - if not entities: - raise ValueError("没有从脚本里识别到角色 / 场景,可调整脚本后重试") - seg_refs = _normalize_extracted_segment_refs(data.get("segments"), {e["id"] for e in entities}) - - # 落库:覆盖 project.metadata + 回填 adopted 脚本每镜 entity_refs(提取为唯一权威来源) - from apps.ai.script_agent import _map_entities_to_project_metadata - - _map_entities_to_project_metadata(project, entities) - # 标记「已走过正式提取步」:脚本生成期也可能吐过 entities(不稳的旧来源),不算正式提取; - # 前端资产页提取闸门据此标记显隐(没走过 → 盖蒙版露三按钮;走过 → 露卡片)。 - md_flag = dict(project.metadata or {}) - md_flag["entities_extracted"] = True - project.metadata = md_flag - project.save(update_fields=["metadata", "updated_at"]) - refs_by_index = {r["index"]: r["entity_refs"] for r in seg_refs} - for i, seg in enumerate(segments): - new_refs = refs_by_index.get(i, []) - if (seg.entity_refs or []) != new_refs: - seg.entity_refs = new_refs - seg.save(update_fields=["entity_refs", "updated_at"]) - - md = project.metadata or {} - return { - "entities": entities, - "segments": seg_refs, - "cast": md.get("cast", []), - "cast_prompts": md.get("cast_prompts", {}), - "scenes": md.get("scenes", []), - "scene_prompts": md.get("scene_prompts", {}), - } + release_credit(reservation=reservation, reason=msg) def split_script_into_segments(content: str, count: int = 4) -> list[str]: diff --git a/core/backend/apps/ai/tasks.py b/core/backend/apps/ai/tasks.py index c01db60..07cef5a 100644 --- a/core/backend/apps/ai/tasks.py +++ b/core/backend/apps/ai/tasks.py @@ -21,6 +21,16 @@ def generate_standalone_image_task(self, task_id: str) -> str: return task_id +@app.task(bind=True, max_retries=0) +def extract_entities_task(self, task_id: str) -> str: + """实体提取的慢活(豆包思考模型流式,可达数十秒)在 worker 内跑,Web 层秒回不被占住、不再 502。 + 幂等且失败自退费(见 run_extract_entities_task),故 max_retries=0,不向上抛重试。""" + from apps.ai.services import run_extract_entities_task + + run_extract_entities_task(task_id=task_id) + return task_id + + @app.task(bind=True, max_retries=0) def generate_base_asset_task(self, task_id: str) -> str: """基础资产(商品/人物/场景立绘)的慢出图在 worker 内跑,Web 层不被占住。 diff --git a/core/backend/apps/projects/tests.py b/core/backend/apps/projects/tests.py index 9dc4796..88f78fb 100644 --- a/core/backend/apps/projects/tests.py +++ b/core/backend/apps/projects/tests.py @@ -4,7 +4,7 @@ from unittest.mock import patch from rest_framework.test import APIClient from apps.accounts.models import Team, TeamMember, User -from apps.ai.models import ModelConfig, ModelProvider +from apps.ai.models import AITask, ModelConfig, ModelProvider from apps.assets.models import Asset from apps.billing.models import CreditAccount, CreditLedger from apps.products.models import Product @@ -205,6 +205,53 @@ class ProjectApiTests(TestCase): project.refresh_from_db() self.assertIn("男主", project.metadata.get("cast", [])) + @patch("apps.ai.services.VolcanoArkProvider") + def test_extract_entities_inflight_guard_reuses_task(self, provider_cls): + """已有在途提取任务时再次提交 → 复用同一任务,不新建、不二次预扣、不重跑模型 + (用户刷新后手痒重点 / 并发点击都不会重复扣费)。""" + from apps.ai.services import submit_extract_entities + + project = Project.objects.create(team=self.team, created_by=self.user, product=self.product, name="P3") + script = ScriptVersion.objects.create( + project=project, title="脚本", content="x", is_adopted=True, metadata={"entities": []}, + ) + ScriptSegment.objects.create(script_version=script, sort_order=0, narration="旧0", visual_prompt="画0") + inflight = AITask.objects.create( + team=self.team, created_by=self.user, project=project, model_config=self.model, + task_type=AITask.Type.ENTITY_EXTRACTION, status=AITask.Status.SUBMITTED, + idempotency_key="entity_extraction:inflight:1", + ) + before = AITask.objects.filter(project=project, task_type=AITask.Type.ENTITY_EXTRACTION).count() + + task = submit_extract_entities(project=project, user=self.user) + self.assertEqual(str(task.id), str(inflight.id)) # 复用在途任务 + self.assertEqual( # 没新建任务 + AITask.objects.filter(project=project, task_type=AITask.Type.ENTITY_EXTRACTION).count(), before + ) + provider_cls.return_value.chat_completion_stream.assert_not_called() # 没重跑模型 + # 没二次预扣额度(复用直接 return,根本没走到 create_ai_task) + self.assertEqual(CreditLedger.objects.filter(team=self.team, ledger_type=CreditLedger.Type.RESERVE).count(), 0) + + def test_extract_status_reports_running_then_failure(self): + """extract-status:在途→running=True;失败→running=False 且把可读错误透给前端(刷新后也能看到失败原因)。""" + project = Project.objects.create(team=self.team, created_by=self.user, product=self.product, name="P4") + t = AITask.objects.create( + team=self.team, created_by=self.user, project=project, model_config=self.model, + task_type=AITask.Type.ENTITY_EXTRACTION, status=AITask.Status.SUBMITTED, + idempotency_key="entity_extraction:status:1", + ) + res = self.client.get(f"/api/projects/{project.id}/extract-status/") + self.assertEqual(res.status_code, 200) + self.assertTrue(res.data["running"]) + + t.status = AITask.Status.FAILED + t.error_message = "提取结果解析失败(模型未返回有效 JSON),请重试" + t.save(update_fields=["status", "error_message"]) + res = self.client.get(f"/api/projects/{project.id}/extract-status/") + self.assertFalse(res.data["running"]) + self.assertEqual(res.data["status"], "failed") + self.assertIn("未返回有效 JSON", res.data["error"]) + @patch("apps.ai.services._store_generated_media") @patch("apps.ai.services.get_image_provider") def test_generate_base_asset_stores_tag_label(self, get_provider, store_media): diff --git a/core/backend/apps/projects/views.py b/core/backend/apps/projects/views.py index 9c772c5..1140e83 100644 --- a/core/backend/apps/projects/views.py +++ b/core/backend/apps/projects/views.py @@ -19,13 +19,14 @@ from apps.ai.services import ( DEFAULT_VOICEOVER_VOICE, VOICEOVER_VOICES, create_export_job, - extract_entities_for_project, generate_base_asset, generate_person_triview, generate_storyboard_frame, get_default_model, + get_inflight_extraction, poll_video_segment, regenerate_script_segment, + submit_extract_entities, submit_storyboard, submit_video_segment, synthesize_project_voiceover, @@ -294,15 +295,37 @@ class ProjectViewSet(TeamScopedViewSetMixin, ModelViewSet): @action(detail=True, methods=["post"], url_path="extract-entities") def extract_entities_action(self, request, pk=None): - """从已定稿脚本提取角色 / 场景实体(独立步,进资产阶段时调)。AI 读剧本 → 角色 / 场景 + 每镜 refs, - 落库覆盖 project.metadata + 每镜 entity_refs;**商品不提**(参考图层用真实主图 / 三视图无条件带上)。 - 前端进资产页蒙版上「只提取 / 提取+生成」按钮点的就是它(生成由前端按返回实体循环调 generate-base-asset)。""" + """从已定稿脚本提取角色 / 场景实体(独立步,进资产阶段时调)——**异步**:豆包思考模型流式抽取较慢, + Web 层只建任务秒回(不再 502),慢活交 worker(run_extract_entities_task)。前端拿 task_id 轮询 + extract-status 取结果。已有在途提取则复用(防刷新后重点重复扣费)。落库覆盖 project.metadata + + 每镜 entity_refs;**商品不提**(参考图层用真实主图)。""" + require_worker() # 异步提取依赖 worker 兜底执行,没 worker 直接拒绝(否则任务永远 RESERVED) project = self.get_object() try: - result = extract_entities_for_project(project=project, user=request.user) - except ValueError as exc: # 无脚本 / 无模型 / 额度不足 / 解析失败 等,立即反馈 + task = submit_extract_entities(project=project, user=request.user) + except ValueError as exc: # 无脚本 / 无分镜 / 无模型 / 额度不足,立即反馈 return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) - return Response(result) + return Response({"task_id": str(task.id), "status": task.status}) + + @action(detail=True, methods=["get"], url_path="extract-status") + def extract_status_action(self, request, pk=None): + """实体提取进度(异步)。前端进资产页 / 点提取后据此轮询 + 刷新后重建「提取中」loading: + running=有在途任务;否则读最近一条提取任务的成败。成功时回带 entities(供「提取+生成」继续出参考图)。""" + project = self.get_object() + inflight = get_inflight_extraction(project) + if inflight is not None: + return Response({"running": True, "status": inflight.status, "error": "", "entities": []}) + last = ( + AITask.objects.filter(project=project, task_type=AITask.Type.ENTITY_EXTRACTION) + .order_by("-created_at") + .first() + ) + if last is None: + return Response({"running": False, "status": "", "error": "", "entities": []}) + entities = (project.metadata or {}).get("script_entities", []) if last.status == AITask.Status.SUCCEEDED else [] + return Response( + {"running": False, "status": last.status, "error": last.error_message or "", "entities": entities} + ) @action(detail=True, methods=["post"], url_path="generate-base-asset") def generate_base_asset_action(self, request, pk=None): diff --git a/core/frontend/src/api.ts b/core/frontend/src/api.ts index d8b822e..c7d86a4 100644 --- a/core/frontend/src/api.ts +++ b/core/frontend/src/api.ts @@ -327,16 +327,19 @@ export const api = { body: JSON.stringify({ script_version_id }) }); }, - // 独立实体提取:读已定稿脚本 → AI 识别角色/场景 + 每镜 refs,落库覆盖 metadata。商品不提(参考图层用真实主图)。 + // 独立实体提取(**异步**):只提交任务、秒回 task_id(慢活在 worker 跑,不再 502);已有在途则复用(防重复扣费)。 + // 进度/结果用 extractStatus 轮询。商品不提(参考图层用真实主图)。 extractEntities(projectId: string) { + return request<{ task_id: string; status: string }>(`/api/projects/${projectId}/extract-entities/`, { method: "POST" }); + }, + // 实体提取进度:running=有在途任务(刷新后据此重建 loading);否则读最近一次成败,成功回带 entities 供「提取+生成」继续出图。 + extractStatus(projectId: string) { return request<{ + running: boolean; + status: string; + error: string; entities: Array<{ id: string; type: "character" | "scene"; name: string; visual_prompt: string; ref_index: number }>; - segments: Array<{ index: number; entity_refs: string[] }>; - cast: string[]; - cast_prompts: Record; - scenes: string[]; - scene_prompts: Record; - }>(`/api/projects/${projectId}/extract-entities/`, { method: "POST" }); + }>(`/api/projects/${projectId}/extract-status/`); }, updateScriptSegment(projectId: string, payload: { segment_id: string; narration?: string; visual_prompt?: string; duration_seconds?: number }) { return request(`/api/projects/${projectId}/update-script-segment/`, { method: "POST", body: JSON.stringify(payload) }); diff --git a/core/frontend/src/routes/pipeline.tsx b/core/frontend/src/routes/pipeline.tsx index d48a0d4..d34fd9c 100644 --- a/core/frontend/src/routes/pipeline.tsx +++ b/core/frontend/src/routes/pipeline.tsx @@ -680,6 +680,53 @@ export function PipelinePage(props: { const [extractState, setExtractState] = useState<"idle" | "running">("idle"); const [extractMsg, setExtractMsg] = useState(""); const [extractErr, setExtractErr] = useState(""); + const extractPollRef = useRef(0); // 提取轮询定时器句柄 + const extractStartedRef = useRef(false); // 是否已有一条轮询在跑(防认领与手点双轮询) + type ExtractEntity = { id: string; type: "character" | "scene"; name: string; visual_prompt: string; ref_index: number }; + // 提取完成后(mode≠only)按返回实体循环出参考图(立绘/三视图/场景图) + async function runGenForEntities(entities: ExtractEntity[], mode: "gen" | "full") { + setExtractMsg("已认出角色 / 场景,正在生成参考图…"); + for (const e of entities) { + const bk = `seed:${e.type === "character" ? "person" : "scene"}:${e.name}`; + if (e.type === "character") { + if (mode === "full") await genPersonWithTri(e.visual_prompt, e.name, bk); // 立绘 + 三视图 + else await genBaseAsset("person", e.visual_prompt, e.name, bk); // 立绘 + } else { + await genBaseAsset("scene", e.visual_prompt, e.name, bk); // 场景图 + } + } + await onRefreshProject(); + } + // 轮询提取进度直到结束:成功→刷新(可选继续出图)→idle;失败→显错→idle。 + // mode=null 表示「刷新后认领」(只恢复 loading,不自动出图 —— 出图态由 pending-assets 单独认领)。 + function pollExtractUntilDone(mode: "only" | "gen" | "full" | null) { + window.clearInterval(extractPollRef.current); + extractStartedRef.current = true; + const finish = () => { window.clearInterval(extractPollRef.current); extractStartedRef.current = false; }; + const tick = async () => { + try { + const s = await api.extractStatus(project.id); + if (s.running) return; // 还在跑,下一轮再看 + finish(); + if (s.status === "succeeded") { + await onRefreshProject(); // 拉回 metadata(cast/scenes/script_entities)+ 回填的每镜 refs + if ((mode === "gen" || mode === "full") && (s.entities?.length ?? 0) > 0) { + await runGenForEntities(s.entities, mode); + } + setExtractState("idle"); setExtractMsg(""); // 闸门据 entities_extracted 已落库 → 渲染层自动隐藏 + } else if (s.status === "failed") { + setExtractState("idle"); setExtractMsg(""); + setExtractErr(s.error || "提取失败,请重试"); + } else { + setExtractState("idle"); setExtractMsg(""); // 既不在途也无成败(理论不至于):收尾,别把 loading 永久挂住 + } + } catch { + /* 网络抖动:忽略,下一轮再试 */ + } + }; + extractPollRef.current = window.setInterval(() => { void tick(); }, 3000); + void tick(); + } // only=只提取关键词(不出图) / gen=提取+生成角色场景(立绘+场景图) / full=提取+全套(立绘+三视图+场景图) async function runExtract(mode: "only" | "gen" | "full") { if (extractState === "running") return; @@ -687,23 +734,8 @@ export function PipelinePage(props: { setExtractState("running"); setExtractMsg("正在从剧本认出角色 / 场景…"); try { - const res = await api.extractEntities(project.id); - await onRefreshProject(); // 拉回 metadata(cast/scenes/script_entities)+ 回填的每镜 refs - if (mode !== "only") { - setExtractMsg("已认出角色 / 场景,正在生成参考图…"); - for (const e of res.entities || []) { - const bk = `seed:${e.type === "character" ? "person" : "scene"}:${e.name}`; - if (e.type === "character") { - if (mode === "full") await genPersonWithTri(e.visual_prompt, e.name, bk); // 立绘 + 三视图 - else await genBaseAsset("person", e.visual_prompt, e.name, bk); // 立绘 - } else { - await genBaseAsset("scene", e.visual_prompt, e.name, bk); // 场景图 - } - } - await onRefreshProject(); - } - setExtractState("idle"); // 闸门据 metadata.script_entities 已落库 → 渲染层自动隐藏 - setExtractMsg(""); + await api.extractEntities(project.id); // 异步提交(已有在途则后端复用,不重复扣费),慢活在 worker 跑 + pollExtractUntilDone(mode); // 轮询直到 worker 跑完 } catch (err) { setExtractState("idle"); setExtractMsg(""); @@ -2171,6 +2203,28 @@ export function PipelinePage(props: { return () => { stopped = true; window.clearTimeout(timer); }; }, [activeDot, viewStage, project.id, onRefreshProject]); + // 流程步骤4 · 刷新后认领在途「提取」:内存态 extractState 一刷新就丢,据后端 extract-status 重建「提取中」loading + // (和上面出图占位卡同思路)。已有轮询在跑则不重复起。进资产趴时检查一次。 + useEffect(() => { + if (activeDot !== 2 && viewStage !== 2) return; + let cancelled = false; + (async () => { + try { + const s = await api.extractStatus(project.id); + if (cancelled || !s.running || extractStartedRef.current) return; + setExtractState("running"); + setExtractMsg("正在从剧本认出角色 / 场景…"); + pollExtractUntilDone(null); // 认领:mode 未知,只恢复 loading,完成后刷新即见卡片 + } catch { + /* 忽略,无在途就不重建 */ + } + })(); + return () => { cancelled = true; }; + // eslint-disable-next-line react-hooks/exhaustive-deps + }, [activeDot, viewStage, project.id]); + // 卸载/离开页面:停掉提取轮询,别泄漏定时器 + useEffect(() => () => window.clearInterval(extractPollRef.current), []); + function goStage(n: number) { setViewStage(n); setNavigated(true);