"""奇成 MySQL 只读查询封装。

接入 kyhl 库,仅 SELECT,强制 READ ONLY 事务防止误写。
所有查询按 ICCID/agent_id 批量分块,避免单条 SQL 过大。
"""
from __future__ import annotations

import contextlib
from dataclasses import dataclass
from decimal import Decimal
from typing import Any, Iterable, Iterator, Optional


BATCH_SIZE = 500


def _normalize_iccid_pairs(iccid_pairs: Iterable[tuple]) -> list[tuple[str, str, bool]]:
    """标准化 ICCID 查询参数。

    tuple 第三项表示是否允许用 iccid_19 查询奇成 tbl_card 主表:
    - 只填 iccid_20 的固定 20 位卡:false,避免 19 位前缀误撞其他卡
    - 显式填了 iccid_19:true,允许作为主表查询键或 20 位缺失时兜底
    """
    out: list[tuple[str, str, bool]] = []
    for item in iccid_pairs:
        i19 = str(item[0] or "").strip() if len(item) > 0 else ""
        i20 = str(item[1] or "").strip() if len(item) > 1 else ""
        allow_i19_lookup = bool(item[2]) if len(item) > 2 else True
        if i19:
            out.append((i19, i20, allow_i19_lookup))
    return out


def _import_pymysql():
    """延迟导入 pymysql,这样脚本1(纯离线)不需要安装 pymysql 也能运行。"""
    try:
        import pymysql
        from pymysql.cursors import DictCursor
        return pymysql, DictCursor
    except ImportError as exc:
        raise ImportError("缺少 pymysql,请先 pip install -r requirements.txt") from exc


@dataclass(frozen=True)
class LegacyPackage:
    """奇成当前生效套餐快照(取自 tbl_card_life)。"""

    iccid: str
    meal_id: str
    meal_name: str
    start_date: Optional[str]
    expire_date: Optional[str]
    flow_size_mb: int  # 套餐总量,单位 MB
    status: int


@dataclass(frozen=True)
class LegacyCardUsage:
    """奇成卡累计流量快照。

    奇成 ``tbl_card.total_bytes_cnt`` 存的是经过当前套餐 ``flow_add_discount``
    加成后的"虚用量",我们这里同时拉取当前生效套餐(``tbl_card_life`` status=1
    且 ``expire_date`` 最大)的 ``flow_add_discount``,反算回新系统期望的"真用量"。

    精度说明:跨多个套餐周期累计时,``total_bytes_cnt`` 是按各时段对应套餐
    虚比加权累加的,这里只用"当前生效套餐"的虚比反算,**有误差但可接受**——
    奇成没有按套餐分段的用量明细,无更好做法。
    """

    iccid: str                       # 业务方视角的 iccid_19
    virtual_used_mb: Decimal         # 奇成 total_bytes_cnt 原始值(虚 MB)
    flow_add_discount_pct: Decimal   # 当前生效套餐虚比百分比(0~100)
    has_active_package: bool         # 是否在 tbl_card_life 中找到当前生效套餐

    @property
    def real_used_mb(self) -> Decimal:
        """反算真用量(Decimal MB,保留小数)。

        虚 = 真 × (1 + pct/100) → 真 = 虚 / (1 + pct/100)
        """
        if self.virtual_used_mb <= 0:
            return Decimal("0")
        ratio = Decimal(1) + self.flow_add_discount_pct / Decimal(100)
        return self.virtual_used_mb / ratio

    @property
    def real_used_mb_floor(self) -> int:
        """适配新系统 ``data_usage_mb``(BIGINT)的整数 MB。

        取整规则:0 → 0,(0, 1) → 1,≥1 → 向下取整。
        sub-MB 上取整为 1,避免"明明用过的卡显示 0"。
        """
        real = self.real_used_mb
        if real <= 0:
            return 0
        if real < 1:
            return 1
        return int(real)


@dataclass(frozen=True)
class LegacyCommissionAccount:
    """奇成代理佣金账户快照。"""

    agent_id: str
    agent_name: str
    can_draw_amount_fen: int      # 当前可提现佣金(分)
    applying_draw_amount_fen: int # 申请提现中佣金(分)
    total_commission_amount_fen: int  # 累计总佣金(分,做参考)
    total_draw_amount_fen: int    # 累计已提现(分,做参考)


@dataclass(frozen=True)
class LegacyAgentBalance:
    """奇成代理主钱包余额快照(取自 tbl_agent.balance_money)。"""

    agent_id: str
    balance_fen: int


@dataclass(frozen=True)
class LegacyCardMeta:
    """奇成卡基础元数据(由 scan_legacy / migrate_assets 共享)。

    业务方提供的 iccid_19 + iccid_20 才是新系统的权威值,奇成的 iccid_mark 长度不可信。
    legacy_raw_iccid 仅作调试参考。
    """

    legacy_raw_iccid: str        # 奇成实际存的 iccid_mark(可能 19 或 20 位)
    account_id: str              # tbl_card.account_id(开卡公司/运营商账号 ID)
    account_name: str            # tbl_card.account_name(开卡公司/运营商账号名称)
    category: int                # tbl_card.category(运营商分类编号)
    category_name: str           # tbl_vendor_category.category_name
    agent_id: str
    agent_name: str
    msisdn: str                  # tbl_card.phone
    virtual_no: str              # tbl_virtual_number.vcode(如有)
    current_meal_id: str         # 当前生效套餐 ID(可空,取自 tbl_card_life status=1 expire_date 最大)
    current_meal_name: str
    current_expire_date: Optional[str]  # ISO 字符串


@dataclass(frozen=True)
class LegacyMealMeta:
    """奇成套餐 + 系列元数据(由 scan_legacy 用)。"""

    meal_id: str
    meal_name: str
    series_id: str
    series_name: str


# ---------------- 连接管理 ----------------


@contextlib.contextmanager
def connect_readonly(dsn: dict) -> Iterator[Any]:
    """打开只读连接。

    用法:
        with connect_readonly(dsn) as conn:
            rows = fetch_current_packages(conn, iccids)
    """
    pymysql, DictCursor = _import_pymysql()
    conn = pymysql.connect(
        host=dsn["host"],
        port=int(dsn.get("port", 3306)),
        user=dsn["user"],
        password=dsn["password"],
        database=dsn["database"],
        charset=dsn.get("charset", "utf8mb4"),
        connect_timeout=int(dsn.get("connect_timeout", 10)),
        cursorclass=DictCursor,
        autocommit=False,
    )
    try:
        with conn.cursor() as cur:
            cur.execute("SET SESSION TRANSACTION READ ONLY")
            cur.execute("SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ")
        conn.begin()
        yield conn
        conn.rollback()  # 只读事务,回滚即结束
    finally:
        conn.close()


def _chunks(items: list[str], size: int = BATCH_SIZE) -> Iterator[list[str]]:
    for start in range(0, len(items), size):
        yield items[start : start + size]


def _to_fen(value) -> int:
    """元 → 分。奇成的 decimal 是元单位。"""
    if value is None:
        return 0
    try:
        d = Decimal(str(value))
    except Exception:
        return 0
    return int((d * Decimal("100")).quantize(Decimal("1")))


def _to_mb_decimal(value) -> Decimal:
    """字符串/数字 → Decimal MB。奇成 total_bytes_cnt 是小数 MB 读数。"""
    if value is None or value == "":
        return Decimal("0")
    try:
        d = Decimal(str(value))
        if d < 0:
            return Decimal("0")
        return d
    except Exception:
        return Decimal("0")


# ---------------- 查询函数 ----------------


def fetch_current_packages(conn, iccids: Iterable[str]) -> dict[str, LegacyPackage]:
    """查每张卡的当前生效套餐:status=1 且 expire_date 最大。

    返回: iccid → LegacyPackage(找不到的 iccid 不在结果中)
    """
    iccid_list = sorted({i for i in iccids if i})
    out: dict[str, LegacyPackage] = {}
    if not iccid_list:
        return out

    sql = """
        SELECT
            iccid_mark,
            meal_id,
            meal_name,
            start_date,
            expire_date,
            COALESCE(flow_size, 0) AS flow_size,
            status
        FROM tbl_card_life
        WHERE iccid_mark IN %s
          AND status = 1
        ORDER BY iccid_mark ASC, expire_date DESC
    """
    seen: set[str] = set()
    with conn.cursor() as cur:
        for batch in _chunks(iccid_list):
            cur.execute(sql, (tuple(batch),))
            for row in cur.fetchall():
                iccid = row["iccid_mark"]
                if not iccid or iccid in seen:
                    continue
                seen.add(iccid)
                out[iccid] = LegacyPackage(
                    iccid=iccid,
                    meal_id=row.get("meal_id") or "",
                    meal_name=row.get("meal_name") or "",
                    start_date=row["start_date"].strftime("%Y-%m-%d %H:%M:%S") if row.get("start_date") else None,
                    expire_date=row["expire_date"].strftime("%Y-%m-%d %H:%M:%S") if row.get("expire_date") else None,
                    flow_size_mb=int(row.get("flow_size") or 0),
                    status=int(row.get("status") or 0),
                )
    return out


def fetch_card_usage(conn, iccid_pairs: Iterable[tuple]) -> dict[str, LegacyCardUsage]:
    """查每张卡的累计已用流量,并同步取当前生效套餐的 flow_add_discount。

    奇成 ``tbl_card.total_bytes_cnt`` 是虚用量(已被套餐虚比加成)。
    这里 JOIN 当前生效套餐(``tbl_card_life`` status=1 且 ``expire_date`` 最大)
    + ``tbl_set_meal`` 取 ``flow_add_discount``,放进 ``LegacyCardUsage``;
    上层通过 ``real_used_mb_floor`` 反算真用量写入新系统。

    Args:
        iccid_pairs: (iccid_19, iccid_20, allow_iccid_19_lookup) 列表。

    Returns:
        {iccid_19: LegacyCardUsage}。
    """
    pairs = _normalize_iccid_pairs(iccid_pairs)
    out: dict[str, LegacyCardUsage] = {}
    if not pairs:
        return out

    full_to_19: dict[str, tuple[str, int]] = {}
    related_keys_by_19: dict[str, set[str]] = {}
    for i19, i20, allow_i19_lookup in pairs:
        # tbl_card 主表同时查优先键和兜底键,结果选择时按 rank:
        # 0=业务方完整 ICCID 精确命中,1=20 位卡在奇成主表只存 19 位时兜底。
        full_to_19[i20 or i19] = (i19, 0)
        if i20 and allow_i19_lookup:
            full_to_19[i19] = (i19, 1)
        related_keys_by_19.setdefault(i19, set()).add(i19)
        if i20:
            related_keys_by_19[i19].add(i20)
    full_set = sorted(full_to_19.keys())

    # tbl_card 主表用 iccid_mark IN 全长精确匹配;tbl_card_life 从表用
    # LEFT(iccid_mark, 19) 兜底,兼容奇成两张表 iccid 长度不一致的脏数据。
    sql = """
        SELECT
            c.iccid_mark,
            c.total_bytes_cnt,
            COALESCE(sm.flow_add_discount, 0) AS flow_add_discount,
            CASE WHEN cl.iccid_mark IS NOT NULL THEN 1 ELSE 0 END AS has_active_package
        FROM tbl_card c
        LEFT JOIN (
            SELECT cl1.iccid_mark, cl1.meal_id
            FROM tbl_card_life cl1
            INNER JOIN (
                SELECT iccid_mark, MAX(expire_date) AS max_expire
                FROM tbl_card_life
                WHERE status = 1 AND iccid_mark IN %s
                GROUP BY iccid_mark
            ) latest
              ON latest.iccid_mark = cl1.iccid_mark
             AND latest.max_expire = cl1.expire_date
            WHERE cl1.status = 1
        ) cl ON LEFT(cl.iccid_mark, 19) = LEFT(c.iccid_mark, 19)
        LEFT JOIN tbl_set_meal sm ON sm.id = cl.meal_id
        WHERE c.iccid_mark IN %s
    """
    hits: dict[str, list[tuple[int, dict]]] = {}
    with conn.cursor() as cur:
        for batch in _chunks(full_set):
            batch_i19s = {full_to_19[x][0] for x in batch}
            related_batch = sorted({x for i19 in batch_i19s for x in related_keys_by_19[i19]})
            cur.execute(sql, (tuple(related_batch), tuple(batch)))
            for row in cur.fetchall():
                legacy_iccid = row["iccid_mark"]
                if not legacy_iccid:
                    continue
                matched = full_to_19.get(legacy_iccid)
                if matched is None:
                    continue
                i19, rank = matched
                hits.setdefault(i19, []).append((rank, row))
    for i19, rows in hits.items():
        best_rank = min(rank for rank, _row in rows)
        row = next(row for rank, row in rows if rank == best_rank)
        out[i19] = LegacyCardUsage(
            iccid=i19,
            virtual_used_mb=_to_mb_decimal(row.get("total_bytes_cnt")),
            flow_add_discount_pct=_to_mb_decimal(row.get("flow_add_discount")),
            has_active_package=bool(row.get("has_active_package")),
        )
    return out


def fetch_commission_accounts(conn, agent_ids: Iterable[str]) -> dict[str, LegacyCommissionAccount]:
    """查代理佣金账户。"""
    agent_list = sorted({a for a in agent_ids if a})
    out: dict[str, LegacyCommissionAccount] = {}
    if not agent_list:
        return out

    sql = """
        SELECT
            agent_id, agent_name,
            can_draw_amount, applying_draw_amount,
            total_commision_amount, total_draw_amount
        FROM tbl_agent_commission_account
        WHERE agent_id IN %s
    """
    with conn.cursor() as cur:
        for batch in _chunks(agent_list):
            cur.execute(sql, (tuple(batch),))
            for row in cur.fetchall():
                aid = row["agent_id"]
                if not aid:
                    continue
                out[aid] = LegacyCommissionAccount(
                    agent_id=aid,
                    agent_name=row.get("agent_name") or "",
                    can_draw_amount_fen=_to_fen(row.get("can_draw_amount")),
                    applying_draw_amount_fen=_to_fen(row.get("applying_draw_amount")),
                    total_commission_amount_fen=_to_fen(row.get("total_commision_amount")),
                    total_draw_amount_fen=_to_fen(row.get("total_draw_amount")),
                )
    return out


def fetch_card_meta(
    conn,
    iccid_pairs: Iterable[tuple],
) -> tuple[dict[str, LegacyCardMeta], set[str]]:
    """查每张卡的元数据(卡本体 + 当前生效套餐 + 虚拟号)。

    匹配策略:
        - tbl_card 主表同时查优先键和显式允许的兜底键:有 iccid_20 时优先取
          20 位精确命中;只有业务方也显式填了 iccid_19 时,才允许退回 19 位。
          只填 iccid_20 的固定 20 位卡绝不拿 19 位前缀查主表。
        - tbl_card_life / tbl_virtual_number 从表用 LEFT(iccid_mark, 19) 兜底,
          应对奇成两张表 iccid 长度不一致的脏数据

    Args:
        iccid_pairs: (iccid_19, iccid_20, allow_iccid_19_lookup) 列表。

    Returns:
        (metas, duplicate_iccid_19):
            - metas: {iccid_19: LegacyCardMeta} 单匹结果
            - duplicate_iccid_19: 一份业务方 iccid_19 命中奇成 tbl_card 多条记录的 set,
              这些卡不进 metas,sql_builder 会写 errors.csv 阻断
    """
    pairs = _normalize_iccid_pairs(iccid_pairs)
    if not pairs:
        return {}, set()

    by_full: dict[str, tuple[str, int]] = {}     # tbl_card 查询键 → (iccid_19, rank)
    related_keys_by_19: dict[str, set[str]] = {}
    for i19, i20, allow_i19_lookup in pairs:
        by_full[i20 or i19] = (i19, 0)
        if i20 and allow_i19_lookup:
            by_full[i19] = (i19, 1)
        related_keys_by_19.setdefault(i19, set()).add(i19)
        if i20:
            related_keys_by_19[i19].add(i20)
    full_set = sorted(by_full.keys())

    sql = """
        SELECT
            c.id                       AS card_id,
            c.iccid_mark               AS iccid,
            COALESCE(c.account_id, '')  AS account_id,
            COALESCE(c.account_name, '') AS account_name,
            COALESCE(c.category, 0)    AS category,
            COALESCE(vc.category_name, '') AS category_name,
            COALESCE(c.agent_id, '')   AS agent_id,
            COALESCE(c.agent_name, '') AS agent_name,
            COALESCE(c.phone, '')      AS phone,
            COALESCE(vn.vcode, '')     AS vcode,
            COALESCE(cl.meal_id, '')   AS meal_id,
            COALESCE(cl.meal_name, '') AS meal_name,
            cl.expire_date             AS expire_date
        FROM tbl_card c
        LEFT JOIN tbl_vendor_category vc ON vc.category = c.category
        LEFT JOIN tbl_virtual_number vn ON LEFT(vn.iccid_mark, 19) = LEFT(c.iccid_mark, 19)
        LEFT JOIN (
            SELECT cl1.iccid_mark, cl1.meal_id, cl1.meal_name, cl1.expire_date
            FROM tbl_card_life cl1
            INNER JOIN (
                SELECT iccid_mark, MAX(expire_date) AS max_expire
                FROM tbl_card_life
                WHERE status = 1 AND iccid_mark IN %s
                GROUP BY iccid_mark
            ) latest
              ON latest.iccid_mark = cl1.iccid_mark
             AND latest.max_expire = cl1.expire_date
            WHERE cl1.status = 1
        ) cl ON LEFT(cl.iccid_mark, 19) = LEFT(c.iccid_mark, 19)
        WHERE c.iccid_mark IN %s
    """

    # 收集每个 iccid_19 命中的奇成行(可能多条 = 脏数据)
    hits: dict[str, list[tuple[int, dict]]] = {}
    with conn.cursor() as cur:
        for batch in _chunks(full_set):
            batch_i19s = {by_full[x][0] for x in batch}
            related_batch = sorted({x for i19 in batch_i19s for x in related_keys_by_19[i19]})
            cur.execute(sql, (tuple(related_batch), tuple(batch)))
            for row in cur.fetchall():
                legacy_iccid = row["iccid"]
                if not legacy_iccid:
                    continue
                matched = by_full.get(legacy_iccid)
                if matched is None:
                    continue
                i19, rank = matched
                hits.setdefault(i19, []).append((rank, row))

    metas: dict[str, LegacyCardMeta] = {}
    duplicates: set[str] = set()
    for i19, rows in hits.items():
        best_rank = min(rank for rank, _row in rows)
        best_rows = [row for rank, row in rows if rank == best_rank]
        # 同一张 tbl_card 可能因 card_life / virtual_number 前缀兜底 JOIN 出多行,
        # 这不是主表重复;只有不同 card_id 才算同一业务 ICCID 命中多张卡。
        rows_by_card_id: dict[str, dict] = {}
        for row in best_rows:
            rows_by_card_id[str(row.get("card_id") or row.get("iccid") or "")] = row
        if len(rows_by_card_id) > 1:
            duplicates.add(i19)
            continue
        row = next(iter(rows_by_card_id.values()))
        expire = row.get("expire_date")
        expire_str = expire.strftime("%Y-%m-%d %H:%M:%S") if expire else None
        metas[i19] = LegacyCardMeta(
            legacy_raw_iccid=row["iccid"],
            account_id=row.get("account_id") or "",
            account_name=row.get("account_name") or "",
            category=int(row.get("category") or 0),
            category_name=row.get("category_name") or "",
            agent_id=row.get("agent_id") or "",
            agent_name=row.get("agent_name") or "",
            msisdn=row.get("phone") or "",
            virtual_no=row.get("vcode") or "",
            current_meal_id=row.get("meal_id") or "",
            current_meal_name=row.get("meal_name") or "",
            current_expire_date=expire_str,
        )
    return metas, duplicates


def fetch_meal_meta(conn, meal_ids: Iterable[str]) -> dict[str, LegacyMealMeta]:
    """查套餐 + 系列元数据(供 scan_legacy 收集映射 legacy_* 字段)。"""
    id_list = sorted({m for m in meal_ids if m})
    out: dict[str, LegacyMealMeta] = {}
    if not id_list:
        return out
    sql = """
        SELECT
            sm.id                            AS meal_id,
            COALESCE(sm.name, '')            AS meal_name,
            COALESCE(sm.meal_series_id, '')  AS series_id,
            COALESCE(sms.name, '')           AS series_name
        FROM tbl_set_meal sm
        LEFT JOIN tbl_set_meal_series sms ON sms.id = sm.meal_series_id
        WHERE sm.id IN %s
    """
    with conn.cursor() as cur:
        for batch in _chunks(id_list):
            cur.execute(sql, (tuple(batch),))
            for row in cur.fetchall():
                mid = row["meal_id"]
                if not mid:
                    continue
                out[mid] = LegacyMealMeta(
                    meal_id=mid,
                    meal_name=row.get("meal_name") or "",
                    series_id=row.get("series_id") or "",
                    series_name=row.get("series_name") or "",
                )
    return out


def fetch_agent_balances(conn, agent_ids: Iterable[str]) -> dict[str, LegacyAgentBalance]:
    """查代理主钱包余额(tbl_agent.balance_money,元单位)。"""
    agent_list = sorted({a for a in agent_ids if a})
    out: dict[str, LegacyAgentBalance] = {}
    if not agent_list:
        return out

    sql = """
        SELECT id, balance_money
        FROM tbl_agent
        WHERE id IN %s
    """
    with conn.cursor() as cur:
        for batch in _chunks(agent_list):
            cur.execute(sql, (tuple(batch),))
            for row in cur.fetchall():
                aid = row["id"]
                if not aid:
                    continue
                out[aid] = LegacyAgentBalance(agent_id=aid, balance_fen=_to_fen(row.get("balance_money")))
    return out