from __future__ import annotations

import argparse
import re
import sys
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Any, Optional

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

import pandas as pd
import pymysql
import yaml

if str(SCRIPT_DIR) not in sys.path:
    sys.path.insert(0, str(SCRIPT_DIR))

from runtime_config import add_config_source_args, load_runtime_config


VIPDATA_DIR = Path(r"C:\vipdata")
TABLE_NAME = "cep_live_session_data"


DEFAULT_CONFIG_PATH = Path(r"D:\projects\skills\vip-report\config.yaml")


@dataclass(frozen=True)
class MonthTarget:
    month: str  # YYYYMM
    path: Path


def parse_month_from_filename(path: Path) -> Optional[str]:
    m = re.match(r"^(20\d{2})(0[1-9]|1[0-2])_", path.name)
    if not m:
        return None
    return f"{m.group(1)}{m.group(2)}"


def previous_month(year: int, month: int) -> tuple[int, int]:
    if month == 1:
        return year - 1, 12
    return year, month - 1


def resolve_target_months(config: dict[str, Any]) -> set[str]:
    report_cfg = config.get("report", {})
    year = int(report_cfg.get("year", 2026))
    month_raw = str(report_cfg.get("month_cn", "1"))
    mm = re.search(r"(1[0-2]|0?[1-9])", month_raw)
    month = int(mm.group(1)) if mm else 1
    py, pm = previous_month(year, month)
    return {f"{year}{month:02d}", f"{py}{pm:02d}"}


def pick_latest_files(target_months: set[str]) -> dict[str, MonthTarget]:
    chosen: dict[str, MonthTarget] = {}
    files = sorted(VIPDATA_DIR.glob("*.xlsx"), key=lambda p: p.stat().st_mtime, reverse=True)
    for path in files:
        if "直播场次数据" not in path.name:
            continue
        month = parse_month_from_filename(path)
        if not month or month not in target_months:
            continue
        if month not in chosen:
            chosen[month] = MonthTarget(month=month, path=path)
    return chosen


def to_str(value: Any) -> Optional[str]:
    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_float(value: Any) -> Optional[float]:
    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 to_int(value: Any) -> Optional[int]:
    v = to_float(value)
    if v is None:
        return None
    return int(round(v))


def pct_to_ratio(value: Any) -> Optional[float]:
    if value is None or (isinstance(value, float) and pd.isna(value)):
        return None
    s = str(value).strip().replace(",", "")
    if not s:
        return None
    if s.endswith("%"):
        s = s[:-1]
    try:
        return float(s) / 100.0
    except Exception:
        return None


def to_date(value: Any) -> Optional[str]:
    if value is None or (isinstance(value, float) and pd.isna(value)):
        return None
    if isinstance(value, datetime):
        return value.strftime("%Y-%m-%d")
    s = str(value).strip()
    m = re.match(r"^(\d{4})-(\d{2})-(\d{2})", s)
    if m:
        return f"{m.group(1)}-{m.group(2)}-{m.group(3)}"
    return None


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


def create_table_sql() -> str:
    return f"""
CREATE TABLE IF NOT EXISTS `{TABLE_NAME}` (
  `id` BIGINT NOT NULL AUTO_INCREMENT,
  `biz_date` DATE NOT NULL,
  `data_month` CHAR(7) NOT NULL,
  `room_session_id` BIGINT NULL,
  `live_room_name` VARCHAR(128) NULL,
  `brand_name` VARCHAR(64) NULL,
  `related_department` VARCHAR(64) NULL,
  `room_type` VARCHAR(128) NULL,
  `room_tag` VARCHAR(128) NULL,
  `relation_type` VARCHAR(32) NULL,
  `relation_id` BIGINT NULL,
  `live_and_commentary_sales_amount` DECIMAL(18,2) NULL,
  `accumulated_viewers` BIGINT NULL,
  `accumulated_view_count` BIGINT NULL,
  `mini_program_avg_stay_seconds` DECIMAL(10,2) NULL,
  `app_avg_stay_seconds` DECIMAL(10,2) NULL,
  `live_sales_penetration_rate` DECIMAL(12,6) NULL,
  `purchasers_total` INT NULL,
  `purchasers_excluding_coupons` INT NULL,
  `purchasers_including_coupons` INT NULL,
  `purchasers_marketing_coupon` INT NULL,
  `conversion_rate_total` DECIMAL(12,6) NULL,
  `conversion_rate_excluding_coupons` DECIMAL(12,6) NULL,
  `conversion_rate_including_coupons` DECIMAL(12,6) NULL,
  `conversion_rate_marketing_coupon` DECIMAL(12,6) NULL,
  `orders_total` INT NULL,
  `orders_excluding_coupons` INT NULL,
  `orders_including_coupons` INT NULL,
  `orders_marketing_coupon` INT NULL,
  `live_period_sales_total` DECIMAL(18,2) NULL,
  `live_period_sales_excluding_coupons` DECIMAL(18,2) NULL,
  `live_period_sales_including_coupons` DECIMAL(18,2) NULL,
  `live_period_sales_marketing_coupon` DECIMAL(18,2) NULL,
  `guided_sales_total` DECIMAL(18,2) NULL,
  `guided_sales_excluding_coupons` DECIMAL(18,2) NULL,
  `guided_sales_including_coupons` DECIMAL(18,2) NULL,
  `guided_sales_marketing_coupon` DECIMAL(18,2) NULL,
  `live_duration_minutes` INT NULL,
  `live_time_range` VARCHAR(64) NULL,
  `brand_customers_total` INT NULL,
  `brand_customers_excluding_coupons` INT NULL,
  `brand_customers_including_coupons` INT NULL,
  `brand_new_customers_total` INT NULL,
  `brand_new_customers_excluding_coupons` INT NULL,
  `brand_new_customers_including_coupons` INT NULL,
  `new_customer_ratio_total` DECIMAL(12,6) NULL,
  `new_customer_ratio_excluding_coupons` DECIMAL(12,6) NULL,
  `new_customer_ratio_including_coupons` DECIMAL(12,6) NULL,
  `avg_spend_per_customer` DECIMAL(18,2) NULL,
  `new_live_followers` INT NULL,
  `accumulated_live_followers` BIGINT NULL,
  `new_brand_members` INT NULL,
  `accumulated_brand_members` BIGINT NULL,
  `source_file` VARCHAR(255) NULL,
  `created_at` TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  KEY `idx_biz_date` (`biz_date`),
  KEY `idx_data_month` (`data_month`),
  UNIQUE KEY `uk_biz_room` (`biz_date`, `room_session_id`, `live_room_name`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
"""


def row_tuple(r: pd.Series, month: str, source_file: str) -> tuple[Any, ...]:
    return (
        to_date(r.get("日期")),
        month_to_char7(month),
        to_int(r.get("ID")),
        to_str(r.get("直播间")),
        to_str(r.get("品牌名称")),
        to_str(r.get("关联部类")),
        to_str(r.get("房间类型")),
        to_str(r.get("房间标签")),
        to_str(r.get("关联类型")),
        to_int(r.get("关联ID")),
        to_float(r.get("直播+讲解销售额")),
        to_int(r.get("累计观看人数")),
        to_int(r.get("累计观看次数")),
        to_float(r.get("小程序人均停留时长(秒)")),
        to_float(r.get("app人均停留时长(秒)")),
        pct_to_ratio(r.get("直播销售渗透率")),
        to_int(r.get("购买人数(总)")),
        to_int(r.get("购买人数(不含券)")),
        to_int(r.get("购买人数(含券)")),
        to_int(r.get("购买人数(营销券)")),
        pct_to_ratio(r.get("转化率(总)")),
        pct_to_ratio(r.get("转化率(不含券)")),
        pct_to_ratio(r.get("转化率(含券)")),
        pct_to_ratio(r.get("转化率(营销券)")),
        to_int(r.get("订单数(总)")),
        to_int(r.get("订单数(不含券)")),
        to_int(r.get("订单数(含券)")),
        to_int(r.get("订单数(营销券)")),
        to_float(r.get("直播时段销售额(总)")),
        to_float(r.get("直播时段销售额(不含券)")),
        to_float(r.get("直播时段销售额(含券)")),
        to_float(r.get("直播时段销售额(营销券)")),
        to_float(r.get("总引导销售(总)")),
        to_float(r.get("总引导销售(不含券)")),
        to_float(r.get("总引导销售(含券)")),
        to_float(r.get("总引导销售(营销券)")),
        to_int(r.get("开播时长(分钟)")),
        to_str(r.get("开播时段")),
        to_int(r.get("品牌客户数(总)")),
        to_int(r.get("品牌客户数(不含券)")),
        to_int(r.get("品牌客户数(含券)")),
        to_int(r.get("品牌新客数(总)")),
        to_int(r.get("品牌新客数(不含券)")),
        to_int(r.get("品牌新客数(含券)")),
        pct_to_ratio(r.get("新客占比(总)")),
        pct_to_ratio(r.get("新客占比(不含券)")),
        pct_to_ratio(r.get("新客占比(含券)")),
        to_float(r.get("人均消费")),
        to_int(r.get("新增直播关注数")),
        to_int(r.get("累计直播关注数(220705开始累计至直播日期当天)")),
        to_int(r.get("新增品牌入会人数")),
        to_int(r.get("累计品牌入会人数(230114开始累计至直播日期当天)")),
        source_file,
    )


INSERT_SQL = f"""
INSERT INTO `{TABLE_NAME}` (
  `biz_date`,`data_month`,`room_session_id`,`live_room_name`,`brand_name`,`related_department`,`room_type`,`room_tag`,
  `relation_type`,`relation_id`,`live_and_commentary_sales_amount`,`accumulated_viewers`,`accumulated_view_count`,
  `mini_program_avg_stay_seconds`,`app_avg_stay_seconds`,`live_sales_penetration_rate`,`purchasers_total`,
  `purchasers_excluding_coupons`,`purchasers_including_coupons`,`purchasers_marketing_coupon`,`conversion_rate_total`,
  `conversion_rate_excluding_coupons`,`conversion_rate_including_coupons`,`conversion_rate_marketing_coupon`,`orders_total`,
  `orders_excluding_coupons`,`orders_including_coupons`,`orders_marketing_coupon`,`live_period_sales_total`,
  `live_period_sales_excluding_coupons`,`live_period_sales_including_coupons`,`live_period_sales_marketing_coupon`,
  `guided_sales_total`,`guided_sales_excluding_coupons`,`guided_sales_including_coupons`,`guided_sales_marketing_coupon`,
  `live_duration_minutes`,`live_time_range`,`brand_customers_total`,`brand_customers_excluding_coupons`,
  `brand_customers_including_coupons`,`brand_new_customers_total`,`brand_new_customers_excluding_coupons`,
  `brand_new_customers_including_coupons`,`new_customer_ratio_total`,`new_customer_ratio_excluding_coupons`,
  `new_customer_ratio_including_coupons`,`avg_spend_per_customer`,`new_live_followers`,`accumulated_live_followers`,
  `new_brand_members`,`accumulated_brand_members`,`source_file`
) VALUES (
  %s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,
  %s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s
)
ON DUPLICATE KEY UPDATE
  `data_month`=VALUES(`data_month`),
  `brand_name`=VALUES(`brand_name`),
  `related_department`=VALUES(`related_department`),
  `room_type`=VALUES(`room_type`),
  `room_tag`=VALUES(`room_tag`),
  `relation_type`=VALUES(`relation_type`),
  `relation_id`=VALUES(`relation_id`),
  `live_and_commentary_sales_amount`=VALUES(`live_and_commentary_sales_amount`),
  `accumulated_viewers`=VALUES(`accumulated_viewers`),
  `accumulated_view_count`=VALUES(`accumulated_view_count`),
  `mini_program_avg_stay_seconds`=VALUES(`mini_program_avg_stay_seconds`),
  `app_avg_stay_seconds`=VALUES(`app_avg_stay_seconds`),
  `live_sales_penetration_rate`=VALUES(`live_sales_penetration_rate`),
  `purchasers_total`=VALUES(`purchasers_total`),
  `purchasers_excluding_coupons`=VALUES(`purchasers_excluding_coupons`),
  `purchasers_including_coupons`=VALUES(`purchasers_including_coupons`),
  `purchasers_marketing_coupon`=VALUES(`purchasers_marketing_coupon`),
  `conversion_rate_total`=VALUES(`conversion_rate_total`),
  `conversion_rate_excluding_coupons`=VALUES(`conversion_rate_excluding_coupons`),
  `conversion_rate_including_coupons`=VALUES(`conversion_rate_including_coupons`),
  `conversion_rate_marketing_coupon`=VALUES(`conversion_rate_marketing_coupon`),
  `orders_total`=VALUES(`orders_total`),
  `orders_excluding_coupons`=VALUES(`orders_excluding_coupons`),
  `orders_including_coupons`=VALUES(`orders_including_coupons`),
  `orders_marketing_coupon`=VALUES(`orders_marketing_coupon`),
  `live_period_sales_total`=VALUES(`live_period_sales_total`),
  `live_period_sales_excluding_coupons`=VALUES(`live_period_sales_excluding_coupons`),
  `live_period_sales_including_coupons`=VALUES(`live_period_sales_including_coupons`),
  `live_period_sales_marketing_coupon`=VALUES(`live_period_sales_marketing_coupon`),
  `guided_sales_total`=VALUES(`guided_sales_total`),
  `guided_sales_excluding_coupons`=VALUES(`guided_sales_excluding_coupons`),
  `guided_sales_including_coupons`=VALUES(`guided_sales_including_coupons`),
  `guided_sales_marketing_coupon`=VALUES(`guided_sales_marketing_coupon`),
  `live_duration_minutes`=VALUES(`live_duration_minutes`),
  `live_time_range`=VALUES(`live_time_range`),
  `brand_customers_total`=VALUES(`brand_customers_total`),
  `brand_customers_excluding_coupons`=VALUES(`brand_customers_excluding_coupons`),
  `brand_customers_including_coupons`=VALUES(`brand_customers_including_coupons`),
  `brand_new_customers_total`=VALUES(`brand_new_customers_total`),
  `brand_new_customers_excluding_coupons`=VALUES(`brand_new_customers_excluding_coupons`),
  `brand_new_customers_including_coupons`=VALUES(`brand_new_customers_including_coupons`),
  `new_customer_ratio_total`=VALUES(`new_customer_ratio_total`),
  `new_customer_ratio_excluding_coupons`=VALUES(`new_customer_ratio_excluding_coupons`),
  `new_customer_ratio_including_coupons`=VALUES(`new_customer_ratio_including_coupons`),
  `avg_spend_per_customer`=VALUES(`avg_spend_per_customer`),
  `new_live_followers`=VALUES(`new_live_followers`),
  `accumulated_live_followers`=VALUES(`accumulated_live_followers`),
  `new_brand_members`=VALUES(`new_brand_members`),
  `accumulated_brand_members`=VALUES(`accumulated_brand_members`),
  `source_file`=VALUES(`source_file`);
"""


def resolve_mysql_conn_cfg(args: argparse.Namespace, config: dict[str, Any]) -> dict[str, Any]:
    mysql_cfg = config.get("mysql", {}) if isinstance(config, dict) else {}
    host = mysql_cfg.get("host") or args.mysql_host
    port = int(mysql_cfg.get("port") or args.mysql_port)
    user = mysql_cfg.get("username") or mysql_cfg.get("user") or args.mysql_user
    password = mysql_cfg.get("password") or args.mysql_password
    database = mysql_cfg.get("database") or args.mysql_database
    return {
        "host": host,
        "port": port,
        "user": user,
        "password": password,
        "database": database,
        "charset": "utf8mb4",
        "autocommit": False,
    }


def execute(args: argparse.Namespace, config: dict[str, Any]) -> None:
    target_months = resolve_target_months(config)
    selected = pick_latest_files(target_months)
    if set(selected.keys()) != target_months:
        raise RuntimeError(f"live session files not complete, expected={sorted(target_months)}, got={sorted(selected.keys())}")

    all_rows: list[tuple[Any, ...]] = []
    all_dates: set[str] = set()
    for month in sorted(selected.keys()):
        f = selected[month].path
        df = pd.read_excel(f, sheet_name=0)
        for _, r in df.iterrows():
            biz_date = to_date(r.get("日期"))
            if not biz_date:
                continue
            all_dates.add(biz_date)
            all_rows.append(row_tuple(r, month, f.name))

    conn = pymysql.connect(**resolve_mysql_conn_cfg(args, config))
    try:
        with conn.cursor() as cur:
            cur.execute(create_table_sql())
            if all_dates:
                placeholders = ",".join(["%s"] * len(all_dates))
                cur.execute(
                    f"DELETE FROM `{TABLE_NAME}` WHERE `biz_date` IN ({placeholders})",
                    tuple(sorted(all_dates)),
                )
            if all_rows:
                cur.executemany(INSERT_SQL, all_rows)
            cur.execute(f"SELECT COUNT(*) FROM `{TABLE_NAME}`")
            total = cur.fetchone()[0]
        conn.commit()
    finally:
        conn.close()

    print("FILES_USED:")
    for month in sorted(selected):
        print(f"  LIVE[{month}]={selected[month].path}")
    print(f"INSERTED {TABLE_NAME}={len(all_rows)}")
    print(f"TABLE_COUNT {TABLE_NAME}={total}")


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Load VIS live session excel files into MySQL.")
    add_config_source_args(parser, default_config=DEFAULT_CONFIG_PATH)
    return parser.parse_args()


if __name__ == "__main__":
    args = parse_args()
    cfg = load_runtime_config(args, default_config=Path(args.config) if getattr(args, "config", "") else DEFAULT_CONFIG_PATH)
    execute(args, cfg)
