content-find-agent-new/content_agent/business_modules/search_intent.py

from __future__ import annotations

import logging
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

from content_agent.constants import RUNTIME_RECORD_SCHEMA_VERSION
from content_agent.errors import ContentAgentError, ErrorCode
from content_agent.integrations import config_store
from content_agent.integrations.query_prompt_config import DEFAULT_PROFILE
from content_agent.interfaces import QueryVariantClient, QueryVariantResult, RuntimeFileStore
from content_agent.record_payload import with_raw_payload

GENERIC_QUERIES = {
    "内容",
    "视频",
    "热门",
    "推荐",
    "短视频",
    "热门视频",
    "推荐视频",
    "热门内容",
    "推荐内容",
    "相关视频",
    "相关内容",
    "热" + "点视频",
    "热" + "点内容",
}
GENERIC_QUERY_TOKENS = (
    "短视频",
    "热门",
    "推荐",
    "相关",
    "热" + "点",
    "内容",
    "视频",
    "素材",
    "资料",
    "信息",
    "话题",
)
SEARCH_INTENT_POLICY_PATH = Path("tech_documents/数据接口与来源/search_intent_policy.json")
DEFAULT_V4_QUERY_SEED_POINTS_LIMIT = 2


def run(
    run_id: str,
    policy_run_id: str,
    pattern_seed_pack: dict[str, Any],
    runtime: RuntimeFileStore,
    query_variant_client: QueryVariantClient,
    *,
    strategy_version: str | None = None,
    platform: str = "",
) -> list[dict[str, Any]]:
    created_at = datetime.now(timezone.utc).isoformat()
    seed_terms = _terms(pattern_seed_pack.get("seed_terms"))
    if not seed_terms:
        raise _query_generation_error("seed_terms_empty")
    if str(strategy_version or "").upper() == "V4":
        return _run_v4(
            run_id,
            policy_run_id,
            pattern_seed_pack,
            runtime,
            created_at,
            seed_terms,
            query_variant_client,
            platform,
        )

    profile = getattr(query_variant_client, "profile", DEFAULT_PROFILE)
    variants_per_seed = int(profile.get("variants_per_seed", 1))
    if variants_per_seed != 1:
        raise _query_generation_error(
            "variants_per_seed_unsupported",
            {"variants_per_seed": variants_per_seed},
        )
    evidence_fields = profile.get("evidence_fields")
    generic_filter = profile.get("generic_filter")

    search_queries: list[dict[str, Any]] = []
    seen_queries: set[str] = set()

    for seed_index, seed_term in enumerate(seed_terms):
        item_query_id = f"q_{seed_index * 2 + 1:03d}"
        variant_query_id = f"q_{seed_index * 2 + 2:03d}"
        pattern_seed_ref = _pattern_seed_ref(pattern_seed_pack, seed_term, seed_index)
        item_single = _base_query_record(
            run_id=run_id,
            policy_run_id=policy_run_id,
            search_query_id=item_query_id,
            search_query=seed_term,
            generation_method="item_single",
            seed_term=seed_term,
            pattern_seed_ref=pattern_seed_ref,
            created_at=created_at,
        )
        _reserve_query(item_single, seen_queries, seed_term=seed_term, method="item_single")
        search_queries.append(item_single)

        evidence_context = _llm_input_evidence(
            pattern_seed_pack,
            seed_terms,
            seed_term,
            seed_index,
            sorted(seen_queries),
            evidence_fields=evidence_fields,
        )
        variant = _generate_variant(query_variant_client, seed_term, evidence_context)
        variant_query = _normalize_query(variant.query)
        _validate_variant_query(variant_query, seed_term, seen_queries, generic_filter=generic_filter)
        llm_variant = _base_query_record(
            run_id=run_id,
            policy_run_id=policy_run_id,
            search_query_id=variant_query_id,
            search_query=variant_query,
            generation_method="llm_variant",
            seed_term=seed_term,
            pattern_seed_ref=pattern_seed_ref,
            created_at=created_at,
        )
        llm_variant.update(
            {
                "llm_variant_of": item_query_id,
                "llm_input_evidence": variant.input_evidence,
                "llm_prompt_version": variant.prompt_version,
                "llm_generation_model": variant.model,
            }
        )
        _reserve_query(llm_variant, seen_queries, seed_term=seed_term, method="llm_variant")
        search_queries.append(llm_variant)

    expected_count = len(seed_terms) * 2
    if len(search_queries) != expected_count:
        raise _query_generation_error(
            "query_count_mismatch",
            {
                "seed_terms_count": len(seed_terms),
                "expected_query_count": expected_count,
                "actual_query_count": len(search_queries),
            },
        )

    search_queries = [with_raw_payload(row) for row in search_queries]
    runtime.append_jsonl(run_id, "search_queries.jsonl", search_queries)
    return search_queries


def _run_v4(
    run_id: str,
    policy_run_id: str,
    pattern_seed_pack: dict[str, Any],
    runtime: RuntimeFileStore,
    created_at: str,
    seed_terms: list[str],
    query_variant_client: QueryVariantClient,
    platform: str,
) -> list[dict[str, Any]]:
    # 首轮搜索池三路并行:pattern seed_terms + 上游 query_seed_points + 分类叶子元素。
    seed_candidates = _v4_seed_candidates(pattern_seed_pack, seed_terms)
    query_seed_point_candidates = _v4_query_seed_point_candidates(pattern_seed_pack)
    category_candidates = _v4_category_element_candidates(pattern_seed_pack)
    if not seed_candidates and not query_seed_point_candidates and not category_candidates:
        raise _query_generation_error("v4_query_candidates_empty")

    # M9D Gate 2:仅非抖音,AI 判 query 是否易搜 50+;拿不准/异常从宽放过。
    apply_gate2 = bool(platform) and platform != "douyin"
    search_queries: list[dict[str, Any]] = []
    seen_queries: set[str] = set()

    def _admit(candidate_list: list[dict[str, Any]], *, use_gate2: bool, fallback: bool = False) -> None:
        for candidate in candidate_list:
            query = _normalize_query(candidate.get("text"))
            if not query:
                continue
            normalized_key = _normalize_query_key(query)
            if normalized_key in seen_queries:
                continue
            if use_gate2 and not _gate2_keep(query, query_variant_client):
                continue
            if fallback:
                candidate["source_ref"]["gate2_fallback"] = True
            seen_queries.add(normalized_key)
            query_id = f"q_{len(search_queries) + 1:03d}"
            row = _base_query_record(
                run_id=run_id,
                policy_run_id=policy_run_id,
                search_query_id=query_id,
                search_query=query,
                generation_method=candidate["generation_method"],
                seed_term=query,
                pattern_seed_ref=_v4_pattern_seed_ref(pattern_seed_pack, candidate, query),
                created_at=created_at,
                query_source_terms=[query],
                query_source_fields=candidate["query_source_fields"],
            )
            row["query_source_refs"] = [_v4_query_source_ref(candidate, query)]
            search_queries.append(row)

    # seed_terms 和分类叶子元素是需求侧必搜来源;query_seed_points 仍保留非抖音 Gate2。
    _admit(seed_candidates, use_gate2=False)
    _admit(query_seed_point_candidates, use_gate2=apply_gate2)
    _admit(category_candidates, use_gate2=False)

    if not search_queries:
        raise _query_generation_error("v4_search_queries_empty")

    search_queries = [with_raw_payload(row) for row in search_queries]
    runtime.append_jsonl(run_id, "search_queries.jsonl", search_queries)
    return search_queries


def _v4_seed_candidates(
    pattern_seed_pack: dict[str, Any],
    seed_terms: list[str],
) -> list[dict[str, Any]]:
    return [
        {
            "text": seed_term,
            "query_source_type": "seed_term",
            "generation_method": "seed_term",
            "query_source_fields": ["seed_terms"],
            "rank": index + 1,
            "source_ref": {
                "source_field": "seed_terms",
                "source_index": index,
                "seed_term": seed_term,
                "query_source_ref_id": f"seed_terms:{index}",
            },
        }
        for index, seed_term in enumerate(seed_terms)
    ]


def _v4_query_seed_point_candidates(pattern_seed_pack: dict[str, Any]) -> list[dict[str, Any]]:
    """首轮搜索池:吃上游 query_seed_points(已按覆盖度排序的灵感/目的点),按 rank 升序。
    目的点剥动作词前缀(复用 _cleanup_purpose_point),原始 point_text 存进 source_ref 不覆盖。"""
    points = pattern_seed_pack.get("query_seed_points")
    if not isinstance(points, list):
        return []
    ordered = sorted(
        (point for point in points if isinstance(point, dict)),
        key=lambda point: point.get("rank") if isinstance(point.get("rank"), int) else 1_000_000,
    )
    limit = _v4_query_seed_points_limit()
    if limit is not None:
        ordered = ordered[:limit]
    candidates: list[dict[str, Any]] = []
    for point in ordered:
        point_type = str(point.get("point_type") or "").strip()
        if point_type not in {"目的点", "灵感点"}:
            continue
        point_text = _normalize_query(point.get("point_text"))
        if not point_text:
            continue
        text = _cleanup_purpose_point(point_text) if point_type == "目的点" else point_text
        if not text or _is_generic_query(text):
            continue
        rank = len(candidates) + 1
        post_id = point.get("post_id")
        point_id = point.get("id") or point.get("topic_point_id")
        candidates.append(
            {
                "text": text,
                "query_source_type": "query_seed_point",
                "generation_method": "query_seed_point",
                "query_source_fields": ["query_seed_points.point_text"],
                "rank": rank,
                "source_ref": {
                    "query_source_ref_id": f"qsp:post:{post_id or 'unknown_post'}#point:{point_id or rank}",
                    "point_type": point_type,
                    "point_text": point_text,
                    "cleaned_text": text,
                    "coverage_post_count": point.get("coverage_post_count"),
                    "qsp_rank": point.get("rank"),
                    "post_id": post_id,
                    "category_id": point.get("category_id"),
                    "element_id": point.get("source_element_id"),
                },
            }
        )
    return candidates


def _v4_query_seed_points_limit() -> int | None:
    try:
        config, _raw = config_store.load_json(SEARCH_INTENT_POLICY_PATH)
    except FileNotFoundError:
        return DEFAULT_V4_QUERY_SEED_POINTS_LIMIT
    value = config.get("v4_query_seed_points_limit") if isinstance(config, dict) else None
    if value is None:
        return DEFAULT_V4_QUERY_SEED_POINTS_LIMIT
    try:
        parsed = int(value)
    except (TypeError, ValueError):
        return DEFAULT_V4_QUERY_SEED_POINTS_LIMIT
    return parsed if parsed >= 0 else DEFAULT_V4_QUERY_SEED_POINTS_LIMIT


def _v4_category_element_candidates(pattern_seed_pack: dict[str, Any]) -> list[dict[str, Any]]:
    candidates: list[dict[str, Any]] = []
    for binding_index, binding in enumerate(_bindings(pattern_seed_pack), start=1):
        for element_index, sample in enumerate(_sample_elements(binding), start=1):
            name = _normalize_query(sample.get("name"))
            if not name or _is_generic_query(name):
                continue
            rank = len(candidates) + 1
            category_id = sample.get("category_id") or binding.get("category_id")
            element_id = sample.get("source_element_id") or sample.get("id")
            source_ref = {
                "query_source_ref_id": _element_source_ref_id(category_id, element_id, name, rank),
                "binding_rank": binding_index,
                "element_rank": element_index,
                "element_name": name,
                "element_id": element_id,
                "category_id": category_id,
                "itemset_item_id": binding.get("itemset_item_id"),
                "post_id": sample.get("post_id"),
                "point_type": sample.get("point_type"),
            }
            candidates.append(
                {
                    "text": name,
                    "query_source_type": "category_terminal_element",
                    "generation_method": "category_leaf_element",
                    "query_source_fields": ["element_bindings.sample_elements.name"],
                    "rank": rank,
                    "source_ref": source_ref,
                }
            )
    return candidates


def _bindings(pattern_seed_pack: dict[str, Any]) -> list[dict[str, Any]]:
    values = pattern_seed_pack.get("element_bindings")
    return [item for item in values if isinstance(item, dict)] if isinstance(values, list) else []


def _sample_elements(binding: dict[str, Any]) -> list[dict[str, Any]]:
    values = binding.get("sample_elements")
    return [item for item in values if isinstance(item, dict)] if isinstance(values, list) else []


def _element_source_ref_id(category_id: Any, element_id: Any, name: str, rank: int) -> str:
    category = category_id or "unknown_category"
    element = element_id or name or rank
    return f"category:{category}#element:{element}"


_LOG = logging.getLogger(__name__)
_PURPOSE_PREFIX_PATH = Path("product_documents/配置/purpose_point_prefixes.v1.json")


def _purpose_point_prefixes() -> tuple[str, ...]:
    """目的点要剥的动作词前缀:**全部来自 JSON 配置**(config_store 按 mtime 缓存),代码不写死任何前缀表。
    以后加词只改 `product_documents/配置/purpose_point_prefixes.v1.json`。
    读不到/为空 → 不剥前缀并**告警**(而非静默退回旧表,免得配置没生效却看着像正常)。"""
    try:
        data, _ = config_store.load_json(_PURPOSE_PREFIX_PATH)
        prefixes = data.get("prefixes") if isinstance(data, dict) else None
        if isinstance(prefixes, list):
            cleaned = tuple(str(p) for p in prefixes if p)
            if cleaned:
                return cleaned
    except (FileNotFoundError, OSError, ValueError) as exc:
        _LOG.warning("目的点前缀配置读取失败(%s),本次不剥前缀:%s", _PURPOSE_PREFIX_PATH, exc)
        return ()
    _LOG.warning("目的点前缀配置缺 prefixes 或为空(%s),本次不剥前缀", _PURPOSE_PREFIX_PATH)
    return ()


def _cleanup_purpose_point(value: str) -> str:
    text = _normalize_query(value)
    for prefix in _purpose_point_prefixes():
        if text.startswith(prefix) and len(text) > len(prefix):
            text = text[len(prefix):].strip(" ，,。:：")
            break
    return _normalize_query(text)


def _v4_pattern_seed_ref(
    pattern_seed_pack: dict[str, Any],
    candidate: dict[str, Any],
    query: str,
) -> dict[str, Any]:
    source_ref = dict(candidate.get("source_ref") or {})
    return {
        "query_source_type": candidate["query_source_type"],
        "query_source_ref_id": source_ref.get("query_source_ref_id"),
        "query_source_text": query,
        "query_source_rank": candidate["rank"],
        "pattern_execution_id": pattern_seed_pack.get("pattern_execution_id"),
        "mining_config_id": pattern_seed_pack.get("mining_config_id"),
        "source_post_id": pattern_seed_pack.get("source_post_id"),
        "matched_post_ids": pattern_seed_pack.get("matched_post_ids") or [],
        "itemset_ids": _itemset_ids(pattern_seed_pack),
        "source_ref": source_ref,
    }


def _v4_query_source_ref(candidate: dict[str, Any], query: str) -> dict[str, Any]:
    source_ref = dict(candidate.get("source_ref") or {})
    return {
        "query_source_type": candidate["query_source_type"],
        "query_source_ref_id": source_ref.get("query_source_ref_id"),
        "query_source_text": query,
        "query_source_rank": candidate["rank"],
        "generation_method": candidate["generation_method"],
        "source_ref": source_ref,
    }


def _terms(values: Any) -> list[str]:
    if not isinstance(values, list):
        return []
    unique: list[str] = []
    seen: set[str] = set()
    for value in values:
        if not isinstance(value, str):
            continue
        term = " ".join(value.split()).strip()
        if not term or term in seen:
            continue
        seen.add(term)
        unique.append(term)
    return unique


def _base_query_record(
    *,
    run_id: str,
    policy_run_id: str,
    search_query_id: str,
    search_query: str,
    generation_method: str,
    seed_term: str,
    pattern_seed_ref: dict[str, Any],
    created_at: str,
    query_source_terms: list[str] | None = None,
    query_source_fields: list[str] | None = None,
) -> dict[str, Any]:
    return {
        "record_schema_version": RUNTIME_RECORD_SCHEMA_VERSION,
        "run_id": run_id,
        "policy_run_id": policy_run_id,
        "search_query_id": search_query_id,
        "search_query": search_query,
        "search_query_generation_method": generation_method,
        "discovery_start_source": "pattern_itemset",
        "previous_discovery_step": "pattern_search_query",
        "search_query_effect_status": "pending",
        "query_source_terms": query_source_terms or [seed_term],
        "query_source_fields": query_source_fields or ["seed_terms"],
        "pattern_seed_ref": pattern_seed_ref,
        "created_at": created_at,
    }


def _generate_variant(
    query_variant_client: QueryVariantClient,
    seed_term: str,
    evidence_context: dict[str, Any],
) -> QueryVariantResult:
    try:
        result = query_variant_client.generate_variant(
            seed_term=seed_term,
            evidence_context=evidence_context,
        )
    except ContentAgentError:
        raise
    except Exception as exc:
        raise _query_generation_error(
            "llm_variant_exception",
            {
                "seed_term": seed_term,
                "exception_type": type(exc).__name__,
            },
        ) from exc
    if not isinstance(result, QueryVariantResult):
        raise _query_generation_error(
            "llm_variant_result_invalid",
            {
                "seed_term": seed_term,
                "result_type": type(result).__name__,
            },
        )
    return result


def _validate_variant_query(
    query: str,
    seed_term: str,
    seen_queries: set[str],
    *,
    generic_filter: dict[str, Any] | None = None,
) -> None:
    if not query:
        raise _query_generation_error("llm_variant_empty", {"seed_term": seed_term})
    if query == seed_term:
        raise _query_generation_error("llm_variant_same_as_seed", {"seed_term": seed_term})
    if query in seen_queries:
        raise _query_generation_error(
            "llm_variant_duplicate",
            {
                "seed_term": seed_term,
                "search_query": query,
            },
        )
    if _is_generic_query(query, generic_filter=generic_filter):
        raise _query_generation_error(
            "llm_variant_generic",
            {
                "seed_term": seed_term,
                "search_query": query,
            },
        )


def _reserve_query(
    row: dict[str, Any],
    seen_queries: set[str],
    *,
    seed_term: str,
    method: str,
) -> None:
    query = _normalize_query(row.get("search_query", ""))
    if not query:
        raise _query_generation_error(
            "search_query_empty",
            {
                "seed_term": seed_term,
                "search_query_generation_method": method,
            },
        )
    if query in seen_queries:
        raise _query_generation_error(
            "search_query_duplicate",
            {
                "seed_term": seed_term,
                "search_query": query,
                "search_query_generation_method": method,
            },
        )
    row["search_query"] = query
    seen_queries.add(query)


def _pattern_seed_ref(
    pattern_seed_pack: dict[str, Any],
    seed_term: str,
    seed_index: int,
) -> dict[str, Any]:
    return {
        "source_field": "seed_terms",
        "source_index": seed_index,
        "seed_term": seed_term,
        "pattern_execution_id": pattern_seed_pack.get("pattern_execution_id"),
        "mining_config_id": pattern_seed_pack.get("mining_config_id"),
        "source_post_id": pattern_seed_pack.get("source_post_id"),
        "matched_post_ids": pattern_seed_pack.get("matched_post_ids") or [],
        "itemset_ids": _itemset_ids(pattern_seed_pack),
    }


def _llm_input_evidence(
    pattern_seed_pack: dict[str, Any],
    seed_terms: list[str],
    seed_term: str,
    seed_index: int,
    existing_search_queries: list[str],
    evidence_fields: list[str] | None = None,
) -> dict[str, Any]:
    evidence = {
        "seed_term": seed_term,
        "seed_terms": seed_terms,
        "existing_search_queries": existing_search_queries,
        "source_field": "seed_terms",
        "source_index": seed_index,
        "itemset_items": pattern_seed_pack.get("itemset_items")
        or pattern_seed_pack.get("itemsets")
        or [],
        "category_bindings": pattern_seed_pack.get("category_bindings") or [],
        "element_bindings": pattern_seed_pack.get("element_bindings") or [],
        "pattern_source_system": pattern_seed_pack.get("pattern_source_system"),
        "pattern_execution_id": pattern_seed_pack.get("pattern_execution_id"),
        "mining_config_id": pattern_seed_pack.get("mining_config_id"),
        "source_post_id": pattern_seed_pack.get("source_post_id"),
        "matched_post_ids": pattern_seed_pack.get("matched_post_ids") or [],
        "itemset_ids": _itemset_ids(pattern_seed_pack),
        "support": pattern_seed_pack.get("support"),
        "absolute_support": pattern_seed_pack.get("absolute_support"),
        "confidence": pattern_seed_pack.get("confidence"),
    }
    if evidence_fields is None:
        return evidence
    return {field: evidence[field] for field in evidence_fields if field in evidence}


def _itemset_ids(pattern_seed_pack: dict[str, Any]) -> list[Any]:
    direct = pattern_seed_pack.get("itemset_ids")
    if isinstance(direct, list):
        return direct

    itemsets = pattern_seed_pack.get("itemsets")
    if not isinstance(itemsets, list):
        return []

    ids: list[Any] = []
    for itemset in itemsets:
        if not isinstance(itemset, dict):
            continue
        itemset_id = itemset.get("itemset_id")
        if itemset_id is not None:
            ids.append(itemset_id)
    return ids


def _normalize_query(value: Any) -> str:
    if not isinstance(value, str):
        return ""
    return " ".join(value.split()).strip()


def _normalize_query_key(value: Any) -> str:
    return _normalize_query(value).casefold()


def _is_generic_query(query: str, generic_filter: dict[str, Any] | None = None) -> bool:
    generic_queries = set((generic_filter or {}).get("queries") or GENERIC_QUERIES)
    generic_tokens = tuple((generic_filter or {}).get("tokens") or GENERIC_QUERY_TOKENS)
    compact = "".join(query.split())
    if not compact or len(compact) <= 1:
        return True
    if not any(char.isalnum() for char in compact):
        return True
    if compact in generic_queries:
        return True
    remainder = compact
    for token in generic_tokens:
        remainder = remainder.replace(token, "")
    return not remainder


def _gate2_keep(query_text: str, query_variant_client: QueryVariantClient) -> bool:
    """M9D Gate 2:True=保留;client 无 judge 方法(mock)/异常 → 从宽保留。"""
    judge = getattr(query_variant_client, "judge_query_fifty_plus", None)
    if not callable(judge):
        return True
    try:
        return bool(judge(query_text))
    except Exception:
        return True


def _query_generation_error(
    reason: str,
    detail: dict[str, Any] | None = None,
) -> ContentAgentError:
    return ContentAgentError(
        ErrorCode.QUERY_GENERATION_FAILED,
        "query generation failed",
        {
            "reason": reason,
            **(detail or {}),
        },
    )