Files
junhong_cmp_fiber/scripts/migration/lib/legacy_query.py
2026-06-09 11:29:40 +08:00

572 lines
21 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""奇成 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 = 200
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
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: LegacyCardUsage}。
"""
pairs = _normalize_iccid_pairs(iccid_pairs)
out: dict[str, LegacyCardUsage] = {}
if not pairs:
return out
query_to_full: dict[str, tuple[str, int]] = {}
related_keys_by_full: dict[str, set[str]] = {}
for i19, i20, allow_i19_lookup in pairs:
card_key = i20 or i19
# tbl_card 主表同时查优先键和兜底键,结果选择时按 rank:
# 0=业务方完整 ICCID 精确命中,1=20 位卡在奇成主表只存 19 位时兜底。
query_to_full[card_key] = (card_key, 0)
if i20 and allow_i19_lookup:
query_to_full[i19] = (card_key, 1)
if not i20 or allow_i19_lookup:
related_keys_by_full.setdefault(card_key, set()).add(i19)
else:
related_keys_by_full.setdefault(card_key, set())
if i20:
related_keys_by_full[card_key].add(i20)
full_set = sorted(query_to_full.keys())
# tbl_card 主表用 iccid_mark IN 全长精确匹配;从表只在主表本身是 19 位时
# 按前缀兜底,避免同前缀 20 位卡互相串套餐。
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 (
cl.iccid_mark = c.iccid_mark
OR (CHAR_LENGTH(c.iccid_mark) = 19 AND LEFT(cl.iccid_mark, 19) = c.iccid_mark)
)
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_keys = {query_to_full[x][0] for x in batch}
related_batch = sorted({x for key in batch_keys for x in related_keys_by_full[key]})
cur.execute(sql, (tuple(related_batch), tuple(batch)))
for row in cur.fetchall():
legacy_iccid = row["iccid_mark"]
if not legacy_iccid:
continue
matched = query_to_full.get(legacy_iccid)
if matched is None:
continue
card_key, rank = matched
hits.setdefault(card_key, []).append((rank, row))
for card_key, 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[card_key] = LegacyCardUsage(
iccid=card_key,
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 从表在主表本身是 19 位时才用
LEFT(iccid_mark, 19) 兜底,避免同前缀 20 位卡互相串数据
Args:
iccid_pairs: (iccid_19, iccid_20, allow_iccid_19_lookup) 列表。
Returns:
(metas, duplicate_iccids):
- metas: {完整 ICCID: LegacyCardMeta} 单匹结果
- duplicate_iccids: 一份业务方完整 ICCID 命中奇成 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, rank)
related_keys_by_full: dict[str, set[str]] = {}
for i19, i20, allow_i19_lookup in pairs:
card_key = i20 or i19
by_full[card_key] = (card_key, 0)
if i20 and allow_i19_lookup:
by_full[i19] = (card_key, 1)
if not i20 or allow_i19_lookup:
related_keys_by_full.setdefault(card_key, set()).add(i19)
else:
related_keys_by_full.setdefault(card_key, set())
if i20:
related_keys_by_full[card_key].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 (
vn.iccid_mark = c.iccid_mark
OR (CHAR_LENGTH(c.iccid_mark) = 19 AND LEFT(vn.iccid_mark, 19) = c.iccid_mark)
)
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 (
cl.iccid_mark = c.iccid_mark
OR (CHAR_LENGTH(c.iccid_mark) = 19 AND LEFT(cl.iccid_mark, 19) = c.iccid_mark)
)
WHERE c.iccid_mark IN %s
"""
# 收集每个完整 ICCID 命中的奇成行(可能多条 = 脏数据)
hits: dict[str, list[tuple[int, dict]]] = {}
with conn.cursor() as cur:
for batch in _chunks(full_set):
batch_keys = {by_full[x][0] for x in batch}
related_batch = sorted({x for key in batch_keys for x in related_keys_by_full[key]})
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
card_key, rank = matched
hits.setdefault(card_key, []).append((rank, row))
metas: dict[str, LegacyCardMeta] = {}
duplicates: set[str] = set()
for card_key, 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(card_key)
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[card_key] = 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