test: replace sqlite local data assumptions

tests/test_acquisition_runner.py

@@ -0,0 +1,296 @@
+from __future__ import annotations
+
+from uuid import UUID, uuid4
+
+from acquisition.classification.coarse import ClassificationResult
+from acquisition.domain import (
+    AcquisitionJob,
+    AcquisitionRun,
+    CandidateItem,
+    ItemClassification,
+    MediaAsset,
+    Query,
+    QueryBatch,
+)
+from acquisition.media.service import StabilizedMedia
+from acquisition.platforms.base import PlatformCandidate, PlatformItem
+from acquisition.runner import run_batch
+from core.config import PgConfig, Settings
+
+
+def _settings() -> Settings:
+    return Settings(
+        pg=PgConfig(host="h", port=5432, user="u", password="p", database="d"),
+        aiddit_crawler_base_url="http://crawler.test",
+        crawler_timeout=30,
+        openrouter_timeout_seconds=90,
+        openrouter_model="m",
+        openrouter_base_url="http://openrouter.test",
+        openrouter_api_key="k",
+        llm_model="m",
+        max_cards=12,
+        frames_dir="f",
+        douyin_ratio="540p",
+        data_dir="data",
+    )
+
+
+class FakeRepo:
+    def __init__(self, *, job_status: str = "pending") -> None:
+        self.batch_id = uuid4()
+        self.query = Query(
+            id=uuid4(),
+            batch_id=self.batch_id,
+            query_text="脚本 开头 怎么做",
+            keep=True,
+            status="ready",
+        )
+        self.job_status = job_status
+        self.runs: list[AcquisitionRun] = []
+        self.jobs: list[AcquisitionJob] = []
+        self.updated_jobs: list[AcquisitionJob] = []
+        self.items: list[CandidateItem] = []
+        self.media: list[MediaAsset] = []
+        self.classifications: list[ItemClassification] = []
+
+    def create_query_batch(self, **kwargs):
+        return QueryBatch(id=self.batch_id, **kwargs)
+
+    def add_query(self, **kwargs):
+        return Query(id=uuid4(), **kwargs)
+
+    def list_queries_for_batch(self, batch_id: UUID, *, keep: bool | None = None):
+        assert batch_id == self.batch_id
+        assert keep is True
+        return [self.query]
+
+    def create_acquisition_run(self, **kwargs):
+        run = AcquisitionRun(id=uuid4(), **kwargs)
+        self.runs.append(run)
+        return run
+
+    def ensure_acquisition_job(self, **kwargs):
+        kwargs["status"] = self.job_status
+        job = AcquisitionJob(id=uuid4(), **kwargs)
+        self.jobs.append(job)
+        return job
+
+    def update_acquisition_job(self, job_id: UUID, **kwargs):
+        original = next(job for job in self.jobs if job.id == job_id)
+        updated = original.model_copy(update=kwargs)
+        self.updated_jobs.append(updated)
+        return updated
+
+    def upsert_candidate_item(self, **kwargs):
+        item = CandidateItem(id=uuid4(), **kwargs)
+        self.items.append(item)
+        return item
+
+    def add_media_asset(self, **kwargs):
+        row = MediaAsset(id=uuid4(), **kwargs)
+        self.media.append(row)
+        return row
+
+    def add_item_classification(self, **kwargs):
+        row = ItemClassification(id=uuid4(), **kwargs)
+        self.classifications.append(row)
+        return row
+
+    def get_run_summary(self, run_id: UUID):
+        return {}
+
+    def get_query_detail(self, *, run_id: UUID, query_id: UUID):
+        return {}
+
+
+class FakeAdapter:
+    platform = "weixin"
+
+    def __init__(self) -> None:
+        self.search_calls = 0
+        self.detail_calls = 0
+
+    def search(self, query, *, settings, limit, rate_limiter):
+        self.search_calls += 1
+        assert query == "脚本 开头 怎么做"
+        assert limit == 2
+        return [
+            PlatformCandidate(
+                rank=1,
+                platform=self.platform,
+                source_id="wx1",
+                url="https://mp.weixin.qq.com/s/a",
+                title="公众号脚本",
+                author="作者",
+                cover_url="https://img.test/a.jpg",
+            )
+        ]
+
+    def fetch_detail(self, candidate, *, settings, rate_limiter):
+        self.detail_calls += 1
+        return PlatformItem(
+            platform=self.platform,
+            source_id=candidate.source_id,
+            url=candidate.url,
+            content_type="图文",
+            title="公众号脚本",
+            author="作者",
+            body_text="先定受众再写开头",
+            image_urls=["https://img.test/a.jpg"],
+            raw={"ok": True},
+        )
+
+
+class PartialAdapter(FakeAdapter):
+    def search(self, query, *, settings, limit, rate_limiter):
+        self.search_calls += 1
+        return [
+            PlatformCandidate(rank=1, platform=self.platform, source_id="bad", url="https://bad"),
+            PlatformCandidate(rank=2, platform=self.platform, source_id="good", url="https://good"),
+        ]
+
+    def fetch_detail(self, candidate, *, settings, rate_limiter):
+        self.detail_calls += 1
+        if candidate.source_id == "bad":
+            raise RuntimeError("detail boom")
+        return PlatformItem(
+            platform=self.platform,
+            source_id=candidate.source_id,
+            url=candidate.url,
+            content_type="图文",
+            title="可用详情",
+            author="作者",
+            body_text="先定受众",
+            image_urls=[],
+            raw={},
+        )
+
+
+class FailingSearchAdapter(FakeAdapter):
+    def search(self, query, *, settings, limit, rate_limiter):
+        self.search_calls += 1
+        raise RuntimeError("search boom")
+
+
+def test_run_batch_writes_jobs_items_media_and_classification():
+    repo = FakeRepo()
+    adapter = FakeAdapter()
+
+    def media_stabilizer(**kwargs):
+        assert kwargs["image_urls"] == ["https://img.test/a.jpg"]
+        return [
+            StabilizedMedia(
+                media_type="image",
+                source_url="https://img.test/a.jpg",
+                cdn_url="https://cdn.test/a.jpg",
+                position=1,
+            )
+        ]
+
+    def classifier(**kwargs):
+        assert kwargs["image_urls"] == ["https://cdn.test/a.jpg"]
+        return ClassificationResult(
+            is_creation_knowledge=True,
+            label="creation",
+            confidence=1.0,
+            reason="是创作知识",
+            knowledge="先定受众",
+            prompt_version="pv1",
+            result_payload={"knowledge": "先定受众"},
+        )
+
+    result = run_batch(
+        repo,
+        batch_id=repo.batch_id,
+        settings=_settings(),
+        platforms=("weixin",),
+        search_limit=2,
+        display_limit=1,
+        adapter_factory=lambda platform: adapter,
+        media_stabilizer=media_stabilizer,
+        classifier=classifier,
+        rate_limiter_factory=lambda platform: object(),
+    )
+
+    assert result.done == 1
+    assert result.failed == 0
+    assert result.total_jobs == 1
+    assert adapter.search_calls == 1
+    assert adapter.detail_calls == 1
+    assert repo.jobs[0].query_id == repo.query.id
+    assert repo.items[0].platform_item_id == "wx1"
+    assert repo.media[0].cdn_url == "https://cdn.test/a.jpg"
+    assert repo.classifications[0].is_creation_knowledge is True
+    assert repo.updated_jobs[-1].status == "done"
+
+
+def test_run_batch_skip_done_does_not_call_platform():
+    repo = FakeRepo(job_status="done")
+    adapter = FakeAdapter()
+
+    result = run_batch(
+        repo,
+        batch_id=repo.batch_id,
+        settings=_settings(),
+        platforms=("weixin",),
+        adapter_factory=lambda platform: adapter,
+    )
+
+    assert result.skipped == 1
+    assert adapter.search_calls == 0
+    assert repo.updated_jobs == []
+
+
+def test_run_batch_marks_partial_when_some_details_fail():
+    repo = FakeRepo()
+    adapter = PartialAdapter()
+
+    result = run_batch(
+        repo,
+        batch_id=repo.batch_id,
+        settings=_settings(),
+        platforms=("weixin",),
+        search_limit=2,
+        display_limit=2,
+        adapter_factory=lambda platform: adapter,
+        media_stabilizer=lambda **kwargs: [],
+        classifier=lambda **kwargs: ClassificationResult(
+            is_creation_knowledge=True,
+            label="creation",
+            confidence=1.0,
+            reason="ok",
+        ),
+        rate_limiter_factory=lambda platform: object(),
+    )
+
+    assert result.partial == 1
+    assert result.done == 0
+    assert result.failed == 0
+    assert repo.updated_jobs[-1].status == "partial"
+    assert repo.updated_jobs[-1].metadata["display_count"] == 1
+    assert repo.updated_jobs[-1].metadata["errors"] == ["detail boom"]
+
+
+def test_run_batch_marks_failed_when_search_fails_before_any_item():
+    repo = FakeRepo()
+    adapter = FailingSearchAdapter()
+
+    result = run_batch(
+        repo,
+        batch_id=repo.batch_id,
+        settings=_settings(),
+        platforms=("weixin",),
+        adapter_factory=lambda platform: adapter,
+        rate_limiter_factory=lambda platform: object(),
+    )
+
+    assert result.failed == 1
+    assert result.done == 0
+    assert repo.updated_jobs[-1].status == "failed"
+    assert "search boom" in (repo.updated_jobs[-1].error_message or "")
+
+
+def test_formal_runner_does_not_import_legacy_store():
+    source = __import__("pathlib").Path("acquisition/runner.py").read_text(encoding="utf-8")
+    assert "acquisition.store" not in source
+    assert "creation_demo.json" not in source

tests/test_app_api.py

@@ -0,0 +1,279 @@
+from __future__ import annotations
+
+from pathlib import Path
+from uuid import uuid4
+
+from fastapi.testclient import TestClient
+
+from acquisition.domain import Query, QueryBatch
+from app.api import app
+from app.dependencies import (
+    get_acquisition_repository,
+    get_decode_repository,
+    get_pipeline_repository,
+)
+from decode_content.models import DecodeJob, DecodeResult, PayloadDraft
+from pipeline.models import PipelineRun
+
+
+class FakeAcquisitionRepo:
+    def __init__(self):
+        self.batch_id = uuid4()
+        self.run_id = uuid4()
+        self.query_id = uuid4()
+        self.item_id = uuid4()
+        self.media_id = uuid4()
+        self.classification_id = uuid4()
+
+    def get_query_batch(self, batch_id):
+        assert batch_id == self.batch_id
+        return QueryBatch(
+            id=batch_id,
+            name="正式 query 批次",
+            source_type="manual",
+            target_platforms=["xiaohongshu", "weixin", "douyin"],
+            status="ready",
+        )
+
+    def list_queries_for_batch(self, batch_id, *, keep=None):
+        assert batch_id == self.batch_id
+        return [
+            Query(
+                id=self.query_id,
+                batch_id=batch_id,
+                query_text="短视频脚本 开头 怎么写",
+                keep=True,
+                status="ready",
+            )
+        ]
+
+    def get_run_summary(self, run_id):
+        assert run_id == self.run_id
+        return {
+            "id": run_id,
+            "run_key": "run-1",
+            "batch_id": self.batch_id,
+            "status": "running",
+            "query_count": 1,
+            "job_count": 3,
+            "candidate_count": 5,
+            "creation_hit_count": 2,
+            "queries": [
+                {
+                    "query_id": self.query_id,
+                    "query_text": "短视频脚本 开头 怎么写",
+                    "candidate_count": 5,
+                    "creation_hit_count": 2,
+                    "platforms": {"xiaohongshu": {"status": "done"}},
+                }
+            ],
+        }
+
+    def get_query_detail(self, *, run_id, query_id):
+        assert run_id == self.run_id
+        assert query_id == self.query_id
+        return {
+            "query": {
+                "id": query_id,
+                "batch_id": self.batch_id,
+                "query_text": "短视频脚本 开头 怎么写",
+                "axes": {},
+                "keep": True,
+                "filter_reason": None,
+                "status": "ready",
+                "sort_order": 0,
+            },
+            "jobs": [{"id": uuid4(), "platform": "xiaohongshu", "status": "done"}],
+            "items": [
+                {
+                    "id": self.item_id,
+                    "platform": "xiaohongshu",
+                    "platform_item_id": "xhs1",
+                    "canonical_url": "https://xhs.test/1",
+                    "content_type": "图文",
+                    "title": "脚本开头",
+                    "author_name": "作者",
+                    "raw_summary": "先定受众",
+                    "status": "candidate",
+                }
+            ],
+            "media_assets": [
+                {
+                    "id": self.media_id,
+                    "item_id": self.item_id,
+                    "media_type": "image",
+                    "source_url": "https://origin.test/1.jpg",
+                    "oss_url": "https://oss.test/1.jpg",
+                    "cdn_url": "https://cdn.test/1.jpg",
+                    "position": 1,
+                    "status": "done",
+                }
+            ],
+            "classifications": [
+                {
+                    "id": self.classification_id,
+                    "item_id": self.item_id,
+                    "is_creation_knowledge": True,
+                    "label": "creation",
+                    "confidence": 0.98,
+                    "reason": "可复用",
+                    "status": "done",
+                }
+            ],
+        }
+
+
+class FakeDecodeRepo:
+    def __init__(self):
+        self.decode_result_id = uuid4()
+        self.decode_job_id = uuid4()
+        self.payload_id = uuid4()
+
+    def get_decode_result_for_item(self, item_id):
+        return DecodeResult(
+            id=self.decode_result_id,
+            item_id=item_id,
+            status="decoded",
+            read_result={"text": "读懂后的帖子内容"},
+            gate_result={"passed": True, "reason": "具备可复用创作方法"},
+            framing_result={"particles": []},
+        )
+
+    def create_decode_job(self, *, item_id, status="pending", metadata=None):
+        return DecodeJob(
+            id=self.decode_job_id,
+            item_id=item_id,
+            status=status,
+            metadata=metadata or {},
+        )
+
+    def list_payload_drafts(self, item_id=None):
+        return [
+            PayloadDraft(
+                id=self.payload_id,
+                item_id=item_id or uuid4(),
+                payload={"title": "可审核 payload"},
+                review_status="pending",
+                ingest_ready=False,
+                status="draft",
+            )
+        ]
+
+
+class FakePipelineRepo:
+    def get_pipeline_run(self, run_id):
+        return PipelineRun(
+            id=run_id,
+            run_key="pipeline-run-1",
+            status="running",
+            current_stage="decode",
+        )
+
+
+def _client(repo: FakeAcquisitionRepo):
+    app.dependency_overrides[get_acquisition_repository] = lambda: repo
+    return TestClient(app)
+
+
+def test_app_health_and_formal_acquisition_routes():
+    repo = FakeAcquisitionRepo()
+    client = _client(repo)
+    try:
+        assert client.get("/health").json()["entrypoint"] == "app.api:app"
+
+        batch = client.get(f"/api/query-batches/{repo.batch_id}").json()
+        assert batch["batch"]["name"] == "正式 query 批次"
+        assert batch["queries"][0]["id"] == str(repo.query_id)
+
+        summary = client.get(f"/api/acquisition/runs/{repo.run_id}/summary").json()
+        assert summary["creation_hit_count"] == 2
+        assert summary["queries"][0]["query_id"] == str(repo.query_id)
+
+        detail = client.get(f"/api/acquisition/runs/{repo.run_id}/queries/{repo.query_id}").json()
+        item = detail["platforms"]["xiaohongshu"]["items"][0]
+        assert item["media_assets"][0]["cdn_url"] == "https://cdn.test/1.jpg"
+        assert item["classification"]["is_creation_knowledge"] is True
+    finally:
+        app.dependency_overrides.clear()
+
+
+def test_formal_app_does_not_expose_legacy_static_mounts():
+    paths = {route.path for route in app.routes if hasattr(route, "path")}
+    assert "/data" not in paths
+    assert "/frames" not in paths
+    assert "/api/creation-search/summary" not in paths
+    assert "/api/creation-search/query" not in paths
+
+
+def test_unimplemented_formal_routes_return_501_instead_of_legacy_data():
+    client = TestClient(app)
+    item_id = uuid4()
+    run_id = uuid4()
+
+    assert client.get(f"/api/decode/items/{item_id}/result").status_code == 501
+    assert client.post(f"/api/decode/items/{item_id}/jobs").status_code == 501
+    assert client.get("/api/payloads/drafts").status_code == 501
+    assert client.get(f"/api/pipeline/runs/{run_id}").status_code == 501
+
+
+def test_formal_decode_payload_pipeline_routes_have_override_success_paths():
+    decode_repo = FakeDecodeRepo()
+    pipeline_repo = FakePipelineRepo()
+    item_id = uuid4()
+    run_id = uuid4()
+    app.dependency_overrides[get_decode_repository] = lambda: decode_repo
+    app.dependency_overrides[get_pipeline_repository] = lambda: pipeline_repo
+    client = TestClient(app)
+
+    try:
+        result = client.get(f"/api/decode/items/{item_id}/result").json()
+        assert result["item_id"] == str(item_id)
+        assert result["read_result"]["text"] == "读懂后的帖子内容"
+
+        job = client.post(f"/api/decode/items/{item_id}/jobs").json()["job"]
+        assert job["item_id"] == str(item_id)
+        assert job["status"] == "pending"
+
+        drafts = client.get(f"/api/payloads/drafts?item_id={item_id}").json()["items"]
+        assert drafts[0]["item_id"] == str(item_id)
+        assert drafts[0]["payload"]["title"] == "可审核 payload"
+
+        run = client.get(f"/api/pipeline/runs/{run_id}").json()["run"]
+        assert run["id"] == str(run_id)
+        assert run["current_stage"] == "decode"
+    finally:
+        app.dependency_overrides.clear()
+
+
+def test_no_legacy_api_dependencies_in_formal_app_sources():
+    roots = [Path("app"), Path("pipeline")]
+    text = "\n".join(
+        path.read_text(encoding="utf-8")
+        for root in roots
+        for path in root.rglob("*")
+        if path.is_file() and path.suffix in {".py", ".jsx", ".js"}
+    )
+    assert "acquisition.store" not in text
+    assert "CK_SQLITE_PATH" not in text
+    assert "/data/queries" not in text
+    assert "/api/creation-search" not in text
+    assert "creation_knowledge.api" not in text
+
+
+def test_legacy_top_level_package_is_archived():
+    assert not Path("creation_knowledge").exists()
+    assert Path("archive/2026-06-30-step6/legacy_creation_knowledge/creation_knowledge").exists()
+
+
+def test_active_sources_do_not_import_legacy_store_or_package():
+    roots = [Path("app"), Path("pipeline"), Path("acquisition"), Path("decode_content"), Path("scripts")]
+    text = "\n".join(
+        path.read_text(encoding="utf-8")
+        for root in roots
+        for path in root.rglob("*")
+        if path.is_file() and path.suffix in {".py", ".jsx", ".js"}
+    )
+    assert "from acquisition import store" not in text
+    assert "import acquisition.store" not in text
+    assert "from creation_knowledge" not in text
+    assert "import creation_knowledge" not in text

tests/test_classification_coarse.py

@@ -0,0 +1,90 @@
+from __future__ import annotations
+
+import acquisition.classification.coarse as coarse
+import acquisition.classify as legacy_classify
+from core.config import PgConfig, Settings
+
+
+def _settings() -> Settings:
+    return Settings(
+        pg=PgConfig(host="h", port=5432, user="u", password="p", database="d"),
+        aiddit_crawler_base_url="http://crawler.test",
+        crawler_timeout=30,
+        openrouter_timeout_seconds=90,
+        openrouter_model="m",
+        openrouter_base_url="http://openrouter.test",
+        openrouter_api_key="k",
+        llm_model="m",
+        max_cards=12,
+        frames_dir="f",
+        douyin_ratio="540p",
+        data_dir="data",
+    )
+
+
+def test_formal_classify_imgtext_passes_http_image_urls(monkeypatch):
+    captured = {}
+
+    def fake_judge(messages, settings, timeout):
+        captured["messages"] = messages
+        return 1, "ok", "先定受众", ""
+
+    monkeypatch.setattr(coarse, "_judge", fake_judge)
+
+    result = coarse.classify_imgtext(
+        {
+            "platform": "xiaohongshu",
+            "title": "脚本创作",
+            "body_text": "讲怎么设计开头",
+            "images": ["https://cdn.test/a.jpg"],
+        },
+        _settings(),
+    )
+
+    assert result[0] == 1
+    user_content = captured["messages"][1]["content"]
+    assert {"type": "image_url", "image_url": {"url": "https://cdn.test/a.jpg"}} in user_content
+
+
+def test_legacy_classify_imgtext_also_passes_http_image_urls(monkeypatch):
+    captured = {}
+
+    def fake_judge(messages, settings, timeout):
+        captured["messages"] = messages
+        return 1, "ok", "先定受众", ""
+
+    monkeypatch.setattr(legacy_classify, "_judge", fake_judge)
+
+    result = legacy_classify.classify_imgtext(
+        {
+            "platform": "weixin",
+            "title": "公众号选题",
+            "body_text": "讲怎么做标题",
+            "images": ["https://cdn.test/w.jpg"],
+        },
+        _settings(),
+    )
+
+    assert result[0] == 1
+    user_content = captured["messages"][1]["content"]
+    assert {"type": "image_url", "image_url": {"url": "https://cdn.test/w.jpg"}} in user_content
+
+
+def test_coarse_classify_item_maps_judge_result(monkeypatch):
+    monkeypatch.setattr(
+        coarse,
+        "classify_imgtext",
+        lambda payload, settings: (0, "不是创作知识", "", ""),
+    )
+
+    result = coarse.coarse_classify_item(
+        platform="weixin",
+        title="普通知识",
+        body_text="只是作品介绍",
+        image_urls=["https://cdn.test/a.jpg"],
+        settings=_settings(),
+    )
+
+    assert result.is_creation_knowledge is False
+    assert result.label == "not_creation"
+    assert result.status == "classified"

tests/test_db_migration_contract.py

@@ -0,0 +1,72 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+
+MIGRATION = Path("db/migrations/001_creation_knowledge_schema.sql")
+
+
+def _migration_sql() -> str:
+    return MIGRATION.read_text(encoding="utf-8")
+
+
+def test_creation_knowledge_migration_declares_formal_business_tables():
+    sql = _migration_sql()
+    expected_tables = [
+        "query_batches",
+        "queries",
+        "acquisition_runs",
+        "acquisition_jobs",
+        "candidate_items",
+        "media_assets",
+        "item_classifications",
+        "decode_jobs",
+        "decode_results",
+        "knowledge_particles",
+        "scope_results",
+        "payload_drafts",
+        "ingest_records",
+        "contract_snapshots",
+    ]
+
+    assert "CREATE SCHEMA IF NOT EXISTS creation_knowledge" in sql
+    assert "CREATE EXTENSION IF NOT EXISTS pgcrypto" in sql
+    for table in expected_tables:
+        assert f"CREATE TABLE IF NOT EXISTS creation_knowledge.{table}" in sql
+
+
+def test_migration_is_replayable_and_not_sqlite_shaped():
+    sql = _migration_sql()
+
+    assert "CREATE TABLE IF NOT EXISTS" in sql
+    assert "CREATE INDEX IF NOT EXISTS" in sql
+    assert "ON CONFLICT (version) DO UPDATE" in sql
+    assert "CREATE OR REPLACE FUNCTION creation_knowledge.touch_updated_at" in sql
+    assert "CREATE TRIGGER" in sql
+
+    sqlite_only_tokens = ["AUTOINCREMENT", "PRAGMA", "sqlite_master", "WITHOUT ROWID"]
+    assert not any(token in sql.upper() for token in sqlite_only_tokens)
+
+
+def test_migration_keeps_state_rows_traceable_and_resumable():
+    sql = _migration_sql()
+
+    for table in [
+        "acquisition_runs",
+        "acquisition_jobs",
+        "candidate_items",
+        "decode_jobs",
+        "decode_results",
+        "payload_drafts",
+        "ingest_records",
+    ]:
+        assert f"creation_knowledge.{table}" in sql
+        assert "status text NOT NULL" in sql
+
+    assert "attempt_count integer NOT NULL DEFAULT 0" in sql
+    assert "error_message text" in sql
+    assert "metadata jsonb NOT NULL DEFAULT '{}'::jsonb" in sql
+    assert "source_payload jsonb NOT NULL DEFAULT '{}'::jsonb" in sql
+    assert "response_payload jsonb NOT NULL DEFAULT '{}'::jsonb" in sql
+    assert "UNIQUE (run_id, query_id, platform)" in sql
+    assert "idx_candidate_items_platform_item" in sql

tests/test_decode_contracts.py

@@ -0,0 +1,47 @@
+from __future__ import annotations
+
+from decode_content.contracts import (
+    BUSINESS_STAGES,
+    CREATION_STAGES,
+    DIM_ATTRIBUTE_BY_TYPE,
+    KNOWLEDGE_TYPES,
+    SCOPE_TYPES,
+    WHAT_KINDS,
+    compute_contract_hash,
+    iter_contract_snapshots,
+    load_decode_contracts,
+)
+
+
+def test_load_decode_contracts_covers_skill_prompts_and_cache():
+    contract = load_decode_contracts()
+
+    assert "判颗" in contract.phase1
+    assert "作用域" in contract.phase2
+    assert "payload" in contract.phase3.lower()
+    assert contract.schema_for_particle_type("how")["$id"] == "ingest-payload-how.schema.json"
+    assert any(src["relative_path"].startswith("prompts/") for src in contract.prompt_sources())
+    assert contract.scope_tree_sources()
+    assert len(contract.content_hash) == 64
+    assert compute_contract_hash(contract) == contract.content_hash
+
+
+def test_contract_constants_record_formal_business_terms():
+    assert KNOWLEDGE_TYPES == ("how", "what", "why")
+    assert BUSINESS_STAGES == ("灵感", "选题", "脚本")
+    assert CREATION_STAGES == ("定向", "构思", "结构", "成文", "打磨")
+    assert WHAT_KINDS == ("子集", "多维关系", "序列")
+    assert SCOPE_TYPES == ("substance", "form", "feeling", "effect", "intent")
+    assert DIM_ATTRIBUTE_BY_TYPE == {"how": "how工序", "what": "what构成", "why": "why原理"}
+
+
+def test_contract_snapshots_can_feed_repository():
+    snapshots = iter_contract_snapshots()
+
+    assert snapshots[0].contract_name == "创作知识提取-skill"
+    assert snapshots[0].contract_type == "skill"
+    assert snapshots[0].content_hash and len(snapshots[0].content_hash) == 64
+    assert snapshots[0].snapshot["constants"]["dim_attribute_by_type"]["how"] == "how工序"
+    assert any(s.contract_type == "prompt_phase" for s in snapshots)
+    assert any(s.contract_type == "schema" for s in snapshots)
+    assert any(s.contract_type == "prompt" for s in snapshots)

tests/test_decode_gates_framing_scoping.py

@@ -0,0 +1,77 @@
+from __future__ import annotations
+
+from decode_content import gates
+from decode_content.framing import normalize_knowledge_shapes
+from decode_content.scoping import apply_scopes, split_scope_items, strip_verb_tail
+
+
+def test_creation_gate_uses_admit_refute_tiebreak():
+    calls = []
+
+    def fake_chat(system, user, timeout=None):
+        calls.append(system)
+        if system == gates.GATE_ADMIT:
+            return {"in_scope": True}
+        if system == gates.GATE_REFUTE:
+            return {"out_of_scope": True}
+        if system == gates.GATE_TIEBREAK:
+            return {"in_scope": False}
+        raise AssertionError(system)
+
+    result = gates.creation_gate("读懂内容", chat_json_fn=fake_chat)
+
+    assert result.passed is False
+    assert "裁决=False" in result.reason
+    assert calls == [gates.GATE_ADMIT, gates.GATE_REFUTE, gates.GATE_TIEBREAK]
+
+
+def test_normalize_knowledge_shapes_guards_stage_kind_and_parent():
+    knowledges = [
+        {
+            "id": "w1",
+            "type": "what",
+            "role": "组件",
+            "parent": {"how_id": "missing", "step": 1},
+            "title": "构成",
+            "kind": "集合",
+            "业务阶段": ["脚本", "错误"],
+        },
+        {
+            "id": "h1",
+            "type": "how",
+            "title": "流程",
+            "业务阶段": ["选题"],
+            "steps": [{"创作阶段": "定稿"}],
+        },
+    ]
+
+    out = normalize_knowledge_shapes(knowledges)
+
+    assert out[0]["kind"] == "子集"
+    assert out[0]["业务阶段"] == ["脚本"]
+    assert out[0]["role"] == "主"
+    assert out[0]["parent"] is None
+    assert out[1]["steps"][0]["创作阶段"] is None
+
+
+def test_scoping_split_strip_and_apply_without_linker():
+    assert strip_verb_tail("寻找爆款选题") == "爆款选题"
+    assert split_scope_items([{"scope_type": "form", "value": "脚本结构与标题结构"}]) == [
+        {"scope_type": "form", "value": "脚本结构"},
+        {"scope_type": "form", "value": "标题结构"},
+    ]
+    knowledges = [{"id": "h1", "type": "how", "steps": [{"id": "s1"}]}]
+
+    apply_scopes(
+        knowledges,
+        [
+            {
+                "knowledge_id": "h1",
+                "step_id": "s1",
+                "items": [{"scope_type": "intent", "value": "吸引注意"}],
+            }
+        ],
+        linker=None,
+    )
+
+    assert knowledges[0]["steps"][0]["作用域"][0]["link"] == "候选"

tests/test_decode_payloads.py

@@ -0,0 +1,79 @@
+from __future__ import annotations
+
+import json
+
+from core.models import Post
+from decode_content.payloads import build_payload, build_payloads
+
+
+def _post() -> Post:
+    return Post(
+        id="xhs_1",
+        platform="xiaohongshu",
+        url="https://xhs/1",
+        content_id="1",
+        title="脚本教程",
+        author_name="作者",
+    )
+
+
+def test_build_how_payload_uses_formal_dim_attribute_and_scope_union():
+    knowledge = {
+        "id": "k1",
+        "type": "how",
+        "title": "先定受众再写脚本",
+        "purpose": "写出稳定脚本",
+        "业务阶段": ["脚本"],
+        "steps": [
+            {
+                "input": "选题",
+                "directive": "收窄受众",
+                "output": "受众画像",
+                "创作阶段": "定向",
+                "动作": "收窄",
+                "作用域": [
+                    {"scope_type": "intent", "value": "吸引注意", "link": "复用"},
+                    {"scope_type": "intent", "value": "吸引注意", "link": "复用"},
+                ],
+            }
+        ],
+    }
+
+    payload = build_payload(_post(), knowledge)
+
+    assert payload["dim_attributes"] == ["how工序"]
+    assert "输入：选题" in payload["content"]
+    assert "指引：收窄受众" in payload["content"]
+    assert payload["scopes"] == [{"scope_type": "intent", "value": "吸引注意"}]
+    assert {"key": "创作阶段", "type": "str", "value": "定向"} in payload["custom_ext"]
+
+
+def test_build_what_and_why_payload_content_is_string_contract():
+    what = {
+        "id": "w1",
+        "type": "what",
+        "title": "脚本三要素",
+        "kind": "子集",
+        "概要": "脚本由开头中段结尾构成",
+        "维度拆分规则": "按表达顺序",
+        "body": [{"item_name": "开头", "item_desc": "抓注意"}],
+        "业务阶段": ["脚本"],
+        "作用域": [{"scope_type": "form", "value": "脚本结构", "top": []}],
+    }
+    why = {
+        "id": "y1",
+        "type": "why",
+        "title": "冲突带来停留",
+        "阐述": "冲突能制造期待，所以提升停留。",
+        "业务阶段": ["选题"],
+        "作用域": [{"scope_type": "effect", "value": "提升停留", "score": 0.95}],
+    }
+
+    what_payload, why_payload = build_payloads(_post(), [what, why])
+
+    assert what_payload["dim_attributes"] == ["what构成"]
+    assert json.loads(what_payload["content"])["body"][0]["item_name"] == "开头"
+    assert {"key": "概要", "type": "str", "value": "脚本由开头中段结尾构成"} in what_payload["custom_ext"]
+    assert why_payload["dim_attributes"] == ["why原理"]
+    assert why_payload["content"] == "冲突能制造期待，所以提升停留。"
+    assert "score" not in why_payload["scopes"][0]

tests/test_decode_readers_service.py

@@ -0,0 +1,75 @@
+from __future__ import annotations
+
+from uuid import uuid4
+
+from acquisition.domain import CandidateItem, MediaAsset
+from core.models import Card, CardExtract, ExtractedContent, Post
+from decode_content.readers.service import post_from_candidate_item, read_item, read_post
+
+
+def test_post_from_candidate_item_prefers_cdn_then_oss_then_source():
+    item_id = uuid4()
+    item = CandidateItem(
+        id=item_id,
+        platform="xiaohongshu",
+        platform_item_id="abc",
+        canonical_url="https://xhs/item/abc",
+        title="脚本结构",
+        author_name="sam",
+        raw_summary="正文",
+        source_payload={"topic_list": ["脚本"]},
+    )
+    media = [
+        MediaAsset(item_id=item_id, media_type="image", source_url="source1", cdn_url="cdn1", position=2),
+        MediaAsset(item_id=item_id, media_type="image", source_url="source0", oss_url="oss0", position=1),
+    ]
+
+    post = post_from_candidate_item(item, media)
+
+    assert post.id == "xiaohongshu_abc"
+    assert post.image_urls == ["oss0", "cdn1"]
+    assert [card.index for card in post.cards] == [1, 2]
+    assert post.body_text == "正文"
+    assert post.topic_list == ["脚本"]
+
+
+def test_read_post_imgtext_merges_text_cards_and_media():
+    post = Post(
+        id="xhs_1",
+        platform="xiaohongshu",
+        url="u",
+        content_id="1",
+        image_urls=["img1", "img2"],
+        cards=[Card(index=1, kind="image", url="img1"), Card(index=2, kind="image", url="img2")],
+    )
+
+    def image_reader(p: Post) -> ExtractedContent:
+        assert p.id == "xhs_1"
+        return ExtractedContent(
+            text="主体",
+            from_image="图片知识",
+            cards=[CardExtract(index=1, content="卡1"), CardExtract(index=2, content="卡2")],
+        )
+
+    result = read_post(post, image_reader=image_reader)
+
+    assert "图片知识" in result.text
+    assert result.cards[0].content == "卡1"
+    assert result.media == {"type": "image", "video_url": None, "images": ["img1", "img2"]}
+
+
+def test_read_item_routes_video_to_video_reader():
+    item_id = uuid4()
+    item = CandidateItem(id=item_id, platform="douyin", platform_item_id="v1", title="视频")
+    media = [MediaAsset(item_id=item_id, media_type="video", cdn_url="https://cdn/video.mp4", position=1)]
+
+    def video_reader(post: Post) -> ExtractedContent:
+        assert post.video_urls == ["https://cdn/video.mp4"]
+        post.cards = [Card(index=1, kind="segment", url=post.video_urls[0], start=0, end=12)]
+        return ExtractedContent(text="视频理解", cards=[CardExtract(index=1, content="片段知识")])
+
+    result = read_item(item, media, video_reader=video_reader)
+
+    assert result.media["type"] == "video"
+    assert result.cards[0].kind == "segment"
+    assert result.cards[0].content == "片段知识"

tests/test_decode_service.py

@@ -0,0 +1,140 @@
+from __future__ import annotations
+
+from uuid import UUID, uuid4
+
+from core.models import Post
+from decode_content import gates
+from decode_content.contracts import load_decode_contracts
+from decode_content.models import ContractSnapshot, DecodeJob, DecodeResult, KnowledgeParticle, PayloadDraft, ReadResult, ScopeResult
+from decode_content.scoping import NORMALIZE
+from decode_content.service import DecodeService, slugify
+
+
+class FakeRepo:
+    def __init__(self):
+        self.jobs = []
+        self.results = []
+        self.particles = []
+        self.scopes = []
+        self.payloads = []
+        self.snapshots = []
+
+    def create_decode_job(self, *, item_id: UUID, status: str = "pending", metadata=None):
+        job = DecodeJob(id=uuid4(), item_id=item_id, status=status, metadata=metadata or {})
+        self.jobs.append(job)
+        return job
+
+    def save_decode_result(self, **kwargs):
+        result = DecodeResult(id=uuid4(), **kwargs)
+        self.results.append(result)
+        return result
+
+    def save_knowledge_particle(self, **kwargs):
+        particle = KnowledgeParticle(id=uuid4(), **kwargs)
+        self.particles.append(particle)
+        return particle
+
+    def save_scope_result(self, **kwargs):
+        scope = ScopeResult(id=uuid4(), **kwargs)
+        self.scopes.append(scope)
+        return scope
+
+    def save_payload_draft(self, **kwargs):
+        payload = PayloadDraft(id=uuid4(), **kwargs)
+        self.payloads.append(payload)
+        return payload
+
+    def save_contract_snapshot(self, **kwargs):
+        snapshot = ContractSnapshot(id=uuid4(), **kwargs)
+        self.snapshots.append(snapshot)
+        return snapshot
+
+
+def test_slugify_formal_helper():
+    assert slugify("分镜脚本") == "分镜脚本"
+    assert slugify("A B!c") == "a-b-c"
+    assert slugify("") == "run"
+
+
+def test_decode_service_short_circuits_empty_read():
+    item_id = uuid4()
+    repo = FakeRepo()
+    service = DecodeService(repository=repo, reader=lambda post: ReadResult(is_empty=True))
+
+    out = service.decode_post(item_id=item_id, post=Post(id="xhs_1", platform="xiaohongshu", url="u", content_id="1"))
+
+    assert out.status == "skipped"
+    assert repo.results[0].status == "skipped"
+    assert repo.particles == []
+    assert repo.snapshots
+
+
+def test_decode_service_full_flow_with_fake_llm():
+    item_id = uuid4()
+    repo = FakeRepo()
+    contract = load_decode_contracts()
+
+    def fake_chat(system, user, timeout=None):
+        if system == gates.GATE_ADMIT:
+            return {"in_scope": True}
+        if system == gates.GATE_REFUTE:
+            return {"out_of_scope": False}
+        if system == gates.GATE_HOW_ADMIT:
+            return {"is_real_how": True}
+        if system == gates.GATE_HOW_REFUTE:
+            return {"is_fake": False}
+        if system == gates.GATE_WHY_REFUTE:
+            return {"not_why": False}
+        if system == NORMALIZE:
+            return {"映射": {}}
+        if system == contract.phase1:
+            return {
+                "knowledges": [
+                    {
+                        "id": "k1",
+                        "type": "how",
+                        "role": "主",
+                        "title": "脚本搭建流程",
+                        "purpose": "搭出脚本",
+                        "业务阶段": ["脚本"],
+                        "steps": [
+                            {
+                                "id": "s1",
+                                "input": "选题",
+                                "directive": "先定受众",
+                                "output": "受众画像",
+                                "创作阶段": "定向",
+                            }
+                        ],
+                    }
+                ]
+            }
+        if system == contract.phase2:
+            return {
+                "scopes": [
+                    {
+                        "knowledge_id": "k1",
+                        "step_id": "s1",
+                        "items": [{"scope_type": "intent", "value": "吸引注意"}],
+                    }
+                ]
+            }
+        raise AssertionError(system[:80])
+
+    service = DecodeService(
+        repository=repo,
+        contract=contract,
+        chat_json_fn=fake_chat,
+        reader=lambda post: ReadResult(text="这是一条脚本创作知识"),
+    )
+    out = service.decode_post(
+        item_id=item_id,
+        post=Post(id="xhs_1", platform="xiaohongshu", url="u", content_id="1", title="脚本教程"),
+    )
+
+    assert out.status == "decoded"
+    assert repo.results[0].status == "decoded"
+    assert repo.particles[0].particle_type == "how"
+    assert repo.scopes[0].scope_type == "intent"
+    assert repo.payloads[0].payload["dim_attributes"] == ["how工序"]
+    assert repo.snapshots[0].contract_name == "创作知识提取-skill"

tests/test_decode_stage_full_chain.py

@@ -0,0 +1,173 @@
+from __future__ import annotations
+
+from uuid import UUID, uuid4
+
+from acquisition.domain import CandidateItem, MediaAsset
+from decode_content import gates
+from decode_content.contracts import load_decode_contracts
+from decode_content.models import (
+    ContractSnapshot,
+    DecodeJob,
+    DecodeResult,
+    KnowledgeParticle,
+    PayloadDraft,
+    ReadResult,
+    ScopeResult,
+)
+from decode_content.scoping import NORMALIZE
+from decode_content.service import DecodeService
+from pipeline.decode_runner import run_decode_stage
+
+
+class CandidateRepo:
+    def __init__(self, item_id: UUID):
+        self.item = CandidateItem(
+            id=item_id,
+            platform="xiaohongshu",
+            platform_item_id="xhs-1",
+            canonical_url="https://xhs.test/1",
+            content_type="图文",
+            title="脚本开头教程",
+            author_name="作者",
+            raw_summary="先定受众，再写开头钩子",
+            source_payload={"topic_list": ["脚本"]},
+        )
+        self.media = [
+            MediaAsset(
+                id=uuid4(),
+                item_id=item_id,
+                media_type="image",
+                source_url="https://origin.test/1.jpg",
+                oss_url="https://oss.test/1.jpg",
+                cdn_url="https://cdn.test/1.jpg",
+                position=1,
+                status="done",
+            )
+        ]
+
+    def list_creation_candidate_items(self, *, run_id=None, limit=100):
+        return [self.item]
+
+    def list_media_assets_for_item(self, item_id):
+        assert item_id == self.item.id
+        return self.media
+
+
+class DecodeRepo:
+    def __init__(self):
+        self.jobs: list[DecodeJob] = []
+        self.results: list[DecodeResult] = []
+        self.particles: list[KnowledgeParticle] = []
+        self.scopes: list[ScopeResult] = []
+        self.payloads: list[PayloadDraft] = []
+        self.snapshots: list[ContractSnapshot] = []
+
+    def create_decode_job(self, *, item_id: UUID, status: str = "pending", metadata=None):
+        job = DecodeJob(id=uuid4(), item_id=item_id, status=status, metadata=metadata or {})
+        self.jobs.append(job)
+        return job
+
+    def save_decode_result(self, **kwargs):
+        result = DecodeResult(id=uuid4(), **kwargs)
+        self.results.append(result)
+        return result
+
+    def save_knowledge_particle(self, **kwargs):
+        particle = KnowledgeParticle(id=uuid4(), **kwargs)
+        self.particles.append(particle)
+        return particle
+
+    def save_scope_result(self, **kwargs):
+        scope = ScopeResult(id=uuid4(), **kwargs)
+        self.scopes.append(scope)
+        return scope
+
+    def save_payload_draft(self, **kwargs):
+        payload = PayloadDraft(id=uuid4(), **kwargs)
+        self.payloads.append(payload)
+        return payload
+
+    def save_contract_snapshot(self, **kwargs):
+        snapshot = ContractSnapshot(id=uuid4(), **kwargs)
+        self.snapshots.append(snapshot)
+        return snapshot
+
+
+def _fake_chat(contract):
+    def chat(system, user, timeout=None):
+        if system == gates.GATE_ADMIT:
+            return {"in_scope": True}
+        if system == gates.GATE_REFUTE:
+            return {"out_of_scope": False}
+        if system == gates.GATE_HOW_ADMIT:
+            return {"is_real_how": True}
+        if system == gates.GATE_HOW_REFUTE:
+            return {"is_fake": False}
+        if system == gates.GATE_WHY_REFUTE:
+            return {"not_why": False}
+        if system == NORMALIZE:
+            return {"映射": {}}
+        if system == contract.phase1:
+            return {
+                "knowledges": [
+                    {
+                        "id": "k1",
+                        "type": "how",
+                        "role": "主",
+                        "title": "脚本开头的受众钩子",
+                        "purpose": "提升开头吸引力",
+                        "业务阶段": ["脚本"],
+                        "steps": [
+                            {
+                                "id": "s1",
+                                "input": "目标受众",
+                                "directive": "先说出痛点",
+                                "output": "开头钩子",
+                                "创作阶段": "开头",
+                            }
+                        ],
+                    }
+                ]
+            }
+        if system == contract.phase2:
+            return {
+                "scopes": [
+                    {
+                        "knowledge_id": "k1",
+                        "step_id": "s1",
+                        "items": [{"scope_type": "intent", "value": "吸引注意"}],
+                    }
+                ]
+            }
+        raise AssertionError(system[:80])
+
+    return chat
+
+
+def test_decode_stage_connects_candidate_media_to_payload_drafts():
+    item_id = uuid4()
+    candidate_repo = CandidateRepo(item_id)
+    decode_repo = DecodeRepo()
+    contract = load_decode_contracts()
+    seen_posts = []
+
+    def reader(post):
+        seen_posts.append(post)
+        return ReadResult(text="这条帖子讲脚本开头创作方法")
+
+    service = DecodeService(
+        repository=decode_repo,
+        contract=contract,
+        chat_json_fn=_fake_chat(contract),
+        reader=reader,
+    )
+
+    batch = run_decode_stage(candidate_repo=candidate_repo, decode_service=service)
+
+    assert batch.total == 1
+    assert batch.decoded == 1
+    assert seen_posts[0].image_urls == ["https://cdn.test/1.jpg"]
+    assert decode_repo.results[0].status == "decoded"
+    assert decode_repo.particles[0].title == "脚本开头的受众钩子"
+    assert decode_repo.scopes[0].scope_value == "吸引注意"
+    assert decode_repo.payloads[0].payload["title"] == "脚本开头的受众钩子"

tests/test_extractor.py

@@ -6,7 +6,7 @@ from __future__ import annotations
 
 import json
 
-from creation_knowledge.integrations.extractor import GeminiExtractor
+from decode_content.readers.imgtext import GeminiExtractor
 from core.models import Post
 
 

tests/test_formal_state_models.py

@@ -0,0 +1,125 @@
+from __future__ import annotations
+
+from uuid import uuid4
+
+import pytest
+from pydantic import ValidationError
+
+from app.schemas import (
+    AcquisitionRunSummarySchema,
+    CandidateItemSchema,
+    ErrorResponse,
+    Page,
+)
+from core.config import CreationDbConfig
+from decode_content.models import (
+    IngestRecord,
+    KnowledgeParticle,
+    PayloadDraft,
+    ReadResult,
+    ScopeResult,
+)
+from pipeline.models import PipelineJob, PipelineRun, ResumeCursor
+
+
+def test_creation_db_config_reads_ck_db_without_open_aigc(tmp_path):
+    env_file = tmp_path / ".env"
+    env_file.write_text(
+        "\n".join(
+            [
+                "OPEN_AIGC_PG_HOST=wrong-open-aigc",
+                "OPEN_AIGC_PG_USER=wrong-user",
+                "CK_DB_HOST=127.0.0.1",
+                "CK_DB_PORT=6543",
+                "CK_DB_NAME=creation_knowledge_prod",
+                "CK_DB_USER=ck_app",
+                "CK_DB_PASSWORD=secret",
+                "CK_DB_SCHEMA=creation_knowledge",
+            ]
+        ),
+        encoding="utf-8",
+    )
+
+    cfg = CreationDbConfig.from_env(env_file)
+
+    assert cfg.host == "127.0.0.1"
+    assert cfg.port == 6543
+    assert cfg.user == "ck_app"
+    assert cfg.database == "creation_knowledge_prod"
+    assert cfg.schema == "creation_knowledge"
+
+
+def test_decode_content_models_are_contract_shaped():
+    item_id = uuid4()
+    read = ReadResult(
+        text="一条创作判断",
+        cards=[{"index": 1, "kind": "image", "content": "构图说明"}],
+        media={"platform": "xiaohongshu"},
+    )
+    knowledge = KnowledgeParticle(
+        item_id=item_id,
+        particle_type="how",
+        title="先定受众再写脚本",
+        content={"steps": [{"input": "目标受众", "directive": "收窄", "output": "脚本方向"}]},
+    )
+    scope = ScopeResult(
+        item_id=item_id,
+        particle_id=uuid4(),
+        scope_type="意图",
+        scope_value="吸引注意",
+        is_reused=True,
+    )
+    payload = PayloadDraft(item_id=item_id, particle_id=knowledge.id, payload={"kind": "how"})
+    ingest = IngestRecord(
+        payload_draft_id=payload.id,
+        target_system="creation-knowledge-db",
+        target_id="remote-1",
+    )
+
+    assert read.cards[0].content == "构图说明"
+    assert knowledge.particle_type == "how"
+    assert scope.scope_value == "吸引注意"
+    assert payload.review_status == "pending"
+    assert ingest.status == "pending"
+
+
+def test_decode_content_rejects_unknown_particle_type():
+    with pytest.raises(ValidationError):
+        KnowledgeParticle(particle_type="idea", title="错误类型")
+
+
+def test_pipeline_models_capture_resume_boundary():
+    run_id = uuid4()
+    run = PipelineRun(id=run_id, run_key="formal-run", current_stage="decode")
+    job = PipelineJob(run_id=run_id, stage="decode", target_id=uuid4())
+    cursor = ResumeCursor(stage="decode", last_successful_job_id=job.id)
+
+    assert run.status == "pending"
+    assert job.status == "pending"
+    assert cursor.stage == "decode"
+
+
+def test_app_schemas_cover_formal_workbench_shapes():
+    run_id = uuid4()
+    item_id = uuid4()
+    summary = AcquisitionRunSummarySchema(
+        id=run_id,
+        status="running",
+        query_count=3,
+        job_count=9,
+        candidate_count=21,
+        creation_hit_count=5,
+    )
+    item = CandidateItemSchema(
+        id=item_id,
+        platform="douyin",
+        title="短视频脚本创作",
+        status="candidate",
+    )
+    page = Page(total=1, items=[{"id": str(item_id)}])
+    error = ErrorResponse(code="not_found", message="missing")
+
+    assert summary.creation_hit_count == 5
+    assert item.media_assets == []
+    assert page.total == 1
+    assert error.detail == {}

tests/test_legacy_cleanup_contract.py

@@ -0,0 +1,51 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+
+FORMAL_SOURCE_ROOTS = [
+    Path("app"),
+    Path("pipeline"),
+    Path("acquisition"),
+    Path("decode_content"),
+    Path("core"),
+]
+
+LEGACY_RUNTIME_TOKENS = [
+    "CK_SQLITE_PATH",
+    "data/app.db",
+    "/data/queries",
+    "/api/creation-search",
+    "creation_knowledge.api",
+    "from acquisition import store",
+    "import acquisition.store",
+    "from creation_knowledge",
+    "import creation_knowledge",
+]
+
+
+def _active_source_files():
+    for root in FORMAL_SOURCE_ROOTS:
+        for path in root.rglob("*"):
+            if path.is_file() and path.suffix in {".py", ".js", ".jsx", ".ts", ".tsx"}:
+                yield path
+
+
+def test_formal_sources_do_not_depend_on_archived_legacy_runtime():
+    offenders: list[str] = []
+    for path in _active_source_files():
+        text = path.read_text(encoding="utf-8")
+        for token in LEGACY_RUNTIME_TOKENS:
+            if token in text:
+                offenders.append(f"{path}:{token}")
+
+    assert offenders == []
+
+
+def test_legacy_debug_decompose_entry_is_explicitly_labeled():
+    text = Path("scripts/decompose.py").read_text(encoding="utf-8")
+
+    assert "LEGACY/debug" in text[:500]
+    assert "正式云端 pipeline 入口" in text[:500]
+    assert "web/frameworks" in text
+    assert "web/payloads" in text

tests/test_media_service.py

@@ -0,0 +1,48 @@
+from __future__ import annotations
+
+from acquisition.media.service import stabilize_media_urls
+
+
+def test_stabilize_media_urls_keeps_order_and_positions():
+    calls = []
+
+    def uploader(url: str, media_type: str) -> str:
+        calls.append((url, media_type))
+        return f"https://cdn.test/{media_type}/{len(calls)}"
+
+    rows = stabilize_media_urls(
+        image_urls=["https://src.test/a.jpg", "", "https://src.test/b.jpg"],
+        video_urls=["https://src.test/c.mp4"],
+        uploader=uploader,
+    )
+
+    assert [(row.media_type, row.position) for row in rows] == [
+        ("image", 1),
+        ("image", 2),
+        ("video", 3),
+    ]
+    assert rows[0].source_url == "https://src.test/a.jpg"
+    assert rows[0].cdn_url == "https://cdn.test/image/1"
+    assert rows[-1].cdn_url == "https://cdn.test/video/3"
+    assert calls == [
+        ("https://src.test/a.jpg", "image"),
+        ("https://src.test/b.jpg", "image"),
+        ("https://src.test/c.mp4", "video"),
+    ]
+
+
+def test_stabilize_media_urls_falls_back_to_source_on_upload_error():
+    def uploader(url: str, media_type: str) -> str:
+        raise RuntimeError(f"{media_type} upload failed")
+
+    rows = stabilize_media_urls(
+        image_urls=["https://src.test/a.jpg"],
+        video_urls=["https://src.test/c.mp4"],
+        uploader=uploader,
+    )
+
+    assert [row.cdn_url for row in rows] == [
+        "https://src.test/a.jpg",
+        "https://src.test/c.mp4",
+    ]
+    assert [row.status for row in rows] == ["ready", "ready"]

tests/test_pipeline_formal.py

@@ -0,0 +1,204 @@
+from __future__ import annotations
+
+from uuid import UUID, uuid4
+
+import pytest
+
+from acquisition.domain import CandidateItem, MediaAsset
+from acquisition.runner import RunBatchResult
+from core.models import Post
+from decode_content.models import GateResult, ReadResult
+from decode_content.service import DecodeWorkflowOutput
+from pipeline import acquisition_runner, creation_pipeline
+from pipeline.acquisition_runner import AcquisitionStageResult
+from pipeline.decode_runner import run_decode_stage
+from pipeline.decode_runner import DecodeBatchResult
+from pipeline.dedupe import dedupe_candidate_items, item_dedupe_key, should_decode_item
+from pipeline.models import PipelineJob, PipelineRun
+
+
+class FakePipelineRepo:
+    def __init__(self):
+        self.jobs = []
+
+    def create_pipeline_run(self, **kwargs):
+        return PipelineRun(id=uuid4(), **kwargs)
+
+    def save_pipeline_job(self, **kwargs):
+        job = PipelineJob(id=uuid4(), **kwargs)
+        self.jobs.append(job)
+        return job
+
+    def mark_job_status(self, job_id, **kwargs):
+        job = next(row for row in self.jobs if row.id == job_id)
+        updated = job.model_copy(update=kwargs)
+        self.jobs.append(updated)
+        return updated
+
+    def get_resume_cursor(self, run_id):
+        return None
+
+
+def test_dedupe_candidate_items_prefers_platform_item_id_then_url():
+    first = CandidateItem(id=uuid4(), platform="xiaohongshu", platform_item_id="x1", canonical_url="https://a")
+    dup = CandidateItem(id=uuid4(), platform="xiaohongshu", platform_item_id="x1", canonical_url="https://b")
+    url_only = CandidateItem(id=uuid4(), platform="weixin", canonical_url="HTTPS://MP.TEST/A")
+
+    assert item_dedupe_key(first) == "xiaohongshu:id:x1"
+    assert item_dedupe_key(url_only) == "url:https://mp.test/a"
+    assert dedupe_candidate_items([first, dup, url_only]) == [first, url_only]
+    assert should_decode_item(first, decoded_item_ids={str(first.id)}).keep is False
+
+
+def test_acquisition_stage_wraps_existing_runner(monkeypatch):
+    batch_id = uuid4()
+    pipeline_run = PipelineRun(id=uuid4(), status="running", current_stage="search")
+    pipeline_repo = FakePipelineRepo()
+
+    def fake_run_batch(acquisition_repo, *, batch_id, settings, **kwargs):
+        return RunBatchResult(run_id=uuid4(), total_jobs=1, done=1, partial=0, failed=0, skipped=0)
+
+    monkeypatch.setattr(acquisition_runner, "run_batch", fake_run_batch)
+    result = acquisition_runner.run_acquisition_stage(
+        acquisition_repo=object(),
+        batch_id=batch_id,
+        settings=object(),
+        pipeline_repo=pipeline_repo,
+        pipeline_run=pipeline_run,
+    )
+
+    assert result.acquisition.done == 1
+    assert pipeline_repo.jobs[0].stage == "search"
+    assert pipeline_repo.jobs[-1].status == "done"
+
+
+def test_acquisition_stage_marks_pipeline_job_failed_and_reraises(monkeypatch):
+    pipeline_run = PipelineRun(id=uuid4(), status="running", current_stage="search")
+    pipeline_repo = FakePipelineRepo()
+
+    def fake_run_batch(*args, **kwargs):
+        raise RuntimeError("acquisition exploded")
+
+    monkeypatch.setattr(acquisition_runner, "run_batch", fake_run_batch)
+
+    with pytest.raises(RuntimeError, match="acquisition exploded"):
+        acquisition_runner.run_acquisition_stage(
+            acquisition_repo=object(),
+            batch_id=uuid4(),
+            settings=object(),
+            pipeline_repo=pipeline_repo,
+            pipeline_run=pipeline_run,
+        )
+
+    assert pipeline_repo.jobs[0].status == "running"
+    assert pipeline_repo.jobs[-1].status == "failed"
+    assert pipeline_repo.jobs[-1].error_message == "acquisition exploded"
+
+
+class FakeCandidateRepo:
+    def __init__(self):
+        self.item_id = uuid4()
+        self.items = [
+            CandidateItem(
+                id=self.item_id,
+                platform="xiaohongshu",
+                platform_item_id="x1",
+                canonical_url="https://xhs/1",
+                title="脚本",
+                status="candidate",
+            ),
+            CandidateItem(
+                id=uuid4(),
+                platform="xiaohongshu",
+                platform_item_id="x1",
+                title="重复",
+                status="candidate",
+            ),
+        ]
+
+    def list_creation_candidate_items(self, *, run_id: UUID | None = None, limit: int = 100):
+        return self.items
+
+    def list_media_assets_for_item(self, item_id: UUID):
+        return [
+            MediaAsset(
+                id=uuid4(),
+                item_id=item_id,
+                media_type="image",
+                cdn_url="https://cdn.test/1.jpg",
+                position=1,
+                status="done",
+            )
+        ]
+
+
+class FakeDecodeService:
+    def __init__(self):
+        self.posts: list[Post] = []
+
+    def decode_post(self, *, item_id: UUID, post: Post):
+        self.posts.append(post)
+        return DecodeWorkflowOutput(
+            item_id=item_id,
+            read_result=ReadResult(text="ok"),
+            gate_result=GateResult(passed=True),
+            status="decoded",
+        )
+
+
+def test_decode_stage_loads_candidate_media_and_dedupes():
+    repo = FakeCandidateRepo()
+    service = FakeDecodeService()
+
+    result = run_decode_stage(candidate_repo=repo, decode_service=service)
+
+    assert result.total == 1
+    assert result.decoded == 1
+    assert result.failed == 0
+    assert service.posts[0].image_urls == ["https://cdn.test/1.jpg"]
+
+
+def test_creation_pipeline_orchestrates_acquisition_then_decode(monkeypatch):
+    batch_id = uuid4()
+    acquisition_run_id = uuid4()
+    pipeline_repo = FakePipelineRepo()
+    calls = {}
+
+    def fake_acquisition_stage(**kwargs):
+        calls["acquisition"] = kwargs
+        return AcquisitionStageResult(
+            pipeline_job=None,
+            acquisition=RunBatchResult(
+                run_id=acquisition_run_id,
+                total_jobs=3,
+                done=3,
+                partial=0,
+                failed=0,
+                skipped=0,
+            ),
+        )
+
+    def fake_decode_stage(**kwargs):
+        calls["decode"] = kwargs
+        return DecodeBatchResult(total=2, decoded=2, skipped=0, failed=0, outputs=[])
+
+    monkeypatch.setattr(creation_pipeline, "run_acquisition_stage", fake_acquisition_stage)
+    monkeypatch.setattr(creation_pipeline, "run_decode_stage", fake_decode_stage)
+
+    result = creation_pipeline.run_creation_pipeline(
+        acquisition_repo=object(),
+        batch_id=batch_id,
+        settings=object(),
+        pipeline_repo=pipeline_repo,
+        decode_service=object(),
+        decode_limit=20,
+    )
+
+    assert result.pipeline_run is not None
+    assert result.pipeline_run.batch_id == batch_id
+    assert result.acquisition.acquisition.done == 3
+    assert result.decode is not None
+    assert result.decode.decoded == 2
+    assert calls["acquisition"]["pipeline_run"] == result.pipeline_run
+    assert calls["decode"]["run_id"] == acquisition_run_id
+    assert calls["decode"]["limit"] == 20

tests/test_platform_adapters.py

@@ -0,0 +1,125 @@
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+from acquisition.platforms.douyin import DouyinAdapter
+from acquisition.platforms import douyin as douyin_module
+from acquisition.platforms import weixin as weixin_module
+from acquisition.platforms import xiaohongshu as xhs_module
+from acquisition.platforms.weixin import WeixinAdapter
+from acquisition.platforms.xiaohongshu import XiaohongshuAdapter
+from core.models import Post
+
+
+def test_xiaohongshu_adapter_maps_search_and_detail(monkeypatch):
+    settings = SimpleNamespace(search_content_type="图文")
+
+    def fake_search(query, *, content_type, limit, settings, rate_limiter):
+        assert query == "脚本 开头"
+        assert content_type == "图文"
+        assert limit == 3
+        return [
+            {
+                "id": "xhs-1",
+                "url": "https://xhs.test/1",
+                "title": "脚本开头",
+                "nick_name": "作者 A",
+                "cover_url": "https://img.test/cover.jpg",
+            }
+        ]
+
+    def fake_fetch(source_id, *, settings, rate_limiter):
+        assert source_id == "xhs-1"
+        return Post(
+            id="xhs_xhs-1",
+            platform="xiaohongshu",
+            url="https://xhs.test/detail/1",
+            content_id="xhs-1",
+            title="详情标题",
+            content_type="图文",
+            body_text="详情正文",
+            image_urls=["https://img.test/1.jpg"],
+            author_name="详情作者",
+            raw={"source": "mock"},
+        )
+
+    monkeypatch.setattr(xhs_module, "search_xiaohongshu", fake_search)
+    monkeypatch.setattr(xhs_module, "fetch_post_detail", fake_fetch)
+
+    adapter = XiaohongshuAdapter()
+    candidate = adapter.search("脚本 开头", settings=settings, limit=3, rate_limiter=None)[0]
+    item = adapter.fetch_detail(candidate, settings=settings, rate_limiter=None)
+
+    assert candidate.source_id == "xhs-1"
+    assert candidate.author == "作者 A"
+    assert item.title == "详情标题"
+    assert item.image_urls == ["https://img.test/1.jpg"]
+    assert item.raw["source"] == "mock"
+
+
+def test_weixin_adapter_hashes_url_and_reads_detail(monkeypatch):
+    settings = SimpleNamespace()
+
+    def fake_search(query, *, limit, settings, rate_limiter):
+        assert query == "公众号 选题"
+        assert limit == 2
+        return [
+            {
+                "url": "https://mp.weixin.qq.com/s/abc",
+                "title": "公众号选题",
+                "nick_name": "公众号",
+                "cover_url": "https://img.test/wx.jpg",
+            }
+        ]
+
+    def fake_detail(url, *, settings, rate_limiter):
+        assert url == "https://mp.weixin.qq.com/s/abc"
+        return "公众号正文", ["https://img.test/wx-1.jpg"]
+
+    monkeypatch.setattr(weixin_module, "search_weixin", fake_search)
+    monkeypatch.setattr(weixin_module, "fetch_weixin_detail", fake_detail)
+
+    adapter = WeixinAdapter()
+    candidate = adapter.search("公众号 选题", settings=settings, limit=2, rate_limiter=None)[0]
+    item = adapter.fetch_detail(candidate, settings=settings, rate_limiter=None)
+
+    assert candidate.platform == "weixin"
+    assert len(candidate.source_id) == 16
+    assert item.body_text == "公众号正文"
+    assert item.image_urls == ["https://img.test/wx-1.jpg"]
+
+
+def test_douyin_adapter_maps_video_search_and_detail(monkeypatch):
+    settings = SimpleNamespace()
+
+    def fake_search(query, *, platform, content_type, limit, settings, rate_limiter):
+        assert query == "短视频 讲法"
+        assert platform == "douyin"
+        assert content_type == "视频"
+        assert limit == 1
+        return ["dy-1"]
+
+    def fake_fetch(source_id, *, settings, rate_limiter):
+        assert source_id == "dy-1"
+        return Post(
+            id="dy_dy-1",
+            platform="douyin",
+            url="https://douyin.test/1",
+            content_id="dy-1",
+            title="短视频讲法",
+            content_type="video",
+            body_text="视频文案",
+            video_urls=["https://video.test/1.mp4"],
+            raw={"source": "mock"},
+        )
+
+    monkeypatch.setattr(douyin_module, "search_keyword", fake_search)
+    monkeypatch.setattr(douyin_module, "fetch_post_detail", fake_fetch)
+
+    adapter = DouyinAdapter()
+    candidate = adapter.search("短视频 讲法", settings=settings, limit=1, rate_limiter=None)[0]
+    item = adapter.fetch_detail(candidate, settings=settings, rate_limiter=None)
+
+    assert candidate.source_id == "dy-1"
+    assert item.content_type == "video"
+    assert item.video_urls == ["https://video.test/1.mp4"]

tests/test_postgres_repository_contract.py

@@ -0,0 +1,121 @@
+from __future__ import annotations
+
+from uuid import uuid4
+
+from acquisition.repositories.postgres import PostgresAcquisitionRepository
+
+
+class RecordingPostgresRepo(PostgresAcquisitionRepository):
+    def __init__(self):
+        super().__init__(conn=None)
+        self.one_calls = []
+        self.all_calls = []
+        self.one_results = []
+        self.all_results = []
+
+    def _one(self, sql, params):
+        self.one_calls.append((sql, params))
+        return self.one_results.pop(0)
+
+    def _all(self, sql, params):
+        self.all_calls.append((sql, params))
+        return self.all_results.pop(0)
+
+
+def test_postgres_summary_uses_formal_run_job_item_tables():
+    run_id = uuid4()
+    batch_id = uuid4()
+    query_id = uuid4()
+    repo = RecordingPostgresRepo()
+    repo.one_results.append(
+        {
+            "id": run_id,
+            "run_key": "r1",
+            "batch_id": batch_id,
+            "status": "done",
+            "query_count": 1,
+            "job_count": 3,
+            "candidate_count": 7,
+            "creation_hit_count": 2,
+            "started_at": None,
+            "finished_at": None,
+        }
+    )
+    repo.all_results.append(
+        [
+            {
+                "query_id": query_id,
+                "query_text": "脚本 开头",
+                "job_count": 3,
+                "candidate_count": 7,
+                "creation_hit_count": 2,
+                "platforms": {"xiaohongshu": {"status": "done"}},
+            }
+        ]
+    )
+
+    summary = repo.get_run_summary(run_id)
+
+    first_sql, first_params = repo.one_calls[0]
+    second_sql, second_params = repo.all_calls[0]
+    assert "FROM acquisition_runs ar" in first_sql
+    assert "LEFT JOIN acquisition_jobs aj" in first_sql
+    assert "LEFT JOIN candidate_items ci" in first_sql
+    assert "LEFT JOIN item_classifications ic" in first_sql
+    assert first_params == (run_id,)
+    assert "jsonb_object_agg" in second_sql
+    assert "WHERE aj.run_id = %s" in second_sql
+    assert second_params == (run_id,)
+    assert summary["queries"][0]["query_id"] == query_id
+
+
+def test_postgres_repository_lists_only_creation_candidates_for_decode():
+    run_id = uuid4()
+    item_id = uuid4()
+    repo = RecordingPostgresRepo()
+    repo.all_results.append(
+        [
+            {
+                "id": item_id,
+                "platform": "xiaohongshu",
+                "platform_item_id": "x1",
+                "canonical_url": "https://xhs.test/1",
+                "status": "candidate",
+            }
+        ]
+    )
+
+    items = repo.list_creation_candidate_items(run_id=run_id, limit=20)
+
+    sql, params = repo.all_calls[0]
+    assert "JOIN acquisition_jobs aj ON aj.id = ci.job_id" in sql
+    assert "JOIN item_classifications ic ON ic.item_id = ci.id" in sql
+    assert "ic.is_creation_knowledge IS TRUE" in sql
+    assert "LIMIT %s" in sql
+    assert params == (run_id, 20)
+    assert items[0].id == item_id
+    assert items[0].platform == "xiaohongshu"
+
+
+def test_postgres_query_detail_loads_media_and_classifications_only_when_items_exist():
+    run_id = uuid4()
+    query_id = uuid4()
+    item_id = uuid4()
+    repo = RecordingPostgresRepo()
+    repo.one_results.append({"id": query_id, "query_text": "q", "status": "ready"})
+    repo.all_results.extend(
+        [
+            [{"id": uuid4(), "run_id": run_id, "query_id": query_id, "platform": "weixin"}],
+            [{"id": item_id, "platform": "weixin", "query_id": query_id, "status": "candidate"}],
+            [{"id": uuid4(), "item_id": item_id, "media_type": "image", "status": "done"}],
+            [{"id": uuid4(), "item_id": item_id, "status": "classified"}],
+        ]
+    )
+
+    detail = repo.get_query_detail(run_id=run_id, query_id=query_id)
+
+    assert len(repo.all_calls) == 4
+    assert "SELECT * FROM media_assets" in repo.all_calls[2][0]
+    assert "SELECT * FROM item_classifications" in repo.all_calls[3][0]
+    assert detail["items"][0]["id"] == item_id
+    assert detail["media_assets"][0]["item_id"] == item_id

tests/test_query_builder.py

@@ -0,0 +1,60 @@
+from __future__ import annotations
+
+from uuid import uuid4
+
+from acquisition.domain import Query, QueryBatch
+from acquisition.queries.builder import persist_query_batch
+
+
+class FakeRepo:
+    def __init__(self) -> None:
+        self.batch_kwargs = None
+        self.queries = []
+
+    def create_query_batch(self, **kwargs):
+        self.batch_kwargs = kwargs
+        return QueryBatch(id=uuid4(), **kwargs)
+
+    def add_query(self, **kwargs):
+        self.queries.append(kwargs)
+        return Query(id=uuid4(), **kwargs)
+
+
+def test_persist_query_batch_writes_formal_batch_and_query_contract():
+    repo = FakeRepo()
+    generated = {
+        "metadata": {"query_filter_prompt_version": "abc123"},
+        "families": [
+            {
+                "key": "f1",
+                "name": "实质 × 模态 × 业务阶段 × 知识类型",
+                "axes": ["实质", "模态"],
+                "items": [
+                    {
+                        "query": "反转 视频 脚本 怎么做",
+                        "parts": {"实质": "反转", "模态": "视频"},
+                        "keep": True,
+                        "valid": 8,
+                        "relevant": True,
+                        "reason": "可搜创作方法",
+                    }
+                ],
+            }
+        ],
+    }
+
+    batch, count = persist_query_batch(repo, generated, name="formal-demo")
+
+    assert batch.name == "formal-demo"
+    assert count == 1
+    assert repo.batch_kwargs["status"] == "ready"
+    assert repo.batch_kwargs["target_platforms"] == ["xiaohongshu", "weixin", "douyin"]
+    row = repo.queries[0]
+    assert row["batch_id"] == batch.id
+    assert row["query_text"] == "反转 视频 脚本 怎么做"
+    assert row["axes"] == {"实质": "反转", "模态": "视频"}
+    assert row["keep"] is True
+    assert row["filter_reason"] == "可搜创作方法"
+    assert row["metadata"]["family_key"] == "f1"
+    assert row["metadata"]["valid"] == 8
+    assert row["metadata"]["query_filter_prompt_version"] == "abc123"

tests/test_run_acquisition_cli.py

@@ -0,0 +1,74 @@
+from __future__ import annotations
+
+from uuid import uuid4
+
+import scripts.run_acquisition as cli
+from acquisition.runner import RunBatchResult
+
+
+class DummyTransaction:
+    def __init__(self, conn):
+        self.conn = conn
+
+    def __enter__(self):
+        return self.conn
+
+    def __exit__(self, exc_type, exc, tb):
+        return False
+
+
+def test_run_acquisition_cli_is_thin_delegate(monkeypatch, capsys):
+    captured = {}
+    batch_id = uuid4()
+    settings = object()
+    db_config = object()
+    conn = object()
+
+    monkeypatch.setattr(cli.Settings, "from_env", lambda env_file: settings)
+    monkeypatch.setattr(cli.CreationDbConfig, "from_env", lambda env_file: db_config)
+    monkeypatch.setattr(cli, "transaction", lambda config: DummyTransaction(conn))
+    monkeypatch.setattr(cli, "PostgresAcquisitionRepository", lambda c: ("repo", c))
+
+    def fake_run_batch(repo, **kwargs):
+        captured["repo"] = repo
+        captured.update(kwargs)
+        return RunBatchResult(
+            run_id=uuid4(),
+            total_jobs=1,
+            done=1,
+            partial=0,
+            failed=0,
+            skipped=0,
+        )
+
+    monkeypatch.setattr(cli, "run_batch", fake_run_batch)
+
+    rc = cli.main(
+        [
+            "--batch-id",
+            str(batch_id),
+            "--platform",
+            "weixin",
+            "--search-limit",
+            "2",
+            "--display-limit",
+            "1",
+            "--no-classify",
+            "--run-key",
+            "rk",
+            "--env-file",
+            ".env.test",
+        ]
+    )
+
+    out = capsys.readouterr().out
+    assert rc == 0
+    assert '"done": 1' in out
+    assert captured["repo"] == ("repo", conn)
+    assert captured["batch_id"] == batch_id
+    assert captured["settings"] is settings
+    assert captured["platforms"] == ("weixin",)
+    assert captured["search_limit"] == 2
+    assert captured["display_limit"] == 1
+    assert captured["classify"] is False
+    assert captured["run_key"] == "rk"

tests/test_video_extract.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 import json
 
 from core.config import PgConfig, Settings
-from creation_knowledge.integrations.video_extract import extract_video
+from decode_content.readers.video import extract_video
 from core.models import Post
 
 SEGMENTS = json.dumps({

tests/test_video_frames.py

@@ -24,7 +24,7 @@ FF = _ffmpeg()
 
 @pytest.mark.skipif(not FF, reason="ffmpeg 不可用")
 def test_extract_frames_from_generated_clip(tmp_path):
-    from creation_knowledge.integrations.video_frames import extract_frames
+    from decode_content.readers.video_frames import extract_frames
     from core.models import Card
 
     clip = tmp_path / "clip.mp4"