SupplyAgent/agents/demand_video_expand_agent/tools/batch_save_demand_expansions.py

"""
批量保存视频点位拓展判断结果。
"""
from __future__ import annotations

import logging
import re
import uuid
from typing import Any

from supply_agent.tools import tool
from supply_infra.db.models.multi_demand_video_point import POINT_TYPES
from supply_infra.db.repositories.demand_video_expansion_repo import (
    DemandVideoExpansionRepository,
)
from supply_infra.db.repositories.multi_demand_video_point_repo import (
    MultiDemandVideoPointRepository,
)
from supply_infra.db.session import get_session

logger = logging.getLogger(__name__)

_VALID_GRADES = frozenset({"S", "A"})


def _optional_str(value: Any) -> str | None:
    if value is None:
        return None
    text = str(value).strip()
    return text or None


def _normalize_items(
    items: list[dict[str, Any]],
    *,
    default_run_id: str,
    biz_dt: str,
    source_demand_grade_id: int,
    source_demand_name: str,
    source_grade: str,
) -> tuple[list[dict[str, Any]], list[str]]:
    rows: list[dict[str, Any]] = []
    errors: list[str] = []
    seen_keys: set[tuple[str, str]] = set()

    for idx, item in enumerate(items):
        if not isinstance(item, dict):
            errors.append(f"第 {idx} 项不是对象")
            continue

        expanded_text = _optional_str(item.get("expanded_text"))
        point_type = _optional_str(item.get("point_type"))
        video_id = _optional_str(item.get("video_id"))
        reason = _optional_str(item.get("reason"))

        if not expanded_text:
            errors.append(f"第 {idx} 项缺少 expanded_text")
            continue
        if not point_type or point_type not in POINT_TYPES:
            allowed = " / ".join(POINT_TYPES)
            errors.append(f"第 {idx} 项 point_type 无效（只能是 {allowed}）: {point_type!r}")
            continue
        if not video_id:
            errors.append(f"第 {idx} 项缺少 video_id")
            continue
        if not reason:
            errors.append(f"第 {idx} 项缺少 reason")
            continue

        dedupe_key = (expanded_text, video_id)
        if dedupe_key in seen_keys:
            errors.append(f"第 {idx} 项在本次请求中重复: {expanded_text!r} / {video_id!r}")
            continue
        seen_keys.add(dedupe_key)

        rows.append(
            {
                "biz_dt": biz_dt,
                "run_id": _optional_str(item.get("run_id")) or default_run_id,
                "source_demand_grade_id": source_demand_grade_id,
                "source_demand_name": source_demand_name,
                "source_grade": source_grade,
                "expanded_text": expanded_text,
                "point_type": point_type,
                "point_desc": _optional_str(item.get("point_desc")),
                "video_id": video_id,
                "reason": reason,
                "is_delete": 0,
            }
        )

    return rows, errors


def _normalize_demand_name(name: str) -> str:
    text = re.sub(r"\s+", "", name.strip().lower())
    return re.sub(r"[，,。．.！!？?；;：:""''\"'、/\\|·—_（）()【】\[\]《》<>]", "", text)


def _fill_missing_point_descs(rows: list[dict[str, Any]], session) -> None:
    """point_desc 为空时，从 multi_demand_video_point 按 video_id/point_type/point_data 补全。

    查不到或源表 point_desc 也为空时，保持原值（仍为 None）。
    """
    missing = [row for row in rows if not row.get("point_desc")]
    if not missing:
        return

    video_ids = sorted({str(row["video_id"]) for row in missing})
    points_by_vid = MultiDemandVideoPointRepository(session).list_by_video_ids(video_ids)

    lookup: dict[tuple[str, str, str], str] = {}
    for vid, points in points_by_vid.items():
        for point in points:
            point_data = _optional_str(point.get("point_data"))
            point_type = _optional_str(point.get("point_type"))
            point_desc = _optional_str(point.get("point_desc"))
            if not point_data or not point_type or not point_desc:
                continue
            lookup[(vid, point_type, point_data)] = point_desc

    for row in missing:
        key = (str(row["video_id"]), str(row["point_type"]), str(row["expanded_text"]))
        desc = lookup.get(key)
        if desc:
            row["point_desc"] = desc
        # 未命中或源表无描述：不改动，保留空值


@tool
def batch_save_demand_expansions(
    items: list[dict[str, Any]],
    biz_dt: str,
    source_demand_grade_id: int,
    source_demand_name: str,
    source_grade: str,
    run_id: str | None = None,
) -> str:
    """
    保存视频点位拓展判断结果到 demand_video_expansion 表。

    Args:
        items: 拓展候选列表。每项必填：
            - expanded_text: 拓展需求文本（来自 point_data）
            - point_type: inspiration / purpose / key
            - video_id: 来源视频 id
            - reason: 为何与原需求相近、可作为拓展
          选填：
            - point_desc: 点位描述快照
        biz_dt: 业务日 YYYYMMDD。
        source_demand_grade_id: 来源 demand_grade.id（由用户消息提供，原样传入）。
        source_demand_name: 来源需求名。
        source_grade: 来源等级 S 或 A。
        run_id: 任务批次 id；省略则自动生成。

    Returns:
        保存结果摘要。若无合适拓展，传 items=[] 即可。
    """
    biz_dt_text = _optional_str(biz_dt)
    if not biz_dt_text or len(biz_dt_text) != 8 or not biz_dt_text.isdigit():
        return f"biz_dt 格式无效，应为 YYYYMMDD: {biz_dt!r}"

    demand_name = _optional_str(source_demand_name)
    if not demand_name:
        return "source_demand_name 不能为空"

    grade = _optional_str(source_grade)
    if grade not in _VALID_GRADES:
        return f"source_grade 无效，只能是 S 或 A: {source_grade!r}"

    try:
        grade_id = int(source_demand_grade_id)
    except (TypeError, ValueError):
        return f"source_demand_grade_id 无效: {source_demand_grade_id!r}"

    if not items:
        return "无拓展候选，跳过落库"

    default_run_id = _optional_str(run_id) or uuid.uuid4().hex
    rows, errors = _normalize_items(
        items,
        default_run_id=default_run_id,
        biz_dt=biz_dt_text,
        source_demand_grade_id=grade_id,
        source_demand_name=demand_name,
        source_grade=grade,
    )

    normalized_demand = _normalize_demand_name(demand_name)
    filtered_rows: list[dict[str, Any]] = []
    skipped_same: list[str] = []
    for row in rows:
        if _normalize_demand_name(row["expanded_text"]) == normalized_demand:
            skipped_same.append(row["expanded_text"])
            continue
        filtered_rows.append(row)

    if not filtered_rows:
        detail = "；".join(errors) if errors else "无有效数据"
        same_note = ""
        if skipped_same:
            same_note = f"；剔除与原需求相同 {len(skipped_same)} 条"
        return f"没有可保存的数据: {detail}{same_note}"

    try:
        with get_session() as session:
            _fill_missing_point_descs(filtered_rows, session)
            saved = DemandVideoExpansionRepository(session).bulk_upsert(filtered_rows)

        parts = [f"成功保存 {saved} 条拓展需求", f"biz_dt={biz_dt_text}", f"run_id={default_run_id}"]
        if skipped_same:
            parts.append(f"剔除与原需求相同 {len(skipped_same)} 条")
        if errors:
            parts.append(f"校验失败 {len(errors)} 条: " + "；".join(errors[:10]))

        message = "。".join(parts)
        logger.info("batch_save_demand_expansions completed: %s", message)
        return message

    except Exception as e:
        logger.error("batch_save_demand_expansions failed: %s", e, exc_info=True)
        return f"保存拓展需求失败: {e}"