content-find-agent-new/tests/test_search_intent.py

import copy

import pytest

from content_agent.business_modules import search_intent
from content_agent.errors import ContentAgentError
from content_agent.integrations.query_prompt_config import DEFAULT_PROFILE
from content_agent.run_service import RunService
from content_agent.schemas import RunStartRequest
from tests.p1_helpers import FakeQueryVariantClient, REAL_SOURCE_FIXTURE


FORBIDDEN_FIXED_BUSINESS_TERMS = [
    "\u8d2a\u8150",
    "\u57fa\u5c42\u516c\u804c\u4eba\u5458",
    "\u6848\u4f8b",
    "\u89e3\u8bfb",
    "\u8b66\u793a",
]


class _Runtime:
    def __init__(self):
        self.rows = {}

    def append_jsonl(self, _run_id, filename, rows):
        self.rows[filename] = rows


def _seed_pack():
    return {
        "seed_terms": ["中医养生"],
        "query_seed_points": [
            {
                "point_text": "分享气血食疗",
                "point_type": "目的点",
                "rank": 1,
                "post_id": "p1",
                "id": "qp1",
                "coverage_post_count": 5,
                "category_id": "c1",
            },
            {
                "point_text": "办公室八段锦",
                "point_type": "灵感点",
                "rank": 2,
                "post_id": "p2",
                "id": "qp2",
                "coverage_post_count": 3,
                "category_id": "c1",
            },
        ],
        "itemset_items": ["补气血"],
        "category_bindings": [{"category_id": "c1"}],
        "element_bindings": [
            {
                "element_id": "e1",
                "category_id": "c1",
                "sample_elements": [
                    {
                        "id": "point_1",
                        "name": "食疗补气血",
                        "point_type": "目的点",
                        "point_text": "分享气血食疗方法",
                        "post_id": "p1",
                    },
                    {
                        "id": "point_2",
                        "name": "八段锦",
                        "point_type": "灵感点",
                        "point_text": "办公室八段锦",
                        "post_id": "p2",
                    },
                    {
                        "id": "point_3",
                        "name": "日常保健",
                        "point_type": "关键点",
                        "point_text": "不应该进入搜索词",
                        "post_id": "p3",
                    },
                ],
            }
        ],
        "pattern_source_system": "pg_pattern_v2",
        "pattern_execution_id": 1987,
        "mining_config_id": 58,
        "source_post_id": "60219550",
        "matched_post_ids": ["60219550"],
        "itemset_ids": [1607977],
        "support": 0.2,
        "absolute_support": 31,
        "confidence": 0.8,
    }


def test_search_seed_and_queries_do_not_inject_fixed_business_terms(tmp_path):
    service = RunService(
        runtime_root=tmp_path / "runtime" / "v1",
        query_variant_client=FakeQueryVariantClient(
            {
                "爱国情感": "家国叙事素材",
                "人物故事": "榜样人物素材",
            }
        ),
    )
    state = service.start_run(
        RunStartRequest(platform_mode="mock", source=str(REAL_SOURCE_FIXTURE), strategy_version="V1")
    )
    run_id = state["run_id"]

    pattern_seed_pack = service.read_json(run_id, "pattern_seed_pack.json")
    queries = service.read_jsonl(run_id, "search_queries.jsonl")
    p2_queries = [
        row
        for row in queries
        if row["search_query_generation_method"] in {"item_single", "llm_variant"}
    ]

    assert pattern_seed_pack["seed_terms"] == ["爱国情感", "人物故事"]
    assert [row["search_query_id"] for row in p2_queries] == ["q_001", "q_002", "q_003", "q_004"]
    assert [row["search_query"] for row in p2_queries] == [
        "爱国情感",
        "家国叙事素材",
        "人物故事",
        "榜样人物素材",
    ]
    assert [row["search_query_generation_method"] for row in p2_queries] == [
        "item_single",
        "llm_variant",
        "item_single",
        "llm_variant",
    ]
    assert p2_queries[1]["llm_variant_of"] == "q_001"
    assert p2_queries[3]["llm_variant_of"] == "q_003"
    for value in [
        *pattern_seed_pack["seed_terms"],
        *(row["search_query"] for row in p2_queries),
    ]:
        assert not any(term in value for term in FORBIDDEN_FIXED_BUSINESS_TERMS)


def test_search_queries_preserve_source_terms_for_replay(tmp_path):
    service = RunService(
        runtime_root=tmp_path / "runtime" / "v1",
        query_variant_client=FakeQueryVariantClient(
            {
                "爱国情感": "家国叙事素材",
                "人物故事": "榜样人物素材",
            }
        ),
    )
    state = service.start_run(
        RunStartRequest(platform_mode="mock", source=str(REAL_SOURCE_FIXTURE), strategy_version="V1")
    )
    queries = service.read_jsonl(state["run_id"], "search_queries.jsonl")
    p2_queries = [
        query
        for query in queries
        if query["search_query_generation_method"] in {"item_single", "llm_variant"}
    ]

    expected_source_terms = [["爱国情感"], ["爱国情感"], ["人物故事"], ["人物故事"]]
    for query, source_terms in zip(p2_queries, expected_source_terms, strict=True):
        assert query["query_source_terms"] == source_terms
        assert query["query_source_fields"] == ["seed_terms"]
        assert query["raw_payload"]["query_source_terms"] == source_terms
        assert query["pattern_seed_ref"]["source_field"] == "seed_terms"
        assert query["pattern_seed_ref"]["seed_term"] == source_terms[0]
        assert query["raw_payload"]["pattern_seed_ref"]["seed_term"] == source_terms[0]

    llm_queries = [
        query
        for query in p2_queries
        if query["search_query_generation_method"] == "llm_variant"
    ]
    assert len(llm_queries) == 2
    for query in llm_queries:
        assert query["raw_payload"]["llm_prompt_version"] == "fake-query-prompt-v1"
        assert query["raw_payload"]["llm_generation_model"] == "fake-query-model"
        assert query["raw_payload"]["llm_input_evidence"]["source_field"] == "seed_terms"
        assert query["raw_payload"]["llm_input_evidence"]["itemset_items"]


def test_search_intent_custom_evidence_fields_whitelist():
    client = FakeQueryVariantClient({"中医养生": "气血食疗"})
    client.profile = copy.deepcopy(DEFAULT_PROFILE)
    client.profile["evidence_fields"] = ["seed_term", "support"]
    runtime = _Runtime()

    queries = search_intent.run("run_1", "policy_1", _seed_pack(), runtime, client)

    llm_query = [row for row in queries if row["search_query_generation_method"] == "llm_variant"][0]
    assert list(llm_query["llm_input_evidence"].keys()) == ["seed_term", "support"]
    assert list(llm_query["raw_payload"]["llm_input_evidence"].keys()) == ["seed_term", "support"]
    assert llm_query["query_source_fields"] == ["seed_terms"]


def test_v4_search_intent_uses_three_direct_sources_without_llm_variant():
    client = FakeQueryVariantClient({"中医养生": "气血食疗"})
    runtime = _Runtime()

    queries = search_intent.run(
        "run_1",
        "policy_1",
        _seed_pack(),
        runtime,
        client,
        strategy_version="V4",
    )

    assert client.calls == []
    # 三路首轮搜索池:seed_terms、query_seed_points、分类叶子元素;不再调 LLM 变体。
    assert [row["search_query"] for row in queries] == [
        "中医养生",
        "气血食疗",
        "办公室八段锦",
        "食疗补气血",
        "八段锦",
        "日常保健",
    ]
    assert [row["search_query_generation_method"] for row in queries] == [
        "seed_term",
        "query_seed_point",
        "query_seed_point",
        "category_leaf_element",
        "category_leaf_element",
        "category_leaf_element",
    ]
    assert all("llm_variant_of" not in row for row in queries)
    assert runtime.rows["search_queries.jsonl"] == queries

    first = next(row for row in queries if row["search_query"] == "气血食疗")
    assert first["pattern_seed_ref"]["query_source_type"] == "query_seed_point"
    assert first["pattern_seed_ref"]["query_source_text"] == "气血食疗"
    assert first["pattern_seed_ref"]["query_source_rank"] == 1
    source_ref = first["query_source_refs"][0]["source_ref"]
    assert source_ref["point_type"] == "目的点"
    assert source_ref["point_text"] == "分享气血食疗"  # 原文不被覆盖
    assert source_ref["cleaned_text"] == "气血食疗"
    assert source_ref["qsp_rank"] == 1
    assert first["raw_payload"]["query_source_refs"][0]["query_source_text"] == "气血食疗"


def test_v4_falls_back_to_seed_terms_when_query_seed_points_empty():
    runtime = _Runtime()
    seed_pack = _seed_pack()
    seed_pack["query_seed_points"] = []
    seed_pack["element_bindings"] = []

    queries = search_intent.run(
        "run_1",
        "policy_1",
        seed_pack,
        runtime,
        FakeQueryVariantClient(),
        strategy_version="V4",
    )

    assert [row["search_query"] for row in queries] == ["中医养生"]
    assert all(row["search_query_generation_method"] == "seed_term" for row in queries)


def test_v4_search_intent_dedupes_query_seed_points():
    runtime = _Runtime()
    seed_pack = _seed_pack()
    seed_pack["query_seed_points"] = [
        {"point_text": "气血食疗", "point_type": "灵感点", "rank": 1, "post_id": "p1", "id": "x1"},
        {"point_text": "气血食疗", "point_type": "灵感点", "rank": 2, "post_id": "p2", "id": "x2"},
    ]

    queries = search_intent.run(
        "run_1",
        "policy_1",
        seed_pack,
        runtime,
        FakeQueryVariantClient(),
        strategy_version="V4",
    )

    matching = [row for row in queries if row["search_query"] == "气血食疗"]
    assert len(matching) == 1
    assert matching[0]["search_query_generation_method"] == "query_seed_point"


def test_search_intent_custom_generic_filter_blocks_query():
    client = FakeQueryVariantClient({"中医养生": "禁用泛词"})
    client.profile = copy.deepcopy(DEFAULT_PROFILE)
    client.profile["generic_filter"] = {"queries": ["禁用泛词"], "tokens": []}

    with pytest.raises(ContentAgentError) as exc:
        search_intent.run("run_1", "policy_1", _seed_pack(), _Runtime(), client)

    assert exc.value.error_code == "QUERY_GENERATION_FAILED"
    assert exc.value.detail["reason"] == "llm_variant_generic"


def test_search_intent_rejects_unsupported_variants_per_seed():
    client = FakeQueryVariantClient({"中医养生": "气血食疗"})
    client.profile = copy.deepcopy(DEFAULT_PROFILE)
    client.profile["variants_per_seed"] = 2

    with pytest.raises(ContentAgentError) as exc:
        search_intent.run("run_1", "policy_1", _seed_pack(), _Runtime(), client)

    assert exc.value.error_code == "QUERY_GENERATION_FAILED"
    assert exc.value.detail == {"reason": "variants_per_seed_unsupported", "variants_per_seed": 2}