"""ODPS 数据读取与日聚合 SQL。"""

from __future__ import annotations

from datetime import datetime, timedelta
from typing import Tuple
from zoneinfo import ZoneInfo

import pandas as pd

from .metrics import GZH_CHANNEL, SELF_CHANNEL
from .odps_client import ODPSClient


TABLE_NAME = "loghubods.opengid_base_data"
SHANGHAI = ZoneInfo("Asia/Shanghai")


class SourceDataNotReadyError(RuntimeError):
    """本轮所需的精确 ROI 源分区尚未就绪。"""


def parse_yyyymmdd(value: str) -> datetime:
    try:
        return datetime.strptime(value, "%Y%m%d")
    except ValueError as exc:
        raise ValueError(f"日期必须是 YYYYMMDD，实际为: {value}") from exc


def date_window(end_date: str) -> Tuple[str, str]:
    end = parse_yyyymmdd(end_date)
    return (end - timedelta(days=2)).strftime("%Y%m%d"), end.strftime("%Y%m%d")


def build_source_readiness_sql(end_date: str) -> str:
    parse_yyyymmdd(end_date)
    return f"""
SELECT
  COUNT(1) AS row_count,
  SUM(CASE WHEN channel = '{SELF_CHANNEL}' THEN 1 ELSE 0 END) AS self_rows,
  SUM(CASE WHEN channel = '{GZH_CHANNEL}' THEN 1 ELSE 0 END) AS gzh_rows
FROM {TABLE_NAME}
WHERE dt = '{end_date}'
  AND usersharedepth <= 1
  AND videoid IS NOT NULL
  AND NVL(hotsencetype, '') <> '1167'
""".strip()


def validate_source_ready(client: ODPSClient, end_date: str) -> dict[str, int]:
    """要求每个报表渠道都存在精确且非空的 T-1 分区。"""

    max_partition = client.odps.get_table("opengid_base_data").get_max_partition()
    if max_partition is None:
        raise SourceDataNotReadyError(
            f"ROI来源表没有可用分区: table={TABLE_NAME}, required_dt={end_date}"
        )
    latest = max_partition.partition_spec["dt"]
    parse_yyyymmdd(latest)
    if latest < end_date:
        raise SourceDataNotReadyError(
            f"ROI来源表未同步目标分区: table={TABLE_NAME}, "
            f"required_dt={end_date}, latest_dt={latest}"
        )

    frame = client.execute_sql(build_source_readiness_sql(end_date))
    if frame.empty:
        raise SourceDataNotReadyError(
            f"ROI来源表目标分区为空: table={TABLE_NAME}, dt={end_date}"
        )
    row = frame.iloc[0]
    counts = {}
    for name in ("row_count", "self_rows", "gzh_rows"):
        value = row.get(name)
        counts[name] = 0 if pd.isna(value) else int(value)
    missing = [name for name, value in counts.items() if value <= 0]
    if missing:
        raise SourceDataNotReadyError(
            f"ROI来源表目标分区业务数据未就绪: table={TABLE_NAME}, "
            f"dt={end_date}, missing={','.join(missing)}"
        )
    return counts


def resolve_end_date(
    client: ODPSClient,
    requested: str | None = None,
    *,
    now: datetime | None = None,
) -> str:
    """精确解析指定日期或 T-1，并校验数据源就绪状态。"""

    if requested:
        parse_yyyymmdd(requested)
        target = requested
    else:
        current = now or datetime.now(SHANGHAI)
        if current.tzinfo is None:
            current = current.replace(tzinfo=SHANGHAI)
        target = (current.astimezone(SHANGHAI) - timedelta(days=1)).strftime(
            "%Y%m%d"
        )
    validate_source_ready(client, target)
    return target


def build_daily_sql(start_date: str, end_date: str) -> str:
    parse_yyyymmdd(start_date)
    parse_yyyymmdd(end_date)
    common_filter = f"""
      dt BETWEEN '{start_date}' AND '{end_date}'
      AND usersharedepth <= 1
      AND videoid IS NOT NULL
      AND NVL(hotsencetype, '') <> '1167'
    """
    optimize_goal = (
        "CASE WHEN 广告优化目标 IS NULL OR 广告优化目标='' "
        "OR 广告优化目标='null' THEN '' ELSE 广告优化目标 END"
    )

    return f"""
SELECT
  dt,
  'self' AS entity_type,
  channel,
  MAX(NVL(代理名称, '')) AS 代理名称,
  NVL(账号id, '') AS 账号id,
  MAX(NVL(账号名称, '')) AS 账号名称,
  NVL(广告id, '') AS 广告id,
  MAX(NVL(广告名称, '')) AS 广告名称,
  NVL(包名, '') AS 包名,
  {optimize_goal} AS 广告优化目标,
  NVL(创意id, '') AS 创意id,
  '' AS 合作方名,
  '' AS 公众号名,
  COUNT(DISTINCT mid) AS 首层UV,
  SUM(NVL(t0_fission_uv_root, 0)) AS T0裂变数,
  SUM(NVL(成本, 0)) AS 成本,
  SUM(NVL(效率收入, 0)) AS 效率收入,
  SUM(NVL(裂变效率收入, 0)) AS 裂变效率收入
FROM {TABLE_NAME}
WHERE {common_filter}
  AND channel = '{SELF_CHANNEL}'
  AND 广告id IS NOT NULL
  AND 创意id IS NOT NULL
GROUP BY
  dt, channel, 账号id, 广告id, 包名, {optimize_goal}, 创意id

UNION ALL

SELECT
  dt,
  'self_ad' AS entity_type,
  channel,
  MAX(NVL(代理名称, '')) AS 代理名称,
  NVL(账号id, '') AS 账号id,
  MAX(NVL(账号名称, '')) AS 账号名称,
  NVL(广告id, '') AS 广告id,
  MAX(NVL(广告名称, '')) AS 广告名称,
  NVL(包名, '') AS 包名,
  {optimize_goal} AS 广告优化目标,
  '' AS 创意id,
  '' AS 合作方名,
  '' AS 公众号名,
  COUNT(DISTINCT mid) AS 首层UV,
  SUM(NVL(t0_fission_uv_root, 0)) AS T0裂变数,
  SUM(NVL(成本, 0)) AS 成本,
  SUM(NVL(效率收入, 0)) AS 效率收入,
  SUM(NVL(裂变效率收入, 0)) AS 裂变效率收入
FROM {TABLE_NAME}
WHERE {common_filter}
  AND channel = '{SELF_CHANNEL}'
  AND 广告id IS NOT NULL
GROUP BY
  dt, channel, 账号id, 广告id, 包名, {optimize_goal}

UNION ALL

SELECT
  dt,
  'gzh' AS entity_type,
  channel,
  '' AS 代理名称,
  '' AS 账号id,
  '' AS 账号名称,
  '' AS 广告id,
  '' AS 广告名称,
  '' AS 包名,
  '' AS 广告优化目标,
  '' AS 创意id,
  NVL(合作方名, '') AS 合作方名,
  NVL(公众号名, '') AS 公众号名,
  COUNT(DISTINCT mid) AS 首层UV,
  SUM(NVL(t0_fission_uv_root, 0)) AS T0裂变数,
  SUM(NVL(成本, 0)) AS 成本,
  SUM(NVL(效率收入, 0)) AS 效率收入,
  SUM(NVL(裂变效率收入, 0)) AS 裂变效率收入
FROM {TABLE_NAME}
WHERE {common_filter}
  AND channel = '{GZH_CHANNEL}'
  AND 公众号名 IS NOT NULL
GROUP BY dt, channel, 合作方名, 公众号名
""".strip()


def build_recent_spend_accounts_sql(start_date: str, end_date: str) -> str:
    """构建包含三个完整消耗分区的账户范围查询。"""

    parse_yyyymmdd(start_date)
    parse_yyyymmdd(end_date)
    return f"""
SELECT
  账号id AS account_id,
  MAX(NVL(账号名称, '')) AS account_name,
  SUM(NVL(成本, 0)) AS cost_yuan
FROM {TABLE_NAME}
WHERE dt BETWEEN '{start_date}' AND '{end_date}'
  AND 账号id IS NOT NULL
  AND NVL(账号id, '') <> ''
GROUP BY 账号id
HAVING SUM(NVL(成本, 0)) > 0
ORDER BY account_id
""".strip()


def fetch_recent_spend_accounts(
    client: ODPSClient,
    start_date: str,
    end_date: str,
) -> list[dict[str, object]]:
    frame = client.execute_sql(
        build_recent_spend_accounts_sql(start_date, end_date)
    )
    accounts: list[dict[str, object]] = []
    for row in frame.to_dict(orient="records"):
        raw_account_id = row.get("account_id")
        try:
            account_id = int(raw_account_id)
        except (TypeError, ValueError):
            continue
        if account_id <= 0:
            continue
        accounts.append(
            {
                "account_id": account_id,
                "account_name": str(row.get("account_name") or ""),
                "cost_yuan": float(row.get("cost_yuan") or 0),
            }
        )
    return accounts


def build_account_agency_fallback_sql(account_ids: list[int]) -> str:
    """为已知数字账户 ID 构建账户级代理商兜底查询。"""

    normalized_ids: set[int] = set()
    for value in account_ids:
        try:
            account_id = int(value)
        except (TypeError, ValueError):
            continue
        if account_id > 0:
            normalized_ids.add(account_id)
    sorted_ids = sorted(normalized_ids)
    if not sorted_ids:
        raise ValueError("account_ids must contain at least one positive account ID")
    account_filter = ", ".join(f"'{value}'" for value in sorted_ids)
    return f"""
SELECT account_id, account_name, agent_name
FROM (
  SELECT
    account_id,
    account_name,
    agent_name,
    is_delete,
    status,
    ROW_NUMBER() OVER (
      PARTITION BY account_id
      ORDER BY id DESC
    ) AS row_number
  FROM loghubods.ad_put_tencent_account
  WHERE account_id IN ({account_filter})
) latest
WHERE row_number = 1
  AND NVL(is_delete, 0) = 0
  AND status = 1
  AND agent_name IS NOT NULL
  AND TRIM(agent_name) <> ''
""".strip()


def fetch_account_agency_fallbacks(
    client: ODPSClient,
    account_ids: list[int],
) -> dict[int, str]:
    """读取各账户最新、未删除且非空的代理商名称。"""

    if not account_ids:
        return {}
    frame = client.execute_sql(build_account_agency_fallback_sql(account_ids))
    agencies: dict[int, str] = {}
    for row in frame.to_dict(orient="records"):
        try:
            account_id = int(row.get("account_id"))
        except (TypeError, ValueError):
            continue
        raw_agency_name = row.get("agent_name")
        agency_name = (
            ""
            if raw_agency_name is None or pd.isna(raw_agency_name)
            else "".join(str(raw_agency_name).split())
        )
        if account_id > 0 and agency_name:
            agencies[account_id] = agency_name
    return agencies


def build_ad_age_sql(end_date: str, lookback_days: int = 30) -> str:
    end = parse_yyyymmdd(end_date)
    start_date = (end - timedelta(days=lookback_days - 1)).strftime("%Y%m%d")
    return f"""
SELECT
  广告id,
  MIN(dt) AS 首次出现日期
FROM {TABLE_NAME}
WHERE dt BETWEEN '{start_date}' AND '{end_date}'
  AND usersharedepth = '0'
  AND channel = '{SELF_CHANNEL}'
  AND videoid IS NOT NULL
  AND NVL(hotsencetype, '') <> '1167'
  AND 广告id IS NOT NULL
GROUP BY 广告id
""".strip()


def fetch_daily_data(
    client: ODPSClient,
    start_date: str,
    end_date: str,
) -> pd.DataFrame:
    return client.execute_sql(build_daily_sql(start_date, end_date))


def fetch_ad_age(
    client: ODPSClient,
    end_date: str,
    lookback_days: int = 30,
) -> pd.DataFrame:
    raw = client.execute_sql(build_ad_age_sql(end_date, lookback_days))
    if raw.empty:
        return pd.DataFrame(columns=["广告id", "广告age"])

    end = parse_yyyymmdd(end_date)
    result = raw.copy()
    result["广告id"] = result["广告id"].fillna("").astype(str)
    first_seen = pd.to_datetime(result["首次出现日期"].astype(str), format="%Y%m%d")
    result["广告age"] = (end - first_seen).dt.days + 1
    return result[["广告id", "广告age"]]