from __future__ import annotations

import argparse
import re
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Any

DEPS_DIR = Path(__file__).resolve().parents[2] / ".pydeps"
if DEPS_DIR.exists():
    sys.path.insert(0, str(DEPS_DIR))
SCRIPT_DIR = Path(__file__).resolve().parent
if str(SCRIPT_DIR) not in sys.path:
    sys.path.insert(0, str(SCRIPT_DIR))

import pandas as pd
import pymysql
from runtime_config import add_config_source_args, load_runtime_config


VIPDATA_DIR = Path(r"C:\vipdata")
DEFAULT_CONFIG = Path(r"D:\projects\skills\vip-report\config.yaml")


@dataclass
class MonthlyFile:
    month: str  # YYYYMM
    path: Path


def parse_month_number(raw: Any) -> int:
    if raw is None:
        return 1
    if isinstance(raw, int):
        return min(12, max(1, raw))
    text = str(raw).strip()
    m = re.search(r"(1[0-2]|0?[1-9])", text)
    if m:
        return int(m.group(1))
    return 1


def resolve_target_months(cfg: dict[str, Any]) -> set[str]:
    report_cfg = cfg.get("report", {})
    year = int(report_cfg.get("year", 2026))
    compare_year = int(report_cfg.get("compare_year", year - 1))
    month = parse_month_number(report_cfg.get("month_cn", "1"))
    return {f"{year}{month:02d}", f"{compare_year}{month:02d}"}


def resolve_mysql_config(cfg: dict[str, Any]) -> dict[str, Any]:
    mysql = cfg.get("mysql", {})
    return {
        "host": mysql.get("host", "192.168.138.182"),
        "port": int(mysql.get("port", 3306)),
        "user": mysql.get("username") or mysql.get("user") or "ceptest",
        "password": mysql.get("password", "ck@123456"),
        "database": mysql.get("database", "ckc_cep_db_test"),
        "charset": "utf8mb4",
        "autocommit": False,
    }


def parse_month_from_filename(path: Path) -> str | None:
    m = re.search(r"(20\d{6})-(20\d{6})_", path.name)
    if not m:
        return None
    return m.group(1)[:6]


def pick_latest_files(pattern: str, target_months: set[str]) -> dict[str, MonthlyFile]:
    files = sorted(VIPDATA_DIR.glob(pattern), key=lambda p: p.stat().st_mtime, reverse=True)
    chosen: dict[str, MonthlyFile] = {}
    for path in files:
        month = parse_month_from_filename(path)
        if not month or month not in target_months:
            continue
        if month not in chosen:
            chosen[month] = MonthlyFile(month=month, path=path)
    return chosen


def pct_to_float(value: Any) -> float | None:
    if value is None or (isinstance(value, float) and pd.isna(value)):
        return None
    s = str(value).strip()
    if not s:
        return None
    s = s.replace(",", "").replace("%", "")
    try:
        return float(s)
    except Exception:
        return None


def num_to_float(value: Any) -> float | None:
    if value is None or (isinstance(value, float) and pd.isna(value)):
        return None
    s = str(value).strip().replace(",", "")
    if not s:
        return None
    try:
        return float(s)
    except Exception:
        return None


def num_to_int(value: Any) -> int | None:
    v = num_to_float(value)
    if v is None:
        return None
    return int(round(v))


def str_or_none(value: Any) -> str | None:
    if value is None or (isinstance(value, float) and pd.isna(value)):
        return None
    s = str(value).strip()
    return s if s else None


def to_char8_date(value: Any) -> str | None:
    if value is None or (isinstance(value, float) and pd.isna(value)):
        return None
    s = str(value).strip()
    if not s:
        return None
    s = re.sub(r"\D", "", s)
    if len(s) >= 8:
        return s[:8]
    return None


def month_to_char7(month: str) -> str:
    return f"{month[:4]}-{month[4:6]}"


def load_ceptrafficanalysis_rows(monthly_files: dict[str, MonthlyFile]) -> tuple[list[tuple[Any, ...]], set[str]]:
    def pick(row: pd.Series, *aliases: str, index: int | None = None) -> Any:
        for alias in aliases:
            if alias in row.index:
                return row.get(alias)
        if index is not None and 0 <= index < len(row):
            return row.iloc[index]
        return None

    rows: list[tuple[Any, ...]] = []
    dates: set[str] = set()
    for item in monthly_files.values():
        df = pd.read_excel(item.path, sheet_name=0)
        for _, r in df.iterrows():
            biz_date = to_char8_date(
                pick(
                    r,
                    "日期",
                    "鏃ユ湡",
                    index=0,
                )
            )
            if not biz_date:
                continue
            dates.add(biz_date)
            rows.append(
                (
                    biz_date,
                    str_or_none(pick(r, "站内流量", "绔欏唴娴侀噺", index=1)),
                    str_or_none(pick(r, "站内付费流量", "绔欏唴浠樿垂娴侀噺", index=2)),
                    str_or_none(pick(r, "站内自然流量", "绔欏唴鑷劧娴侀噺", index=3)),
                    str_or_none(pick(r, "站外流量", "绔欏娴侀噺", index=4)),
                    str_or_none(pick(r, "站外付费流量", "绔欏浠樿垂娴侀噺", index=5)),
                    str_or_none(pick(r, "站外免费流量", "绔欏鍏嶈垂娴侀噺", index=6)),
                )
            )
    return rows, dates


def load_site_entry_rows(monthly_files: dict[str, MonthlyFile]) -> tuple[list[tuple[Any, ...]], set[str]]:
    rows: list[tuple[Any, ...]] = []
    dates: set[str] = set()
    for item in monthly_files.values():
        df = pd.read_excel(item.path, sheet_name=0)
        for _, r in df.iterrows():
            vals = r.tolist()
            biz_date = to_char8_date(vals[0] if len(vals) > 0 else None)
            if not biz_date:
                continue
            dates.add(biz_date)
            rows.append(
                (
                    biz_date,
                    str_or_none(vals[1] if len(vals) > 1 else None),
                    str_or_none(vals[2] if len(vals) > 2 else None),
                    str_or_none(vals[3] if len(vals) > 3 else None),
                    num_to_float(vals[4] if len(vals) > 4 else None),
                    pct_to_float(vals[5] if len(vals) > 5 else None),
                    num_to_float(vals[6] if len(vals) > 6 else None),
                    pct_to_float(vals[7] if len(vals) > 7 else None),
                    num_to_float(vals[8] if len(vals) > 8 else None),
                    pct_to_float(vals[9] if len(vals) > 9 else None),
                    num_to_float(vals[10] if len(vals) > 10 else None),
                    num_to_int(vals[11] if len(vals) > 11 else None),
                    num_to_int(vals[12] if len(vals) > 12 else None),
                    num_to_int(vals[13] if len(vals) > 13 else None),
                    num_to_int(vals[14] if len(vals) > 14 else None),
                    num_to_float(vals[15] if len(vals) > 15 else None),
                    num_to_float(vals[16] if len(vals) > 16 else None),
                )
            )
    return rows, dates


def load_traffic_summary_rows(monthly_files: dict[str, MonthlyFile]) -> tuple[list[tuple[Any, ...]], set[str]]:
    rows: list[tuple[Any, ...]] = []
    month_keys: set[str] = set()
    for item in monthly_files.values():
        daily = month_to_char7(item.month)
        month_keys.add(daily)
        df = pd.read_excel(item.path, sheet_name=0)
        for _, r in df.iterrows():
            vals = r.tolist()
            rows.append(
                (
                    str_or_none(vals[0] if len(vals) > 0 else None),
                    str_or_none(vals[1] if len(vals) > 1 else None),
                    str_or_none(vals[2] if len(vals) > 2 else None),
                    num_to_int(vals[3] if len(vals) > 3 else None),
                    pct_to_float(vals[4] if len(vals) > 4 else None),
                    num_to_float(vals[5] if len(vals) > 5 else None),
                    pct_to_float(vals[6] if len(vals) > 6 else None),
                    num_to_int(vals[7] if len(vals) > 7 else None),
                    pct_to_float(vals[8] if len(vals) > 8 else None),
                    num_to_float(vals[9] if len(vals) > 9 else None),
                    num_to_int(vals[10] if len(vals) > 10 else None),
                    num_to_int(vals[11] if len(vals) > 11 else None),
                    num_to_int(vals[12] if len(vals) > 12 else None),
                    num_to_int(vals[13] if len(vals) > 13 else None),
                    num_to_float(vals[14] if len(vals) > 14 else None),
                    num_to_float(vals[15] if len(vals) > 15 else None),
                    daily,
                )
            )
    return rows, month_keys


def parse_args() -> argparse.Namespace:
    p = argparse.ArgumentParser(description="Load traffic exports into MySQL by configured report month + compare year month.")
    add_config_source_args(p, default_config=DEFAULT_CONFIG)
    return p.parse_args()


def execute(cfg: dict[str, Any]) -> None:
    target_months = resolve_target_months(cfg)
    mysql_config = resolve_mysql_config(cfg)

    summary_files: dict[str, MonthlyFile] = {}
    summary_candidates = sorted(VIPDATA_DIR.glob("202*_summary_*.xlsx"), key=lambda p: p.stat().st_mtime, reverse=True)
    for path in summary_candidates:
        month = parse_month_from_filename(path)
        if not month or month not in target_months:
            continue
        if not path.name.startswith(f"{month[:4]}_summary_"):
            continue
        if month not in summary_files:
            summary_files[month] = MonthlyFile(month=month, path=path)

    oneclick_daily_files = pick_latest_files("202*_oneclick_daily_*.xlsx", target_months)
    oneclick_summary_files = pick_latest_files("202*_oneclick_summary_*.xlsx", target_months)

    if set(summary_files.keys()) != target_months:
        raise RuntimeError(f"summary files not complete: {summary_files}, target_months={target_months}")
    if set(oneclick_daily_files.keys()) != target_months:
        raise RuntimeError(f"oneclick daily files not complete: {oneclick_daily_files}, target_months={target_months}")
    if set(oneclick_summary_files.keys()) != target_months:
        raise RuntimeError(f"oneclick summary files not complete: {oneclick_summary_files}, target_months={target_months}")

    analysis_rows, analysis_dates = load_ceptrafficanalysis_rows(summary_files)
    site_rows, site_dates = load_site_entry_rows(oneclick_daily_files)
    traffic_summary_rows, traffic_months = load_traffic_summary_rows(oneclick_summary_files)

    conn = pymysql.connect(**mysql_config)
    try:
        with conn.cursor() as cur:
            if analysis_dates:
                placeholders = ",".join(["%s"] * len(analysis_dates))
                cur.execute(
                    f"DELETE FROM `ceptrafficanalysis` WHERE `daily` IN ({placeholders})",
                    tuple(sorted(analysis_dates)),
                )
            if analysis_rows:
                cur.executemany(
                    """
                    INSERT INTO `ceptrafficanalysis`
                    (`daily`,`stationTraffic`,`paidTraffic`,`organicTraffic`,`externalTraffic`,`off_site_paid_traffic`,`off_site_free_traffic`)
                    VALUES (%s,%s,%s,%s,%s,%s,%s)
                    """,
                    analysis_rows,
                )

            if site_dates:
                placeholders = ",".join(["%s"] * len(site_dates))
                cur.execute(
                    f"DELETE FROM `cep_site_traffic_entry` WHERE `biz_date` IN ({placeholders})",
                    tuple(sorted(site_dates)),
                )
            if site_rows:
                cur.executemany(
                    """
                    INSERT INTO `cep_site_traffic_entry`
                    (`biz_date`,`traffic_type`,`primary_entry`,`secondary_entry`,`product_detail_uv`,
                     `product_detail_uv_ratio`,`sales_amount_incl_reject_return`,`sales_amount_ratio`,`sales_volume`,
                     `purchase_conversion_rate`,`unit_price`,`customer_count`,`new_customer_count`,
                     `marketing_new_customer_count`,`add_to_cart_user_count`,`product_detail_uv_value`,`arpu`)
                    VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
                    """,
                    site_rows,
                )

            if traffic_months:
                placeholders = ",".join(["%s"] * len(traffic_months))
                cur.execute(
                    f"DELETE FROM `cep_traffic_summary` WHERE `daily` IN ({placeholders})",
                    tuple(sorted(traffic_months)),
                )
            if traffic_summary_rows:
                cur.executemany(
                    """
                    INSERT INTO `cep_traffic_summary`
                    (`traffic_type`,`primary_entry`,`secondary_entry`,`product_detail_uv`,`product_detail_uv_ratio`,
                     `sales_amount`,`sales_ratio`,`sales_volume`,`purchase_conversion_rate`,`unit_price`,
                     `customer_count`,`new_customer_count`,`marketing_new_customer_count`,`add_to_cart_users`,
                     `product_detail_uv_value`,`arpu`,`daily`)
                    VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
                    """,
                    traffic_summary_rows,
                )

            cur.execute("SELECT COUNT(*) FROM `ceptrafficanalysis`")
            c1 = cur.fetchone()[0]
            cur.execute("SELECT COUNT(*) FROM `cep_site_traffic_entry`")
            c2 = cur.fetchone()[0]
            cur.execute("SELECT COUNT(*) FROM `cep_traffic_summary`")
            c3 = cur.fetchone()[0]

        conn.commit()

        print("TARGET_MONTHS:", sorted(target_months))
        print("FILES_USED:")
        for k in sorted(summary_files):
            print(f"  SUMMARY[{k}]={summary_files[k].path}")
        for k in sorted(oneclick_daily_files):
            print(f"  ONECLICK_DAILY[{k}]={oneclick_daily_files[k].path}")
        for k in sorted(oneclick_summary_files):
            print(f"  ONECLICK_SUMMARY[{k}]={oneclick_summary_files[k].path}")
        print(f"INSERTED ceptrafficanalysis={len(analysis_rows)}")
        print(f"INSERTED cep_site_traffic_entry={len(site_rows)}")
        print(f"INSERTED cep_traffic_summary={len(traffic_summary_rows)}")
        print(f"TABLE_COUNTS ceptrafficanalysis={c1}, cep_site_traffic_entry={c2}, cep_traffic_summary={c3}")
    finally:
        conn.close()


if __name__ == "__main__":
    args = parse_args()
    execute(load_runtime_config(args, default_config=DEFAULT_CONFIG))
