external_demand/app/hot_content/demand_quality.py

"""需求质量判断：事件性、老年性 LLM 评分。

流程（串行两次 LLM，评分阶段互不截断）：
1. 微信指数达标后，构建待评需求（特征点组合 + 有匹配的短语）
2. 对全部待评需求执行事件性 LLM 评分
3. 对同一批全部待评需求执行老年性 LLM 评分
4. 导出表 / ODPS 写入时再过滤：标题保留（微信指数 + 灵感/目的点匹配）+ 事件性、老年性双达标
"""

from __future__ import annotations

import json
import re
import time
from typing import Any

from app.core.open_router_llm import OpenRouterCallError, create_chat_completion
from app.hot_content.demand_export import ITEM_TYPE_ELEMENT, ITEM_TYPE_PHRASE
from app.hot_content.exceptions import HotContentFlowError

TYPE_FEATURE_POINT = "特征点"
TYPE_PHRASE = "短语"


def _normalize_demand_key(value: str) -> str:
    return "".join(str(value or "").split())


def _dedupe_texts(texts: list[str]) -> list[str]:
    deduped: list[str] = []
    seen: set[str] = set()
    for raw in texts:
        text = str(raw).strip()
        if not text:
            continue
        keys = {text, _normalize_demand_key(text)}
        if keys & seen:
            continue
        seen.update(keys)
        deduped.append(text)
    return deduped


def _has_matched_demand(row: dict[str, Any]) -> bool:
    return bool(str(row.get("matched_demand") or "").strip())


def _record_wxindex_score(export_rows: list[dict[str, Any]]) -> float:
    scores: list[float] = []
    for row in export_rows:
        try:
            scores.append(float(row.get("wxindex_latest_score") or 0))
        except (TypeError, ValueError):
            continue
    return max(scores) if scores else 0.0


def passes_wxindex_gate(
    export_rows: list[dict[str, Any]],
    *,
    wxindex_threshold: float,
) -> bool:
    """记录级微信指数是否达标，用于决定是否进入质量判断。"""
    return _record_wxindex_score(export_rows) >= wxindex_threshold


def _extract_json_object(text: str) -> dict[str, Any]:
    raw = text.strip()
    if raw.startswith("```"):
        raw = re.sub(r"^```(?:json)?\s*", "", raw)
        raw = re.sub(r"\s*```$", "", raw)
    try:
        parsed = json.loads(raw)
        if isinstance(parsed, dict):
            return parsed
    except json.JSONDecodeError:
        pass
    match = re.search(r"\{[\s\S]*\}", raw)
    if not match:
        raise HotContentFlowError("llm output is not json object")
    try:
        parsed = json.loads(match.group(0))
    except json.JSONDecodeError as exc:
        raise HotContentFlowError(f"llm output invalid json: {exc}") from exc
    if not isinstance(parsed, dict):
        raise HotContentFlowError("llm output is not json object")
    return parsed


def _candidate_key(demand_type: str, demand_text: str) -> tuple[str, str]:
    return demand_type.strip(), _normalize_demand_key(demand_text)


def build_feature_combo_text(export_rows: list[dict[str, Any]]) -> str:
    element_texts = _dedupe_texts(
        [
            str(row.get("item_text") or "").strip()
            for row in export_rows
            if str(row.get("item_type") or "") == ITEM_TYPE_ELEMENT
            and _has_matched_demand(row)
        ]
    )
    return " ".join(element_texts)


def build_quality_candidates(
    export_rows: list[dict[str, Any]],
    *,
    wxindex_threshold: float,
) -> list[dict[str, str]]:
    """微信指数达标时，构建特征点组合与短语两类待评需求。"""
    if not passes_wxindex_gate(export_rows, wxindex_threshold=wxindex_threshold):
        return []

    candidates: list[dict[str, str]] = []
    seen: set[tuple[str, str]] = set()

    feature_combo = build_feature_combo_text(export_rows)
    if feature_combo:
        key = _candidate_key(TYPE_FEATURE_POINT, feature_combo)
        if key not in seen:
            seen.add(key)
            candidates.append(
                {
                    "demand_type": TYPE_FEATURE_POINT,
                    "demand_text": feature_combo,
                }
            )

    for row in export_rows:
        if str(row.get("item_type") or "") != ITEM_TYPE_PHRASE:
            continue
        if not _has_matched_demand(row):
            continue
        phrase_text = str(row.get("item_text") or "").strip()
        if not phrase_text:
            continue
        key = _candidate_key(TYPE_PHRASE, phrase_text)
        if key in seen:
            continue
        seen.add(key)
        candidates.append(
            {
                "demand_type": TYPE_PHRASE,
                "demand_text": phrase_text,
            }
        )
    return candidates


def _normalize_score(value: Any) -> float | None:
    try:
        score = float(value)
    except (TypeError, ValueError):
        return None
    if score < 0:
        return 0.0
    if score > 10:
        return 10.0
    return score


def _build_score_lookup(result_json: dict[str, Any] | None) -> dict[tuple[str, str], dict[str, Any]]:
    lookup: dict[tuple[str, str], dict[str, Any]] = {}
    if not isinstance(result_json, dict):
        return lookup
    items = result_json.get("items") or []
    if not isinstance(items, list):
        return lookup
    for item in items:
        if not isinstance(item, dict):
            continue
        demand_type = str(item.get("demand_type") or "").strip()
        demand_text = str(item.get("demand_text") or "").strip()
        if not demand_type or not demand_text:
            continue
        lookup[_candidate_key(demand_type, demand_text)] = item
    return lookup


def lookup_quality_scores(
    *,
    demand_type: str,
    demand_text: str,
    event_sense_json: dict[str, Any] | None,
    senior_fit_json: dict[str, Any] | None,
) -> tuple[float | None, float | None]:
    key = _candidate_key(demand_type, demand_text)
    event_item = _build_score_lookup(event_sense_json).get(key)
    senior_item = _build_score_lookup(senior_fit_json).get(key)
    event_score = _normalize_score(event_item.get("score")) if event_item else None
    senior_score = _normalize_score(senior_item.get("score")) if senior_item else None
    return event_score, senior_score


def quality_passed(
    *,
    demand_type: str,
    demand_text: str,
    event_sense_json: dict[str, Any] | None,
    senior_fit_json: dict[str, Any] | None,
    event_threshold: float,
    senior_threshold: float,
) -> bool:
    event_score, senior_score = lookup_quality_scores(
        demand_type=demand_type,
        demand_text=demand_text,
        event_sense_json=event_sense_json,
        senior_fit_json=senior_fit_json,
    )
    if event_score is None or senior_score is None:
        return False
    return event_score >= event_threshold and senior_score >= senior_threshold


def attach_quality_scores_to_export_rows(
    export_rows: list[dict[str, Any]],
    *,
    event_sense_json: dict[str, Any] | None,
    senior_fit_json: dict[str, Any] | None,
) -> list[dict[str, Any]]:
    feature_combo = build_feature_combo_text(export_rows)
    rows: list[dict[str, Any]] = []
    for row in export_rows:
        item_type = str(row.get("item_type") or "")
        item_text = str(row.get("item_text") or "").strip()
        if item_type == ITEM_TYPE_ELEMENT and feature_combo:
            event_score, senior_score = lookup_quality_scores(
                demand_type=TYPE_FEATURE_POINT,
                demand_text=feature_combo,
                event_sense_json=event_sense_json,
                senior_fit_json=senior_fit_json,
            )
        elif item_type == ITEM_TYPE_PHRASE and item_text:
            event_score, senior_score = lookup_quality_scores(
                demand_type=TYPE_PHRASE,
                demand_text=item_text,
                event_sense_json=event_sense_json,
                senior_fit_json=senior_fit_json,
            )
        else:
            event_score, senior_score = None, None
        rows.append(
            {
                **row,
                "event_sense_score": event_score,
                "senior_fit_score": senior_score,
            }
        )
    return rows


def _normalize_llm_items(
    parsed: dict[str, Any],
    candidates: list[dict[str, str]],
) -> list[dict[str, Any]]:
    candidate_lookup = {
        _candidate_key(item["demand_type"], item["demand_text"]): item
        for item in candidates
    }
    raw_items = parsed.get("items") or []
    if not isinstance(raw_items, list):
        raw_items = []

    items: list[dict[str, Any]] = []
    seen: set[tuple[str, str]] = set()
    for raw in raw_items:
        if not isinstance(raw, dict):
            continue
        demand_type = str(raw.get("demand_type") or "").strip()
        demand_text = str(raw.get("demand_text") or "").strip()
        if not demand_type or not demand_text:
            continue
        key = _candidate_key(demand_type, demand_text)
        if key not in candidate_lookup or key in seen:
            continue
        seen.add(key)
        score = _normalize_score(raw.get("score"))
        if score is None:
            continue
        items.append(
            {
                "demand_type": demand_type,
                "demand_text": demand_text,
                "score": score,
                "reason": str(raw.get("reason") or "").strip(),
            }
        )
    return items


def _llm_score_demands(
    *,
    channel_content_id: str,
    candidates: list[dict[str, str]],
    system_prompt: str,
    model: str,
    max_attempts: int,
    retry_sleep_seconds: float,
    max_tokens: int,
    score_field: str,
) -> dict[str, Any]:
    if not candidates:
        return {"source": channel_content_id, "items": []}

    user_payload = {
        "source": channel_content_id,
        "demands": candidates,
        "output_schema": {
            "source": "string",
            "items": [
                {
                    "demand_type": "string, 特征点 or 短语",
                    "demand_text": "string, must match one demand in demands",
                    "score": "number, 0-10",
                    "reason": "string",
                }
            ],
        },
        "constraints": [
            "仅对给定 demands 逐项评分，不得新增或遗漏",
            "score 为 0-10 的数字，越大表示越符合判断标准",
            "demand_type 与 demand_text 必须与输入完全一致",
            "仅输出 JSON 对象，不要 markdown 代码块",
        ],
    }

    last_error: Exception | None = None
    for attempt in range(1, max(max_attempts, 1) + 1):
        try:
            resp = create_chat_completion(
                [
                    {"role": "system", "content": system_prompt},
                    {
                        "role": "user",
                        "content": json.dumps(user_payload, ensure_ascii=False),
                    },
                ],
                model=model or None,
                temperature=0,
                max_tokens=max(max_tokens, 1),
            )
            parsed = _extract_json_object(str(resp.get("content") or ""))
            parsed.setdefault("source", channel_content_id)
            items = _normalize_llm_items(parsed, candidates)
            return {
                "source": channel_content_id,
                "score_field": score_field,
                "items": items,
            }
        except (OpenRouterCallError, HotContentFlowError) as exc:
            last_error = exc
            if attempt < max(max_attempts, 1):
                time.sleep(max(retry_sleep_seconds, 0))
    raise HotContentFlowError(
        f"llm {score_field} scoring failed for {channel_content_id}: {last_error}"
    ) from last_error


def llm_score_event_sense(
    *,
    channel_content_id: str,
    candidates: list[dict[str, str]],
    model: str,
    max_attempts: int,
    retry_sleep_seconds: float,
    max_tokens: int,
) -> dict[str, Any]:
    system_prompt = """
你是一个事件表达精确度评估专家。
# 任务
我会提供若干短语或词组组合（可以是特征词的拼接）。
请逐项判断：该短语/词组能否准确表达出一个具体的事件。
表达越确切、事件越具体，得分越高。
# 评分标准（0-10）
9-10：
精准指向某一具体事件，无歧义，可直接还原事件内容
7-8：
大体可判断是某类事件，但存在少量歧义或信息不完整
4-6：
有一定事件指向，但过于泛化，无法锁定具体事件
1-3：
偏属性/概念描述，几乎无法对应具体事件
0：
完全无法表达任何具体事件
# 评估维度（综合考量）
- 主体明确性：是否点出了事件涉及的人/物/组织
- 动作/结果明确性：是否体现了发生了什么
- 时空限定性：是否暗示了特定时间或地点
- 可还原性：仅凭该短语，能否在脑中重建出事件场景
# 输出格式
严格输出 JSON，禁止输出任何其他内容。
    """
    return _llm_score_demands(
        channel_content_id=channel_content_id,
        candidates=candidates,
        system_prompt=system_prompt,
        model=model,
        max_attempts=max_attempts,
        retry_sleep_seconds=retry_sleep_seconds,
        max_tokens=max_tokens,
        score_field="event_sense",
    )


def llm_score_senior_fit(
    *,
    channel_content_id: str,
    candidates: list[dict[str, str]],
    model: str,
    max_attempts: int,
    retry_sleep_seconds: float,
    max_tokens: int,
) -> dict[str, Any]:
    system_prompt = """
# 角色

你是一名严格的中老年内容适老性评分专家，专门评估词组/短语对中国50岁以上中老年用户的吸引力与相关性。你的判断基于严格的用户画像，而非主观感受。你会识别并拒绝一切看似"老年"实则属于年轻群体、中产焦虑、高认知门槛或语义模糊的伪适老词组。

# 核心任务

对输入的每个词组/短语，输出一个0-10的适老性评分，并给出简短判断依据。

---

# 一、基础定义（严格遵守，不可修改）

## 用户画像：中国50岁以上中老年人

### 认知特点

- 追求"确定性"和"安全感"，偏好简单直白，拒绝烧脑与推理
- 不关注新事物、抽象宏观经济、复杂金融博弈、枯燥行政程序
- 对网络梗、亚文化、职场黑话不敏感甚至反感

### 文化背景

- 成长于上世纪50-70年代，传统观念根深蒂固
- 深受儒家文化影响，强烈的孝道观念与集体主义倾向
- 处于"安享期"而非"奋斗期"
- 关注"保命"（三高/心脏/防骗）而非"塑形"（减肥/发际线）
- 关注"存量财产安全"而非"增量资产博弈"

### 情感需求

- 核心情感：安逸、从容、被尊重
- 偏好：正能量、民族自豪感、家庭温情、传统文化、同龄人故事
- 反感：贩卖焦虑、激烈矛盾冲突、血腥暴力、悲惨负面内容

### 场景偏好

- 接受：菜市场、公园、家庭、医院、老友聚会、怀旧场景
- 排斥：写字楼、夜店、高端消费场所、极限运动

---

# 二、适老品类白名单（命中即有基础分）

以下品类的词组，具备"适老性基础系数"，可在此基础上评估具体程度：

- 国家力量/民族自豪：阅兵、基建、外交胜利、撤侨、领土主权、中国强大
- 健康养生：三高管理、心脑血管、养生食疗、长寿、防病（严格排除减肥/塑形/医美/脱发）
- 防骗安全：电信诈骗、保健品骗局、新型骗局案例（极高优先级）
- 惠民政策：养老金、医保报销、现金支付保障、物价、天气预警
- 怀旧时光：70年代及以前老照片、经典老歌、老电影、童年记忆
- 家庭亲情：隔辈亲、孝道、家庭互助（排除婆媳恶斗/剧烈伦理冲突）
- 传统文化：节气、民俗、戏曲、国学、非遗
- 正能量：见义勇为、拾金不昧、平凡善举、反腐倡廉
- 人文科普：文化/历史/人文/健康等社科知识（非科技/自然猎奇）
- 自然惊奇：自然奇观、动物趣闻（排除血腥/恐怖/猎奇阴暗）

---

# 三、低分黑名单信号

## 强制低分信号（命中任一，评分不超过3）

- 职场类：升职、副业、内卷、打工人、绩效、裁员
- 年轻文化：网络梗、二次元、潮流、追星、发际线、颜值
- 金融投资：炒股、基金、加密货币、理财产品、资产配置
- 房产相关：买房、贷款、学区房、房价涨跌
- 健身塑形：减肥、健身、马甲线、体脂率、增肌
- 科技数码：手机评测、AI工具、电脑配置、游戏硬件
- 高消费场景：奢侈品、出境游、米其林、高端健身房
- 情绪贩卖：焦虑、内耗、emo、迷茫、躺平
- 模糊悬念：无具体信息的"千万别做这件事"类表达

## 中度扣分信号（命中使评分下浮1-2分）

- 内容偏泛人群，缺乏老年专属场景
- 认知门槛较高，需要背景知识才能理解
- 表达方式年轻化，但内容本身不排斥老年人

---

# 四、评分标准（0-10）

- 9-10：高度契合中老年用户核心关注点、典型生活场景或强情感诉求
- 7-8：对中老年用户有较强吸引力或实用价值，场景清晰
- 5-6：有一定相关性，但中老年专属属性一般，泛人群居多
- 3-4：偏年轻群体或泛人群，老年性弱，中老年用户兴趣低
- 1-2：明显面向年轻群体，中老年用户几乎不感兴趣
- 0：与中老年用户完全无关，或存在强烈排斥信号

---

# 五、输出规则

严格输出 JSON 对象（含 items 数组），禁止输出 JSON 之外的任何内容（无前缀、无解释、无markdown格式）。
    """
    return _llm_score_demands(
        channel_content_id=channel_content_id,
        candidates=candidates,
        system_prompt=system_prompt,
        model=model,
        max_attempts=max_attempts,
        retry_sleep_seconds=retry_sleep_seconds,
        max_tokens=max_tokens,
        score_field="senior_fit",
    )


def filter_candidates_by_event_sense(
    candidates: list[dict[str, str]],
    event_sense_json: dict[str, Any],
    *,
    event_threshold: float,
) -> list[dict[str, str]]:
    lookup = _build_score_lookup(event_sense_json)
    passed: list[dict[str, str]] = []
    for candidate in candidates:
        key = _candidate_key(candidate["demand_type"], candidate["demand_text"])
        item = lookup.get(key)
        score = _normalize_score(item.get("score")) if item else None
        if score is not None and score >= event_threshold:
            passed.append(candidate)
    return passed


def run_demand_quality_pipeline(
    *,
    channel_content_id: str,
    export_rows: list[dict[str, Any]],
    wxindex_threshold: float,
    event_threshold: float,
    senior_threshold: float,
    model: str,
    max_attempts: int,
    retry_sleep_seconds: float,
    max_tokens: int,
) -> tuple[dict[str, Any], dict[str, Any]]:
    """微信指数达标的需求：串行执行事件性、老年性 LLM，均对全量候选评分。"""
    candidates = build_quality_candidates(
        export_rows,
        wxindex_threshold=wxindex_threshold,
    )
    if not candidates:
        return {"source": channel_content_id, "items": []}, {"source": channel_content_id, "items": []}

    llm_kwargs = {
        "channel_content_id": channel_content_id,
        "candidates": candidates,
        "model": model,
        "max_attempts": max_attempts,
        "retry_sleep_seconds": retry_sleep_seconds,
        "max_tokens": max_tokens,
    }

    event_sense_json = llm_score_event_sense(**llm_kwargs)
    event_sense_json["threshold"] = event_threshold

    senior_fit_json = llm_score_senior_fit(**llm_kwargs)
    senior_fit_json["threshold"] = senior_threshold
    return event_sense_json, senior_fit_json