补充下缀数据

supply_infra/db/models/multi_demand_video_detail.py

@@ -22,6 +22,21 @@ class MultiDemandVideoDetail(Base):
     final_topic_json: Mapped[str] = mapped_column(
         Text, nullable=False, comment="最终选题JSON"
     )
+    inspiration_points_json: Mapped[str | None] = mapped_column(
+        Text,
+        nullable=True,
+        comment="灵感点列表JSON（decode_result.灵感点，每项保留 点/点描述）",
+    )
+    purpose_points_json: Mapped[str | None] = mapped_column(
+        Text,
+        nullable=True,
+        comment="目的点列表JSON（decode_result.目的点，每项保留 点/点描述）",
+    )
+    key_points_json: Mapped[str | None] = mapped_column(
+        Text,
+        nullable=True,
+        comment="关键点列表JSON（decode_result.关键点，每项保留 点/点描述）",
+    )
     create_time: Mapped[datetime] = mapped_column(
         nullable=False,
         server_default=func.now(),

supply_infra/db/repositories/multi_demand_video_detail_repo.py

@@ -55,6 +55,15 @@ class MultiDemandVideoDetailRepository(BaseRepository[MultiDemandVideoDetail]):
         )
         return [str(v) for v in self.session.scalars(stmt).all() if v]
 
+    def list_vids_missing_points(self) -> list[str]:
+        """返回灵感点/目的点/关键点任一字段为空的全部 vid。"""
+        stmt = select(MultiDemandVideoDetail.vid).where(
+            MultiDemandVideoDetail.inspiration_points_json.is_(None)
+            | MultiDemandVideoDetail.purpose_points_json.is_(None)
+            | MultiDemandVideoDetail.key_points_json.is_(None)
+        )
+        return [str(v) for v in self.session.scalars(stmt).all() if v]
+
     def bulk_insert_ignore(self, rows: list[dict]) -> int:
         """批量插入，MySQL 自动忽略 vid 重复行。"""
         if not rows:
@@ -83,3 +92,21 @@ class MultiDemandVideoDetailRepository(BaseRepository[MultiDemandVideoDetail]):
             result = self.session.execute(stmt)
             updated += result.rowcount or 0
         return updated
+
+    def update_points(self, points_by_vid: dict[str, dict[str, str | None]]) -> int:
+        """按 vid 批量更新灵感点/目的点/关键点（values 为落库字段名 → JSON 文本）。"""
+        if not points_by_vid:
+            return 0
+
+        updated = 0
+        for vid, values in points_by_vid.items():
+            if not values:
+                continue
+            stmt = (
+                update(MultiDemandVideoDetail)
+                .where(MultiDemandVideoDetail.vid == vid)
+                .values(**values)
+            )
+            result = self.session.execute(stmt)
+            updated += result.rowcount or 0
+        return updated

supply_infra/scheduler/jobs/sync_multi_demand_videos.py

@@ -21,6 +21,14 @@ logger = logging.getLogger(__name__)
 _FINAL_TOPIC_KEY = "最终选题"
 _TARGET_POST_KEY = "target_post"
 _TITLE_KEY = "title"
+_POINT_KEY = "点"
+_POINT_DESC_KEY = "点描述"
+# decode_result 中的点位 key → 落库字段名
+_POINT_FIELD_MAP = {
+    "灵感点": "inspiration_points_json",
+    "目的点": "purpose_points_json",
+    "关键点": "key_points_json",
+}
 VIDEO_SYNC_BATCH_SIZE = 100
 
 
@@ -75,6 +83,35 @@ def _extract_final_topic_json(payload: dict[str, Any]) -> str | None:
     return json.dumps(final_topic, ensure_ascii=False)
 
 
+def _extract_points(payload: dict[str, Any], key: str) -> str | None:
+    """从 decode_result[key]（灵感点/目的点/关键点）取每项的 点/点描述，重组为 JSON 文本。"""
+    items = payload.get(key)
+    if not isinstance(items, list):
+        return None
+
+    points: list[dict[str, Any]] = []
+    for item in items:
+        if not isinstance(item, dict):
+            continue
+        point = item.get(_POINT_KEY)
+        point_desc = item.get(_POINT_DESC_KEY)
+        if point is None and point_desc is None:
+            continue
+        points.append({_POINT_KEY: point, _POINT_DESC_KEY: point_desc})
+
+    if not points:
+        return None
+    return json.dumps(points, ensure_ascii=False)
+
+
+def _extract_all_points(payload: dict[str, Any]) -> dict[str, str | None]:
+    """从 decode_result 提取灵感点/目的点/关键点三个字段，返回 落库字段名 → JSON 文本。"""
+    return {
+        field: _extract_points(payload, key)
+        for key, field in _POINT_FIELD_MAP.items()
+    }
+
+
 def _extract_title(payload: dict[str, Any]) -> str | None:
     """从 decode_result.target_post.title 取标题。"""
     target_post = payload.get(_TARGET_POST_KEY)
@@ -130,6 +167,7 @@ def _sync_one_batch(
                 "vid": vid,
                 "title": _extract_title(payload),
                 "final_topic_json": topic_json,
+                **_extract_all_points(payload),
             }
         )
 
@@ -344,3 +382,87 @@ def backfill_video_titles(
     }
     logger.info("Backfill video titles completed: %s", result)
     return result
+
+
+def backfill_video_points(
+    batch_size: int = VIDEO_SYNC_BATCH_SIZE,
+) -> dict[str, Any]:
+    """为灵感点/目的点/关键点字段缺失的已有行，从 ODPS 昨天分区回填。"""
+    decode_dt = (datetime.now() - timedelta(days=1)).strftime("%Y%m%d")
+    chunk = max(1, int(batch_size))
+    logger.info("Backfill video points: decode_dt=%s batch_size=%d", decode_dt, chunk)
+
+    with get_session() as session:
+        pending = MultiDemandVideoDetailRepository(session).list_vids_missing_points()
+
+    if not pending:
+        result = {
+            "decode_dt": decode_dt,
+            "pending": 0,
+            "batches": 0,
+            "updated": 0,
+            "missing_in_odps": 0,
+            "skipped_no_decode": 0,
+        }
+        logger.info("No vids missing points: %s", result)
+        return result
+
+    odps = get_odps_client()
+    batches = [pending[i : i + chunk] for i in range(0, len(pending), chunk)]
+    total_updated = 0
+    total_missing = 0
+    total_skipped = 0
+
+    for idx, batch_vids in enumerate(batches, start=1):
+        logger.info(
+            "Points backfill batch %d/%d: %d vids",
+            idx,
+            len(batches),
+            len(batch_vids),
+        )
+        odps_rows = odps.fetch_topic_decode_results(
+            decode_dt, batch_vids, batch_size=len(batch_vids)
+        )
+        points_by_vid: dict[str, dict[str, str | None]] = {}
+        for row in odps_rows:
+            raw_vid = row.get("vid")
+            if raw_vid is None:
+                continue
+            vid = str(raw_vid).strip()
+            if not vid or vid in points_by_vid:
+                continue
+            payload = _parse_decode_result(row.get("decode_result"))
+            if payload is None:
+                total_skipped += 1
+                continue
+            points_by_vid[vid] = _extract_all_points(payload)
+
+        with get_session() as session:
+            updated = MultiDemandVideoDetailRepository(session).update_points(
+                points_by_vid
+            )
+        total_updated += updated
+
+        odps_vids = {
+            str(r.get("vid")).strip()
+            for r in odps_rows
+            if r.get("vid") is not None and str(r.get("vid")).strip()
+        }
+        total_missing += len(set(batch_vids) - odps_vids)
+        logger.info(
+            "Points backfill batch %d/%d done: updated=%d",
+            idx,
+            len(batches),
+            updated,
+        )
+
+    result = {
+        "decode_dt": decode_dt,
+        "pending": len(pending),
+        "batches": len(batches),
+        "updated": total_updated,
+        "missing_in_odps": total_missing,
+        "skipped_no_decode": total_skipped,
+    }
+    logger.info("Backfill video points completed: %s", result)
+    return result