external_demand/app/festival_demand/matching.py

"""需求词与当天活跃节日/节点事件的 LLM 语义匹配。"""

from __future__ import annotations

import json
import time
from typing import Any

from app.core.open_router_llm import OpenRouterCallError, create_chat_completion
from app.festival_demand.exceptions import FestivalDemandError
from app.festival_demand.llm_json import extract_json_object
from app.festival_demand.types import FestivalDemandConfig

MATCH_SYSTEM_PROMPT = """
你是一个专业的视频内容主题匹配判断助手。你的任务是根据给定的"节点事件"及其类型，判断"视频特征词"是否符合该节点事件的核心主题，从而确定视频是否属于此节点事件。

判断规则
请严格参照以下节点事件类型与主题对应关系：
- 类型为"节日/节气"：主题以祝福、知识分享为主（如节日祝福、民俗科普、节气养生知识等）。
- 类型为"纪念日"：主题以家国仇恨、勿忘国耻、怀念英烈、歌颂伟人事迹为主（如历史事件回顾、英雄人物纪念等）。
- 类型为"特定天气时段"：主题以描述天气、健康养生为主（如天气变化提醒、时令养生方法等）。
- 类型为"特定民生热点"：主题以民生、知识分享为主（如政策解读、考试备考、消费警示等）。

输入信息
用户会提供以下内容：
- 节点事件名称：<事件名称>
- 节点事件类型：<从"节日/节气、纪念日、特定天气时段、特定民生热点"中选择其一>
- 视频特征词列表：多个待判断的特征词

判断逻辑
1. 首先明确该节点事件所属类型对应的允许主题范围。
2. 逐条分析每个视频特征词所表达的核心内容。若特征词明显指向上述主题范围（不需要完全一致，语义相近、属正常延伸即可），则判定匹配；若特征词与主题无关（如纯娱乐、无关的日常生活、其他不相干领域），则判定不匹配。
3. 如遇模棱两可的情况，倾向于严格判断，即只有当特征词明显属于主题方向时才匹配，否则不匹配。

输出要求
严格只输出一个 JSON 对象，禁止输出 JSON 之外的任何内容。固定格式如下：
{
  "matched_demand_names": ["匹配成功的特征词1", "匹配成功的特征词2"]
}

约束：
- 只返回判定为匹配的特征词，不匹配的不要放入数组。
- matched_demand_names 中的每一项必须与用户提供的视频特征词原文完全一致，不得改写、不得编造。
- 若没有任何特征词匹配，返回 {"matched_demand_names": []}。

示例1：
节点事件名称：春节
节点事件类型：节日/节气
视频特征词列表：拜年祝福、年夜饭做法、搞笑模仿
输出：{"matched_demand_names": ["拜年祝福", "年夜饭做法"]}

示例2：
节点事件名称：九一八事变纪念日
节点事件类型：纪念日
视频特征词列表：搞笑模仿、宠物日常
输出：{"matched_demand_names": []}

示例3：
节点事件名称：三伏
节点事件类型：特定天气时段
视频特征词列表：防暑小妙招、三伏贴教程、明星八卦
输出：{"matched_demand_names": ["防暑小妙招", "三伏贴教程"]}

现在，请根据上述规则处理用户输入。
"""


def _chunked(items: list[str], batch_size: int) -> list[list[str]]:
    size = max(batch_size, 1)
    return [items[index : index + size] for index in range(0, len(items), size)]


def _build_demand_lookup(demand_names: list[str]) -> dict[str, str]:
    lookup: dict[str, str] = {}
    for item in demand_names:
        demand_name = str(item).strip()
        if not demand_name:
            continue
        lookup.setdefault(demand_name, demand_name)
        compact_key = "".join(demand_name.split())
        if compact_key:
            lookup.setdefault(compact_key, demand_name)
    return lookup


def _resolve_demand_name(demand_name: str, demand_lookup: dict[str, str]) -> str | None:
    value = demand_name.strip()
    if not value:
        return None
    return demand_lookup.get(value) or demand_lookup.get("".join(value.split()))


def _build_batch_user_message(
    *,
    festival_name: str,
    event_type: str,
    demand_names: list[str],
) -> str:
    feature_lines = "\n".join(
        f"{index}. {demand_name}" for index, demand_name in enumerate(demand_names, start=1)
    )
    return f"""节点事件名称：{festival_name}
节点事件类型：{event_type}

视频特征词列表：
{feature_lines}"""


def _extract_matched_name_list(parsed: Any) -> list[Any] | None:
    if not isinstance(parsed, dict):
        return None

    for key in ("matched_demand_names", "matched", "demand_names", "results"):
        value = parsed.get(key)
        if isinstance(value, list):
            return value
    return None


def _normalize_matched_demand_names(
    parsed: Any,
    *,
    festival_name: str,
    event_type: str,
    demand_names: list[str],
    demand_lookup: dict[str, str],
) -> list[dict[str, Any]]:
    matched_raw = _extract_matched_name_list(parsed)
    if matched_raw is None:
        raise FestivalDemandError("llm output missing matched_demand_names")

    allowed_demand_names = set(demand_names)
    normalized: list[dict[str, Any]] = []
    seen: set[str] = set()
    for item in matched_raw:
        if isinstance(item, str):
            raw_name = item
        elif isinstance(item, dict):
            raw_name = str(
                item.get("demand_name")
                or item.get("feature_word")
                or item.get("视频特征词")
                or item.get("name")
                or ""
            )
        else:
            continue

        demand_name = _resolve_demand_name(str(raw_name), demand_lookup)
        if not demand_name or demand_name not in allowed_demand_names:
            continue
        if demand_name in seen:
            continue
        seen.add(demand_name)
        normalized.append(
            {
                "demand_name": demand_name,
                "festival_name": festival_name,
                "event_type": event_type,
            }
        )
    return normalized


def _llm_match_festival_batch(
    *,
    festival: dict[str, Any],
    demand_names: list[str],
    config: FestivalDemandConfig,
) -> list[dict[str, Any]]:
    festival_name = str(festival.get("name") or "").strip()
    event_type = str(festival.get("event_type") or "").strip()
    if not festival_name or not demand_names:
        return []

    demand_lookup = _build_demand_lookup(demand_names)
    user_message = _build_batch_user_message(
        festival_name=festival_name,
        event_type=event_type,
        demand_names=demand_names,
    )

    last_error: Exception | None = None
    for attempt in range(1, config.llm_max_attempts + 1):
        try:
            resp = create_chat_completion(
                [
                    {"role": "system", "content": MATCH_SYSTEM_PROMPT.strip()},
                    {"role": "user", "content": user_message},
                ],
                model=config.llm_model,
                temperature=config.llm_temperature,
                max_tokens=config.llm_max_tokens,
            )
            parsed = extract_json_object(str(resp.get("content") or ""))
            return _normalize_matched_demand_names(
                parsed,
                festival_name=festival_name,
                event_type=event_type,
                demand_names=demand_names,
                demand_lookup=demand_lookup,
            )
        except (OpenRouterCallError, FestivalDemandError, ValueError) as exc:
            last_error = exc
            if attempt < config.llm_max_attempts:
                time.sleep(config.llm_retry_sleep_seconds)

    raise FestivalDemandError(
        f"festival demand match failed for {festival_name} "
        f"after {config.llm_max_attempts} attempts: {last_error}"
    ) from last_error


def match_demand_names_to_festivals(
    *,
    demand_names: list[str],
    festivals: list[dict[str, Any]],
    config: FestivalDemandConfig,
) -> list[dict[str, Any]]:
    """按批次将需求词与活跃节日列表做 LLM 匹配。"""
    if not demand_names or not festivals:
        return []

    all_matches: list[dict[str, Any]] = []
    seen: set[tuple[str, str]] = set()
    batches = _chunked(demand_names, config.match_batch_size)
    total_batches = len(batches)
    for batch_index, batch in enumerate(batches, start=1):
        for festival in festivals:
            festival_name = str(festival.get("name") or "").strip()
            print(
                f"festival demand: matching batch {batch_index}/{total_batches} "
                f"festival={festival_name} size={len(batch)}",
                flush=True,
            )
            batch_matches = _llm_match_festival_batch(
                festival=festival,
                demand_names=batch,
                config=config,
            )
            for row in batch_matches:
                dedupe_key = (row["demand_name"], row["festival_name"])
                if dedupe_key in seen:
                    continue
                seen.add(dedupe_key)
                all_matches.append(row)
    return all_matches


def collect_matched_demand_names(matches: list[dict[str, Any]]) -> list[str]:
    """收集去重后的 demand_name 列表。"""
    names: list[str] = []
    seen: set[str] = set()
    for item in matches:
        demand_name = str(item.get("demand_name") or "").strip()
        if not demand_name or demand_name in seen:
            continue
        seen.add(demand_name)
        names.append(demand_name)
    return names