"""去年同期阳历策略 ODPS 查询。""" from app.odps.client import get_odps_client from app.strategies.odps._date_utils import resolve_solar_same_period_window, today_shanghai_date, partition_yyyymmdd from app.strategies.odps._utils import normalize_scalar, parse_video_list def build_solar_same_period_sql( *, bizdate: str, start_date: str, end_date: str, view_pv_count: int, min_contribution_score: float, rov_avg: float, ) -> str: return f""" WITH cleaned_video_metrics AS ( SELECT CAST(视频id AS STRING) AS vid, rov_t0 FROM loghubods.video_dimension_detail_add_column WHERE dt >= '{start_date}' AND dt <= '{end_date}' AND COALESCE(`当日分发曝光pv`, 0) >= {int(view_pv_count)} ), video_avg_metrics AS ( SELECT vid, AVG(CASE WHEN rov_t0 = 0 THEN NULL ELSE rov_t0 END) AS vid_avg_rov FROM cleaned_video_metrics GROUP BY vid ), tag_vid_dedup AS ( SELECT DISTINCT CAST(vid AS STRING) AS vid, 原始元素, 原始元素描述, 点类型, 元素维度, 短语, 选题, `extend` FROM loghubods.dwd_topic_decode_result_detail_di WHERE dt = MAX_PT('loghubods.dwd_topic_decode_result_detail_di') AND TRIM(原始元素) <> '' AND 原始元素 IS NOT NULL AND 元素维度 = '实质' AND 贡献分 >= {float(min_contribution_score)} ) SELECT '去年同期阳历' AS strategy, md5(CONCAT('去年同期阳历', t1.原始元素, '{bizdate}')) AS demand_id, t1.原始元素 AS demand_name, COALESCE(ROUND(AVG(t2.vid_avg_rov), 6), 0) AS weight, '特征点' AS type, COUNT(DISTINCT t1.vid) AS video_count, COLLECT_SET(t1.vid) AS video_list, '{{}}' AS extend FROM tag_vid_dedup t1 LEFT JOIN video_avg_metrics t2 ON t1.vid = t2.vid GROUP BY t1.原始元素 HAVING COALESCE(ROUND(AVG(t2.vid_avg_rov), 6), 0) >= {float(rov_avg)} ORDER BY weight DESC """ def query_solar_same_period_demands( *, bizdate: str | None = None, period_days: int = 7, view_pv_count: int, min_contribution_score: float, rov_avg: float, ) -> list[dict[str, object]]: if bizdate is None: bizdate = partition_yyyymmdd(today_shanghai_date()) bizdate_value, start_date, end_date = resolve_solar_same_period_window( bizdate=bizdate, period_days=period_days, ) sql = build_solar_same_period_sql( bizdate=bizdate_value, start_date=start_date, end_date=end_date, view_pv_count=view_pv_count, min_contribution_score=min_contribution_score, rov_avg=rov_avg, ) odps_client = get_odps_client() instance = odps_client.execute_sql( sql, hints={ "odps.sql.submit.mode": "script", "odps.sql.decimal.odps2": "true", }, ) rows: list[dict[str, object]] = [] with instance.open_reader(tunnel=True) as reader: for record in reader: rows.append( { "strategy": record["strategy"], "demand_id": record["demand_id"], "demand_name": record["demand_name"], "weight": normalize_scalar(record["weight"]), "type": record["type"], "video_count": record["video_count"], "video_list": parse_video_list(record["video_list"]), "extend": record["extend"], } ) return rows