from __future__ import annotations

import argparse
import re
import sys
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))
if str(SCRIPT_DIR) not in sys.path:
    sys.path.insert(0, str(SCRIPT_DIR))

import openpyxl
import pymysql
import yaml
from openpyxl.worksheet import _reader as ws_reader

from runtime_config import add_config_source_args, load_runtime_config


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


def patch_openpyxl_column_dimension_error() -> None:
    """Ignore malformed column-dimension metadata in some exported xlsx files."""
    orig = ws_reader.WorkSheetParser.parse_column_dimensions

    def patched(self: Any, col: Any) -> Any:
        try:
            return orig(self, col)
        except ValueError as e:
            if "Invalid column index" in str(e):
                return None
            raise

    ws_reader.WorkSheetParser.parse_column_dimensions = patched


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(cfg: dict[str, Any], report_year_arg: int, report_month_arg: str) -> str:
    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 pick_latest_file(download_dir: Path, month_key: str) -> Path:
    patterns = [
        f"{month_key}_goods_effect_*MID*.xlsx",
        f"{month_key}_goods_effect_*mid*.xlsx",
        f"{month_key}_goods_effect_*.xlsx",
    ]
    files: list[Path] = []
    for pattern in patterns:
        files = sorted(download_dir.glob(pattern), key=lambda p: p.stat().st_mtime, reverse=True)
        if files:
            break
    if not files:
        raise RuntimeError(f"No goods-effect file found under: {download_dir}")
    return files[0]


def to_str(value: Any) -> Optional[str]:
    if value is None:
        return None
    s = str(value).strip()
    return s or None


def to_int(value: Any) -> Optional[int]:
    if value is None or value == "":
        return None
    try:
        if isinstance(value, str):
            value = value.replace(",", "").strip()
        return int(float(value))
    except Exception:
        return None


def to_decimal(value: Any) -> Optional[float]:
    if value is None or value == "":
        return None
    try:
        if isinstance(value, str):
            value = value.replace(",", "").strip()
        return float(value)
    except Exception:
        return None


def to_date(value: Any) -> Optional[str]:
    if value is None or value == "":
        return None
    if isinstance(value, datetime):
        return value.strftime("%Y-%m-%d")
    s = str(value).strip()
    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}"
    return None


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,
  `vendor_code` VARCHAR(32) NULL,
  `live_type` VARCHAR(64) NULL,
  `room_id` BIGINT NULL,
  `room_name` VARCHAR(255) NULL,
  `room_type` VARCHAR(128) NULL,
  `room_tags` VARCHAR(255) NULL,
  `room_related_brand` VARCHAR(128) NULL,
  `related_sn` VARCHAR(64) NULL,
  `related_brand` VARCHAR(128) NULL,
  `related_brand_en` VARCHAR(128) NULL,
  `product_title` VARCHAR(512) NULL,
  `product_spu_id` VARCHAR(64) NULL,
  `mid` VARCHAR(64) NULL,
  `platform_spu_id` VARCHAR(64) NULL,
  `department` VARCHAR(128) NULL,
  `category_level_1` VARCHAR(128) NULL,
  `category_level_2` VARCHAR(128) NULL,
  `category_level_3` VARCHAR(128) NULL,
  `buyers_total` INT NULL,
  `buyers_excluding_coupon` INT NULL,
  `buyers_including_coupon` INT NULL,
  `buyers_marketing_coupon` INT NULL,
  `order_count_total` INT NULL,
  `order_count_excluding_coupon` INT NULL,
  `order_count_including_coupon` INT NULL,
  `order_count_marketing_coupon` INT NULL,
  `item_count_total` INT NULL,
  `item_count_excluding_coupon` INT NULL,
  `item_count_including_coupon` INT NULL,
  `item_count_marketing_coupon` INT NULL,
  `sales_total` DECIMAL(18,2) NULL,
  `sales_excluding_coupon` DECIMAL(18,2) NULL,
  `sales_including_coupon` DECIMAL(18,2) NULL,
  `sales_marketing_coupon` DECIMAL(18,2) NULL,
  `brand_new_customer_count` INT NULL,
  `source_file` VARCHAR(255) NOT NULL,
  `created_at` TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  KEY `idx_month` (`data_month`),
  KEY `idx_date` (`biz_date`),
  KEY `idx_room_mid` (`room_id`, `mid`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
"""


INSERT_SQL = f"""
INSERT INTO `{TABLE_NAME}` (
  `biz_date`,`data_month`,`vendor_code`,`live_type`,`room_id`,`room_name`,`room_type`,`room_tags`,
  `room_related_brand`,`related_sn`,`related_brand`,`related_brand_en`,`product_title`,`product_spu_id`,`mid`,
  `platform_spu_id`,`department`,`category_level_1`,`category_level_2`,`category_level_3`,
  `buyers_total`,`buyers_excluding_coupon`,`buyers_including_coupon`,`buyers_marketing_coupon`,
  `order_count_total`,`order_count_excluding_coupon`,`order_count_including_coupon`,`order_count_marketing_coupon`,
  `item_count_total`,`item_count_excluding_coupon`,`item_count_including_coupon`,`item_count_marketing_coupon`,
  `sales_total`,`sales_excluding_coupon`,`sales_including_coupon`,`sales_marketing_coupon`,
  `brand_new_customer_count`,`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
);
"""


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


def build_rows(file_path: Path, month_key: str) -> tuple[list[tuple[Any, ...]], set[str]]:
    patch_openpyxl_column_dimension_error()
    wb = openpyxl.load_workbook(file_path, read_only=True, data_only=True)
    ws = wb[wb.sheetnames[0]]
    iterator = ws.iter_rows(values_only=True)
    header_row = next(iterator, None)
    if not header_row:
        return [], set()
    headers = [str(h).strip() if h is not None else "" for h in header_row]
    idx = {name: i for i, name in enumerate(headers)}

    def cell(row: tuple[Any, ...], name: str) -> Any:
        i = idx.get(name)
        if i is None or i >= len(row):
            return None
        return row[i]

    rows: list[tuple[Any, ...]] = []
    dates: set[str] = set()
    data_month = month_to_char7(month_key)

    for row in iterator:
        biz_date = to_date(cell(row, "日期"))
        if not biz_date:
            continue
        dates.add(biz_date)
        rows.append(
            (
                biz_date,
                data_month,
                to_str(cell(row, "供应商编码")),
                to_str(cell(row, "直播类型")),
                to_int(cell(row, "房间ID")),
                to_str(cell(row, "房间名称")),
                to_str(cell(row, "房间类型")),
                to_str(cell(row, "房间标签")),
                to_str(cell(row, "房间关联品牌")),
                to_str(cell(row, "商品关联SN")),
                to_str(cell(row, "商品关联品牌")),
                to_str(cell(row, "商品关联品牌En")),
                to_str(cell(row, "商品标题")),
                to_str(cell(row, "商品spuID")),
                to_str(cell(row, "MID")),
                to_str(cell(row, "p-spuID")),
                to_str(cell(row, "部类")),
                to_str(cell(row, "一级分类")),
                to_str(cell(row, "二级分类")),
                to_str(cell(row, "三级分类")),
                to_int(cell(row, "购买人数(总)")),
                to_int(cell(row, "购买人数(不含券)")),
                to_int(cell(row, "购买人数(含券)")),
                to_int(cell(row, "购买人数(营销券)")),
                to_int(cell(row, "购买订单数(总)")),
                to_int(cell(row, "购买订单数(不含券)")),
                to_int(cell(row, "购买订单数(含券)")),
                to_int(cell(row, "购买订单数(营销券)")),
                to_int(cell(row, "购买件数(总)")),
                to_int(cell(row, "购买件数(不含券)")),
                to_int(cell(row, "购买件数(含券)")),
                to_int(cell(row, "购买件数(营销券)")),
                to_decimal(cell(row, "销售额(总)")),
                to_decimal(cell(row, "销售额(不含券)")),
                to_decimal(cell(row, "销售额(含券)")),
                to_decimal(cell(row, "销售额(营销券)")),
                to_int(cell(row, "品牌新客数")),
                file_path.name,
            )
        )
    return rows, dates


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Load VIS goods-effect xlsx into MySQL.")
    add_config_source_args(parser, default_config=Path(r"D:\projects\skills\vip-report\config.yaml"))
    parser.add_argument("--download-dir", default=DEFAULT_DOWNLOAD_DIR)
    parser.add_argument("--report-year", type=int, default=0)
    parser.add_argument("--report-month", default="")
    parser.add_argument("--file", default="")
    return parser.parse_args()


def resolve_mysql_conn_cfg(args: argparse.Namespace, cfg: dict[str, Any]) -> dict[str, Any]:
    mysql_cfg = cfg.get("mysql", {})
    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
    conn_cfg = {
        "host": host,
        "port": port,
        "user": user,
        "password": password,
        "database": database,
        "charset": "utf8mb4",
        "autocommit": False,
    }
    return conn_cfg


def run_loader(args: argparse.Namespace, cfg: dict[str, Any], download_dir: Path, report_year: int, report_month: str, file_arg: str) -> None:
    conn_cfg = resolve_mysql_conn_cfg(args, cfg)
    target_month = resolve_target_month(cfg, report_year, report_month)
    source_file = Path(file_arg) if file_arg else pick_latest_file(download_dir, target_month)
    if not source_file.exists():
        raise RuntimeError(f"source file not found: {source_file}")
    if "MID" not in source_file.name.upper():
        fallback = pick_latest_file(download_dir, target_month)
        if "MID" in fallback.name.upper():
            source_file = fallback
        else:
            raise RuntimeError(f"selected file is not MID detail file: {source_file.name}")

    rows, dates = build_rows(source_file, target_month)
    if not rows:
        raise RuntimeError(f"no valid rows parsed from file: {source_file}")

    conn = pymysql.connect(**conn_cfg)
    try:
        with conn.cursor() as cur:
            cur.execute(create_table_sql())
            cur.execute(
                "SELECT COUNT(*) FROM information_schema.statistics "
                "WHERE table_schema=%s AND table_name=%s AND index_name='uk_biz_room_mid'",
                (conn_cfg["database"], TABLE_NAME),
            )
            if cur.fetchone()[0]:
                cur.execute(f"ALTER TABLE `{TABLE_NAME}` DROP INDEX `uk_biz_room_mid`")

            placeholders = ",".join(["%s"] * len(dates))
            cur.execute(
                f"DELETE FROM `{TABLE_NAME}` WHERE `biz_date` IN ({placeholders}) AND `data_month`=%s",
                tuple(sorted(dates)) + (month_to_char7(target_month),),
            )
            cur.executemany(INSERT_SQL, rows)
            cur.execute(
                f"SELECT COUNT(*) FROM `{TABLE_NAME}` WHERE `data_month`=%s",
                (month_to_char7(target_month),),
            )
            month_count = cur.fetchone()[0]
        conn.commit()
    finally:
        conn.close()

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


def main() -> None:
    args = parse_args()
    config_path = Path(args.config or r"D:\projects\skills\vip-report\config.yaml")
    cfg = load_runtime_config(args, default_config=config_path)
    run_loader(
        args=args,
        cfg=cfg,
        download_dir=Path(args.download_dir),
        report_year=args.report_year,
        report_month=args.report_month,
        file_arg=args.file,
    )


if __name__ == "__main__":
    main()
