#!/usr/bin/env python
"""Export historical hourly spend, revenue, CPM and ROI to a Feishu sheet."""

from __future__ import annotations

import argparse
import math
import re
from datetime import date, datetime, timedelta
from pathlib import Path
from typing import TYPE_CHECKING, Any
from zoneinfo import ZoneInfo

import pandas as pd
from openpyxl import Workbook
from openpyxl.styles import Alignment, Font, PatternFill
from openpyxl.utils import get_column_letter

if TYPE_CHECKING:
    from odps import ODPS


SHANGHAI = ZoneInfo("Asia/Shanghai")
ROOT = Path(__file__).resolve().parent
MINIAPP_CHANNEL = "小程序投流-稳定"
REVENUE_TABLE = "ads_ad_own_package_detail_15min"
COST_TABLE = "opengid_base_data"
MINIAPP_REVENUE_TABLE = "ad_pqtid_behavior_info"
DEFAULT_START_HOUR = 6
DEFAULT_END_HOUR = 22

REPORT_HEADERS = [
    "日期",
    "小时",
    "总投放成本",
    "小程序投流成本",
    "商业化收入",
    "小程序投流收入",
    "CPM",
    "收入/总投放",
    "收入/小程序成本",
    "小程序投流收入/小程序成本",
    "当日累计总收入",
    "当日累计小程序投流收入",
    "当日累计总投放成本",
    "当日累计小程序投流成本",
    "当日累计收入/总投放",
    "当日累计收入/小程序成本",
    "当日累计小程序投流收入/小程序成本",
    "总去重UV",
    "总单用户成本",
    "小程序去重UV",
    "小程序单用户成本",
    "收入15分钟窗口数",
    "最新收入分区",
]


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--end-date",
        default=datetime.now(SHANGHAI).strftime("%Y%m%d"),
        help="Inclusive end date in YYYYMMDD format; defaults to today.",
    )
    parser.add_argument(
        "--days",
        type=int,
        default=7,
        help="Number of calendar days to export, inclusive of --end-date.",
    )
    parser.add_argument(
        "--miniapp-channel",
        default=MINIAPP_CHANNEL,
        help="Channel counted as miniapp delivery cost.",
    )
    parser.add_argument(
        "--start-hour",
        type=int,
        default=DEFAULT_START_HOUR,
        help="First hourly bucket to export; defaults to 6.",
    )
    parser.add_argument(
        "--end-hour",
        type=int,
        default=DEFAULT_END_HOUR,
        help="Last hourly bucket to export, inclusive; defaults to 22.",
    )
    parser.add_argument(
        "--output",
        type=Path,
        help="Local xlsx output path; defaults to outputs/historical_hourly_roi_*.xlsx.",
    )
    parser.add_argument(
        "--skip-upload",
        action="store_true",
        help="Only generate the local xlsx; do not import it to Feishu.",
    )
    return parser.parse_args()


def _parse_date(value: str) -> date:
    if not re.fullmatch(r"\d{8}", value):
        raise ValueError("date must use YYYYMMDD format")
    return datetime.strptime(value, "%Y%m%d").date()


def _date_values(start_date: date, end_date: date) -> list[date]:
    days = (end_date - start_date).days
    return [start_date + timedelta(days=offset) for offset in range(days + 1)]


def fetch_revenue_rows(
    client: "ODPS",
    *,
    start_date: date,
    end_date: date,
) -> pd.DataFrame:
    sql = f"""
SELECT
  dt,
  report_date,
  MIN(NVL(daily_exposure_cnt, 0)) AS daily_exposure_cnt,
  MIN(NVL(overall_cpm, 0)) AS overall_cpm,
  MIN(NVL(package_cost_times_today, 0)) AS package_cost_times_today
FROM loghubods.{REVENUE_TABLE}
WHERE dt >= '{start_date:%Y%m%d}000000'
  AND dt <= '{end_date:%Y%m%d}235959'
  AND data_type = 'today'
  AND report_date IS NOT NULL
GROUP BY dt, report_date
ORDER BY report_date
""".strip()
    instance = client.execute_sql(sql, hints={"odps.sql.submit.mode": "script"})
    with instance.open_reader(tunnel=True) as reader:
        return reader.to_pandas()


def fetch_cost_rows(
    client: "ODPS",
    *,
    start_date: date,
    end_date: date,
    miniapp_channel: str,
) -> pd.DataFrame:
    escaped_channel = miniapp_channel.replace("'", "''")
    sql = f"""
SELECT
  dt AS data_date,
  CAST(`点击小时` AS BIGINT) AS hour_of_day,
  COUNT(DISTINCT mid) AS total_uv,
  SUM(NVL(`成本`, 0)) AS total_cost,
  CASE
    WHEN COUNT(DISTINCT mid) > 0
    THEN SUM(NVL(`成本`, 0)) / COUNT(DISTINCT mid)
    ELSE 0
  END AS total_unit_user_cost,
  COUNT(DISTINCT CASE WHEN channel = '{escaped_channel}' THEN mid ELSE NULL END)
    AS miniapp_uv,
  SUM(CASE WHEN channel = '{escaped_channel}' THEN NVL(`成本`, 0) ELSE 0 END)
    AS miniapp_cost,
  CASE
    WHEN COUNT(DISTINCT CASE WHEN channel = '{escaped_channel}' THEN mid ELSE NULL END) > 0
    THEN SUM(CASE WHEN channel = '{escaped_channel}' THEN NVL(`成本`, 0) ELSE 0 END)
      / COUNT(DISTINCT CASE WHEN channel = '{escaped_channel}' THEN mid ELSE NULL END)
    ELSE 0
  END AS miniapp_unit_user_cost
FROM loghubods.{COST_TABLE}
WHERE dt >= '{start_date:%Y%m%d}'
  AND dt <= '{end_date:%Y%m%d}'
  AND usersharedepth = '0'
  AND videoid IS NOT NULL
  AND NVL(hotsencetype, '') <> '1167'
GROUP BY dt, CAST(`点击小时` AS BIGINT)
ORDER BY dt, hour_of_day
""".strip()
    instance = client.execute_sql(sql, hints={"odps.sql.submit.mode": "script"})
    with instance.open_reader(tunnel=True) as reader:
        return reader.to_pandas()


def fetch_miniapp_revenue_rows(
    client: "ODPS",
    *,
    start_date: date,
    end_date: date,
    miniapp_channel: str,
) -> pd.DataFrame:
    escaped_channel = miniapp_channel.replace("'", "''")
    sql = f"""
SELECT
  dt AS data_date,
  CAST(hh AS BIGINT) AS hour_of_day,
  SUM(NVL(`真实收入`, 0)) AS miniapp_revenue
FROM loghubods.{MINIAPP_REVENUE_TABLE}
WHERE dt >= '{start_date:%Y%m%d}'
  AND dt <= '{end_date:%Y%m%d}'
  AND `增长渠道来源` = '{escaped_channel}'
GROUP BY dt, CAST(hh AS BIGINT)
ORDER BY dt, hour_of_day
""".strip()
    instance = client.execute_sql(sql, hints={"odps.sql.submit.mode": "script"})
    with instance.open_reader(tunnel=True) as reader:
        return reader.to_pandas()


def hourly_revenue(revenue_rows: pd.DataFrame) -> pd.DataFrame:
    columns = [
        "data_date",
        "hour_of_day",
        "commercial_revenue",
        "cpm",
        "revenue_window_count",
        "latest_revenue_partition",
    ]
    if revenue_rows.empty:
        return pd.DataFrame(columns=columns)

    rows = revenue_rows.copy()
    rows["window_start"] = pd.to_datetime(
        rows["report_date"].astype(str),
        format="%Y%m%d%H%M%S",
        errors="coerce",
    )
    rows = rows.dropna(subset=["window_start"])
    rows["data_date"] = rows["window_start"].dt.strftime("%Y%m%d")
    rows["hour_of_day"] = rows["window_start"].dt.hour.astype("int64")
    rows["commercial_revenue"] = pd.to_numeric(
        rows["package_cost_times_today"],
        errors="coerce",
    ).fillna(0.0)
    rows["daily_exposure_cnt"] = pd.to_numeric(
        rows["daily_exposure_cnt"],
        errors="coerce",
    ).fillna(0.0)
    rows["overall_cpm"] = pd.to_numeric(rows["overall_cpm"], errors="coerce")
    rows["weighted_cpm"] = rows["overall_cpm"].fillna(0.0) * rows[
        "daily_exposure_cnt"
    ]

    grouped = (
        rows.groupby(["data_date", "hour_of_day"], as_index=False)
        .agg(
            commercial_revenue=("commercial_revenue", "sum"),
            exposure=("daily_exposure_cnt", "sum"),
            weighted_cpm=("weighted_cpm", "sum"),
            avg_cpm=("overall_cpm", "mean"),
            revenue_window_count=("report_date", "count"),
            latest_revenue_partition=("dt", "max"),
        )
        .sort_values(["data_date", "hour_of_day"])
    )
    grouped["cpm"] = grouped.apply(
        lambda row: (
            row["weighted_cpm"] / row["exposure"]
            if row["exposure"] and not math.isnan(row["exposure"])
            else row["avg_cpm"]
        ),
        axis=1,
    )
    return grouped[columns]


def hourly_cost(cost_rows: pd.DataFrame) -> pd.DataFrame:
    columns = [
        "data_date",
        "hour_of_day",
        "total_cost",
        "miniapp_cost",
        "total_uv",
        "total_unit_user_cost",
        "miniapp_uv",
        "miniapp_unit_user_cost",
    ]
    if cost_rows.empty:
        return pd.DataFrame(columns=columns)

    rows = cost_rows.copy()
    rows["data_date"] = rows["data_date"].astype(str)
    rows["hour_of_day"] = pd.to_numeric(rows["hour_of_day"], errors="coerce")
    rows = rows.dropna(subset=["hour_of_day"])
    rows["hour_of_day"] = rows["hour_of_day"].astype("int64")
    for column in columns[2:]:
        rows[column] = pd.to_numeric(rows[column], errors="coerce").fillna(0.0)
    return rows[columns].sort_values(["data_date", "hour_of_day"])


def hourly_miniapp_revenue(revenue_rows: pd.DataFrame) -> pd.DataFrame:
    columns = ["data_date", "hour_of_day", "miniapp_revenue"]
    if revenue_rows.empty:
        return pd.DataFrame(columns=columns)

    rows = revenue_rows.copy()
    rows["data_date"] = rows["data_date"].astype(str)
    rows["hour_of_day"] = pd.to_numeric(rows["hour_of_day"], errors="coerce")
    rows = rows.dropna(subset=["hour_of_day"])
    rows["hour_of_day"] = rows["hour_of_day"].astype("int64")
    rows["miniapp_revenue"] = pd.to_numeric(
        rows["miniapp_revenue"], errors="coerce"
    ).fillna(0.0)
    return rows[columns].sort_values(["data_date", "hour_of_day"])


def build_hourly_report(
    *,
    revenue_rows: pd.DataFrame,
    cost_rows: pd.DataFrame,
    miniapp_revenue_rows: pd.DataFrame | None = None,
    start_date: date,
    end_date: date,
    start_hour: int = DEFAULT_START_HOUR,
    end_hour: int = DEFAULT_END_HOUR,
) -> pd.DataFrame:
    if not 0 <= start_hour <= end_hour <= 23:
        raise ValueError("hours must satisfy 0 <= start_hour <= end_hour <= 23")
    # Build the full day first so cumulative values at 06:00 include 00:00-05:00.
    grid = pd.DataFrame(
        [
            {"data_date": f"{item:%Y%m%d}", "hour_of_day": hour}
            for item in _date_values(start_date, end_date)
            for hour in range(24)
        ]
    )
    report = grid.merge(
        hourly_cost(cost_rows),
        on=["data_date", "hour_of_day"],
        how="left",
    ).merge(
        hourly_revenue(revenue_rows),
        on=["data_date", "hour_of_day"],
        how="left",
    ).merge(
        hourly_miniapp_revenue(
            miniapp_revenue_rows
            if miniapp_revenue_rows is not None
            else pd.DataFrame()
        ),
        on=["data_date", "hour_of_day"],
        how="left",
    )

    numeric_defaults = {
        "total_cost": 0.0,
        "miniapp_cost": 0.0,
        "commercial_revenue": 0.0,
        "miniapp_revenue": 0.0,
        "total_uv": 0.0,
        "total_unit_user_cost": 0.0,
        "miniapp_uv": 0.0,
        "miniapp_unit_user_cost": 0.0,
        "revenue_window_count": 0.0,
    }
    for column, default in numeric_defaults.items():
        report[column] = pd.to_numeric(report[column], errors="coerce").fillna(default)

    report["收入/总投放"] = report.apply(
        lambda row: (
            row["commercial_revenue"] / row["total_cost"]
            if row["total_cost"] > 0
            else None
        ),
        axis=1,
    )
    report["收入/小程序成本"] = report.apply(
        lambda row: (
            row["commercial_revenue"] / row["miniapp_cost"]
            if row["miniapp_cost"] > 0
            else None
        ),
        axis=1,
    )
    report["小程序投流收入/小程序成本"] = report.apply(
        lambda row: (
            row["miniapp_revenue"] / row["miniapp_cost"]
            if row["miniapp_cost"] > 0
            else None
        ),
        axis=1,
    )
    report = report.sort_values(["data_date", "hour_of_day"]).reset_index(drop=True)
    report["当日累计总收入"] = report.groupby("data_date")[
        "commercial_revenue"
    ].cumsum()
    report["当日累计小程序投流收入"] = report.groupby("data_date")[
        "miniapp_revenue"
    ].cumsum()
    report["当日累计总投放成本"] = report.groupby("data_date")[
        "total_cost"
    ].cumsum()
    report["当日累计小程序投流成本"] = report.groupby("data_date")[
        "miniapp_cost"
    ].cumsum()
    report["当日累计收入/总投放"] = report.apply(
        lambda row: (
            row["当日累计总收入"] / row["当日累计总投放成本"]
            if row["当日累计总投放成本"] > 0
            else None
        ),
        axis=1,
    )
    report["当日累计收入/小程序成本"] = report.apply(
        lambda row: (
            row["当日累计总收入"] / row["当日累计小程序投流成本"]
            if row["当日累计小程序投流成本"] > 0
            else None
        ),
        axis=1,
    )
    report["当日累计小程序投流收入/小程序成本"] = report.apply(
        lambda row: (
            row["当日累计小程序投流收入"] / row["当日累计小程序投流成本"]
            if row["当日累计小程序投流成本"] > 0
            else None
        ),
        axis=1,
    )
    report["日期"] = pd.to_datetime(report["data_date"], format="%Y%m%d").dt.strftime(
        "%Y-%m-%d"
    )
    report["小时"] = report["hour_of_day"].map(lambda value: f"{int(value):02d}:00")
    report = report.rename(
        columns={
            "total_cost": "总投放成本",
            "miniapp_cost": "小程序投流成本",
            "commercial_revenue": "商业化收入",
            "miniapp_revenue": "小程序投流收入",
            "cpm": "CPM",
            "total_uv": "总去重UV",
            "total_unit_user_cost": "总单用户成本",
            "miniapp_uv": "小程序去重UV",
            "miniapp_unit_user_cost": "小程序单用户成本",
            "revenue_window_count": "收入15分钟窗口数",
            "latest_revenue_partition": "最新收入分区",
        }
    )
    return report[
        report["hour_of_day"].between(start_hour, end_hour)
    ][REPORT_HEADERS]


def _clean_cell(value: Any) -> Any:
    if pd.isna(value):
        return ""
    if isinstance(value, float) and math.isfinite(value):
        return float(value)
    return value


def create_workbook(
    *,
    report: pd.DataFrame,
    start_date: date,
    end_date: date,
    miniapp_channel: str,
    output_path: Path,
    start_hour: int = DEFAULT_START_HOUR,
    end_hour: int = DEFAULT_END_HOUR,
) -> Path:
    output_path.parent.mkdir(parents=True, exist_ok=True)
    workbook = Workbook()
    sheet = workbook.active
    sheet.title = "历史小时ROI"
    sheet.append(REPORT_HEADERS)
    for row in report.itertuples(index=False):
        sheet.append([_clean_cell(value) for value in row])

    header_fill = PatternFill("solid", fgColor="1F4E78")
    for cell in sheet[1]:
        cell.fill = header_fill
        cell.font = Font(color="FFFFFF", bold=True)
        cell.alignment = Alignment(horizontal="center", vertical="center")

    widths = [
        13, 9, 14, 16, 14, 16, 10, 14, 17, 22, 16, 20, 18, 22,
        20, 24, 28, 11, 14, 13, 16, 17, 18,
    ]
    for index, width in enumerate(widths, start=1):
        sheet.column_dimensions[get_column_letter(index)].width = width
    for row in sheet.iter_rows(min_row=2):
        for cell in row:
            cell.alignment = Alignment(horizontal="center", vertical="center")
    money_headers = (
        "总投放成本",
        "小程序投流成本",
        "商业化收入",
        "小程序投流收入",
        "CPM",
        "当日累计总收入",
        "当日累计小程序投流收入",
        "当日累计总投放成本",
        "当日累计小程序投流成本",
        "总单用户成本",
        "小程序单用户成本",
    )
    ratio_headers = (
        "收入/总投放",
        "收入/小程序成本",
        "小程序投流收入/小程序成本",
        "当日累计收入/总投放",
        "当日累计收入/小程序成本",
        "当日累计小程序投流收入/小程序成本",
    )
    count_headers = ("总去重UV", "小程序去重UV", "收入15分钟窗口数")
    for header in money_headers:
        column = get_column_letter(REPORT_HEADERS.index(header) + 1)
        for cell in sheet[column][1:]:
            cell.number_format = "0.00"
    for header in ratio_headers:
        column = get_column_letter(REPORT_HEADERS.index(header) + 1)
        for cell in sheet[column][1:]:
            cell.number_format = "0.0000"
    for header in count_headers:
        column = get_column_letter(REPORT_HEADERS.index(header) + 1)
        for cell in sheet[column][1:]:
            cell.number_format = "0"
    sheet.freeze_panes = "A2"
    sheet.auto_filter.ref = sheet.dimensions

    summary = workbook.create_sheet("运行摘要")
    summary_rows = [
        ("生成时间", datetime.now(SHANGHAI).strftime("%Y-%m-%d %H:%M:%S")),
        ("统计开始日期", start_date.strftime("%Y-%m-%d")),
        ("统计结束日期", end_date.strftime("%Y-%m-%d")),
        ("每日统计时段", f"{start_hour:02d}:00-{end_hour:02d}:59"),
        ("统计小时数", len(report)),
        ("小程序成本渠道", miniapp_channel),
        (
            "小程序收入来源",
            f"loghubods.{MINIAPP_REVENUE_TABLE}.真实收入",
        ),
        ("总投放成本", float(report["总投放成本"].sum())),
        ("小程序投流成本", float(report["小程序投流成本"].sum())),
        ("商业化收入", float(report["商业化收入"].sum())),
        ("小程序投流收入", float(report["小程序投流收入"].sum())),
    ]
    summary.append(["字段", "值"])
    for row in summary_rows:
        summary.append(list(row))
    for cell in summary[1]:
        cell.fill = header_fill
        cell.font = Font(color="FFFFFF", bold=True)
    summary.column_dimensions["A"].width = 22
    summary.column_dimensions["B"].width = 28
    for cell in summary["B"][1:]:
        if isinstance(cell.value, float):
            cell.number_format = "0.00"

    workbook.save(output_path)
    return output_path


def main() -> None:
    args = parse_args()
    if args.days <= 0:
        raise ValueError("--days must be positive")
    if not 0 <= args.start_hour <= args.end_hour <= 23:
        raise ValueError(
            "hours must satisfy 0 <= --start-hour <= --end-hour <= 23"
        )

    end_date = _parse_date(args.end_date)
    start_date = end_date - timedelta(days=args.days - 1)

    from dotenv import load_dotenv
    from feishu_notifier import FeishuNotifier
    from odps_source import build_odps_client

    load_dotenv(ROOT.parent / "auto_put_ad_mini" / ".env", override=False)
    load_dotenv(Path.cwd() / ".env", override=False)

    client = build_odps_client()
    revenue_rows = fetch_revenue_rows(
        client,
        start_date=start_date,
        end_date=end_date,
    )
    cost_rows = fetch_cost_rows(
        client,
        start_date=start_date,
        end_date=end_date,
        miniapp_channel=args.miniapp_channel,
    )
    miniapp_revenue_rows = fetch_miniapp_revenue_rows(
        client,
        start_date=start_date,
        end_date=end_date,
        miniapp_channel=args.miniapp_channel,
    )
    report = build_hourly_report(
        revenue_rows=revenue_rows,
        cost_rows=cost_rows,
        miniapp_revenue_rows=miniapp_revenue_rows,
        start_date=start_date,
        end_date=end_date,
        start_hour=args.start_hour,
        end_hour=args.end_hour,
    )

    output_path = args.output or (
        ROOT
        / "outputs"
        / (
            "historical_hourly_roi_"
            f"{start_date:%Y%m%d}_{end_date:%Y%m%d}.xlsx"
        )
    )
    path = create_workbook(
        report=report,
        start_date=start_date,
        end_date=end_date,
        miniapp_channel=args.miniapp_channel,
        output_path=output_path,
        start_hour=args.start_hour,
        end_hour=args.end_hour,
    )

    print(f"rows={len(report)}")
    print(f"date_range={start_date:%Y%m%d}-{end_date:%Y%m%d}")
    print(f"output={path}")
    if args.skip_upload:
        print("sheet_url=SKIPPED")
        return

    url = FeishuNotifier().import_spreadsheet(path)
    print(f"sheet_url={url}")


if __name__ == "__main__":
    main()
