demand_server/app/strategies/odps/monthly_demands.py

"""逐月策略 ODPS 查询。"""

import re
from datetime import datetime
from zoneinfo import ZoneInfo

from app.odps.client import get_odps_client
from app.strategies.odps._utils import normalize_scalar, parse_video_list

SHANGHAI_TZ = ZoneInfo("Asia/Shanghai")

_DATE_PARTITION_RE = re.compile(r"^\d{8}$")

_EXCLUDED_ELEMENTS = (
    "元旦", "腊八节", "小年", "除夕", "春节", "正月初一", "正月初二", "正月初三",
    "正月初四", "正月初五", "情人节", "元宵节", "龙抬头", "妇女节", "植树节", "劳动节",
    "母亲节", "儿童节", "端午节", "父亲节", "建党节", "建军节", "七夕节", "中元节",
    "中秋节", "国庆节", "重阳节", "感恩节", "公祭日", "平安夜", "圣诞节", "小寒",
    "大寒", "立春", "雨水", "惊蛰", "春分", "清明", "谷雨", "立夏", "小满", "芒种",
    "夏至", "小暑", "大暑", "立秋", "处暑", "白露", "秋分", "寒露", "霜降", "立冬",
    "小雪", "大雪", "冬至", "早上好", "中午好", "下午好", "晚上好", "晚安", "祝福",
    "祝愿", "祝你", "祝贺", "祝大家", "祝您", "祝好运", "祝群主", "祝朋友",
)


def _validate_bizdate(bizdate: str) -> str:
    value = bizdate.strip()
    if not _DATE_PARTITION_RE.match(value):
        raise ValueError(f"bizdate 须为 YYYYMMDD 格式，当前为 {value!r}")
    return value


def _sql_string_list(values: tuple[str, ...]) -> str:
    return ", ".join(f"'{item}'" for item in values)


def build_monthly_demands_sql(
    *,
    bizdate: str,
    strategy_label: str,
    view_pv_count: int,
    month_total_pv_threshold: float,
    min_contribution_score: float,
    rov_avg: float,
    min_frequency: int,
) -> str:
    bizdate_value = _validate_bizdate(bizdate)
    label = strategy_label.strip()
    if not label:
        raise ValueError("strategy_label cannot be empty")
    excluded_sql = _sql_string_list(_EXCLUDED_ELEMENTS)

    return f"""
WITH biz_day AS (
    SELECT TO_DATE(
        CONCAT(
            SUBSTR('{bizdate_value}', 1, 4), '-',
            SUBSTR('{bizdate_value}', 5, 2), '-',
            SUBSTR('{bizdate_value}', 7, 2)
        )
    ) AS biz_dt
),
yesterday AS (
    SELECT DATE_SUB((SELECT biz_dt FROM biz_day), 1) AS yest
),
window_bounds AS (
    SELECT
        CAST((SELECT yest FROM yesterday) AS DATETIME) AS end_dt,
        CAST(DATE_SUB((SELECT yest FROM yesterday), 359) AS DATETIME) AS start_dt
),
cleaned_video_metrics AS (
    SELECT
        CAST(视频id AS STRING) AS vid,
        CAST(FLOOR(DATEDIFF(
            (SELECT yest FROM yesterday),
            TO_DATE(REGEXP_REPLACE(CAST(dt AS STRING), '-', ''), 'yyyyMMdd')
        ) / 30) AS STRING) AS ym,
        rov_t0,
        COALESCE(`当日分发曝光pv`, 0) AS day_dist_pv
    FROM loghubods.video_dimension_detail_add_column
    WHERE TO_DATE(REGEXP_REPLACE(CAST(dt AS STRING), '-', ''), 'yyyyMMdd')
        BETWEEN (SELECT start_dt FROM window_bounds) AND (SELECT end_dt FROM window_bounds)
      AND COALESCE(`当日分发曝光pv`, 0) >= {int(view_pv_count)}
),
video_monthly_avg_metrics AS (
    SELECT
        ym,
        vid,
        AVG(CASE WHEN rov_t0 = 0 THEN NULL ELSE rov_t0 END) AS vid_avg_rov,
        SUM(day_dist_pv) AS month_total_pv
    FROM cleaned_video_metrics
    GROUP BY ym, vid
    HAVING SUM(day_dist_pv) > {float(month_total_pv_threshold)}
),
tag_vid_dedup AS (
    SELECT DISTINCT
        CAST(vid AS STRING) AS vid,
        原始元素
    FROM loghubods.dwd_topic_decode_result_detail_di
    WHERE dt = MAX_PT('loghubods.dwd_topic_decode_result_detail_di')
      AND 元素维度 = '实质'
      AND 贡献分 >= {float(min_contribution_score)}
),
element_monthly_metrics AS (
    SELECT
        t1.原始元素,
        t2.ym,
        COALESCE(ROUND(AVG(t2.vid_avg_rov), 6), 0) AS month_avg_rov
    FROM tag_vid_dedup t1
    JOIN video_monthly_avg_metrics t2
      ON t1.vid = t2.vid
    GROUP BY t1.原始元素, t2.ym
    HAVING COALESCE(ROUND(AVG(t2.vid_avg_rov), 6), 0) >= {float(rov_avg)}
),
element_total_rov AS (
    SELECT
        原始元素,
        ROUND(SUM(month_avg_rov), 6) AS avg_rov
    FROM element_monthly_metrics
    GROUP BY 原始元素
),
element_vid_dedup AS (
    SELECT DISTINCT
        em.原始元素,
        vm.vid
    FROM element_monthly_metrics em
    JOIN tag_vid_dedup tv
      ON em.原始元素 = tv.原始元素
    JOIN video_monthly_avg_metrics vm
      ON tv.vid = vm.vid
     AND em.ym = vm.ym
),
element_vid_stats AS (
    SELECT
        原始元素,
        COUNT(DISTINCT vid) AS vid_count,
        COLLECT_SET(vid) AS vid_list
    FROM element_vid_dedup
    GROUP BY 原始元素
),
element_freq AS (
    SELECT
        原始元素,
        COUNT(1) AS 频次
    FROM element_monthly_metrics
    GROUP BY 原始元素
)
SELECT
    '{label}' AS strategy,
    md5(CONCAT('{label}', r.原始元素, '{bizdate_value}')) AS demand_id,
    r.原始元素 AS demand_name,
    r.avg_rov AS weight,
    '特征点' AS type,
    COALESCE(v.vid_count, 0) AS video_count,
    v.vid_list AS video_list,
    '{{}}' AS extend
FROM element_total_rov r
LEFT JOIN element_vid_stats v
  ON r.原始元素 = v.原始元素
LEFT JOIN element_freq f
  ON r.原始元素 = f.原始元素
WHERE r.原始元素 NOT IN ({excluded_sql})
  AND COALESCE(f.频次, 0) >= {int(min_frequency)}
ORDER BY weight DESC
"""


def query_monthly_demands(
    *,
    bizdate: str | None = None,
    strategy_label: str = "逐月",
    view_pv_count: int,
    month_total_pv_threshold: float,
    min_contribution_score: float,
    rov_avg: float,
    min_frequency: int,
) -> list[dict[str, object]]:
    if bizdate is None:
        bizdate = datetime.now(SHANGHAI_TZ).strftime("%Y%m%d")
    sql = build_monthly_demands_sql(
        bizdate=bizdate,
        strategy_label=strategy_label,
        view_pv_count=view_pv_count,
        month_total_pv_threshold=month_total_pv_threshold,
        min_contribution_score=min_contribution_score,
        rov_avg=rov_avg,
        min_frequency=min_frequency,
    )

    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