fix(extract): 实体提取改异步(worker+轮询),根治 502 + 刷新丢 loading + 重复扣费

问题:进资产页点"提取人物/场景"——同步死等豆包思考模型(数十秒)→ 撞网关超时
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) <noreply@anthropic.com>

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,
+            ),
+        ),
+    ]

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"

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]:

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 层不被占住。

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):

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):