feat(documentary): add shared frame analysis contract

.gitignore

@@ -46,3 +46,4 @@ openspec/*
 AGENTS.md
 CLAUDE.md
 tests/*
+!tests/test_documentary_frame_analysis_service.py

app/services/documentary/__init__.py

@@ -0,0 +1,13 @@
+from app.services.documentary.frame_analysis_models import (
+    DocumentaryAnalysisConfig,
+    FrameBatchResult,
+)
+from app.services.documentary.frame_analysis_service import (
+    DocumentaryFrameAnalysisService,
+)
+
+__all__ = [
+    "DocumentaryAnalysisConfig",
+    "FrameBatchResult",
+    "DocumentaryFrameAnalysisService",
+]

app/services/documentary/frame_analysis_models.py

@@ -0,0 +1,25 @@
+from dataclasses import dataclass, field
+
+
+@dataclass(slots=True)
+class DocumentaryAnalysisConfig:
+    video_path: str
+    frame_interval_seconds: float
+    vision_batch_size: int
+    vision_llm_provider: str
+    vision_model_name: str
+    custom_prompt: str = ""
+    max_concurrency: int = 2
+
+
+@dataclass(slots=True)
+class FrameBatchResult:
+    batch_index: int
+    status: str
+    time_range: str
+    raw_response: str
+    frame_paths: list[str] = field(default_factory=list)
+    observations: list[dict] = field(default_factory=list)
+    summary: str = ""
+    fallback_summary: str = ""
+    error_message: str = ""

app/services/documentary/frame_analysis_service.py

@@ -0,0 +1,37 @@
+from app.services.documentary.frame_analysis_models import FrameBatchResult
+
+
+class DocumentaryFrameAnalysisService:
+    PROMPT_TEMPLATE = """
+我提供了 {frame_count} 张视频帧，它们按时间顺序排列，代表一个连续的视频片段。
+首先，请详细描述每一帧的关键视觉信息（包含：主要内容、人物、动作和场景）。
+然后，基于所有帧的分析，请用简洁的语言总结整个视频片段中发生的主要活动或事件流程。
+请务必使用 JSON 格式输出。
+请务必不要遗漏视频帧，我提供了 {frame_count} 张视频帧，frame_observations 必须包含 {frame_count} 个元素
+""".strip()
+
+    def _build_analysis_prompt(self, frame_count: int) -> str:
+        return self.PROMPT_TEMPLATE.format(frame_count=frame_count)
+
+    def _build_failed_batch_result(
+        self,
+        *,
+        batch_index: int,
+        raw_response: str,
+        error_message: str,
+        frame_paths: list[str],
+        time_range: str,
+    ) -> FrameBatchResult:
+        fallback_summary = (raw_response or "").strip()[:200]
+        if not fallback_summary:
+            fallback_summary = f"Batch {batch_index} analysis failed: {error_message or 'unknown error'}"
+
+        return FrameBatchResult(
+            batch_index=batch_index,
+            status="failed",
+            time_range=time_range,
+            raw_response=raw_response,
+            frame_paths=list(frame_paths),
+            fallback_summary=fallback_summary,
+            error_message=error_message,
+        )

tests/test_documentary_frame_analysis_service.py

@@ -0,0 +1,30 @@
+import unittest
+
+from app.services.documentary.frame_analysis_models import DocumentaryAnalysisConfig
+from app.services.documentary.frame_analysis_service import DocumentaryFrameAnalysisService
+
+
+class DocumentaryFrameAnalysisServiceTests(unittest.TestCase):
+    def test_build_analysis_prompt_formats_real_frame_count(self):
+        service = DocumentaryFrameAnalysisService()
+
+        prompt = service._build_analysis_prompt(frame_count=3)
+
+        self.assertIn("我提供了 3 张视频帧", prompt)
+        self.assertNotIn("%s", prompt)
+
+    def test_parse_failed_batch_keeps_raw_response_and_time_range(self):
+        service = DocumentaryFrameAnalysisService()
+
+        batch = service._build_failed_batch_result(
+            batch_index=2,
+            raw_response="not-json",
+            error_message="JSON decode failed",
+            frame_paths=["/tmp/keyframe_000000_000000000.jpg"],
+            time_range="00:00:00,000-00:00:03,000",
+        )
+
+        self.assertEqual("failed", batch.status)
+        self.assertEqual("not-json", batch.raw_response)
+        self.assertEqual("00:00:00,000-00:00:03,000", batch.time_range)
+        self.assertTrue(batch.fallback_summary)