from __future__ import annotations

import argparse
import calendar
import json
import re
import sys
from dataclasses import dataclass
from decimal import Decimal, ROUND_HALF_UP
from pathlib import Path
from typing import Any

import pymysql

try:
    import openpyxl
    from openpyxl.worksheet import _reader as ws_reader
except Exception:  # optional offline Excel fallback
    openpyxl = None
    ws_reader = None

SCRIPT_DIR = Path(__file__).resolve().parent
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


TABLE_SESSION = "cep_live_session_data"
TABLE_GOODS = "cep_live_goods_effect"

MONTH_ABBR = {
    1: "Jan",
    2: "Feb",
    3: "Mar",
    4: "Apr",
    5: "May",
    6: "Jun",
    7: "Jul",
    8: "Aug",
    9: "Sep",
    10: "Oct",
    11: "Nov",
    12: "Dec",
}

CATEGORY_ORDER = ["箱包", "鞋", "配饰"]

TOP5_NOTE_RULES = [
    "ESS独家5折",
    "ESS独家7折",
    "ESS独家",
    "正价RP",
    "礼赠款",
]


@dataclass(frozen=True)
class MonthKey:
    year: int
    month: int

    @property
    def ym(self) -> str:
        return f"{self.year}-{self.month:02d}"

    @property
    def label(self) -> str:
        return MONTH_ABBR[self.month]


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 previous_month(m: MonthKey) -> MonthKey:
    if m.month == 1:
        return MonthKey(m.year - 1, 12)
    return MonthKey(m.year, m.month - 1)


def same_month_last_year(m: MonthKey) -> MonthKey:
    return MonthKey(m.year - 1, m.month)


def to_decimal(value: Any) -> Decimal:
    if value is None:
        return Decimal("0")
    text = str(value).strip().replace(",", "")
    if not text:
        return Decimal("0")
    try:
        return Decimal(text)
    except Exception:
        return Decimal("0")


def fmt_int(v: Decimal) -> str:
    return f"{int(v.quantize(Decimal('1'), rounding=ROUND_HALF_UP)):,}"


def fmt_amount(v: Decimal) -> str:
    return f"{v.quantize(Decimal('0.01'), rounding=ROUND_HALF_UP):,}"


def fmt_pct(v: Decimal, digits: int = 0, signed: bool = False) -> str:
    q = Decimal("1") if digits == 0 else Decimal("1." + "0" * digits)
    pct = (v * Decimal("100")).quantize(q, rounding=ROUND_HALF_UP)
    sign = ""
    if signed and pct > 0:
        sign = "+"
    return f"{sign}{pct}%"


def ratio(curr: Decimal, prev: Decimal) -> Decimal:
    if prev == 0:
        return Decimal("0")
    return (curr - prev) / prev


def share(part: Decimal, total: Decimal) -> Decimal:
    if total == 0:
        return Decimal("0")
    return part / total


def extract_article(title: str) -> str:
    m = re.search(r"(CK\d-[0-9]{8}(?:-[0-9])?)", title.upper())
    return m.group(1) if m else ""


def normalize_note(text: str) -> str:
    for key in TOP5_NOTE_RULES:
        if key in text:
            return key
    return ""


def choose_sales_axis(values: list[Decimal]) -> tuple[int, int]:
    max_v = max([int(v) for v in values] or [0])
    if max_v <= 30000:
        step = 5000
    elif max_v <= 90000:
        step = 10000
    else:
        step = 20000
    y_max = ((max_v + step - 1) // step) * step
    if y_max <= 0:
        y_max = step * 6
    return y_max, step


def choose_conv_axis(values: list[Decimal]) -> tuple[float, float]:
    max_v = float(max(values or [Decimal("0")]))
    if max_v <= 0.08:
        y_max = 0.10
    elif max_v <= 0.12:
        y_max = 0.15
    elif max_v <= 0.18:
        y_max = 0.20
    else:
        y_max = min(0.40, (int(max_v * 100 / 5) + 1) * 0.05)
    return y_max, 0.02


def fetch_live_session_agg(conn: pymysql.connections.Connection, month_key: MonthKey) -> dict[str, Decimal]:
    sql = f"""
SELECT
  SUM(IFNULL(accumulated_viewers, 0)) AS accumulated_viewers,
  SUM(IFNULL(purchasers_total, 0)) AS purchasers_total,
  SUM(IFNULL(live_period_sales_total, 0)) AS live_period_sales_total,
  SUM(IFNULL(guided_sales_total, 0)) AS guided_sales_total,
  SUM(IFNULL(brand_customers_total, 0)) AS brand_customers_total,
  SUM(IFNULL(brand_new_customers_total, 0)) AS brand_new_customers_total,
  SUM(IFNULL(new_brand_members, 0)) AS new_brand_members,
  SUM(IFNULL(live_sales_penetration_rate, 0) * IFNULL(accumulated_viewers, 0)) AS weighted_penetration,
  SUM(IFNULL(accumulated_viewers, 0)) AS weighted_base
FROM `{TABLE_SESSION}`
WHERE data_month = %s
"""
    with conn.cursor() as cur:
        cur.execute(sql, (month_key.ym,))
        row = cur.fetchone() or {}

    viewers = to_decimal(row.get("accumulated_viewers"))
    purchasers = to_decimal(row.get("purchasers_total"))
    brand_customers = to_decimal(row.get("brand_customers_total"))
    brand_new = to_decimal(row.get("brand_new_customers_total"))
    weighted_pen = to_decimal(row.get("weighted_penetration"))
    weighted_base = to_decimal(row.get("weighted_base"))

    live_pen = (weighted_pen / weighted_base) if weighted_base > 0 else Decimal("0")
    conversion = (purchasers / viewers) if viewers > 0 else Decimal("0")
    new_ratio = (brand_new / brand_customers) if brand_customers > 0 else Decimal("0")

    return {
        "accumulated_viewers": viewers,
        "live_penetration": live_pen,
        "purchasers_total": purchasers,
        "conversion": conversion,
        "live_period_sales_total": to_decimal(row.get("live_period_sales_total")),
        "guided_sales_total": to_decimal(row.get("guided_sales_total")),
        "brand_customers_total": brand_customers,
        "brand_new_customers_total": brand_new,
        "new_customer_ratio": new_ratio,
        "new_brand_members": to_decimal(row.get("new_brand_members")),
    }


def build_slide14_summary_rows(prev_key: MonthKey, prev: dict[str, Decimal], curr_key: MonthKey, curr: dict[str, Decimal]) -> tuple[list[str], list[list[str]]]:
    headers = [
        "Month",
        "Accum. Viewers",
        "Live Penetration",
        "Purchasers",
        "Conversion",
        "Live Sales",
        "Guided Sales",
        "Brand Customers",
        "Brand New Customers",
        "New Customer Ratio",
        "New Brand Members",
    ]

    prev_row = [
        prev_key.label,
        fmt_int(prev["accumulated_viewers"]),
        fmt_pct(prev["live_penetration"], 2),
        fmt_int(prev["purchasers_total"]),
        fmt_pct(prev["conversion"], 2),
        fmt_int(prev["live_period_sales_total"]),
        fmt_int(prev["guided_sales_total"]),
        fmt_int(prev["brand_customers_total"]),
        fmt_int(prev["brand_new_customers_total"]),
        fmt_pct(prev["new_customer_ratio"], 0),
        fmt_int(prev["new_brand_members"]),
    ]
    curr_row = [
        curr_key.label,
        fmt_int(curr["accumulated_viewers"]),
        fmt_pct(curr["live_penetration"], 2),
        fmt_int(curr["purchasers_total"]),
        fmt_pct(curr["conversion"], 2),
        fmt_int(curr["live_period_sales_total"]),
        fmt_int(curr["guided_sales_total"]),
        fmt_int(curr["brand_customers_total"]),
        fmt_int(curr["brand_new_customers_total"]),
        fmt_pct(curr["new_customer_ratio"], 0),
        fmt_int(curr["new_brand_members"]),
    ]
    lfl_row = [
        "LFL",
        fmt_pct(ratio(curr["accumulated_viewers"], prev["accumulated_viewers"]), 0, signed=True),
        fmt_pct(ratio(curr["live_penetration"], prev["live_penetration"]), 0, signed=True),
        fmt_pct(ratio(curr["purchasers_total"], prev["purchasers_total"]), 0, signed=True),
        fmt_pct(ratio(curr["conversion"], prev["conversion"]), 0, signed=True),
        fmt_pct(ratio(curr["live_period_sales_total"], prev["live_period_sales_total"]), 0, signed=True),
        fmt_pct(ratio(curr["guided_sales_total"], prev["guided_sales_total"]), 0, signed=True),
        fmt_pct(ratio(curr["brand_customers_total"], prev["brand_customers_total"]), 0, signed=True),
        fmt_pct(ratio(curr["brand_new_customers_total"], prev["brand_new_customers_total"]), 0, signed=True),
        fmt_pct(ratio(curr["new_customer_ratio"], prev["new_customer_ratio"]), 0, signed=True),
        fmt_pct(ratio(curr["new_brand_members"], prev["new_brand_members"]), 0, signed=True),
    ]
    return headers, [prev_row, curr_row, lfl_row]


def fetch_live_session_daily(conn: pymysql.connections.Connection, month_key: MonthKey) -> tuple[list[int], list[Decimal], list[Decimal]]:
    sql = f"""
SELECT
  biz_date,
  SUM(IFNULL(live_period_sales_total, 0)) AS sales_total,
  SUM(IFNULL(live_sales_penetration_rate, 0) * IFNULL(accumulated_viewers, 0)) AS weighted_penetration,
  SUM(IFNULL(accumulated_viewers, 0)) AS weighted_base
FROM `{TABLE_SESSION}`
WHERE data_month = %s
GROUP BY biz_date
ORDER BY biz_date
"""
    day_map: dict[int, tuple[Decimal, Decimal]] = {}
    with conn.cursor() as cur:
        cur.execute(sql, (month_key.ym,))
        rows = cur.fetchall() or []
    for row in rows:
        biz_date = str(row.get("biz_date") or "")
        if len(biz_date) >= 10 and biz_date[8:10].isdigit():
            day = int(biz_date[8:10])
        else:
            continue
        sales = to_decimal(row.get("sales_total"))
        weighted_pen = to_decimal(row.get("weighted_penetration"))
        weighted_base = to_decimal(row.get("weighted_base"))
        penetration = weighted_pen / weighted_base if weighted_base > 0 else Decimal("0")
        day_map[day] = (sales, penetration)

    days_in_month = calendar.monthrange(month_key.year, month_key.month)[1]
    day_list: list[int] = []
    sales_values: list[Decimal] = []
    conv_values: list[Decimal] = []
    for day in range(1, days_in_month + 1):
        s, c = day_map.get(day, (Decimal("0"), Decimal("0")))
        day_list.append(day)
        sales_values.append(s)
        conv_values.append(c)
    return day_list, sales_values, conv_values


def fetch_goods_category_rows(conn: pymysql.connections.Connection, month_key: MonthKey) -> dict[str, dict[str, Decimal]]:
    sql = f"""
SELECT
  CASE
    WHEN category_level_1 IN ('配饰', '珠宝首饰') THEN '配饰'
    ELSE category_level_1
  END AS cat,
  COUNT(DISTINCT product_spu_id) AS article_cnt,
  SUM(IFNULL(item_count_total, 0)) AS qty,
  SUM(IFNULL(sales_total, 0)) AS sales,
  SUM(IFNULL(brand_new_customer_count, 0)) AS brand_new_customers
FROM `{TABLE_GOODS}`
WHERE data_month = %s
GROUP BY cat
"""
    out: dict[str, dict[str, Decimal]] = {}
    with conn.cursor() as cur:
        cur.execute(sql, (month_key.ym,))
        rows = cur.fetchall() or []

    for row in rows:
        cat = str(row.get("cat") or "").strip() or "未分类"
        out[cat] = {
            "article_cnt": to_decimal(row.get("article_cnt")),
            "qty": to_decimal(row.get("qty")),
            "sales": to_decimal(row.get("sales")),
            "brand_new_customers": to_decimal(row.get("brand_new_customers")),
        }
    return out


def build_slide15_summary_rows(
    curr_rows: dict[str, dict[str, Decimal]],
    prev_rows: dict[str, dict[str, Decimal]],
) -> tuple[list[str], list[list[str]]]:
    for c in CATEGORY_ORDER:
        curr_rows.setdefault(c, {"article_cnt": Decimal("0"), "qty": Decimal("0"), "sales": Decimal("0"), "brand_new_customers": Decimal("0")})
        prev_rows.setdefault(c, {"article_cnt": Decimal("0"), "qty": Decimal("0"), "sales": Decimal("0"), "brand_new_customers": Decimal("0")})

    ttl_curr = {
        "article_cnt": sum(curr_rows[c]["article_cnt"] for c in CATEGORY_ORDER),
        "qty": sum(curr_rows[c]["qty"] for c in CATEGORY_ORDER),
        "sales": sum(curr_rows[c]["sales"] for c in CATEGORY_ORDER),
        "brand_new_customers": sum(curr_rows[c]["brand_new_customers"] for c in CATEGORY_ORDER),
    }
    ttl_prev = {
        "article_cnt": sum(prev_rows[c]["article_cnt"] for c in CATEGORY_ORDER),
        "qty": sum(prev_rows[c]["qty"] for c in CATEGORY_ORDER),
        "sales": sum(prev_rows[c]["sales"] for c in CATEGORY_ORDER),
        "brand_new_customers": sum(prev_rows[c]["brand_new_customers"] for c in CATEGORY_ORDER),
    }

    headers = [
        "Cat",
        "Article",
        "Article%",
        "Live QTY",
        "QTY%",
        "QTY LFL",
        "Sales",
        "Sales%",
        "Sales LFL",
        "New Customers",
        "New Cust %",
        "New Cust LFL",
        "ASP",
        "ASP LFL",
    ]

    def row_values(cat: str, curr: dict[str, Decimal], prev: dict[str, Decimal]) -> list[str]:
        asp_curr = curr["sales"] / curr["qty"] if curr["qty"] > 0 else Decimal("0")
        asp_prev = prev["sales"] / prev["qty"] if prev["qty"] > 0 else Decimal("0")
        return [
            cat,
            f"{int(curr['article_cnt'].quantize(Decimal('1'), rounding=ROUND_HALF_UP))}",
            fmt_pct(share(curr["article_cnt"], ttl_curr["article_cnt"]), 0),
            f"{int(curr['qty'].quantize(Decimal('1'), rounding=ROUND_HALF_UP))}",
            fmt_pct(share(curr["qty"], ttl_curr["qty"]), 0),
            fmt_pct(ratio(curr["qty"], prev["qty"]), 0, signed=True),
            fmt_amount(curr["sales"]),
            fmt_pct(share(curr["sales"], ttl_curr["sales"]), 0),
            fmt_pct(ratio(curr["sales"], prev["sales"]), 0, signed=True),
            f"{int(curr['brand_new_customers'].quantize(Decimal('1'), rounding=ROUND_HALF_UP))}",
            fmt_pct(share(curr["brand_new_customers"], ttl_curr["brand_new_customers"]), 0),
            fmt_pct(ratio(curr["brand_new_customers"], prev["brand_new_customers"]), 0, signed=True),
            f"{int(asp_curr.quantize(Decimal('1'), rounding=ROUND_HALF_UP))}",
            fmt_pct(ratio(asp_curr, asp_prev), 0, signed=True),
        ]

    rows: list[list[str]] = []
    for c in CATEGORY_ORDER:
        rows.append(row_values(c, curr_rows[c], prev_rows[c]))
    rows.append(row_values("TTL", ttl_curr, ttl_prev))
    return headers, rows


def fetch_goods_mix_rows(conn: pymysql.connections.Connection, month_key: MonthKey) -> tuple[list[tuple[str, Decimal]], list[tuple[str, Decimal]]]:
    sql = f"""
SELECT
  CASE
    WHEN category_level_1 IN ('配饰', '珠宝首饰') THEN '配饰'
    ELSE category_level_1
  END AS level1,
  COALESCE(NULLIF(TRIM(category_level_3), ''), '未分类') AS level3,
  SUM(IFNULL(item_count_total, 0)) AS qty
FROM `{TABLE_GOODS}`
WHERE data_month = %s
GROUP BY level1, level3
HAVING SUM(IFNULL(item_count_total, 0)) > 0
"""
    level1_totals: dict[str, Decimal] = {}
    level3_totals: dict[str, Decimal] = {}
    with conn.cursor() as cur:
        cur.execute(sql, (month_key.ym,))
        rows = cur.fetchall() or []

    for row in rows:
        l1 = str(row.get("level1") or "未分类")
        l3 = str(row.get("level3") or "未分类")
        qty = to_decimal(row.get("qty"))
        level1_totals[l1] = level1_totals.get(l1, Decimal("0")) + qty
        level3_totals[l3] = level3_totals.get(l3, Decimal("0")) + qty

    level1_sorted = sorted(level1_totals.items(), key=lambda x: x[1], reverse=True)
    level3_sorted = sorted(level3_totals.items(), key=lambda x: x[1], reverse=True)
    return level1_sorted, level3_sorted



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


def patch_openpyxl_column_dimension_error() -> None:
    if ws_reader is None:
        return
    orig = ws_reader.WorkSheetParser.parse_column_dimensions
    if getattr(orig, "_vip_report_patched", False):
        return

    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

    patched._vip_report_patched = True  # type: ignore[attr-defined]
    ws_reader.WorkSheetParser.parse_column_dimensions = patched


def excel_cell(row: dict[str, Any], name: str) -> Any:
    return row.get(name)


LIVE_COL_DATE = "\u65e5\u671f"
LIVE_COL_VIEWERS = "\u7d2f\u8ba1\u89c2\u770b\u4eba\u6570"
LIVE_COL_PENETRATION = "\u76f4\u64ad\u9500\u552e\u6e17\u900f\u7387"
LIVE_COL_PURCHASERS = "\u8d2d\u4e70\u4eba\u6570(\u603b)"
LIVE_COL_LIVE_SALES = "\u76f4\u64ad\u65f6\u6bb5\u9500\u552e\u989d(\u603b)"
LIVE_COL_GUIDED_SALES = "\u603b\u5f15\u5bfc\u9500\u552e(\u603b)"
LIVE_COL_BRAND_CUSTOMERS = "\u54c1\u724c\u5ba2\u6237\u6570(\u603b)"
LIVE_COL_BRAND_NEW = "\u54c1\u724c\u65b0\u5ba2\u6570(\u603b)"
LIVE_COL_NEW_MEMBERS = "\u65b0\u589e\u54c1\u724c\u5165\u4f1a\u4eba\u6570"

GOODS_COL_LEVEL1 = "\u4e00\u7ea7\u5206\u7c7b"
GOODS_COL_LEVEL3 = "\u4e09\u7ea7\u5206\u7c7b"
GOODS_COL_SPU = "\u5546\u54c1spuID"
GOODS_COL_TITLE = "\u5546\u54c1\u6807\u9898"
GOODS_COL_QTY = "\u8d2d\u4e70\u4ef6\u6570(\u603b)"
GOODS_COL_SALES = "\u9500\u552e\u989d(\u603b)"
GOODS_COL_BRAND_NEW = "\u54c1\u724c\u65b0\u5ba2\u6570"

CATEGORY_BAGS = "\u7bb1\u5305"
CATEGORY_SHOES = "\u978b"
CATEGORY_ACCESSORIES = "\u914d\u9970"
CATEGORY_JEWELRY = "\u73e0\u5b9d\u9996\u9970"
CATEGORY_UNKNOWN = "\u672a\u5206\u7c7b"


def excel_pct_to_ratio(value: Any) -> Decimal:
    if value is None:
        return Decimal("0")
    text = str(value).strip().replace(",", "")
    if not text:
        return Decimal("0")
    if text.endswith("%"):
        text = text[:-1]
        return to_decimal(text) / Decimal("100")
    number = to_decimal(text)
    return number / Decimal("100") if number > 1 else number


def excel_date_to_day(value: Any) -> int | None:
    if value is None:
        return None
    if hasattr(value, "day"):
        return int(value.day)
    text = str(value).strip()
    m = re.search(r"(\d{4})[-/](\d{1,2})[-/](\d{1,2})", text)
    if m:
        return int(m.group(3))
    return None


def read_excel_rows(path: Path) -> list[dict[str, Any]]:
    if openpyxl is None:
        raise RuntimeError("openpyxl is required for offline Excel fallback")
    patch_openpyxl_column_dimension_error()
    wb = openpyxl.load_workbook(path, read_only=True, data_only=True)
    ws = wb[wb.sheetnames[0]]
    iterator = ws.iter_rows(values_only=True)
    headers = next(iterator, None)
    if not headers:
        return []
    names = [str(h).strip() if h is not None else "" for h in headers]
    rows: list[dict[str, Any]] = []
    for raw in iterator:
        row = {names[i]: raw[i] if i < len(raw) else None for i in range(len(names))}
        rows.append(row)
    return rows


def pick_live_excel(month_key: MonthKey) -> Path:
    prefix = f"{month_key.year}{month_key.month:02d}_"
    candidates = [
        p for p in VIPDATA_DIR.glob(f"{prefix}*.xlsx")
        if "goods_effect" not in p.name and p.stat().st_size >= 100000
    ]
    if not candidates:
        raise RuntimeError(f"No live-session Excel found for {month_key.ym} under {VIPDATA_DIR}")
    return sorted(candidates, key=lambda p: p.stat().st_mtime, reverse=True)[0]


def pick_goods_excel(month_key: MonthKey) -> Path:
    prefix = f"{month_key.year}{month_key.month:02d}_goods_effect_"
    candidates = sorted(VIPDATA_DIR.glob(f"{prefix}*MID*.xlsx"), key=lambda p: p.stat().st_mtime, reverse=True)
    if not candidates:
        candidates = sorted(VIPDATA_DIR.glob(f"{prefix}*.xlsx"), key=lambda p: p.stat().st_mtime, reverse=True)
    if not candidates:
        raise RuntimeError(f"No goods-effect Excel found for {month_key.ym} under {VIPDATA_DIR}")
    return candidates[0]


def fetch_live_session_agg_from_excel(month_key: MonthKey) -> dict[str, Decimal]:
    rows = read_excel_rows(pick_live_excel(month_key))
    viewers = Decimal("0")
    purchasers = Decimal("0")
    live_sales = Decimal("0")
    guided_sales = Decimal("0")
    brand_customers = Decimal("0")
    brand_new = Decimal("0")
    new_members = Decimal("0")
    weighted_pen = Decimal("0")
    weighted_base = Decimal("0")
    for row in rows:
        v = to_decimal(excel_cell(row, LIVE_COL_VIEWERS))
        pen = excel_pct_to_ratio(excel_cell(row, LIVE_COL_PENETRATION))
        viewers += v
        purchasers += to_decimal(excel_cell(row, LIVE_COL_PURCHASERS))
        live_sales += to_decimal(excel_cell(row, LIVE_COL_LIVE_SALES))
        guided_sales += to_decimal(excel_cell(row, LIVE_COL_GUIDED_SALES))
        brand_customers += to_decimal(excel_cell(row, LIVE_COL_BRAND_CUSTOMERS))
        brand_new += to_decimal(excel_cell(row, LIVE_COL_BRAND_NEW))
        new_members += to_decimal(excel_cell(row, LIVE_COL_NEW_MEMBERS))
        weighted_pen += pen * v
        weighted_base += v
    return {
        "accumulated_viewers": viewers,
        "live_penetration": weighted_pen / weighted_base if weighted_base > 0 else Decimal("0"),
        "purchasers_total": purchasers,
        "conversion": purchasers / viewers if viewers > 0 else Decimal("0"),
        "live_period_sales_total": live_sales,
        "guided_sales_total": guided_sales,
        "brand_customers_total": brand_customers,
        "brand_new_customers_total": brand_new,
        "new_customer_ratio": brand_new / brand_customers if brand_customers > 0 else Decimal("0"),
        "new_brand_members": new_members,
    }


def fetch_live_session_daily_from_excel(month_key: MonthKey) -> tuple[list[int], list[Decimal], list[Decimal]]:
    rows = read_excel_rows(pick_live_excel(month_key))
    day_totals: dict[int, dict[str, Decimal]] = {}
    for row in rows:
        day = excel_date_to_day(excel_cell(row, LIVE_COL_DATE))
        if not day:
            continue
        bucket = day_totals.setdefault(day, {"sales": Decimal("0"), "weighted_pen": Decimal("0"), "viewers": Decimal("0")})
        viewers = to_decimal(excel_cell(row, LIVE_COL_VIEWERS))
        pen = excel_pct_to_ratio(excel_cell(row, LIVE_COL_PENETRATION))
        bucket["sales"] += to_decimal(excel_cell(row, LIVE_COL_LIVE_SALES))
        bucket["weighted_pen"] += pen * viewers
        bucket["viewers"] += viewers
    days_in_month = calendar.monthrange(month_key.year, month_key.month)[1]
    day_list: list[int] = []
    sales_values: list[Decimal] = []
    conv_values: list[Decimal] = []
    for day in range(1, days_in_month + 1):
        bucket = day_totals.get(day, {"sales": Decimal("0"), "weighted_pen": Decimal("0"), "viewers": Decimal("0")})
        day_list.append(day)
        sales_values.append(bucket["sales"])
        conv_values.append(bucket["weighted_pen"] / bucket["viewers"] if bucket["viewers"] > 0 else Decimal("0"))
    return day_list, sales_values, conv_values


def normalize_category_level1(value: Any) -> str:
    cat = str(value or "").strip() or CATEGORY_UNKNOWN
    return CATEGORY_ACCESSORIES if cat in {CATEGORY_ACCESSORIES, CATEGORY_JEWELRY} else cat


def iter_goods_excel_rows(month_key: MonthKey) -> list[dict[str, Any]]:
    return read_excel_rows(pick_goods_excel(month_key))


def fetch_goods_category_rows_from_excel(month_key: MonthKey) -> dict[str, dict[str, Decimal]]:
    out: dict[str, dict[str, Decimal]] = {}
    for row in iter_goods_excel_rows(month_key):
        cat = normalize_category_level1(excel_cell(row, GOODS_COL_LEVEL1))
        bucket = out.setdefault(cat, {"articles": set(), "qty": Decimal("0"), "sales": Decimal("0"), "brand_new_customers": Decimal("0")})
        article = str(excel_cell(row, GOODS_COL_SPU) or "").strip()
        if article:
            bucket["articles"].add(article)  # type: ignore[union-attr]
        bucket["qty"] += to_decimal(excel_cell(row, GOODS_COL_QTY))
        bucket["sales"] += to_decimal(excel_cell(row, GOODS_COL_SALES))
        bucket["brand_new_customers"] += to_decimal(excel_cell(row, GOODS_COL_BRAND_NEW))
    converted: dict[str, dict[str, Decimal]] = {}
    for cat, values in out.items():
        converted[cat] = {
            "article_cnt": Decimal(len(values.get("articles", set()))),
            "qty": values["qty"],
            "sales": values["sales"],
            "brand_new_customers": values["brand_new_customers"],
        }
    return converted


def fetch_goods_mix_rows_from_excel(month_key: MonthKey) -> tuple[list[tuple[str, Decimal]], list[tuple[str, Decimal]]]:
    level1_totals: dict[str, Decimal] = {}
    level3_totals: dict[str, Decimal] = {}
    for row in iter_goods_excel_rows(month_key):
        qty = to_decimal(excel_cell(row, GOODS_COL_QTY))
        if qty <= 0:
            continue
        l1 = normalize_category_level1(excel_cell(row, GOODS_COL_LEVEL1))
        l3 = str(excel_cell(row, GOODS_COL_LEVEL3) or CATEGORY_UNKNOWN).strip() or CATEGORY_UNKNOWN
        level1_totals[l1] = level1_totals.get(l1, Decimal("0")) + qty
        level3_totals[l3] = level3_totals.get(l3, Decimal("0")) + qty
    return (
        sorted(level1_totals.items(), key=lambda x: x[1], reverse=True),
        sorted(level3_totals.items(), key=lambda x: x[1], reverse=True),
    )


def fetch_goods_top5_from_excel(month_key: MonthKey) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
    bag_map: dict[str, int] = {}
    shoe_map: dict[str, int] = {}
    bag_note_map: dict[str, dict[str, int]] = {}
    shoe_note_map: dict[str, dict[str, int]] = {}
    for row in iter_goods_excel_rows(month_key):
        cat = normalize_category_level1(excel_cell(row, GOODS_COL_LEVEL1))
        title = str(excel_cell(row, GOODS_COL_TITLE) or "")
        qty = int(to_decimal(excel_cell(row, GOODS_COL_QTY)))
        if qty <= 0:
            continue
        article = extract_article(title)
        if not article:
            continue
        note = normalize_note(title)
        if cat == CATEGORY_BAGS:
            bag_map[article] = bag_map.get(article, 0) + qty
            if note:
                bag_note_map.setdefault(article, {})[note] = bag_note_map.setdefault(article, {}).get(note, 0) + qty
        elif cat == CATEGORY_SHOES:
            shoe_map[article] = shoe_map.get(article, 0) + qty
            if note:
                shoe_note_map.setdefault(article, {})[note] = shoe_note_map.setdefault(article, {}).get(note, 0) + qty

    def to_top_rows(base: dict[str, int], notes: dict[str, dict[str, int]]) -> list[dict[str, Any]]:
        rows_out: list[dict[str, Any]] = []
        for article, qty in sorted(base.items(), key=lambda x: x[1], reverse=True)[:5]:
            note = ""
            if article in notes and notes[article]:
                note = sorted(notes[article].items(), key=lambda kv: kv[1], reverse=True)[0][0]
            rows_out.append({"article": article, "qty": qty, "note": note})
        return rows_out

    return to_top_rows(bag_map, bag_note_map), to_top_rows(shoe_map, shoe_note_map)


def fetch_goods_top5(conn: pymysql.connections.Connection, month_key: MonthKey) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
    sql = f"""
SELECT
  category_level_1 AS cat,
  product_title,
  IFNULL(item_count_total, 0) AS qty
FROM `{TABLE_GOODS}`
WHERE data_month = %s
"""
    bag_map: dict[str, int] = {}
    shoe_map: dict[str, int] = {}
    bag_note_map: dict[str, dict[str, int]] = {}
    shoe_note_map: dict[str, dict[str, int]] = {}

    with conn.cursor() as cur:
        cur.execute(sql, (month_key.ym,))
        rows = cur.fetchall() or []

    for row in rows:
        cat = str(row.get("cat") or "").strip()
        title = str(row.get("product_title") or "")
        qty = int(to_decimal(row.get("qty")))
        if qty <= 0:
            continue
        article = extract_article(title)
        if not article:
            continue

        side = ""
        if cat == "箱包":
            side = "bags"
        elif cat == "鞋":
            side = "shoes"
        else:
            continue

        note = normalize_note(title)
        if side == "bags":
            bag_map[article] = bag_map.get(article, 0) + qty
            if note:
                bag_note_map.setdefault(article, {})
                bag_note_map[article][note] = bag_note_map[article].get(note, 0) + qty
        else:
            shoe_map[article] = shoe_map.get(article, 0) + qty
            if note:
                shoe_note_map.setdefault(article, {})
                shoe_note_map[article][note] = shoe_note_map[article].get(note, 0) + qty

    def to_top_rows(base: dict[str, int], notes: dict[str, dict[str, int]]) -> list[dict[str, Any]]:
        top = sorted(base.items(), key=lambda x: x[1], reverse=True)[:5]
        rows_out: list[dict[str, Any]] = []
        for article, qty in top:
            note = ""
            if article in notes and notes[article]:
                note = sorted(notes[article].items(), key=lambda kv: kv[1], reverse=True)[0][0]
            rows_out.append({"article": article, "qty": qty, "note": note})
        return rows_out

    return to_top_rows(bag_map, bag_note_map), to_top_rows(shoe_map, shoe_note_map)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Build editable PPT operations for slides 14 and 15."
    )
    add_config_source_args(parser, default_config=Path(r"D:\projects\skills\vip-report\config.yaml"))
    parser.add_argument("--report-month", default="", help="Month override, e.g. 3 / Mar")
    parser.add_argument("--report-year", type=int, default=0, help="Report year override")
    parser.add_argument("--output-ops", default="", help="Optional output ops path")
    parser.add_argument("--slides", default="14,15", help="Comma-separated subset: 14,15")
    return parser.parse_args()


def find_local_product_image(workdir: Path, article: str) -> str:
    if not article:
        return ""
    search_roots = [
        workdir / "assets" / "top-products",
        workdir / "vip-report" / "assets" / "top-products",
        workdir / "assets" / "campaign-s11" / "top10-images",
    ]
    suffixes = (".png", ".jpg", ".jpeg")
    for root in search_roots:
        if not root.exists():
            continue
        for suffix in suffixes:
            matches = sorted(root.glob(f"*{article}*{suffix}"))
            if matches:
                return str(matches[0].resolve())
    return ""


def attach_product_images(rows: list[dict[str, Any]], workdir: Path) -> list[dict[str, Any]]:
    out: list[dict[str, Any]] = []
    for row in rows:
        copied = dict(row)
        copied["image_path"] = find_local_product_image(workdir, str(copied.get("article") or ""))
        out.append(copied)
    return out


def execute(
    config: dict[str, Any],
    *,
    report_month: str = "",
    report_year: int = 0,
    output_ops: str = "",
    slides: str = "14,15",
) -> dict[str, str]:
    report_cfg = config.get("report", {})
    year = int(report_year or report_cfg.get("year", 2026))
    month = parse_month_number(report_month or report_cfg.get("month_cn", "1"))
    curr_key = MonthKey(year, month)
    prev_key = previous_month(curr_key)
    slide15_curr_key = curr_key
    slide15_compare_key = same_month_last_year(curr_key)

    requested = {token.strip() for token in slides.split(",") if token.strip()}
    include14 = "14" in requested
    include15 = "15" in requested
    if not include14 and not include15:
        raise RuntimeError("No valid slides requested. Use --slides 14,15")

    try:
        mysql_cfg = config["mysql"]
        mysql_user = mysql_cfg.get("username") or mysql_cfg.get("user")
        conn = pymysql.connect(
            host=mysql_cfg["host"],
            port=int(mysql_cfg["port"]),
            user=mysql_user,
            password=mysql_cfg["password"],
            database=mysql_cfg["database"],
            charset="utf8mb4",
            cursorclass=pymysql.cursors.DictCursor,
        )
        try:
            session_curr = fetch_live_session_agg(conn, curr_key)
            session_prev = fetch_live_session_agg(conn, prev_key)
            days, sales_values, penetration_values = fetch_live_session_daily(conn, curr_key)

            goods_curr = fetch_goods_category_rows(conn, slide15_curr_key)
            goods_prev = fetch_goods_category_rows(conn, slide15_compare_key)
            compare_month = slide15_compare_key

            level1_mix, level3_mix = fetch_goods_mix_rows(conn, curr_key)
            bags_top5, shoes_top5 = fetch_goods_top5(conn, slide15_curr_key)
        finally:
            conn.close()
    except Exception as db_error:
        print(f"WARN: MySQL unavailable, using offline Excel fallback: {db_error}", file=sys.stderr)
        session_curr = fetch_live_session_agg_from_excel(curr_key)
        session_prev = fetch_live_session_agg_from_excel(prev_key)
        days, sales_values, penetration_values = fetch_live_session_daily_from_excel(curr_key)

        goods_curr = fetch_goods_category_rows_from_excel(slide15_curr_key)
        goods_prev = fetch_goods_category_rows_from_excel(slide15_compare_key)
        compare_month = slide15_compare_key

        level1_mix, level3_mix = fetch_goods_mix_rows_from_excel(curr_key)
        bags_top5, shoes_top5 = fetch_goods_top5_from_excel(slide15_curr_key)

    s14_headers, s14_rows = build_slide14_summary_rows(prev_key, session_prev, curr_key, session_curr)
    sales_axis_max, sales_axis_step = choose_sales_axis(sales_values)
    penetration_axis_max, penetration_axis_step = choose_conv_axis(penetration_values)

    s15_headers, s15_rows = build_slide15_summary_rows(goods_curr, goods_prev)

    level1_total = sum([v for _, v in level1_mix] or [Decimal("0")])
    level3_total = sum([v for _, v in level3_mix] or [Decimal("0")])
    level3_top = level3_mix[:5]
    level3_tail = level3_mix[5:15]

    vip_workdir = Path(config["paths"]["workdir"]).resolve()
    ops_path = Path(output_ops).resolve() if output_ops else (vip_workdir / "render-ops.flow14-15.editable.json")
    ops_path.parent.mkdir(parents=True, exist_ok=True)

    native_ops: list[dict[str, Any]] = []

    if include14:
        native_ops.extend(
            [
                {
                    "type": "slide14_live_session_summary_table",
                    "slide": 14,
                    "shape_id": 4,
                    "shape_name": "Table 1",
                    "target_box": {
                        "left": 1341272,
                        "top": 1883910,
                        "width": 17421552,
                        "height": 1286012,
                    },
                    "headers": s14_headers,
                    "rows": s14_rows,
                },
                {
                    "type": "slide14_live_session_trend",
                    "slide": 14,
                    "shape_id": 7,
                    "shape_name": "Chart 4",
                    "target_box": {
                        "left": 3350681,
                        "top": 3269323,
                        "width": 13402732,
                        "height": 3134677,
                    },
                    "month": curr_key.month,
                    "days": days,
                    "sales_values": [float(v) for v in sales_values],
                    "penetration_values": [round(float(v), 6) for v in penetration_values],
                    "sales_axis_max": sales_axis_max,
                    "sales_axis_major_unit": sales_axis_step,
                    "penetration_axis_max": penetration_axis_max,
                    "penetration_axis_major_unit": penetration_axis_step,
                },
                {
                    "type": "slide14_goods_category_mix",
                    "slide": 14,
                    "shape_id": 10,
                    "shape_name": "Group 12",
                    "target_box": {
                        "left": 683261,
                        "top": 6920612,
                        "width": 10262215,
                        "height": 4023309,
                    },
                    "level1_rows": [
                        {
                            "name": name,
                            "value": float(value),
                            "share": float(share(value, level1_total)),
                        }
                        for name, value in level1_mix
                    ],
                    "level3_top_rows": [
                        {
                            "name": name,
                            "value": float(value),
                            "share": float(share(value, level3_total)),
                        }
                        for name, value in level3_top
                    ],
                    "level3_tail_rows": [
                        {
                            "name": name,
                            "value": float(value),
                            "share": float(share(value, level3_total)),
                        }
                        for name, value in level3_tail
                    ],
                },
            ]
        )

    if include15:
        native_ops.extend(
            [
                {
                    "type": "slide15_goods_summary_table",
                    "slide": 15,
                    "shape_id": 5,
                    "shape_name": "Table 2",
                    "target_box": {
                        "left": 2171701,
                        "top": 1875631,
                        "width": 15995648,
                        "height": 1539081,
                    },
                    "headers": s15_headers,
                    "rows": s15_rows,
                    "compare_month": compare_month.ym,
                },
                {
                    "type": "slide15_top5_tables",
                    "slide": 15,
                    "shape_id": 7,
                    "shape_name": "Picture 1",
                    "target_box": {
                        "left": 5675312,
                        "top": 3716208,
                        "width": 8753475,
                        "height": 5372100,
                    },
                    "bags_rows": attach_product_images(bags_top5, vip_workdir),
                    "shoes_rows": attach_product_images(shoes_top5, vip_workdir),
                },
            ]
        )

    operations = {
        "replace_text": [],
        "replace_tables": [],
        "replace_charts": [],
        "replace_images": [],
        "replace_native_charts": native_ops,
    }

    ops_path.write_text(json.dumps(operations, ensure_ascii=False, indent=2), encoding="utf-8")
    return {"operations_path": str(ops_path)}


def main() -> None:
    args = parse_args()
    config = load_runtime_config(args, default_config=Path(r"D:\projects\skills\vip-report\config.yaml"))
    result = execute(
        config,
        report_month=args.report_month,
        report_year=args.report_year,
        output_ops=args.output_ops,
        slides=args.slides,
    )
    print(json.dumps(result, ensure_ascii=False))


if __name__ == "__main__":
    main()
