""" 批量保存视频点位拓展判断结果。 """ 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}"