|
|
@@ -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
|