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

"""M4 离线测试：注入假 chat，逐环节断言；assemble 纯函数对齐设计文档 §6。"""
from __future__ import annotations

import json

from creation_knowledge.models import (
    Deconstruction,
    ExtractedContent,
    KnowledgeItem,
    Post,
    Scope,
)
from creation_knowledge.stages import (
    build_ingest_payload,
    deconstruct_item,
    screen_post,
    split_post,
)


def _post() -> Post:
    return Post(
        id="xhs_abc", url="https://www.xiaohongshu.com/explore/abc", content_id="abc",
        title="只要学会这几样，写短视频脚本真的不难", platform="xiaohongshu",
        author_name="海狸教自媒体运营", topic_list=["短视频"],
    )


def _content() -> ExtractedContent:
    return ExtractedContent(
        text="短视频脚本包含标题、拍摄地点、分镜等要素",
        from_image="九宫格图讲了脚本6要素", is_empty=False,
    )


def test_screen_passed():
    r = screen_post(_post(), _content(),
                    chat=lambda s, u: {"passed": True, "score": 8, "reason": "含明确 How"})
    assert r.passed and r.score == 8


def test_screen_feeds_extracted_not_body():
    seen = {}

    def chat(s, u):
        seen["u"] = u
        return {"passed": True, "score": 7, "reason": "x"}

    screen_post(_post(), _content(), chat=chat)
    assert "九宫格" in seen["u"]  # 多模态提取内容进了 prompt，而非空 body_text


def test_split_consistency_and_drop_empty():
    def chat(s, u):
        return {"items": [
            {"title": "脚本要素", "knowledge_types": ["what", "how", "why"],
             "what": "脚本含6要素", "why": "null", "how": "逐个填要素",
             "evidence": ["原句"]},
            {"title": "空", "knowledge_types": ["what"],
             "what": "null", "why": "null", "how": "null", "evidence": []},
        ]}

    items = split_post(_post(), _content(), chat=chat)
    assert len(items) == 1  # 三个都空的片段被丢弃
    it = items[0]
    assert it.why is None  # "null" 归一为 None
    assert set(it.knowledge_types) == {"what", "how"}  # 与非空字段对齐，why 被剔除


def test_deconstruct_filters_invalid():
    def chat(s, u):
        return {"stages": ["选题", "脚本", "乱写"],
                "scopes": [{"scope_type": "form", "value": "操作流程"},
                           {"scope_type": "bad", "value": "x"},
                           {"scope_type": "effect", "value": ""}],
                "stage_reason": "r1", "scope_reason": "r2"}

    d = deconstruct_item(KnowledgeItem(title="t", knowledge_types=["how"], how="做法"),
                         chat=chat)
    assert d.stages == ["选题", "脚本"]  # 非法阶段过滤
    assert [s.scope_type for s in d.scopes] == ["form"]  # 非法 scope / 空 value 过滤


def test_assemble_matches_design_doc():
    item = KnowledgeItem(title="评论区选题法", knowledge_types=["what", "how"],
                         what="从评论提取痛点", why=None, how="收集高赞评论改写",
                         evidence=["原句a"])
    deco = Deconstruction(
        stages=["选题", "脚本"],
        scopes=[Scope(scope_type="form", value="公式框架"),
                Scope(scope_type="effect", value="生成选题")],
        stage_reason="既选题又脚本", scope_reason="形式+作用")
    p = build_ingest_payload(_post(), item, deco)

    assert p.source["id"] == "xhs_abc"
    assert p.source["source_type"] == "post"
    assert p.source["source_metadata"]["platform"] == "xiaohongshu"
    assert p.dim_attributes == ["what", "how"]
    assert p.dim_creations == ["选题", "脚本"]
    assert {"scope_type": "form", "value": "公式框架"} in p.scopes
    assert json.loads(p.content) == {"what": "从评论提取痛点", "why": None,
                                     "how": "收集高赞评论改写"}
    assert "原文证据" in [e["key"] for e in p.custom_ext]