"""去年同期阴历策略 ODPS 查询。"""

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


def build_lunar_same_period_sql(
    *,
    bizdate: str,
    period_days: int,
    view_pv_count: int,
    min_contribution_score: float,
    rov_avg: float,
) -> str:
    if period_days < 1:
        raise ValueError("period_days 须 >= 1")
    days_after = period_days - 1

    return f"""
SET odps.sql.type.system.odps2 = true
;

WITH base_date AS (
    SELECT
        lunar_to_solar(
            lunar_add(
                loghubods.solar_to_lunar('{bizdate}', 'yyyyMMdd'),
                -1,
                'Y'
            )
        ) AS start_date
),
date_params AS (
    SELECT
        start_date,
        DATE_FORMAT(
            CAST(DATE_ADD(TO_DATE(start_date, 'yyyyMMdd'), {int(days_after)}) AS TIMESTAMP),
            'yyyyMMdd'
        ) AS end_date
    FROM base_date
),
cleaned_video_metrics AS (
    SELECT
        CAST(d.视频id AS STRING) AS vid,
        d.rov_t0
    FROM loghubods.video_dimension_detail_add_column d
    WHERE d.dt >= (SELECT start_date FROM date_params)
      AND d.dt <= (SELECT end_date FROM date_params)
      AND COALESCE(d.`当日分发曝光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_lunar_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())

    sql = build_lunar_same_period_sql(
        bizdate=bizdate,
        period_days=period_days,
        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