增加导出。增加视频

supply_infra/db/repositories/multi_demand_pool_di_repo.py

@@ -153,3 +153,26 @@ class MultiDemandPoolDiRepository(BaseRepository[MultiDemandPoolDi]):
             result = self.session.execute(stmt)
             updated += result.rowcount or 0
         return updated
+
+    def update_video_fields(self, biz_dt: str, rows: list[dict]) -> int:
+        """按 (strategy, demand_id) 批量更新 video_list / video_count。"""
+        if not rows:
+            return 0
+
+        updated = 0
+        for row in rows:
+            stmt = (
+                update(MultiDemandPoolDi)
+                .where(
+                    MultiDemandPoolDi.biz_dt == biz_dt,
+                    MultiDemandPoolDi.strategy == row["strategy"],
+                    MultiDemandPoolDi.demand_id == row["demand_id"],
+                )
+                .values(
+                    video_list=row.get("video_list"),
+                    video_count=row.get("video_count"),
+                )
+            )
+            result = self.session.execute(stmt)
+            updated += result.rowcount or 0
+        return updated

supply_infra/odps/client.py

@@ -85,7 +85,7 @@ class ODPSClient:
         return self.execute_sql(sql)
 
     def fetch_multi_demand_pool(self, bizdate: str) -> list[dict[str, Any]]:
-        """拉取 dwd_multi_demand_pool_di 策略需求天级数据（不同步 video_list）。"""
+        """拉取 dwd_multi_demand_pool_di 策略需求天级数据（video_list 仅取前 100 个）。"""
         sql = f"""
         SELECT  strategy
                 ,demand_id
@@ -93,6 +93,7 @@ class ODPSClient:
                 ,weight
                 ,`type`
                 ,video_count
+                ,SLICE(video_list, 1, 100) AS video_list
                 ,extend
         FROM    loghubods.dwd_multi_demand_pool_di
         WHERE   dt = '{bizdate}'

supply_infra/scheduler/jobs/sync_multi_demand_pool_odps_to_mysql.py

@@ -3,16 +3,18 @@
 
 流程：
 1. 比对当天 ODPS / MySQL 行数，相同则跳过写入
-2. 有差异时拉取 ODPS，按 (strategy, demand_id) 只插入缺失、删除多余
-3. 拉取近 7 日真实 ROV/VOV，按特征值匹配回填 real_rov_7d / real_vov_7d
-4. 查询当天全部 demand_name，按空格分词写入 set
-5. 过滤 demand_belong_category 中已存在的词
-6. 剩余词按 100 词一批调用 demand_belong_category_agent
-7. 遍历 demand_belong_category 全部词，按策略与真实 ROV/VOV 写入 demand_popularity_stats（avg/count）
-8. 基于三表计算整棵类目树节点加权平均分，写入 category_tree_weight
+2. 有差异时拉取 ODPS，按 (strategy, demand_id) 只插入缺失、删除多余，并回填已有行 video_list
+3. video_list 每条最多保留前 100 个 video_id，video_count 与之保持一致
+4. 拉取近 7 日真实 ROV/VOV，按特征值匹配回填 real_rov_7d / real_vov_7d
+5. 查询当天全部 demand_name，按空格分词写入 set
+6. 过滤 demand_belong_category 中已存在的词
+7. 剩余词按 100 词一批调用 demand_belong_category_agent
+8. 遍历 demand_belong_category 全部词，按策略与真实 ROV/VOV 写入 demand_popularity_stats（avg/count）
+9. 基于三表计算整棵类目树节点加权平均分，写入 category_tree_weight
 """
 from __future__ import annotations
 
+import json
 import logging
 from datetime import datetime, timedelta
 from decimal import Decimal
@@ -39,6 +41,7 @@ _STATS_UPSERT_BATCH = 200
 _REAL_METRIC_LIMIT = 1000
 _REAL_METRIC_LOOKBACK_DAYS = 7
 _GLOBAL_FEATURE_VALUE = "全局SUM"
+_VIDEO_LIST_LIMIT = 100
 
 # 策略名 → 统计字段前缀；去年同期阳历/阴历合并为 plat_ly_pop
 _STRATEGY_METRIC: dict[str, str] = {
@@ -68,6 +71,39 @@ def _row_key(row: dict[str, Any]) -> RowKey:
     return (row["strategy"], row["demand_id"])
 
 
+def _normalize_video_list(raw: Any) -> tuple[str | None, int]:
+    """取前 N 个 video_id，返回 (JSON 文本, count)，二者保持一致。"""
+    if raw is None:
+        return None, 0
+
+    items: list[Any]
+    if isinstance(raw, str):
+        text = raw.strip()
+        if not text:
+            return None, 0
+        try:
+            parsed = json.loads(text)
+            items = list(parsed) if isinstance(parsed, list) else [text]
+        except json.JSONDecodeError:
+            items = [part.strip() for part in text.split(",") if part.strip()]
+    elif isinstance(raw, (list, tuple)):
+        items = list(raw)
+    else:
+        try:
+            items = list(raw)
+        except TypeError:
+            return None, 0
+
+    truncated = [
+        str(v).strip()
+        for v in items[:_VIDEO_LIST_LIMIT]
+        if v is not None and str(v).strip()
+    ]
+    if not truncated:
+        return None, 0
+    return json.dumps(truncated, ensure_ascii=False), len(truncated)
+
+
 def _to_mysql_rows(raw_rows: list[dict[str, Any]], biz_dt: str) -> list[dict[str, Any]]:
     """转换为 MySQL 行，按 (strategy, demand_id) 去重（保留最后一条）。"""
     by_key: dict[RowKey, dict[str, Any]] = {}
@@ -79,14 +115,15 @@ def _to_mysql_rows(raw_rows: list[dict[str, Any]], biz_dt: str) -> list[dict[str
             logger.warning("Skip row with missing required fields: %s", row)
             continue
 
+        video_list, video_count = _normalize_video_list(row.get("video_list"))
         mapped = {
             "strategy": str(strategy),
             "demand_id": str(demand_id),
             "demand_name": str(demand_name),
             "weight": row.get("weight"),
             "type": str(row["type"]) if row.get("type") is not None else None,
-            "video_count": row.get("video_count"),
-            "video_list": None,
+            "video_count": video_count,
+            "video_list": video_list,
             "extend": str(row["extend"]) if row.get("extend") is not None else None,
             "biz_dt": biz_dt,
         }
@@ -143,7 +180,7 @@ def _classify_words(biz_dt: str) -> dict:
 
 
 def _sync_diff(partition_date: str) -> dict[str, Any]:
-    """行数不同时拉取 ODPS，只同步 (strategy, demand_id) 差异。"""
+    """行数不同时拉取 ODPS，只同步 (strategy, demand_id) 差异，并回填已有行的 video 字段。"""
     odps = get_odps_client()
     raw_rows = odps.fetch_multi_demand_pool(partition_date)
     mysql_rows = _to_mysql_rows(raw_rows, partition_date)
@@ -156,17 +193,21 @@ def _sync_diff(partition_date: str) -> dict[str, Any]:
 
         to_insert_keys = odps_keys - mysql_keys
         to_delete_keys = mysql_keys - odps_keys
+        to_update_keys = odps_keys & mysql_keys
 
         insert_rows = [odps_by_key[k] for k in to_insert_keys]
+        update_rows = [odps_by_key[k] for k in to_update_keys]
         deleted = repo.delete_by_keys(partition_date, list(to_delete_keys))
         inserted = repo.bulk_insert(insert_rows)
+        updated = repo.update_video_fields(partition_date, update_rows)
 
     logger.info(
-        "Diff sync: odps=%d mysql_before=%d insert=%d delete=%d",
+        "Diff sync: odps=%d mysql_before=%d insert=%d delete=%d video_update=%d",
         len(odps_keys),
         len(mysql_keys),
         inserted,
         deleted,
+        updated,
     )
     return {
         "fetched": len(raw_rows),
@@ -174,10 +215,35 @@ def _sync_diff(partition_date: str) -> dict[str, Any]:
         "mysql_before": len(mysql_keys),
         "inserted": inserted,
         "deleted": deleted,
+        "video_updated": updated,
         "skipped_invalid": len(raw_rows) - len(mysql_rows),
     }
 
 
+def backfill_video_list(partition_date: str) -> dict[str, Any]:
+    """从 ODPS 回填指定分区的 video_list / video_count（每条最多前 100 个 video_id）。"""
+    logger.info("Backfill video_list for partition: %s", partition_date)
+    odps = get_odps_client()
+    raw_rows = odps.fetch_multi_demand_pool(partition_date)
+    mysql_rows = _to_mysql_rows(raw_rows, partition_date)
+
+    with get_session() as session:
+        updated = MultiDemandPoolDiRepository(session).update_video_fields(
+            partition_date,
+            mysql_rows,
+        )
+
+    result = {
+        "partition_date": partition_date,
+        "fetched": len(raw_rows),
+        "unique_rows": len(mysql_rows),
+        "updated": updated,
+        "with_video": sum(1 for r in mysql_rows if r.get("video_list")),
+    }
+    logger.info("Backfill video_list completed: %s", result)
+    return result
+
+
 def _to_float(value: Any) -> float | None:
     if value is None:
         return None

web/src/components/CategoryTree.vue

@@ -11,6 +11,7 @@ import {
   maxTreeDepth,
 } from '../types/category'
 import type { DemandBelongItem, DemandsByCategory } from '../types/demand'
+import { downloadCategoryTreeHtml } from '../utils/exportCategoryTreeHtml'
 
 const FULL_TREE_KEY = 'full' as const
 type TreeTabKey = typeof FULL_TREE_KEY | WeightDimKey
@@ -158,6 +159,15 @@ function closeDrawer() {
   drawerOpen.value = false
 }
 
+function exportHtml() {
+  downloadCategoryTreeHtml({
+    nodes: props.nodes,
+    dims: dimTabs.value,
+    bizDt: props.bizDt ?? null,
+    demandsByCategory: props.demandsByCategory,
+  })
+}
+
 function onPanStart(e: MouseEvent) {
   const el = treePanelRef.value
   if (!el || e.button !== 0) return
@@ -209,6 +219,7 @@ function onPanEnd() {
           <span>层</span>
         </label>
         <button type="button" class="btn" @click="expandAll">全部展开</button>
+        <button type="button" class="btn btn-secondary" @click="exportHtml">导出 HTML</button>
         <span class="hint">
           共 {{ treeMaxDepth }} 层 · 空白处按住拖动可平移 · Shift+滚轮左右滚
         </span>
@@ -370,6 +381,17 @@ function onPanEnd() {
   background: #1e293b;
 }
 
+.btn-secondary {
+  background: #fff;
+  color: #0f172a;
+  border-color: #cbd5e1;
+}
+
+.btn-secondary:hover {
+  background: #f8fafc;
+  border-color: #94a3b8;
+}
+
 .hint {
   font-size: 12px;
   color: #94a3b8;