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_goods_analysis"


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 to_int(value: Any) -> int | 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 int(float(s))
    except Exception:
        return None


def 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(",", "").replace("%", "")
    if not s:
        return None
    try:
        return float(s)
    except Exception:
        return None


def to_str(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 create_sql() -> str:
    return f"""
CREATE TABLE IF NOT EXISTS `{TABLE_NAME}` (
  `id` BIGINT NOT NULL AUTO_INCREMENT,
  `data_month` CHAR(7) NOT NULL,
  `time_range` VARCHAR(32) NULL,
  `spu` VARCHAR(64) NULL,
  `reason_level_1` VARCHAR(128) NULL,
  `reason_level_2` VARCHAR(128) NULL,
  `reason_issue_category` VARCHAR(128) NULL,
  `reject_scene_level_1` VARCHAR(128) NULL,
  `reject_scene_level_2` VARCHAR(128) NULL,
  `product_name` VARCHAR(512) NULL,
  `category_level_1` VARCHAR(128) NULL,
  `category_level_2` VARCHAR(128) NULL,
  `category_level_3` VARCHAR(128) NULL,
  `article` VARCHAR(64) NULL,
  `brand_sn` VARCHAR(64) NULL,
  `brand_name` VARCHAR(128) NULL,
  `min_price` DECIMAL(18,2) NULL,
  `sales_qty` INT NULL,
  `reject_qty` INT NULL,
  `reject_amount` DECIMAL(18,2) NULL,
  `reject_rate_qty` DECIMAL(18,6) NULL,
  `reject_rate_amount` DECIMAL(18,6) NULL,
  `source_file` VARCHAR(255) NOT NULL,
  `created_at` TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  KEY `idx_data_month` (`data_month`),
  KEY `idx_article` (`article`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
"""


INSERT_SQL = f"""
INSERT INTO `{TABLE_NAME}` (
  `data_month`,`time_range`,`spu`,`reason_level_1`,`reason_level_2`,`reason_issue_category`,
  `reject_scene_level_1`,`reject_scene_level_2`,`product_name`,`category_level_1`,`category_level_2`,
  `category_level_3`,`article`,`brand_sn`,`brand_name`,`min_price`,`sales_qty`,`reject_qty`,
  `reject_amount`,`reject_rate_qty`,`reject_rate_amount`,`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
);
"""


def parse_rows(df: pd.DataFrame, data_month: str, source_file: str) -> list[tuple[Any, ...]]:
    rows: list[tuple[Any, ...]] = []
    for _, r in df.iterrows():
        rows.append(
            (
                data_month,
                to_str(r.get("时间")),
                to_str(r.get("SPU")),
                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_str(r.get("一级品类名称")),
                to_str(r.get("二级品类名称")),
                to_str(r.get("三级品类名称")),
                to_str(r.get("款号")),
                to_str(r.get("品牌SN")),
                to_str(r.get("品牌名称")),
                to_float(r.get("最低到手价")),
                to_int(r.get("销售量")),
                to_int(r.get("拒退件数")),
                to_float(r.get("拒退金额")),
                to_float(r.get("拒退率(数量)")),
                to_float(r.get("拒退率(金额)")),
                source_file,
            )
        )
    return rows


def parse_args() -> argparse.Namespace:
    p = argparse.ArgumentParser(description="Load Chinese-header rejection goods xlsx into MySQL.")
    p.add_argument("--config", default=r"D:\projects\skills\vip-report\config.yaml")
    p.add_argument("--file", required=True)
    p.add_argument("--report-year", type=int, default=0)
    p.add_argument("--report-month", default="")
    return p.parse_args()


def main() -> None:
    args = parse_args()
    config_path = Path(args.config).resolve()
    source_file = Path(args.file).resolve()
    cfg = yaml.safe_load(config_path.read_text(encoding="utf-8"))
    data_month = resolve_target_month(config_path, args.report_year, args.report_month)

    df = pd.read_excel(source_file, sheet_name=0)
    required = {"时间", "SPU", "一级拒退原因", "款号", "拒退件数", "拒退率(数量)"}
    missing = sorted(required - set(df.columns))
    if missing:
        raise RuntimeError(f"Missing required columns: {missing}")

    time_values = {str(x).strip() for x in df["时间"].dropna().head(20)}
    expected_month = data_month.replace("-", "")
    if not any(expected_month in x for x in time_values):
        raise RuntimeError(f"Unexpected time values for {data_month} file: {sorted(time_values)[:5]}")

    rows = parse_rows(df, data_month, source_file.name)
    mysql_cfg = cfg["mysql"]
    conn = pymysql.connect(
        host=mysql_cfg["host"],
        port=int(mysql_cfg.get("port", 3306)),
        user=mysql_cfg.get("username") or mysql_cfg.get("user"),
        password=mysql_cfg["password"],
        database=mysql_cfg["database"],
        charset="utf8mb4",
        autocommit=False,
    )
    try:
        with conn.cursor() as cur:
            cur.execute(create_sql())
            cur.execute(f"DELETE FROM `{TABLE_NAME}` WHERE `data_month`=%s", (data_month,))
            if rows:
                cur.executemany(INSERT_SQL, rows)
            cur.execute(f"SELECT COUNT(*) FROM `{TABLE_NAME}` WHERE `data_month`=%s", (data_month,))
            month_count = cur.fetchone()[0]
        conn.commit()
    finally:
        conn.close()

    print(f"LOAD_OK table={TABLE_NAME} month={data_month} rows={len(rows)} month_count={month_count} file={source_file}")


if __name__ == "__main__":
    main()
