csxj2026/junhong_cmp_fiber
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects 1 Releases Wiki Activity

迁移脚本的一些问题修复

Browse Source
This commit is contained in:
Break
2026-05-28 17:22:52 +08:00
parent 1550ef3cb1
commit 872769e69e
13 changed files with 328 additions and 114 deletions
Download Patch File Download Diff File Expand all files Collapse all files

128
scripts/migration/lib/legacy_query.py
View File

@@ -14,6 +14,23 @@ 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:
@@ -110,6 +127,8 @@ class LegacyCardMeta:
"""
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
@@ -242,7 +261,7 @@ def fetch_current_packages(conn, iccids: Iterable[str]) -> dict[str, LegacyPacka
return out
def fetch_card_usage(conn, iccid_pairs: Iterable[tuple[str, str]]) -> dict[str, LegacyCardUsage]:
def fetch_card_usage(conn, iccid_pairs: Iterable[tuple]) -> dict[str, LegacyCardUsage]:
"""查每张卡的累计已用流量,并同步取当前生效套餐的 flow_add_discount。
奇成 ``tbl_card.total_bytes_cnt`` 是虚用量(已被套餐虚比加成)。
@@ -251,21 +270,27 @@ def fetch_card_usage(conn, iccid_pairs: Iterable[tuple[str, str]]) -> dict[str,
上层通过 ``real_used_mb_floor`` 反算真用量写入新系统。
Args:
iccid_pairs: (iccid_19, iccid_20) 列表;iccid_20 可为空字符串。
iccid_pairs: (iccid_19, iccid_20, allow_iccid_19_lookup) 列表。
Returns:
{iccid_19: LegacyCardUsage}。
"""
pairs = [(p[0], p[1]) for p in iccid_pairs if p[0]]
pairs = _normalize_iccid_pairs(iccid_pairs)
out: dict[str, LegacyCardUsage] = {}
if not pairs:
return out
full_to_19: dict[str, str] = {}
for i19, i20 in pairs:
full_to_19[i19] = i19
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:
full_to_19[i20] = i19
related_keys_by_19[i19].add(i20)
full_set = sorted(full_to_19.keys())
# tbl_card 主表用 iccid_mark IN 全长精确匹配;tbl_card_life 从表用
@@ -293,25 +318,30 @@ def fetch_card_usage(conn, iccid_pairs: Iterable[tuple[str, str]]) -> dict[str,
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):
cur.execute(sql, (tuple(batch), tuple(batch)))
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
i19 = full_to_19.get(legacy_iccid)
if i19 is None:
matched = full_to_19.get(legacy_iccid)
if matched is None:
continue
# 取第一个匹配,后续 fetch_card_meta 已经会把重复命中的卡标记为 errors
if i19 in out:
continue
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")),
)
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
@@ -350,18 +380,19 @@ def fetch_commission_accounts(conn, agent_ids: Iterable[str]) -> dict[str, Legac
def fetch_card_meta(
conn,
iccid_pairs: Iterable[tuple[str, str]],
iccid_pairs: Iterable[tuple],
) -> tuple[dict[str, LegacyCardMeta], set[str]]:
"""查每张卡的元数据(卡本体 + 当前生效套餐 + 虚拟号)。
匹配策略:
- tbl_card 主表用 `iccid_mark IN (iccid_19 ∪ iccid_20)` 全长精确;
这样不会因前 19 位相同(校验位不同的两张 20 位卡)而误匹配多张
- 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) 列表;iccid_20 可为空字符串(CTCC 卡)。
iccid_pairs: (iccid_19, iccid_20, allow_iccid_19_lookup) 列表。
Returns:
(metas, duplicate_iccid_19):
@@ -369,23 +400,27 @@ def fetch_card_meta(
- duplicate_iccid_19: 一份业务方 iccid_19 命中奇成 tbl_card 多条记录的 set,
这些卡不进 metas,sql_builder 会写 errors.csv 阻断
"""
pairs = [(p[0], p[1]) for p in iccid_pairs if p[0]]
pairs = _normalize_iccid_pairs(iccid_pairs)
if not pairs:
return {}, set()
# 全长 ICCID → 业务方 iccid_19。同时维护 iccid_20 优先级:
# 奇成 iccid_mark 命中时,优先取等于 iccid_20 的映射(20 位精确),
# 找不到再退到 iccid_19(奇成存了 19 位的兜底)。
by_full: dict[str, str] = {} # 任意长度 ICCID → iccid_19
for i19, i20 in pairs:
by_full[i19] = i19
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:
by_full[i20] = i19
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,
@@ -415,33 +450,42 @@ def fetch_card_meta(
"""
# 收集每个 iccid_19 命中的奇成行(可能多条 = 脏数据)
hits: dict[str, list[dict]] = {}
hits: dict[str, list[tuple[int, dict]]] = {}
with conn.cursor() as cur:
for batch in _chunks(full_set):
cur.execute(sql, (tuple(batch), tuple(batch)))
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
i19 = by_full.get(legacy_iccid)
if i19 is None:
# WHERE 已经限定 IN,理论不会出现;保险起见再按前 19 位反查
i19 = by_full.get(legacy_iccid[:19])
if i19 is None:
continue
hits.setdefault(i19, []).append(row)
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():
if len(rows) > 1:
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 = rows[0]
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 "",