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

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"