|
|
@@ -0,0 +1,208 @@
|
|
|
+"""demand_grade_agent 专用的全局树局部热度查询辅助（不对外注册为工具）。"""
|
|
|
+from __future__ import annotations
|
|
|
+
|
|
|
+from collections import defaultdict
|
|
|
+from types import SimpleNamespace
|
|
|
+from typing import Any
|
|
|
+
|
|
|
+from agents.demand_grade_agent.tools.shared import (
|
|
|
+ build_category_heat_summary,
|
|
|
+ build_category_path,
|
|
|
+ format_category_weight_lines,
|
|
|
+)
|
|
|
+from supply_infra.db.repositories.category_tree_weight_repo import CategoryTreeWeightRepository
|
|
|
+from supply_infra.db.repositories.global_tree_category_repo import GlobalTreeCategoryRepository
|
|
|
+from supply_infra.db.session import get_session
|
|
|
+
|
|
|
+
|
|
|
+def _normalize_parent_id(parent_id: int | None) -> int | None:
|
|
|
+ if parent_id is None or parent_id == 0:
|
|
|
+ return None
|
|
|
+ return int(parent_id)
|
|
|
+
|
|
|
+
|
|
|
+def _materialize_category(row: Any) -> SimpleNamespace:
|
|
|
+ """在 session 内抽出标量，避免 DetachedInstanceError。"""
|
|
|
+ return SimpleNamespace(
|
|
|
+ id=int(row.id),
|
|
|
+ name=row.name,
|
|
|
+ parent_id=row.parent_id,
|
|
|
+ level=row.level,
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+def _materialize_weight(row: Any) -> SimpleNamespace:
|
|
|
+ """在 session 内抽出标量，避免 DetachedInstanceError。"""
|
|
|
+ return SimpleNamespace(
|
|
|
+ category_id=int(row.category_id),
|
|
|
+ biz_dt=row.biz_dt,
|
|
|
+ total_score=row.total_score,
|
|
|
+ hung_word_count=row.hung_word_count,
|
|
|
+ real_rov_7d_avg=row.real_rov_7d_avg,
|
|
|
+ real_rov_7d_count=row.real_rov_7d_count,
|
|
|
+ real_vov_7d_avg=row.real_vov_7d_avg,
|
|
|
+ real_vov_7d_count=row.real_vov_7d_count,
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+def _score_value(weight: Any | None) -> float | None:
|
|
|
+ if weight is None or weight.total_score is None:
|
|
|
+ return None
|
|
|
+ return float(weight.total_score)
|
|
|
+
|
|
|
+
|
|
|
+def load_local_tree_state(biz_dt: str) -> tuple[dict[int, Any], dict[int | None, list[int]], dict[int, Any]]:
|
|
|
+ with get_session() as session:
|
|
|
+ categories = [
|
|
|
+ _materialize_category(row)
|
|
|
+ for row in GlobalTreeCategoryRepository(session).list_active_categories()
|
|
|
+ ]
|
|
|
+ weights = [
|
|
|
+ _materialize_weight(row)
|
|
|
+ for row in CategoryTreeWeightRepository(session).list_by_biz_dt(biz_dt)
|
|
|
+ ]
|
|
|
+ by_id = {row.id: row for row in categories}
|
|
|
+ children: dict[int | None, list[int]] = defaultdict(list)
|
|
|
+ for row in categories:
|
|
|
+ children[_normalize_parent_id(row.parent_id)].append(row.id)
|
|
|
+ for ids in children.values():
|
|
|
+ ids.sort()
|
|
|
+ weight_by_id = {row.category_id: row for row in weights}
|
|
|
+ return by_id, children, weight_by_id
|
|
|
+
|
|
|
+
|
|
|
+def _describe_node(
|
|
|
+ category_id: int,
|
|
|
+ by_id: dict[int, Any],
|
|
|
+ weight_by_id: dict[int, Any],
|
|
|
+ *,
|
|
|
+ biz_dt: str,
|
|
|
+) -> dict[str, Any]:
|
|
|
+ category = by_id.get(category_id)
|
|
|
+ weight = weight_by_id.get(category_id)
|
|
|
+ return {
|
|
|
+ "category_id": category_id,
|
|
|
+ "name": category.name if category is not None else None,
|
|
|
+ "path": build_category_path(category_id, by_id),
|
|
|
+ "hung_word_count": int(weight.hung_word_count or 0) if weight is not None else 0,
|
|
|
+ "heat": build_category_heat_summary(weight, biz_dt=biz_dt),
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+def _sibling_ids(category_id: int, parent_id: int | None, children: dict[int | None, list[int]]) -> list[int]:
|
|
|
+ if parent_id is not None:
|
|
|
+ return list(children.get(parent_id, []))
|
|
|
+ return list(children.get(None, []))
|
|
|
+
|
|
|
+
|
|
|
+def build_local_heat_snapshot(
|
|
|
+ biz_dt: str,
|
|
|
+ category_ids: list[int],
|
|
|
+ *,
|
|
|
+ tree_state: tuple[dict[int, Any], dict[int | None, list[int]], dict[int, Any]] | None = None,
|
|
|
+) -> list[dict[str, Any]]:
|
|
|
+ """构建局部环境快照：自身、父节点、全部兄弟节点的全局热度与后验。"""
|
|
|
+ if tree_state is None:
|
|
|
+ by_id, children, weight_by_id = load_local_tree_state(biz_dt)
|
|
|
+ else:
|
|
|
+ by_id, children, weight_by_id = tree_state
|
|
|
+ snapshots: list[dict[str, Any]] = []
|
|
|
+ for category_id in dict.fromkeys(int(value) for value in category_ids):
|
|
|
+ category = by_id.get(category_id)
|
|
|
+ if category is None:
|
|
|
+ snapshots.append({
|
|
|
+ "category_id": category_id,
|
|
|
+ "error": "分类不存在",
|
|
|
+ })
|
|
|
+ continue
|
|
|
+
|
|
|
+ parent_id = _normalize_parent_id(category.parent_id)
|
|
|
+ sibling_ids = _sibling_ids(category_id, parent_id, children)
|
|
|
+ ranked_siblings = sorted(
|
|
|
+ sibling_ids,
|
|
|
+ key=lambda cid: (
|
|
|
+ _score_value(weight_by_id.get(cid)) is None,
|
|
|
+ -(_score_value(weight_by_id.get(cid)) or 0),
|
|
|
+ cid,
|
|
|
+ ),
|
|
|
+ )
|
|
|
+ sibling_total = len(ranked_siblings)
|
|
|
+ siblings = []
|
|
|
+ for index, sibling_id in enumerate(ranked_siblings, start=1):
|
|
|
+ item = _describe_node(sibling_id, by_id, weight_by_id, biz_dt=biz_dt)
|
|
|
+ item["rank_among_siblings"] = index
|
|
|
+ item["sibling_count"] = sibling_total
|
|
|
+ item["is_self"] = sibling_id == category_id
|
|
|
+ siblings.append(item)
|
|
|
+
|
|
|
+ snapshots.append({
|
|
|
+ "category_id": category_id,
|
|
|
+ "biz_dt": biz_dt,
|
|
|
+ "self": _describe_node(category_id, by_id, weight_by_id, biz_dt=biz_dt),
|
|
|
+ "parent": (
|
|
|
+ _describe_node(parent_id, by_id, weight_by_id, biz_dt=biz_dt)
|
|
|
+ if parent_id is not None
|
|
|
+ else None
|
|
|
+ ),
|
|
|
+ "siblings": siblings,
|
|
|
+ "sibling_count": sibling_total,
|
|
|
+ })
|
|
|
+ return snapshots
|
|
|
+
|
|
|
+
|
|
|
+def _append_node_block(
|
|
|
+ lines: list[str],
|
|
|
+ *,
|
|
|
+ title: str,
|
|
|
+ node: dict[str, Any] | None,
|
|
|
+ biz_dt: str,
|
|
|
+ weight_by_id: dict[int, Any] | None = None,
|
|
|
+) -> None:
|
|
|
+ if node is None:
|
|
|
+ lines.append(f"{title}: 无")
|
|
|
+ return
|
|
|
+ lines.append(f"{title}:")
|
|
|
+ weight = weight_by_id.get(int(node["category_id"])) if weight_by_id is not None else None
|
|
|
+ lines.extend(format_category_weight_lines(
|
|
|
+ weight,
|
|
|
+ category_id=int(node["category_id"]),
|
|
|
+ name=node.get("name"),
|
|
|
+ path=node.get("path"),
|
|
|
+ biz_dt=biz_dt,
|
|
|
+ hung_word_count=int(node.get("hung_word_count") or 0),
|
|
|
+ ))
|
|
|
+
|
|
|
+
|
|
|
+def format_local_heat_snapshot(snapshot: dict[str, Any], *, weight_by_id: dict[int, Any] | None = None) -> list[str]:
|
|
|
+ if snapshot.get("error"):
|
|
|
+ return [f"--- category_id={snapshot['category_id']} ---", f"错误: {snapshot['error']}"]
|
|
|
+
|
|
|
+ biz_dt = snapshot["biz_dt"]
|
|
|
+ lines = [f"--- category_id={snapshot['category_id']} ---"]
|
|
|
+ _append_node_block(lines, title="自身节点", node=snapshot["self"], biz_dt=biz_dt, weight_by_id=weight_by_id)
|
|
|
+ _append_node_block(lines, title="父节点", node=snapshot.get("parent"), biz_dt=biz_dt, weight_by_id=weight_by_id)
|
|
|
+
|
|
|
+ siblings = snapshot.get("siblings") or []
|
|
|
+ lines.append(f"全部兄弟节点（共 {len(siblings)} 个，按 total_score 降序，* 为当前节点）:")
|
|
|
+ for sibling in siblings:
|
|
|
+ marker = "*" if sibling.get("is_self") else " "
|
|
|
+ rank_note = f" 兄弟排名 {sibling.get('rank_among_siblings')}/{sibling.get('sibling_count')}"
|
|
|
+ lines.append(f"{marker} {'-' * 8}{rank_note}")
|
|
|
+ _append_node_block(lines, title=" 节点", node=sibling, biz_dt=biz_dt, weight_by_id=weight_by_id)
|
|
|
+ return lines
|
|
|
+
|
|
|
+
|
|
|
+def render_local_heat_report(biz_dt: str, category_ids: list[int]) -> str:
|
|
|
+ tree_state = load_local_tree_state(biz_dt)
|
|
|
+ _, _, weight_by_id = tree_state
|
|
|
+ snapshots = build_local_heat_snapshot(biz_dt, category_ids, tree_state=tree_state)
|
|
|
+ if not snapshots:
|
|
|
+ return "category_ids 不能为空"
|
|
|
+ lines = [
|
|
|
+ f"biz_dt={biz_dt} | 局部环境包含节点自身、父节点、全部兄弟节点的全局热度 total_score 与后验数据；",
|
|
|
+ "不得只参考部分兄弟节点，应以本报告列出的全部节点为准。",
|
|
|
+ ]
|
|
|
+ for snapshot in snapshots:
|
|
|
+ lines.extend(format_local_heat_snapshot(snapshot, weight_by_id=weight_by_id))
|
|
|
+ lines.append("")
|
|
|
+ return "\n".join(lines).rstrip()
|
xueyiming