external_demand/app/gap_script_demand/judging.py

"""品类+解构词 画面改造可改造性 LLM 批量判定。"""

from __future__ import annotations

import time
from typing import Any

from app.core.open_router_llm import OpenRouterCallError, create_chat_completion
from app.gap_script_demand.exceptions import GapScriptDemandError
from app.gap_script_demand.llm_json import extract_json_object
from app.gap_script_demand.types import GapScriptDemandConfig

JUDGE_SYSTEM_PROMPT = """
你是一个专业的视频内容可改造性评估专家。你的任务是对给定的“品类 + 解构词”组合进行判定，筛选出其中适合用于画面改造的条目。
背景定义（已内置，无需输出解释）
- 画面改造：指替换原视频的画面与语音包，达到类似洗稿的效果。
- 解构词：以“品类 解构词”形式呈现的内容标签。
筛选目标
仅保留那些能够通过替换画面和语音包，生成一个主题明确、内容可独立成立的新视频的解构词组合。
淘汰标准（满足任意一条即判定为“不匹配”）
1. 画面主导型内容：原视频的核心价值几乎完全依赖视觉呈现，缺乏可被文字/语音替代的叙事或信息逻辑。典型示例包括但不限于：
  - 美景分享（如风景航拍、城市掠影）
  - 表演类（魔术、舞蹈、杂技、花式运动）
  - 旅行记录（无解说/无主题的Vlog）
  - 纯音乐/演奏（无歌词或无故事线）
  - 祝福类（节日祝贺、生日祝福画面）
  - 宠物日常（无情节的萌宠片段） 此类内容即使替换画面和语音，也无法保留原有吸引力或会彻底改变内容性质，故不可改造。
2. 无意义/弱主题解构词：解构词本身过于宽泛、模糊或与品类关联度极低，无法推导出具体的可替换内容框架。例如：
  - 解构词为“日常”“合集”“精选”“欣赏”“感受”等无法指向具体场景、动作或叙事的词汇；
  - 解构词与品类组合后，仍不能明确该视频要表达什么事件、过程或观点（如“美食 好吃”“旅行 好看”）。
判定原则
- 综合判断：必须同时参考品类和解构词，两者共同决定主题明确性。例如：
  - “美食 教程” → 主题明确（有操作流程），可改造 ✅
  - “美食 展示” → 仅画面陈列，无过程，不可改造 ❌
  - “舞蹈 教学” → 有动作拆解和语音指导，可改造 ✅
  - “舞蹈 表演” → 纯视觉观赏，不可改造 ❌
- 主题可迁移性：如果替换画面和语音后，新视频仍能传递相同的信息价值（如知识、步骤、故事、观点），则判定为可改造；否则为不可改造。

输出要求
严格只输出一个 JSON 对象，禁止输出 JSON 之外的任何内容。固定格式如下：
{
  "matched_demand_names": ["可改造的品类 解构词1", "可改造的品类 解构词2"]
}

约束：
- 只返回判定为可改造（匹配）的条目，不匹配的不要放入数组。
- matched_demand_names 中的每一项必须与用户提供的原文完全一致，不得改写、不得编造。
- 若没有任何条目可改造，返回 {"matched_demand_names": []}。
""".strip()


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(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"""请对以下“品类 解构词”组合逐条判定是否适合画面改造：

{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,
    *,
    demand_names: list[str],
    demand_lookup: dict[str, str],
) -> list[str]:
    matched_raw = _extract_matched_name_list(parsed)
    if matched_raw is None:
        raise GapScriptDemandError("llm output missing matched_demand_names")

    allowed_demand_names = set(demand_names)
    normalized: list[str] = []
    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("name")
                or item.get("品类 解构词")
                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)
    return normalized


def _llm_judge_batch(
    *,
    demand_names: list[str],
    config: GapScriptDemandConfig,
) -> list[str]:
    if not demand_names:
        return []

    demand_lookup = _build_demand_lookup(demand_names)
    user_message = _build_batch_user_message(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": JUDGE_SYSTEM_PROMPT},
                    {"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,
                demand_names=demand_names,
                demand_lookup=demand_lookup,
            )
        except (OpenRouterCallError, GapScriptDemandError, ValueError) as exc:
            last_error = exc
            if attempt < config.llm_max_attempts:
                time.sleep(config.llm_retry_sleep_seconds)

    raise GapScriptDemandError(
        f"gap script demand judge failed after {config.llm_max_attempts} attempts: {last_error}"
    ) from last_error


def judge_remake_suitable_demands(
    *,
    demand_names: list[str],
    config: GapScriptDemandConfig,
) -> list[str]:
    """按批次用 LLM 筛选适合画面改造的品类+解构词。"""
    if not demand_names:
        return []

    matched: list[str] = []
    seen: set[str] = set()
    batches = _chunked(demand_names, config.judge_batch_size)
    total_batches = len(batches)
    for batch_index, batch in enumerate(batches, start=1):
        print(
            f"gap script demand: judging batch {batch_index}/{total_batches} "
            f"size={len(batch)}",
            flush=True,
        )
        batch_matched = _llm_judge_batch(demand_names=batch, config=config)
        for demand_name in batch_matched:
            if demand_name in seen:
                continue
            seen.add(demand_name)
            matched.append(demand_name)
    return matched