"""ODPS video feature lookup for landing-video material recall. videoContentList only chooses landing videos. Material recall features are read from loghubods.dwd_video_element_contribution_analysis by video id. """ from __future__ import annotations import logging from dataclasses import dataclass from pathlib import Path from typing import Iterable logger = logging.getLogger(__name__) _MINI_DIR = Path(__file__).resolve().parent.parent @dataclass(frozen=True) class VideoElementFeature: video_id: int element_dimension: str point_type: str standard_element: str contribution_score: float dt: str CREATE_CACHE_TABLE_SQL = """ CREATE TABLE IF NOT EXISTS video_element_feature_cache ( id BIGINT AUTO_INCREMENT PRIMARY KEY COMMENT '自增主键', video_id BIGINT NOT NULL COMMENT '业务视频ID，对应 ODPS vid', dt VARCHAR(8) NOT NULL COMMENT 'ODPS 分区日期 YYYYMMDD', point_type VARCHAR(50) NOT NULL COMMENT '点类型', standard_element VARCHAR(1024) NOT NULL COMMENT '标准化元素或解构选题文本', element_dimension VARCHAR(50) NOT NULL DEFAULT '实质' COMMENT '元素维度', contribution_score DOUBLE NOT NULL DEFAULT 0 COMMENT '贡献分', is_miss BOOLEAN NOT NULL DEFAULT FALSE COMMENT '是否为无特征负缓存', raw_element VARCHAR(255) DEFAULT NULL COMMENT '原始元素名称', element_id VARCHAR(100) DEFAULT NULL COMMENT '标准化元素ID', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间', UNIQUE KEY uk_dt_video_point_element (dt, video_id, element_dimension, point_type, standard_element(191)), KEY idx_video_dt (video_id, dt), KEY idx_dt (dt), KEY idx_updated_at (updated_at) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='视频元素特征缓存' """ def _quote_sql_string(value: str) -> str: return "'" + value.replace("\\", "\\\\").replace("'", "\\'") + "'" def _chunks(items: list[int], size: int) -> Iterable[list[int]]: for i in range(0, len(items), size): yield items[i:i + size] def _get_odps_client(): import sys tools_dir = _MINI_DIR / "tools" if str(tools_dir) not in sys.path: sys.path.insert(0, str(tools_dir)) from odps_module import get_odps_client return get_odps_client(project="loghubods") def _latest_partition_dt(client) -> str: table_name = "dwd_video_element_contribution_analysis" try: table = client._odps.get_table(table_name) partition = table.get_max_partition() if partition is not None: name = getattr(partition, "name", "") or str(partition) for part in name.split(","): if part.startswith("dt="): return part.split("=", 1)[1].strip("'\"") except Exception as e: logger.warning("[video_feature] get max partition by metadata failed:%s", e) rows = client.query( "SELECT MAX_PT('loghubods.dwd_video_element_contribution_analysis') AS dt", limit=1, ) if rows and rows[0].get("dt"): return str(rows[0]["dt"]) raise RuntimeError("无法获取 dwd_video_element_contribution_analysis 最新分区") def _query_odps_rows(client, sql: str, columns: list[str]) -> list[dict]: """Read ODPS rows without tunnel when possible. The configured tunnel endpoint can be unreachable in some runtime environments. Small feature lookups are better served through the instance reader without tunnel. """ odps = getattr(client, "_odps", None) if odps is None: return client.query(sql) instance = odps.execute_sql(sql) instance.wait_for_success() rows: list[dict] = [] with instance.open_reader(tunnel=False) as reader: for record in reader: rows.append({col: record[i] for i, col in enumerate(columns)}) return rows def _ensure_cache_table() -> None: from db.connection import get_connection conn = get_connection() try: with conn.cursor() as cur: cur.execute(CREATE_CACHE_TABLE_SQL) cur.execute("SHOW COLUMNS FROM video_element_feature_cache LIKE 'element_dimension'") if not cur.fetchone(): cur.execute( """ ALTER TABLE video_element_feature_cache ADD COLUMN element_dimension VARCHAR(50) NOT NULL DEFAULT '实质' COMMENT '元素维度' AFTER standard_element """ ) cur.execute("SHOW INDEX FROM video_element_feature_cache WHERE Key_name = 'uk_dt_video_point_element'") index_rows = cur.fetchall() index_cols = [str(row.get("Column_name") or "") for row in index_rows] index_ok = ( index_cols == ["dt", "video_id", "element_dimension", "point_type", "standard_element"] and str((index_rows[-1] or {}).get("Sub_part") or "") == "191" ) if index_rows and not index_ok: cur.execute("ALTER TABLE video_element_feature_cache DROP INDEX uk_dt_video_point_element") cur.execute("SHOW COLUMNS FROM video_element_feature_cache LIKE 'standard_element'") standard_col = cur.fetchone() if standard_col and "varchar(1024)" not in str(standard_col.get("Type") or "").lower(): cur.execute( """ ALTER TABLE video_element_feature_cache MODIFY COLUMN standard_element VARCHAR(1024) NOT NULL COMMENT '标准化元素或解构选题文本' """ ) cur.execute("SHOW COLUMNS FROM video_element_feature_cache LIKE 'is_miss'") if not cur.fetchone(): cur.execute( """ ALTER TABLE video_element_feature_cache ADD COLUMN is_miss BOOLEAN NOT NULL DEFAULT FALSE COMMENT '是否为无特征负缓存' AFTER contribution_score """ ) if not index_ok: cur.execute( """ ALTER TABLE video_element_feature_cache ADD UNIQUE KEY uk_dt_video_point_element (dt, video_id, element_dimension, point_type, standard_element(191)) """ ) finally: conn.close() def _read_cache(video_ids: list[int], dt: str) -> tuple[dict[int, list[VideoElementFeature]], set[int]]: if not video_ids: return {}, set() from db.connection import get_connection out: dict[int, list[VideoElementFeature]] = {} cached_ids: set[int] = set() conn = get_connection() try: with conn.cursor() as cur: for batch in _chunks(video_ids, 500): placeholders = ",".join(["%s"] * len(batch)) cur.execute( f""" SELECT video_id, dt, element_dimension, point_type, standard_element, contribution_score, is_miss FROM video_element_feature_cache WHERE dt = %s AND video_id IN ({placeholders}) ORDER BY video_id, contribution_score DESC """, [dt, *batch], ) for row in cur.fetchall(): video_id = int(row["video_id"]) cached_ids.add(video_id) if row.get("is_miss"): continue out.setdefault(video_id, []).append(VideoElementFeature( video_id=video_id, element_dimension=str(row["element_dimension"] or ""), point_type=str(row["point_type"] or ""), standard_element=str(row["standard_element"] or ""), contribution_score=float(row["contribution_score"] or 0.0), dt=str(row["dt"] or ""), )) finally: conn.close() return out, cached_ids def _write_cache( features: dict[int, list[VideoElementFeature]], miss_video_ids: Iterable[int], dt: str, ) -> None: rows = [ ( feature.video_id, feature.dt, feature.point_type, feature.standard_element, feature.element_dimension, feature.contribution_score, False, ) for feature_list in features.values() for feature in feature_list ] miss_rows = [ (int(video_id), dt, "", "", "实质", 0.0, True) for video_id in miss_video_ids ] rows.extend(miss_rows) if not rows: return from db.connection import get_connection conn = get_connection() try: with conn.cursor() as cur: cur.executemany( """ INSERT INTO video_element_feature_cache (video_id, dt, point_type, standard_element, element_dimension, contribution_score, is_miss) VALUES (%s, %s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE contribution_score = VALUES(contribution_score), element_dimension = VALUES(element_dimension), is_miss = VALUES(is_miss), updated_at = CURRENT_TIMESTAMP """, rows, ) finally: conn.close() def read_cached_video_element_features( video_ids: Iterable[int], ) -> dict[int, list[VideoElementFeature]]: """Read latest cached feature rows per video id from local DB only.""" ids: list[int] = [] seen: set[int] = set() for raw in video_ids: try: vid = int(raw) except (TypeError, ValueError): continue if vid in seen: continue seen.add(vid) ids.append(vid) if not ids: return {} _ensure_cache_table() from db.connection import get_connection out: dict[int, list[VideoElementFeature]] = {} conn = get_connection() try: with conn.cursor() as cur: for batch in _chunks(ids, 500): placeholders = ",".join(["%s"] * len(batch)) params = [*batch, *batch] cur.execute( f""" SELECT c.video_id, c.dt, c.element_dimension, c.point_type, c.standard_element, c.contribution_score, c.is_miss FROM video_element_feature_cache c JOIN ( SELECT video_id, MAX(dt) AS dt FROM video_element_feature_cache WHERE video_id IN ({placeholders}) GROUP BY video_id ) latest ON latest.video_id = c.video_id AND latest.dt = c.dt WHERE c.video_id IN ({placeholders}) ORDER BY c.video_id, c.contribution_score DESC """, params, ) for row in cur.fetchall(): if row.get("is_miss"): continue video_id = int(row["video_id"]) out.setdefault(video_id, []).append(VideoElementFeature( video_id=video_id, element_dimension=str(row["element_dimension"] or ""), point_type=str(row["point_type"] or ""), standard_element=str(row["standard_element"] or ""), contribution_score=float(row["contribution_score"] or 0.0), dt=str(row["dt"] or ""), )) finally: conn.close() return out def fetch_video_element_features( video_ids: Iterable[int], chunk_size: int = 100, ) -> dict[int, list[VideoElementFeature]]: """Fetch recall feature rows per video id from latest ODPS partition. One video can have multiple usable element rows. Keep all rows where: - 元素维度 = 解构选题 - 元素维度 = 实质 and 点类型 in 灵感点/关键点/目的点 - 贡献分 >= 0.8 - 标准化元素 is non-empty """ ids: list[int] = [] seen: set[int] = set() for raw in video_ids: try: vid = int(raw) except (TypeError, ValueError): continue if vid in seen: continue seen.add(vid) ids.append(vid) if not ids: return {} client = _get_odps_client() if client is None: logger.warning("[video_feature] ODPS client unavailable, skip enrichment") return {} dt = _latest_partition_dt(client) _ensure_cache_table() out, cached_ids = _read_cache(ids, dt) missing_ids = [video_id for video_id in ids if video_id not in cached_ids] if not missing_ids: logger.info( "[video_feature] cache hit %d/%d videos dt=%s", len(cached_ids), len(ids), dt, ) return out fetched: dict[int, list[VideoElementFeature]] = {} for batch in _chunks(missing_ids, max(1, chunk_size)): vid_list = ",".join(_quote_sql_string(str(v)) for v in batch) sql = f""" SELECT vid ,`元素维度` AS element_dimension ,`点类型` AS point_type ,`标准化元素` AS standard_element ,`解构选题` AS deconstruct_topic ,`贡献分` AS contribution_score ,dt FROM loghubods.dwd_video_element_contribution_analysis WHERE dt = {_quote_sql_string(dt)} AND vid IN ({vid_list}) AND ( (`元素维度` = '实质' AND `点类型` IN ('灵感点', '关键点', '目的点')) OR (`解构选题` IS NOT NULL AND `解构选题` <> '') ) AND `贡献分` >= 0.8 ORDER BY vid, `贡献分` DESC """ rows = _query_odps_rows( client, sql, ["vid", "element_dimension", "point_type", "standard_element", "deconstruct_topic", "contribution_score", "dt"], ) seen_features: set[tuple[int, str, str, str]] = set() for row in rows: try: video_id = int(row.get("vid")) except (TypeError, ValueError): continue element_dimension = str(row.get("element_dimension") or "").strip() point_type = str(row.get("point_type") or "").strip() standard_element = str(row.get("standard_element") or "").strip() contribution_score = float(row.get("contribution_score") or 0.0) dt_value = str(row.get("dt") or "") if ( element_dimension == "实质" and point_type in {"灵感点", "关键点", "目的点"} and standard_element ): key = (video_id, element_dimension, point_type, standard_element) if key not in seen_features: seen_features.add(key) fetched.setdefault(video_id, []).append(VideoElementFeature( video_id=video_id, element_dimension=element_dimension, point_type=point_type, standard_element=standard_element, contribution_score=contribution_score, dt=dt_value, )) topic = str(row.get("deconstruct_topic") or "").strip() if topic: key = (video_id, "解构选题", "", topic) if key not in seen_features: seen_features.add(key) fetched.setdefault(video_id, []).append(VideoElementFeature( video_id=video_id, element_dimension="解构选题", point_type="", standard_element=topic, contribution_score=contribution_score, dt=dt_value, )) fetched_ids = set(fetched.keys()) miss_ids = [video_id for video_id in missing_ids if video_id not in fetched_ids] _write_cache(fetched, miss_ids, dt) out.update(fetched) logger.info( "[video_feature] dt=%s cache_hit=%d odps_hit=%d odps_miss=%d/%d element_rows=%d", dt, len(cached_ids), len(fetched), len(miss_ids), len(missing_ids), sum(len(v) for v in out.values()), ) return out