from __future__ import annotations

import argparse
import re
import sys
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))

import pandas as pd
import pymysql
import yaml


TABLE_NAME = "cep_rejection_scene_analysis"
DEFAULT_DOWNLOAD_DIR = r"C:\vipdata"


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_month(config_path: Path, report_year_arg: int, report_month_arg: str) -> str:
    cfg = yaml.safe_load(config_path.read_text(encoding="utf-8"))
    report_cfg = cfg.get("report", {})
    year = int(report_year_arg or report_cfg.get("year", 2026))
    month = parse_month_number(report_month_arg or report_cfg.get("month_cn", "1"))
    return f"{year}{month:02d}"


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


def pick_latest_file(download_dir: Path, month_key: str) -> Path:
    candidates = sorted(download_dir.glob(f"{month_key}_*.xlsx"), key=lambda p: p.stat().st_mtime, reverse=True)
    if not candidates:
        raise RuntimeError(f"No downloaded xlsx found for month {month_key} under {download_dir}")
    return candidates[0]


def normalize_header_name(name: str, idx: int) -> str:
    raw = (name or "").strip()
    if not raw:
        return f"col_{idx:02d}"

    k = raw.lower().replace("（", "(").replace("）", ")")
    mapping = [
        ("日期", "biz_date"),
        ("场景", "scene_name"),
        ("一级", "category_level_1"),
        ("二级", "category_level_2"),
        ("三级", "category_level_3"),
        ("退货件数", "return_item_count"),
        ("拒收件数", "reject_item_count"),
        ("拒退件数", "reject_return_item_count"),
        ("退货率", "return_rate"),
        ("拒收率", "reject_rate"),
        ("拒退率", "reject_return_rate"),
        ("销售额", "sales_amount"),
        ("退款金额", "refund_amount"),
        ("订单数", "order_count"),
        ("商品件数", "item_count"),
        ("件数", "item_count"),
        ("sku", "sku"),
        ("spu", "spu"),
        ("品类", "category_name"),
        ("品牌", "brand_name"),
    ]
    for key, val in mapping:
        if key in raw or key in k:
            return val if val != "item_count" else f"item_count_{idx:02d}"

    ascii_text = re.sub(r"[^a-zA-Z0-9]+", "_", raw).strip("_").lower()
    if ascii_text:
        return ascii_text
    return f"col_{idx:02d}"


def ensure_unique(names: list[str]) -> list[str]:
    used: dict[str, int] = {}
    out: list[str] = []
    for n in names:
        base = n
        if base not in used:
            used[base] = 1
            out.append(base)
            continue
        used[base] += 1
        out.append(f"{base}_{used[base]}")
    return out


def infer_sql_type(values: pd.Series, col_name: str) -> str:
    if col_name == "biz_date":
        return "DATE"
    sample = values.dropna().astype(str).str.strip()
    if sample.empty:
        return "VARCHAR(255)"

    if sample.str.match(r"^\d{4}-\d{2}-\d{2}$").mean() > 0.8:
        return "DATE"
    if sample.str.match(r"^-?\d+$").mean() > 0.8:
        return "BIGINT"
    if sample.str.match(r"^-?\d+(\.\d+)?%?$").mean() > 0.8:
        return "DECIMAL(18,6)"
    max_len = int(sample.str.len().max())
    if max_len <= 255:
        return "VARCHAR(255)"
    if max_len <= 1024:
        return "VARCHAR(1024)"
    return "TEXT"


def clean_value(v: Any, sql_type: str) -> Any:
    if pd.isna(v):
        return None
    s = str(v).strip()
    if not s:
        return None
    if sql_type == "DATE":
        m = re.match(r"^(\d{4})[-/](\d{1,2})[-/](\d{1,2})$", s)
        if m:
            return f"{m.group(1)}-{int(m.group(2)):02d}-{int(m.group(3)):02d}"
        m2 = re.match(r"^(\d{4})(\d{2})(\d{2})$", re.sub(r"\D", "", s))
        if m2:
            return f"{m2.group(1)}-{m2.group(2)}-{m2.group(3)}"
        return None
    if sql_type.startswith("DECIMAL"):
        s2 = s.replace(",", "").replace("%", "")
        try:
            return float(s2)
        except Exception:
            return None
    if sql_type == "BIGINT":
        s2 = s.replace(",", "")
        try:
            return int(float(s2))
        except Exception:
            return None
    return s


def parse_args() -> argparse.Namespace:
    p = argparse.ArgumentParser(description="Create cep rejection-scene table and load latest downloaded xlsx.")
    p.add_argument("--config", default=r"D:\projects\skills\vip-report\config.yaml")
    p.add_argument("--download-dir", default=DEFAULT_DOWNLOAD_DIR)
    p.add_argument("--report-year", type=int, default=0)
    p.add_argument("--report-month", default="")
    p.add_argument("--file", default="")
    return p.parse_args()


def run_loader(config_path: Path, download_dir: Path, report_year: int, report_month: str, file_arg: str) -> None:
    cfg = yaml.safe_load(config_path.read_text(encoding="utf-8"))
    mysql_cfg = cfg.get("mysql", {})
    conn_cfg = {
        "host": mysql_cfg.get("host"),
        "port": int(mysql_cfg.get("port", 3306)),
        "user": mysql_cfg.get("username") or mysql_cfg.get("user"),
        "password": mysql_cfg.get("password"),
        "database": mysql_cfg.get("database"),
        "charset": "utf8mb4",
        "autocommit": False,
    }

    month_key = resolve_target_month(config_path, report_year, report_month)
    source_file = Path(file_arg) if file_arg else pick_latest_file(download_dir, month_key)
    if not source_file.exists():
        raise RuntimeError(f"source file not found: {source_file}")

    df = pd.read_excel(source_file, sheet_name=0)
    if df.empty:
        raise RuntimeError(f"source file has no rows: {source_file}")

    raw_headers = [str(c).strip() for c in df.columns]
    english_headers = ensure_unique([normalize_header_name(h, i + 1) for i, h in enumerate(raw_headers)])
    df.columns = english_headers

    if "biz_date" not in df.columns:
        df.insert(0, "biz_date", None)
    df.insert(1, "data_month", month_to_char7(month_key))
    df.insert(2, "source_file", source_file.name)

    sql_types: dict[str, str] = {}
    for c in df.columns:
        if c in {"data_month", "source_file"}:
            sql_types[c] = "VARCHAR(64)" if c == "data_month" else "VARCHAR(255)"
        else:
            sql_types[c] = infer_sql_type(df[c], c)

    cols_sql = ",\n  ".join([f"`{c}` {sql_types[c]} NULL" for c in df.columns])
    create_sql = f"""
CREATE TABLE IF NOT EXISTS `{TABLE_NAME}` (
  `id` BIGINT NOT NULL AUTO_INCREMENT,
  {cols_sql},
  `created_at` TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  KEY `idx_data_month` (`data_month`),
  KEY `idx_biz_date` (`biz_date`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
"""

    insert_cols = [f"`{c}`" for c in df.columns]
    placeholders = ",".join(["%s"] * len(df.columns))
    insert_sql = f"INSERT INTO `{TABLE_NAME}` ({','.join(insert_cols)}) VALUES ({placeholders})"

    rows: list[tuple[Any, ...]] = []
    for _, r in df.iterrows():
        row_vals: list[Any] = []
        for c in df.columns:
            row_vals.append(clean_value(r[c], sql_types[c]))
        rows.append(tuple(row_vals))

    conn = pymysql.connect(**conn_cfg)
    try:
        with conn.cursor() as cur:
            cur.execute(create_sql)
            cur.execute(f"SHOW COLUMNS FROM `{TABLE_NAME}`")
            existing_cols = {x[0] for x in cur.fetchall()}
            for c in df.columns:
                if c in existing_cols:
                    continue
                cur.execute(f"ALTER TABLE `{TABLE_NAME}` ADD COLUMN `{c}` {sql_types[c]} NULL")

            cur.execute(f"DELETE FROM `{TABLE_NAME}` WHERE `data_month`=%s", (month_to_char7(month_key),))
            if rows:
                cur.executemany(insert_sql, rows)
            cur.execute(f"SELECT COUNT(*) FROM `{TABLE_NAME}` WHERE `data_month`=%s", (month_to_char7(month_key),))
            month_count = cur.fetchone()[0]
        conn.commit()
    finally:
        conn.close()

    print(
        f"LOAD_OK table={TABLE_NAME} month={month_to_char7(month_key)} rows={len(rows)} "
        f"month_count={month_count} file={source_file}"
    )
    print("COLUMN_MAP:")
    for zh, en in zip(raw_headers, english_headers):
        print(f"  {zh} -> {en}")


def main() -> None:
    args = parse_args()
    run_loader(
        config_path=Path(args.config).resolve(),
        download_dir=Path(args.download_dir).resolve(),
        report_year=args.report_year,
        report_month=args.report_month,
        file_arg=args.file,
    )


if __name__ == "__main__":
    main()

