Create-knowledge-find-decode/tests/test_app_api.py

from __future__ import annotations

from contextlib import nullcontext
from pathlib import Path
from uuid import uuid4

from fastapi.testclient import TestClient

from acquisition.domain import AcquisitionRun, Query, QueryBatch
from app.api import app
from app.dependencies import (
    get_acquisition_repository,
    get_creation_db_config,
    get_decode_repository,
    get_pipeline_repository,
)
from app.routes import manual_queries
from decode_content.models import (
    DecodeJob,
    DecodeResult,
    IngestRecord,
    KnowledgeParticle,
    PayloadDraft,
    ScopeResult,
)
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",
                    "unique_key": "xhs:xhs1",
                    "canonical_url": "https://xhs.test/1",
                    "content_type": "图文",
                    "content_mode": "image_post",
                    "title": "脚本开头",
                    "author_name": "作者",
                    "body_text": "先定受众，再写开头钩子",
                    "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",
                    "error_message": None,
                }
            ],
        }

    def get_latest_query_result_list(self, query_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,
            },
            "run": {
                "id": self.run_id,
                "batch_id": self.batch_id,
                "run_key": "run-1",
                "status": "running",
            },
            "jobs": [{"id": uuid4(), "platform": "xiaohongshu", "status": "done"}],
            "items": [
                {
                    "id": self.item_id,
                    "platform": "xiaohongshu",
                    "title": "脚本开头",
                    "raw_summary": "先定受众",
                    "status": "candidate",
                    "content_mode": "image_post",
                    "metadata": {
                        "page_index": 1,
                        "matched_candidate": {"raw": "large"},
                        "source_payload": {"raw": "large"},
                    },
                }
            ],
            "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,
                    "status": "done",
                    "error_message": None,
                }
            ],
            "decode_summaries": [
                {
                    "item_id": self.item_id,
                    "decode_status": "decoded",
                    "particle_count": 1,
                    "payload_count": 1,
                }
            ],
        }

    def get_candidate_item(self, item_id):
        assert item_id == self.item_id
        return self.get_query_detail(run_id=self.run_id, query_id=self.query_id)["items"][0]

    def list_media_assets_for_item(self, item_id):
        assert item_id == self.item_id
        return self.get_query_detail(run_id=self.run_id, query_id=self.query_id)["media_assets"]


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

    def list_knowledge_particles(self, item_id=None):
        return [
            KnowledgeParticle(
                id=uuid4(),
                item_id=item_id or uuid4(),
                particle_type="what",
                title="What I know",
                content={"content": "一个可复用知识点"},
                status="draft",
            )
        ]

    def list_scope_results(self, item_id=None):
        return [
            ScopeResult(
                id=uuid4(),
                item_id=item_id or uuid4(),
                scope_type="intent",
                scope_value="吸引注意",
                status="draft",
            )
        ]

    def list_ingest_records(self, item_id=None):
        return [
            IngestRecord(
                id=uuid4(),
                payload_draft_id=self.payload_id,
                target_system="dry-run",
                status="ingested",
            )
        ]


class FakePipelineRepo:
    def get_pipeline_run(self, run_id):
        return PipelineRun(
            id=run_id,
            run_key="pipeline-run-1",
            status="running",
            current_stage="decode",
        )


class FakeManualRepo:
    def __init__(self):
        self.batch_id = uuid4()
        self.run_id = uuid4()
        self.queries = []
        self.runs = []
        self.updates = []

    def create_query_batch(self, **kwargs):
        self.batch_kwargs = kwargs
        return QueryBatch(id=self.batch_id, **kwargs)

    def add_query(self, **kwargs):
        query = Query(id=uuid4(), **kwargs)
        self.queries.append(query)
        return query

    def create_acquisition_run(self, **kwargs):
        run = AcquisitionRun(id=self.run_id, **kwargs)
        self.runs.append(run)
        return run

    def update_acquisition_run(self, run_id, **kwargs):
        assert run_id == self.run_id
        self.updates.append(kwargs)
        base = self.runs[-1].model_dump()
        base.update(kwargs)
        return AcquisitionRun(**base)


def _client(repo: FakeAcquisitionRepo):
    app.dependency_overrides[get_acquisition_repository] = lambda: repo
    return TestClient(app)


def test_manual_query_batch_api_creates_ready_batch_and_starts_pipeline(monkeypatch, tmp_path):
    repo = FakeManualRepo()
    popen_calls = []

    class FakeProcess:
        pid = 12345

    def fake_popen(cmd, **kwargs):
        popen_calls.append((cmd, kwargs))
        return FakeProcess()

    monkeypatch.setattr(manual_queries, "transaction", lambda _config: nullcontext(object()))
    monkeypatch.setattr(manual_queries, "PostgresAcquisitionRepository", lambda _conn: repo)
    monkeypatch.setattr(manual_queries, "RUNTIME_MANUAL_DIR", tmp_path)
    monkeypatch.setattr(manual_queries.subprocess, "Popen", fake_popen)
    app.dependency_overrides[get_creation_db_config] = lambda: object()
    client = TestClient(app)

    try:
        response = client.post(
            "/api/query-batches/manual",
            json={
                "queries": [
                    "符号 视频 灵感 怎么做",
                    "符号 视频 灵感 怎么做",
                    {"query_text": "叙事体裁 视频 灵感 有哪些", "axes": {"实质": "叙事体裁"}},
                ]
            },
        )
        assert response.status_code == 200
        data = response.json()
        assert data["status"] == "queued"
        assert data["batch_id"] == str(repo.batch_id)
        assert data["run_id"] == str(repo.run_id)
        assert data["pid"] == 12345
        assert data["query_count"] == 2
        assert data["summary_url"] == f"/api/acquisition/runs/{repo.run_id}/summary"
        assert repo.batch_kwargs["generation_method"] == "manual_query_api_v1"
        assert repo.batch_kwargs["target_platforms"] == ["xiaohongshu", "weixin", "douyin"]
        assert [query.query_text for query in repo.queries] == [
            "符号 视频 灵感 怎么做",
            "叙事体裁 视频 灵感 有哪些",
        ]
        assert all(query.keep is True and query.status == "ready" for query in repo.queries)
        assert all(query.metadata["family_key"] == "manual" for query in repo.queries)
        assert repo.runs[0].status == "pending"
        assert repo.runs[0].metadata["log_path"].endswith(".log")
        assert repo.runs[0].metadata["command"]
        assert popen_calls
        cmd, kwargs = popen_calls[0]
        assert "scripts/run_creation_pipeline.py" in cmd[1]
        assert ["--platform", "xiaohongshu"] == cmd[-6:-4]
        assert "--no-dry-ingest-record" not in cmd
        assert kwargs["cwd"] == str(manual_queries.ROOT)
        assert str(manual_queries.ROOT) in kwargs["env"]["PYTHONPATH"]
        assert repo.updates == []
    finally:
        app.dependency_overrides.clear()


def test_manual_query_batch_api_extracts_family_json(monkeypatch, tmp_path):
    repo = FakeManualRepo()
    monkeypatch.setattr(manual_queries, "transaction", lambda _config: nullcontext(object()))
    monkeypatch.setattr(manual_queries, "PostgresAcquisitionRepository", lambda _conn: repo)
    monkeypatch.setattr(manual_queries, "RUNTIME_MANUAL_DIR", tmp_path)
    monkeypatch.setattr(manual_queries.subprocess, "Popen", lambda *args, **kwargs: type("P", (), {"pid": 1})())
    app.dependency_overrides[get_creation_db_config] = lambda: object()
    client = TestClient(app)

    try:
        response = client.post(
            "/api/query-batches/manual",
            json={
                "families": [
                    {
                        "key": "f1",
                        "name": "实质正交",
                        "items": [{"query": "公共安全 视频 灵感 怎么做"}],
                    }
                ]
            },
        )
        assert response.status_code == 200
        assert repo.queries[0].query_text == "公共安全 视频 灵感 怎么做"
        assert repo.queries[0].metadata["family_key"] == "manual"
        assert repo.queries[0].metadata["source_family_key"] == "f1"
    finally:
        app.dependency_overrides.clear()


def test_manual_query_batch_api_rejects_empty_and_invalid_platform():
    client = TestClient(app)
    assert client.post("/api/query-batches/manual", json={"queries": ["  "]}).status_code == 422
    assert client.post(
        "/api/query-batches/manual",
        json={"queries": ["符号 视频 灵感 怎么做"], "target_platforms": ["bilibili"]},
    ).status_code == 422


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["content_mode"] == "image_post"
        assert item["body_text"] == "先定受众，再写开头钩子"
        assert item["media_assets"][0]["cdn_url"] == "https://cdn.test/1.jpg"
        assert item["classification"]["is_creation_knowledge"] is True

        latest = client.get(f"/api/query-generation/latest/queries/{repo.query_id}").json()
        lightweight_item = latest["items"][0]
        assert "body_text" not in lightweight_item
        assert "source_payload" not in lightweight_item
        assert "matched_candidate" not in lightweight_item["metadata"]
        assert "source_payload" not in lightweight_item["metadata"]
        assert lightweight_item["raw_summary"] == "先定受众"
        assert len(lightweight_item["media_assets"]) == 1
        assert lightweight_item["decode_summary"]["particle_count"] == 1
        assert latest["platforms"]["xiaohongshu"]["item_ids"] == [str(repo.item_id)]
        assert "items" not in latest["platforms"]["xiaohongshu"]
    finally:
        app.dependency_overrides.clear()


def test_app_compresses_large_json_responses():
    repo = FakeAcquisitionRepo()
    client = _client(repo)
    try:
        response = client.get(
            f"/api/query-generation/latest/queries/{repo.query_id}",
            headers={"Accept-Encoding": "gzip"},
        )
        assert response.status_code == 200
        assert response.headers.get("content-encoding") == "gzip"
    finally:
        app.dependency_overrides.clear()


def test_query_generation_preview_defaults_to_first_two_families():
    client = TestClient(app)
    data = client.get("/api/query-generation/preview?per=2").json()

    assert data["summary"]["family_keys"] == ["f1", "f2"]
    assert [family["key"] for family in data["families"]] == ["f1", "f2"]
    assert data["summary"]["query_count"] == 4
    assert data["metadata"]["active_family_keys"] == ["f1", "f2"]


def test_query_generation_preview_can_return_full_cartesian_combinations():
    client = TestClient(app)
    data = client.get("/api/query-generation/preview?per=0&batch_n=1").json()

    assert data["summary"]["family_keys"] == ["f1", "f2"]
    assert data["summary"]["query_count"] == 36
    assert [len(family["items"]) for family in data["families"]] == [18, 18]


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_pipeline_route_returns_501_instead_of_legacy_data_when_not_wired():
    client = TestClient(app)
    run_id = uuid4()

    assert client.get(f"/api/pipeline/runs/{run_id}").status_code == 501


def test_formal_decode_payload_pipeline_routes_have_override_success_paths():
    acquisition_repo = FakeAcquisitionRepo()
    decode_repo = FakeDecodeRepo()
    pipeline_repo = FakePipelineRepo()
    item_id = acquisition_repo.item_id
    run_id = uuid4()
    app.dependency_overrides[get_acquisition_repository] = lambda: acquisition_repo
    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"

        detail = client.get(f"/api/decode/items/{item_id}/detail").json()
        assert detail["item"]["id"] == str(item_id)
        assert detail["knowledge_particles"][0]["title"] == "What I know"
        assert detail["ingest_records"][0]["status"] == "ingested"

        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