pod_trend_agent/graph/nodes/screen_node.py

"""节点 4/6：合规筛选（screen）。

调用可插拔 LLM 后端做合规筛查 + 结构化四要素；后端调用失败时降级 MockBackend。
按 topic 把筛查结果映射回 scored 候选（补 score/sources/country），再做最终风险过滤
（blocked 丢弃；review 按 keep_review 决定）。
"""
from pathlib import Path
from typing import Any, Dict, List

from graph.loader import load_system_prompt
from graph.llms import DEFAULT_SYSTEM_PROMPT, get_backend
from graph.style_rules import COUNTRY_AESTHETICS
from graph.validate import with_fallback


@with_fallback("screen")
def screen_node(state: Dict[str, Any]) -> Dict[str, Any]:
    country = state["country"]
    scored: List[Dict[str, Any]] = state.get("scored_rows") or []
    config = state["config"]
    cc = state["country_config"]
    llm_cfg = config.get("llm_screen") or {}
    provider = llm_cfg.get("provider", "mock")
    keep_review = bool(llm_cfg.get("keep_review", False))
    batch_size = int(llm_cfg.get("max_topics_per_call", 12))
    blacklist = [str(b).lower() for b in (config.get("blacklist") or [])]

    prompts_dir = Path(state["prompts_dir"])
    system_prompt = load_system_prompt(prompts_dir, DEFAULT_SYSTEM_PROMPT)
    aesthetic_hint = cc.get("style_hint") or COUNTRY_AESTHETICS.get(country, {}).get("style_hint", "")

    # 排除已用热点（去重生效：已用 topic 不再进入本次简报，每次跑都用新热点）
    try:
        import json as _json
        used_p = Path(state.get("cache_dir") or state.get("output_dir", "")) / "used_designs.json"
        if used_p.exists():
            ud = _json.loads(used_p.read_text(encoding="utf-8")).get("used", []) or []
            used_topics = {str(u.get("topic", "")).strip().lower() for u in ud}
            before = len(scored)
            scored = [c for c in scored if str(c.get("topic", "")).strip().lower() not in used_topics]
            if len(scored) < before:
                print(f"[screen] 排除已用热点 {before - len(scored)} 条（去重），剩余 {len(scored)} 条可选")
    except Exception:  # noqa: BLE001
        pass

    # 简报数量 = 用多少生成多少：llm_screen.max_briefs 配置优先，否则按扩展后的总任务数
    #（每个产品一个热点；数量 N=每个款-颜色条目的设计数 → 总任务=条目数×N）；
    # 前台显示不依赖简报（读 collected 完整池 + used_designs 剔除已用，用完即从前台消失）。
    pcfg = config.get("product") or {}
    ls_cfg = config.get("llm_screen") or {}
    limit = int(ls_cfg.get("max_briefs") or 0)
    if not limit and pcfg.get("spu_tasks"):
        limit = len(pcfg.get("spu_tasks") or [])   # 扩展后总任务数（=产品数）
    if not limit:
        limit = int(pcfg.get("spu_count") or 0)
    if limit > 0:
        ordered = sorted(scored, key=lambda c: -(float(c.get("score") or 0)))
        scored_limited = ordered[:limit]
        print(f"[screen] 简报限量 {limit} → 筛前 {len(scored_limited)} 个高分热点（共 {len(scored)} 个）")
    else:
        scored_limited = scored
        print(f"[screen] 简报全量 {len(scored_limited)} 条（未限量，全部生成简报）")

    topics = [c["topic"] for c in scored_limited]

    backend = get_backend(provider)
    if provider != "mock":
        backend.bind_config(llm_cfg)
        if not backend.has_key:
            print("[screen] 未检测到 LLM api_key（请配置 llm_screen.api_key 或环境变量 "
                  "LLM_API_KEY/OPENAI_API_KEY），降级 Mock 兜底。")
            backend = get_backend("mock")
    try:
        screened = backend.screen(topics, country, aesthetic_hint, system_prompt, blacklist, batch_size)
    except Exception as e:  # noqa: BLE001
        print(f"[screen] {provider} 调用失败，降级 Mock: {e}")
        backend = get_backend("mock")
        screened = backend.screen(topics, country, aesthetic_hint, system_prompt, blacklist, batch_size)

    # 映射回 scored 候选
    by_topic = {s.get("topic", "").lower(): s for s in screened}
    out: List[Dict[str, Any]] = []
    missing: List[Dict[str, Any]] = []
    for it in scored_limited:
        s = by_topic.get(it["topic"].lower())
        if s is None:
            missing.append(it)
            continue
        s = dict(s)
        s["country"] = country
        s["score"] = it.get("score", 0)
        s["sources"] = it.get("sources", "")
        out.append(s)

    # LLM 漏判的主题用 Mock 单独补全，避免丢数据（而非整批降级）
    if missing:
        print(f"[screen] LLM 漏判 {len(missing)} 个主题，用 Mock 单独补全："
              f"{[m['topic'] for m in missing]}")
        mock = get_backend("mock")
        m_res = mock.screen(
            [m["topic"] for m in missing], country, aesthetic_hint,
            system_prompt, blacklist, batch_size,
        )
        m_by = {r.get("topic", "").lower(): r for r in m_res}
        for it in missing:
            s = m_by.get(it["topic"].lower())
            if s is None:
                continue
            s = dict(s)
            s["country"] = country
            s["score"] = it.get("score", 0)
            s["sources"] = it.get("sources", "")
            out.append(s)

    # 最终风险过滤：只滤 blocked（硬拦截）；review（待复核）保留——
    # 由 assign_hotspots 的 allow_review 决定是否参与分配（openai 模式 review+concept 可用），
    # 避免 review 被静默丢弃导致"任务 N 个但简报不足、设计缺失"
    kept: List[Dict[str, Any]] = []
    for r in out:
        lvl = r.get("risk_level", "safe")
        if lvl == "blocked":
            continue
        kept.append(r)

    stats = dict(state.get("stats") or {})
    stats["screen"] = {"screened": len(out), "kept": len(kept)}
    return {"screened": kept, "stats": stats}