完善视频提炼和视频复刻完善提交测试

2026-08-28 10:15:15 +08:00
parent 52c5cdae1d
commit 849b14a0f6
33 changed files with 1475 additions and 208 deletions
@@ -59,6 +59,14 @@ class VideoDigestError(ValueError):
    """用户可见的失败(文件不合格 / ffmpeg 读不动),一律 400。"""


+class VideoDigestInProgress(VideoDigestError):
+    """本团队已有提炼在跑。视图转 409,并把在跑的那条带回去让前端接上。"""
+
+    def __init__(self, job: dict):
+        super().__init__("已有一个视频正在提炼中，完成或取消后才能再提交")
+        self.job = job
+
+
@dataclass(frozen=True)
 class VideoFrame:
    at_seconds: int
@@ -923,6 +931,14 @@ def submit_team_digest(*, team, user, upload=None, reuse_task_id=None, model_con
    from apps.billing.pricing import quote_video_digest
    from apps.billing.services.ledger import reserve_credit

+    # 单飞闸:一个团队同时只允许一条提炼在跑。
+    # 刷新页面时前端要异步拉状态,这个空档用户很容易以为「没任务」再传一次 —— 前端加载态
+    # 只是治标,真正兜底在这里。先跑过期回收,死任务不会把人永久锁住。
+    expire_stale_team_digests(team=team)
+    running = get_inflight_team_digest(team=team)
+    if running:
+        raise VideoDigestInProgress(running)
+
    model_config = resolve_digest_model_config(preferred_id=model_config_id)
    if model_config is None:
        raise VideoDigestError("视频提炼需要 Gemini 3.1 Pro(会看图),当前没有启用,请联系管理员")
@@ -1359,6 +1375,9 @@ def digest_asset_video(*, asset, task, model_config_id=None) -> tuple[str, dict]
    复刻本来就是一次收费,拆解是它的内部工序。模型调用的审计仍记在传入的复刻
    task 上(AIModelAttempt),出问题查得到。
    """
+    import time
+
+    from apps.ai.models import AIModelAttempt
    from apps.ai.services import _collect_extract_text, get_text_provider

    primary = asset.files.filter(is_primary=True).first() or asset.files.first()
@@ -1408,7 +1427,12 @@ def digest_asset_video(*, asset, task, model_config_id=None) -> tuple[str, dict]
        expected_frames = len(frames) or (24 if video else 0)
        digest = ""
        last_error: Exception | None = None
-        for attempt in range(1, DIGEST_MAX_ATTEMPTS + 1):
+        for attempt_no in range(1, DIGEST_MAX_ATTEMPTS + 1):
+            # 这一步不走 execute_model_call(它要求 task.model_config 就是本次主模型,
+            # 而复刻任务的主模型是 Seedance),所以尝试记录得自己落 —— 不落的话
+            # 运营后台的任务详情看不到这次 Gemini 调用,出问题无从查起。
+            record = _open_digest_attempt(task, model_config, attempt_no, duration, expected_frames, video)
+            started = time.monotonic()
            try:
                text, _payload = _collect_extract_text(
                    provider,
@@ -1418,12 +1442,19 @@ def digest_asset_video(*, asset, task, model_config_id=None) -> tuple[str, dict]
                    extra_body={"max_tokens": DIGEST_MAX_TOKENS},
                )
                digest = validate_digest_text(text, duration=duration, frame_count=expected_frames)
+                _close_digest_attempt(
+                    record, AIModelAttempt.Status.SUCCEEDED, started,
+                    summary={"chars": len(digest), "shots": shot_count(digest)},
+                )
                break
            except Exception as exc:  # noqa: BLE001 — 空文/废稿/网络抖动都值得再来一次
                last_error = exc
+                _close_digest_attempt(
+                    record, AIModelAttempt.Status.FAILED, started, error=str(exc),
+                )
                logger.warning(
                    "replace digest attempt %s/%s failed for task %s: %s",
-                    attempt, DIGEST_MAX_ATTEMPTS, task.id, exc,
+                    attempt_no, DIGEST_MAX_ATTEMPTS, task.id, exc,
                )
        if not digest:
            raise VideoDigestError(f"参考视频拆解失败：{last_error}")
@@ -1439,3 +1470,73 @@ def digest_asset_video(*, asset, task, model_config_id=None) -> tuple[str, dict]
    finally:
        if local_path:
            Path(local_path).unlink(missing_ok=True)
+
+
+def _open_digest_attempt(task, model_config, attempt_no: int, duration: float, frames: int, video):
+    """给复刻任务补一条「提炼」尝试记录,让运营后台的尝试链能看到这次 Gemini 调用。
+
+    审计失败绝不能拖垮拆解本身,所以整段吞异常、返回 None。
+    """
+    from django.db import transaction
+    from django.db.models import Max
+    from django.utils import timezone
+
+    from apps.ai.models import AIModelAttempt, AITask
+
+    try:
+        with transaction.atomic():
+            AITask.objects.select_for_update().only("id").get(pk=task.pk)
+            sequence = (
+                AIModelAttempt.objects.filter(task_id=task.pk).aggregate(m=Max("sequence"))["m"] or 0
+            ) + 1
+            provider = model_config.provider
+            return AIModelAttempt.objects.create(
+                task_id=task.pk,
+                sequence=sequence,
+                provider=provider,
+                model_config=model_config,
+                provider_name=provider.name,
+                provider_display_name=provider.display_name,
+                model_name=model_config.name,
+                model_display_name=model_config.display_name,
+                public_model_name=model_config.display_name or model_config.name,
+                capability="text",
+                operation="video_digest",
+                status=AIModelAttempt.Status.STARTED,
+                is_retry=attempt_no > 1,
+                started_at=timezone.now(),
+                request_summary={
+                    "for": "video_replace",
+                    "duration_seconds": round(float(duration or 0), 2),
+                    "frame_count": frames,
+                    "input": "native_video" if video is not None else "frames",
+                },
+            )
+    except Exception:  # noqa: BLE001 — 审计写失败不影响出片
+        logger.warning("video replace digest attempt record failed", exc_info=True)
+        return None
+
+
+def _close_digest_attempt(record, status, started_at: float, *, summary: dict | None = None, error: str = ""):
+    import time
+
+    from django.utils import timezone
+
+    if record is None:
+        return
+    try:
+        record.status = status
+        record.finished_at = timezone.now()
+        record.duration_ms = max(0, round((time.monotonic() - started_at) * 1000))
+        if summary:
+            record.response_summary = summary
+        if error:
+            record.error_type = "processing_failed"
+            record.raw_error = error[:2000]
+            record.safe_error_summary = "参考视频拆解未通过校验"
+        record.save(update_fields=[
+            "status", "finished_at", "duration_ms", "response_summary",
+            "error_type", "raw_error", "safe_error_summary", "updated_at",
+        ])
+    except Exception:  # noqa: BLE001
+        logger.warning("video replace digest attempt close failed", exc_info=True)