demand_server/app/strategies/odps/solar_same_period_demands.py

"""去年同期阳历策略 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