From 1c492fe8db1a5d4973108e1e5b123567eabfc0db Mon Sep 17 00:00:00 2001 From: Break Date: Fri, 12 Jun 2026 18:10:22 +0800 Subject: [PATCH] =?UTF-8?q?=E6=9B=B4=E6=96=B0=E4=B8=80=E7=89=88?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- scripts/migration/README.md | 217 ++++- scripts/migration/config/mapping.yaml.example | 46 +- scripts/migration/lib/csv_loader.py | 84 +- scripts/migration/lib/legacy_query.py | 213 ++++- scripts/migration/lib/mapping_loader.py | 474 +++++++++- scripts/migration/lib/sql_builder.py | 826 ++++++++++++++---- scripts/migration/migrate_assets.py | 2 +- scripts/migration/migrate_runtime.py | 4 +- scripts/migration/resources/README.md | 36 +- scripts/migration/scan_legacy.py | 91 +- 10 files changed, 1735 insertions(+), 258 deletions(-) diff --git a/scripts/migration/README.md b/scripts/migration/README.md index beddb90..f97bdb0 100644 --- a/scripts/migration/README.md +++ b/scripts/migration/README.md @@ -2,13 +2,13 @@ > 详细方案见 [`docs/脚本/奇成数据迁移方案.md`](../../docs/脚本/奇成数据迁移方案.md)。 -## 1. 四步走 +## 1. 两条执行路径 ``` 业务方 我们 运维 │ │ │ │ cards.csv │ scan_legacy.py → mapping.yaml │ - │ devices.csv │ ↓ 人工填 target_* │ + │ devices.csv? │ ↓ 人工填 target_* │ ├─────────────▶│ migrate_assets.py → step1_*.sql │ │ │ migrate_runtime.py → step2_*.sql │ │ ├────────────────────────────────────▶│ psql -1 -f step*.sql @@ -16,6 +16,11 @@ │ │ │ 重建轮询队列/缓存 ``` +`devices.csv` 是可选文件: + +- **只有卡**:只准备 `cards.csv`,不要放 `devices.csv`;脚本按独立卡迁移,不会生成设备绑定关系。 +- **卡 + 设备**:同时准备 `cards.csv` 和 `devices.csv`;设备上的卡跟随设备归属,设备套餐挂到设备资产,旧套餐数据从 `package_source_slot` 对应绑定卡读取。 + ## 2. 文件清单 ``` @@ -28,10 +33,10 @@ scripts/migration/ ├── lib/ ← 公共库 ├── resources/ │ ├── README.md -│ ├── cards.csv.example ← 业务方提供:iccid + 可选覆盖列 -│ └── devices.csv.example ← 业务方提供:设备 + 卡槽 +│ ├── cards.csv.example ← 业务方提供:ICCID + is_industry +│ └── devices.csv.example ← 可选:设备 + 卡槽 + 可选槽位覆盖 ├── config/ -│ ├── mapping.yaml.example ← 唯一映射配置(全局 + carrier + package + series + agent) +│ ├── mapping.yaml.example ← 唯一决策配置(归属/槽位/套餐 + carrier/package/agent) │ └── legacy_dsn.yaml.example← 奇成 MySQL 只读 DSN └── output/ ← 生成的 SQL 与日志 ``` @@ -40,14 +45,14 @@ scripts/migration/ ### 3.1 业务方提供数据 -把两个 CSV 放到 `resources/` 下(命名为 `cards.csv` / `devices.csv`): +把 CSV 放到 `resources/` 下: - **cards.csv**:`iccid_19`/`iccid_20` 至少填一个;仅填 `iccid_20` 时脚本自动派生前 19 位,但奇成主表只按 20 位精确查询;`is_industry` 可选 -- **devices.csv**:`imei` / `virtual_no` 至少填一个;`device_name` / `device_model` / `device_type` / `max_sim_slots` 可选;`sim_iccid_*` 引用 `cards.csv` 中的 ICCID(可填 19 或 20 位;若 19 位前缀对应多张 20 位卡,必须填完整 20 位) +- **devices.csv** 可选。只有卡时不要放这个文件;有设备时再提供:`imei` / `virtual_no` 至少填一个;`device_name` / `device_model` / `device_type` / `max_sim_slots` 可选;`sim_iccid_*` 引用 `cards.csv` 中的 ICCID;`current_slot` / `package_source_slot` 可选,空值由 `mapping.yaml` 批量规则填充 字段说明详见 [`resources/README.md`](resources/README.md)。 -### 3.2 我们扫描 + 填映射 +### 3.2 扫描 + 填映射 ```bash pip install -r requirements.txt @@ -61,18 +66,102 @@ python3 scan_legacy.py vim config/mapping.yaml ``` -`scan_legacy.py` 可以反复跑(增量合并),已填的 `target_*` 不会被覆盖。 +`scan_legacy.py` 可以反复跑(增量合并),已填的 `target_*` 不会被覆盖。写回的 `mapping.yaml` +会带中文注释,说明哪些字段来自奇成、哪些字段需要人工填写。 -### 3.3 生成 SQL +### 3.3 只有卡的执行顺序 + +只有卡时,`resources/` 下只放 `cards.csv`,不要放 `devices.csv`。 ```bash -python3 migrate_assets.py # → output/step1_*.sql + errors.csv -python3 migrate_runtime.py # → output/step2_*.sql + warnings.csv +python3 scan_legacy.py +# 填 config/mapping.yaml: +# - ownership_rules.standalone_card.mode +# - ownership_rules.default_target_shop_code (mode=default_shop 时必填) +# - agents[].target_shop_code (mode=legacy_agent 时按奇成代理分店铺) +# - carriers[].target_* +# - packages[].target_package_id + +python3 migrate_assets.py ``` -如果 `errors.csv` 有内容,意味着 `mapping.yaml` 还没填完整、或者某些 ICCID 在奇成查不到,处理完再重跑。 +先审核: -### 3.4 测试环境 / 线上执行 +- `output/ownership_resolution.csv`:只会有独立卡归属决策。 +- `output/errors.csv`:非空时不要执行 SQL。 +- `output/summary.txt`:确认独立卡数量、分配卡数。 + +再按编号执行 step1。纯卡迁移时 `step1_02_devices.sql` 和 `step1_05_sim_bindings.sql` +只包含文件头或空操作,可以执行也可以跳过;为了减少漏执行,建议仍按编号执行: + +```bash +psql -h $HOST -1 -f output/step1_01_iot_cards.sql +psql -h $HOST -1 -f output/step1_02_devices.sql +psql -h $HOST -1 -f output/step1_03_asset_identifiers.sql +psql -h $HOST -1 -f output/step1_04_asset_wallets.sql +psql -h $HOST -1 -f output/step1_05_sim_bindings.sql +psql -h $HOST -1 -f output/step1_06_shop_allocations.sql +``` + +step1 执行完成后再生成并执行 step2: + +```bash +python3 migrate_runtime.py +``` + +审核 `package_resolution.csv` / `errors.csv` / `warnings.csv` / `summary.txt` 后,按编号执行: + +```bash +psql -h $HOST -1 -f output/step2_01_migration_orders.sql +psql -h $HOST -1 -f output/step2_02_package_usages.sql +psql -h $HOST -1 -f output/step2_03_card_data_usage_update.sql +psql -h $HOST -1 -f output/step2_04_agent_wallet_init.sql +psql -h $HOST -1 -f output/step2_05_agent_wallet_transactions.sql +psql -h $HOST -1 -f output/step2_06_asset_series_update.sql +``` + +### 3.4 卡 + 设备的执行顺序 + +有设备时,同时准备 `cards.csv` 和 `devices.csv`。所有设备槽位里引用的 ICCID 必须存在于 +`cards.csv`。 + +```bash +python3 scan_legacy.py +# 填 config/mapping.yaml: +# - ownership_rules.device.mode/current_slot/package_source_slot +# - ownership_rules.standalone_card.mode +# - agents[].target_shop_code (mode=legacy_agent 时按奇成代理分店铺) +# - overrides.devices/cards (如有特殊设备或特殊卡) +# - carriers[].target_* +# - packages[].target_package_id + +python3 migrate_assets.py +``` + +先审核: + +- `ownership_resolution.csv`:核对每台设备归属、当前槽位、旧套餐读取槽位;核对独立卡归属。 +- `errors.csv`:非空时不要执行 SQL。 +- `summary.txt`:确认卡数、设备数、分配数量。 + +再执行 step1,然后生成 step2,审核 `package_resolution.csv`,最后执行 step2。命令顺序与 +“只有卡”相同。 + +### 3.5 生成 SQL 的审核要求 + +```bash +python3 migrate_assets.py # → output/step1_*.sql + ownership_resolution.csv + errors.csv +python3 migrate_runtime.py # → output/step2_*.sql + package_resolution.csv + errors.csv + warnings.csv +``` + +执行 SQL 前必须先审核: + +- `ownership_resolution.csv`:核对每个设备/独立卡的目标店铺、决策来源、当前槽位和旧套餐读取槽位 +- `package_resolution.csv`:核对 active/pending 套餐、目标套餐、优先级、跳过原因和阻断原因 +- `summary.txt`:核对分配数、active/pending 套餐数、跳过数、错误数 +- `errors.csv`:非空时不执行 SQL,先修正配置或输入后重跑 + +### 3.6 目标库执行 ```bash # 按编号顺序执行 @@ -112,6 +201,28 @@ go run ./cmd/migration-finalize --batch-no MIGRATION-QICHENG-YYYYMMDD --apply migration_user_id: 0 # 后台超级管理员 ID,作为所有 creator/updater migration_batch_no: "MIGRATION-QICHENG-YYYYMMDD" # 本次迁移批次号,线上每批必须唯一 +ownership_rules: + default_target_shop_code: KWTX # default_shop 模式的批量目标店铺;legacy_agent/none 可为 null + device: + mode: none # default_shop / legacy_agent / none;只有卡时保持 none + current_slot: null # 设备当前卡槽,写 is_current;只有卡时 null + package_source_slot: null # 设备旧套餐读取槽位;只有卡时 null + standalone_card: + mode: legacy_agent # default_shop / legacy_agent / none + +package_rules: + migrate_statuses: [active, pending] # 迁移当前生效 + 未生效待生效正式套餐 + +overrides: + devices: + - virtual_no: "862639073940258" + target_shop_code: OTHER + current_slot: 1 + package_source_slot: 1 + cards: + - iccid: "89861590172420360956" + target_shop_code: OTHER + carriers: # 由 scan_legacy 自动发现 legacy_*,你只填 target_* - legacy_account_id: "72493DB64A2A4F01B5E59E49D58A2A34" legacy_account_name: "联通10-1" @@ -131,23 +242,52 @@ series: # 仅保留奇成套餐系列快照,生成器不要求填 legacy_series_name: "国内卡 无预存30天" target_series_id: null -agents: # 决定所有卡/设备的店铺归属 +agents: # legacy_agent 归属模式 + 代理钱包初始化 - legacy_agent_id: "D15EC779767B412A82B8448AC2324F04" legacy_agent_name: "叶发伟" - target_shop_code: "SHOP20260428103424CZZQ" # 留 null = 该代理的卡进平台库存 + target_shop_code: "SHOP20260428103424CZZQ" # legacy_agent 模式下 null=该代理资产进平台库存 ``` ### 店铺归属是怎么决定的 -CSV 里**没有** `target_shop_code` 列,所有归属都从 `mapping.agents` 推导: +CSV 里**没有** `target_shop_code` 列。资产归属由 `mapping.yaml` 显式决定: | 资产 | 归属规则 | |------|---------| -| 独立卡 | 奇成 `tbl_card.agent_id` → `mapping.agents[].target_shop_code` | -| 设备 | 主卡(`sim_iccid_1`)的 agent_id → 同上 | +| 独立卡 | `overrides.cards` 优先,否则 `ownership_rules.standalone_card` | +| 设备 | `overrides.devices` 优先,否则 `ownership_rules.device` | | 设备上的卡 | 跟随设备所属店铺(`shop_id` 与设备一致,设备分配时卡状态同步为已分销) | -如果 agent 在 `mapping.agents` 里 `target_shop_code` 留空 → 该资产进平台库存。 +`ownership_rules.*.mode` 有三种: + +| mode | 含义 | 适用场景 | +|------|------|----------| +| `default_shop` | 全部分到 `default_target_shop_code` | 单店铺批次 | +| `legacy_agent` | 按奇成 `tbl_card.agent_id` 查 `agents[].target_shop_code` | 一批卡分散到多个店铺 | +| `none` | 不分配店铺,进入平台库存 | 先入库后分配 | + +一万张卡分散到多个店铺时,不要写一万条 `overrides.cards`。做法是: + +```yaml +ownership_rules: + default_target_shop_code: null + standalone_card: + mode: legacy_agent + +agents: + - legacy_agent_id: "0072250EBB88422C88A573AB06870D17" + legacy_agent_name: "酷蛙通讯" + target_shop_code: "KWTX" + - legacy_agent_id: "..." + legacy_agent_name: "另一个代理" + target_shop_code: "SHOP_A" +``` + +这样脚本会按每张卡在奇成 `tbl_card.agent_id` 自动分流到不同店铺。设备批次如果也要按代理分店铺,设置 +`ownership_rules.device.mode: legacy_agent`,脚本会用设备 `package_source_slot` 那张绑定卡的 `agent_id` +决定设备归属。 + +设备槽位解析优先级是 `devices.csv` 行级配置 > `overrides.devices` > `ownership_rules.device`。`current_slot` 对应 `tb_device_sim_binding.is_current`;`package_source_slot` 只决定从哪张绑定卡读取奇成旧套餐,新系统套餐仍写到设备资产。 ## 5. 识别迁移数据(后台报表过滤用) @@ -157,9 +297,9 @@ CSV 里**没有** `target_shop_code` 列,所有归属都从 `mapping.agents` 推 | 迁移分配记录 | `WHERE allocation_no LIKE 'MIG-ALLOC-%'` | | 迁移钱包流水 | `WHERE metadata->>'source' = 'qicheng'` | -迁移伪订单号固定为 `MIG-CARD-`: +迁移伪订单号固定为 `MIG--`: -- 20 位卡总长 29,19 位卡总长 28,符合新库 `order_no varchar(30)` 限制。 +- 20 位卡加优先级总长小于 30,符合新库 `order_no varchar(30)` 限制。 - 订单号不包含生成时间,`step2_01_migration_orders.sql` 和 `step2_02_package_usages.sql` 分开重跑也能稳定匹配。 ## 6. 流量用量迁移规则(真用量 / 虚用量) @@ -182,12 +322,12 @@ CSV 里**没有** `target_shop_code` 列,所有归属都从 `mapping.agents` 推 real_used_mb = total_bytes_cnt / (1 + flow_add_discount / 100) ``` -其中 `flow_add_discount` 取自该卡在 `tbl_card_life`(status=1, expire_date 最大)对应的当前生效套餐。 +其中 `flow_add_discount` 取自该卡在 `tbl_card_life`(status=1, type!=1, expire_date 最大)对应的当前生效正式套餐。奇成 `type=1` 是叠加包/加油包,不能作为新系统主套餐迁入。 ### 6.3 精度与近似 -- **跨多个套餐周期的累计**:`total_bytes_cnt` 是按各时段对应套餐虚比加权累加的,这里只用"当前生效套餐"的虚比反算,**有误差但可接受**(奇成没有按套餐分段的用量明细,无更好办法)。 -- **无当前生效套餐却有 `total_bytes_cnt`**:按 `flow_add_discount=0` 处理(真=虚),并写入 `warnings.csv`(`kind=usage_without_active_package`)供人工核查。 +- **跨多个套餐周期的累计**:`total_bytes_cnt` 是按各时段对应套餐虚比加权累加的,这里只用"当前生效正式套餐"的虚比反算,**有误差但可接受**(奇成没有按套餐分段的用量明细,无更好办法)。 +- **无当前生效正式套餐却有 `total_bytes_cnt`**:按 `flow_add_discount=0` 处理(真=虚),并写入 `warnings.csv`(`kind=usage_without_active_package`)供人工核查。 - **整数取整规则**(`data_usage_mb` 是 BIGINT): - `real_used_mb = 0` → 写 0 - `0 < real_used_mb < 1` → **上取整为 1**,避免"明明用过的卡显示 0" @@ -196,7 +336,7 @@ real_used_mb = total_bytes_cnt / (1 + flow_add_discount / 100) ### 6.4 套餐使用记录的虚流量快照 -`tb_package_usage` 的 `virtual_total_mb_snapshot` / `display_gain_ratio_snapshot` / `enable_virtual_data_snapshot` 三个字段**全部从新系统 `tb_package` 当场快照**,与奇成无关: +`tb_package_usage` 的 `virtual_total_mb_snapshot` / `display_gain_ratio_snapshot` / `enable_virtual_data_snapshot` 三个字段**全部从新系统 `tb_package` 当场快照**,与奇成无关。迁移脚本只允许 `target_package_id` 指向新系统 `package_type='formal'` 的套餐;如果误指向 `addon` 加油包,生成的 step2 SQL 会直接报错并回滚: | 快照字段 | 取值 | |---|---| @@ -206,9 +346,20 @@ real_used_mb = total_bytes_cnt / (1 + flow_add_discount / 100) | `data_limit_mb` | 始终 `real_data_mb`(真总量) | | `data_usage_mb` | `LEAST(real_used_mb_floor, virtual_total_mb_snapshot)`(防卫上限) | -### 6.5 设备绑定卡的套餐归属 +### 6.5 当前生效 + 待生效套餐 -设备上的卡在新系统运行态会按设备维度判断有效套餐和复机条件，因此迁移时只用设备当前槽位卡（默认 `sim_iccid_1`）作为奇成套餐来源，生成一条 `usage_type=device`、`device_id=设备ID` 的套餐使用记录。其他槽位绑定卡不重复生成主套餐，避免同一设备出现多条并列生效主套餐。 +脚本读取奇成 `tbl_card_life` 当前/历史正式套餐和 `tbl_next_month_card_life` 次月待生效套餐: + +- `type=1` 加油包不迁移,写入 `package_resolution.csv` 跳过原因 +- 已过期套餐不迁移,写入跳过原因 +- active 写 `tb_package_usage.status=1`,保留 `activated_at` / `expires_at` 和用量快照 +- pending 写 `tb_package_usage.status=0`,保留 `expires_at`,用量为 0,按 `start_date ASC, expire_date ASC, legacy_life_id ASC` 生成稳定 `priority` +- 同一资产出现多个 active 正式套餐时写入 `errors.csv` 并阻断该资产套餐 SQL +- legacy 套餐未映射或目标套餐缺失时写入 `errors.csv` 并阻断对应套餐 SQL + +### 6.6 设备绑定卡的套餐归属 + +设备上的卡在新系统运行态会按设备维度判断有效套餐和复机条件，因此迁移时只用设备 `package_source_slot` 对应绑定卡读取奇成旧套餐生命周期。生成的新系统套餐使用记录仍然是 `usage_type=device`、`device_id=设备ID`,不会挂到这张卡资产上。其他槽位绑定卡不读取旧套餐,避免同一设备重复迁入多份主套餐。 设备和卡的 `series_id` 以新系统目标套餐为准：脚本通过 `mapping.packages[].target_package_id` 对应的 `tb_package.series_id` 写入资产，并额外生成 `step2_06_asset_series_update.sql`，用于修复已经导入但因 `ON CONFLICT DO NOTHING` 没有被 step1 更新到的资产。 @@ -216,6 +367,8 @@ real_used_mb = total_bytes_cnt / (1 + flow_add_discount / 100) - ❗ 所有奇成查询走 `SET TRANSACTION READ ONLY`,严禁误写。 - ❗ 所有 INSERT 用 `ON CONFLICT DO NOTHING`,严禁更新已有数据。 +- ❗ 目标店铺和目标套餐都有 SQL 前置校验,缺失时直接报错并回滚当前 SQL 文件。 +- ❗ `errors.csv` 非空时必须先修复后重跑,不要执行已生成 SQL。 - ❗ 伪订单的 `payment_status=2 commission_status=2 commission_result=2`,避免被支付回调、佣金调度器扫到。 - ❗ 迁移卡默认 `status=1(在库)` / `activation_status=0(未激活)`,激活由后续业务流程驱动。 @@ -228,9 +381,13 @@ real_used_mb = total_bytes_cnt / (1 + flow_add_discount / 100) | `errors.csv` 出现 `carrier_target_missing` | `mapping.yaml.carriers` 里有这一项但 `target_*` 还是 null,手动填 | | `errors.csv` 出现 `not_in_legacy` | 该 ICCID 在奇成 `tbl_card` 查不到,业务方确认 | | `errors.csv` 出现 `legacy_duplicate` | 同一 ICCID 在奇成命中多条 `tbl_card`,业务方人工确认是哪张物理卡 | +| `errors.csv` 出现 `current_slot_empty` | `current_slot` 指向的设备槽位无卡,修正 devices.csv 或 mapping.yaml | +| `errors.csv` 出现 `package_source_slot_empty` | `package_source_slot` 指向的设备槽位无卡,修正后重跑 | +| `errors.csv` 出现 `package_not_in_mapping` | 可迁移 legacy 套餐未配置 `target_package_id`,补齐 mapping.yaml | +| `errors.csv` 出现 `multiple_active_packages` | 同一资产多个 active 主套餐,需人工裁决后用配置或数据修正 | | `errors.csv` 出现 `carrier_length_conflict` | 业务方填的 `iccid_20` 与奇成查到的运营商不匹配(CTCC 不该有 20 位 / 非 CTCC 必须有 20 位) | | `errors.csv` 出现 `iccid_20_prefix_mismatch` | `iccid_20` 前 19 位 ≠ `iccid_19`,业务方校对 | | `warnings.csv` 出现 `agent_no_legacy` | `mapping.agents` 配了但奇成没有 tbl_agent 记录,业务方确认 | -| `warnings.csv` 出现 `usage_without_active_package` | 该卡奇成 `total_bytes_cnt > 0` 但 `tbl_card_life` 中没有 status=1 的当前生效套餐,无法精确反算真用量(按真=虚处理),详见 §6.3 | +| `warnings.csv` 出现 `usage_without_active_package` | 该卡奇成 `total_bytes_cnt > 0` 但 `tbl_card_life` 中没有 status=1 且 type!=1 的当前生效正式套餐,无法精确反算真用量(按真=虚处理),详见 §6.3 | | 线上报 `null value in column "creator"` | `migration_user_id` 是 0,某些约束可能拒绝;改成真实管理员 ID | -| 线上报 `null shop_id` | `mapping.agents.target_shop_code` 指向的店铺在新库不存在,先建店铺 | +| 线上报目标店铺不存在 | `ownership_rules` 或 `overrides` 的 shop_code 在新库不存在,先建店铺或修正配置 | diff --git a/scripts/migration/config/mapping.yaml.example b/scripts/migration/config/mapping.yaml.example index 1652f9a..51a6260 100644 --- a/scripts/migration/config/mapping.yaml.example +++ b/scripts/migration/config/mapping.yaml.example @@ -15,6 +15,43 @@ migration_user_id: 0 # 本次迁移批次号。线上每次迁移必须使用独立批次,用于导入后收尾、验收和回滚定位。 migration_batch_no: "MIGRATION-QICHENG-YYYYMMDD" +# ============== 资产归属与槽位规则 ============== +# mapping.yaml 是迁移决策源。奇成 agent_id 不会默认参与归属;只有显式 legacy_agent 模式才读取 agents。 +ownership_rules: + # default_shop 模式的批量目标店铺。legacy_agent/none 模式可保持 null。 + default_target_shop_code: KWTX + device: + # default_shop=默认店铺;legacy_agent=按旧套餐读取槽位的绑定卡 agent_id 分店铺;none=进平台库存 + mode: default_shop + # 批量默认当前槽位,用于 tb_device_sim_binding.is_current + current_slot: 2 + # 批量默认旧套餐读取槽位。新系统套餐挂设备,这里只决定从奇成哪张绑定卡读套餐生命周期 + package_source_slot: 2 + standalone_card: + # default_shop=默认店铺;legacy_agent=按卡 agent_id 分店铺;none=进平台库存 + mode: legacy_agent + +# ============== 套餐迁移规则 ============== +package_rules: + # active=当前生效正式套餐,pending=未生效待生效正式套餐 + migrate_statuses: + - active + - pending + # 可选:也可把套餐映射写在这里;scan_legacy 仍会增量维护顶层 packages。 + packages: [] + +# ============== 少量资产级覆盖 ============== +# 覆盖优先级:devices.csv 行级槽位 > overrides.devices 槽位 > ownership_rules.device 默认槽位。 +overrides: + devices: + - virtual_no: "862639073940258" + target_shop_code: OTHER + current_slot: 1 + package_source_slot: 1 + cards: + - iccid: "89861590172420360956" + target_shop_code: OTHER + # ============== 运营商映射 ============== # scan_legacy 会按"奇成 tbl_card.account_id/account_name"分组写入,你只需填 target_* # legacy_category 只是奇成运营商大类,用于人工判断 target_carrier_type,不要把它当具体运营商。 @@ -41,10 +78,11 @@ series: legacy_series_name: "国内卡 无预存30天" target_series_id: null # 可选参考字段,生成 SQL 不依赖这里 -# ============== 代理-店铺映射 ============== -# scan_legacy 会按"奇成 tbl_card.agent_id"分组写入,你只需填 target_shop_code -# 留空(null)的代理:其卡进平台库存(除非 cards.csv 里显式指定 target_shop_code) +# ============== 代理-店铺映射(legacy_agent 归属 + 代理钱包) ============== +# scan_legacy 会按"奇成 tbl_card.agent_id"分组写入。 +# standalone_card/device mode=legacy_agent 时,资产按这里的 target_shop_code 分配。 +# 不使用 legacy_agent 时,这里只影响 migrate_runtime.py 的代理钱包初始化。 agents: - legacy_agent_id: "D15EC779767B412A82B8448AC2324F04" legacy_agent_name: "叶发伟" - target_shop_code: null # 新库 tb_shop.shop_code,null=不归任何店铺 + target_shop_code: null # 新库 tb_shop.shop_code;legacy_agent 模式下 null=该代理资产进平台库存 diff --git a/scripts/migration/lib/csv_loader.py b/scripts/migration/lib/csv_loader.py index 113572c..d960320 100644 --- a/scripts/migration/lib/csv_loader.py +++ b/scripts/migration/lib/csv_loader.py @@ -1,9 +1,10 @@ """cards.csv / devices.csv 加载与跨文件一致性校验。 -新设计:CSV 只承载"业务方决策"—— - - cards.csv 只读 iccid + 可选覆盖列(target_shop_code、is_industry) +新设计:CSV 只承载"资产清单和行级槽位决策"—— + - cards.csv 只读 iccid + is_industry - devices.csv 读设备元数据 + sim_iccid_* - 卡的运营商、套餐、msisdn、agent_id 等元数据全部从奇成查(见 legacy_query.fetch_card_meta) + - 归属、默认当前槽位、默认旧套餐读取槽位从 mapping.yaml 解析 """ from __future__ import annotations @@ -47,7 +48,7 @@ class DeviceRow: """devices.csv 单行的解析结果。 简化为业务方核心关心的列:imei / virtual_no / device_* / max_sim_slots / sim_iccid_*。 - 店铺归属由"主卡 sim_iccid_1 的 agent_id → mapping.agents.target_shop_code"推导。 + 店铺归属、批量默认 current_slot 和旧套餐读取槽位 package_source_slot 由 mapping.yaml 决定。 其他字段(manufacturer 等)直接用默认值,不暴露给业务方。 """ @@ -60,6 +61,9 @@ class DeviceRow: device_type: str = "" max_sim_slots: int = 1 # 不填时按非空 sim_iccid_* 的最大槽位推导 current_slot: int = 1 # 默认 slot 1 为当前生效 + package_source_slot: int = 1 # 设备旧套餐读取槽位 + current_slot_source: str = "default" + package_source_slot_source: str = "default" @dataclass @@ -123,7 +127,7 @@ def _parse_int(raw: str, default: int = 0) -> int: # ---------------- 主入口 ---------------- -def load_assets(resources_dir: Path) -> tuple[list[CardRow], list[DeviceRow], list[ErrorRow]]: +def load_assets(resources_dir: Path, mapping=None) -> tuple[list[CardRow], list[DeviceRow], list[ErrorRow]]: """读 cards.csv + devices.csv,做格式校验与跨文件一致性校验。 返回:(合法卡列表, 合法设备列表, 异常行列表)。 @@ -139,6 +143,8 @@ def load_assets(resources_dir: Path) -> tuple[list[CardRow], list[DeviceRow], li devices_path = resources_dir / "devices.csv" if devices_path.exists(): _load_devices(devices_path, result) + if mapping is not None: + _apply_device_slot_rules(result, mapping) _cross_check(result) @@ -276,6 +282,23 @@ def _load_devices(path: Path, result: LoadResult) -> None: ) continue + current_slot, current_slot_source = _parse_slot_with_source(raw.get("current_slot", ""), 1) + if current_slot < 1 or current_slot > 4: + result.errors.append( + ErrorRow("devices.csv", idx, "current_slot", raw.get("current_slot", ""), "invalid_slot", + "current_slot 必须是 1-4 的整数") + ) + continue + package_source_slot, package_source_slot_source = _parse_slot_with_source( + raw.get("package_source_slot", ""), current_slot + ) + if package_source_slot < 1 or package_source_slot > 4: + result.errors.append( + ErrorRow("devices.csv", idx, "package_source_slot", raw.get("package_source_slot", ""), "invalid_slot", + "package_source_slot 必须是 1-4 的整数") + ) + continue + result.devices.append( DeviceRow( line_no=idx, @@ -286,10 +309,63 @@ def _load_devices(path: Path, result: LoadResult) -> None: device_type=device_type, sim_iccids=sim_iccids, max_sim_slots=max_sim_slots, + current_slot=current_slot, + package_source_slot=package_source_slot, + current_slot_source=current_slot_source, + package_source_slot_source=package_source_slot_source, ) ) +def _parse_slot_with_source(raw: str, default: int) -> tuple[int, str]: + """解析槽位并返回来源。""" + v = _normalize_cell(raw) + if not v: + return default, "default" + try: + return int(v), "row" + except ValueError: + return 0, "row" + + +def _apply_device_slot_rules(result: LoadResult, mapping) -> None: + """按 行配置 > 覆盖项 > 批量默认值 解析设备槽位。""" + for device in result.devices: + override = mapping.lookup_device_override(device.virtual_no) + if device.current_slot_source != "row": + if override and override.current_slot is not None: + device.current_slot = override.current_slot + device.current_slot_source = "override" + elif mapping.ownership_rules.device.current_slot is not None: + device.current_slot = mapping.ownership_rules.device.current_slot + device.current_slot_source = "ownership_rules.device" + if device.package_source_slot_source != "row": + if override and override.package_source_slot is not None: + device.package_source_slot = override.package_source_slot + device.package_source_slot_source = "override" + elif mapping.ownership_rules.device.package_source_slot is not None: + device.package_source_slot = mapping.ownership_rules.device.package_source_slot + device.package_source_slot_source = "ownership_rules.device" + else: + device.package_source_slot = device.current_slot + device.package_source_slot_source = device.current_slot_source + + for field_name, slot in ( + ("current_slot", device.current_slot), + ("package_source_slot", device.package_source_slot), + ): + if slot < 1 or slot > 4: + result.errors.append( + ErrorRow("devices.csv", device.line_no, field_name, str(slot), "invalid_slot", + f"{field_name} 必须是 1-4 的整数") + ) + elif slot > device.max_sim_slots: + result.errors.append( + ErrorRow("devices.csv", device.line_no, field_name, str(slot), "slot_overflow", + f"{field_name}={slot} 超出 max_sim_slots={device.max_sim_slots}") + ) + + def _cross_check(result: LoadResult) -> None: """校验设备与卡之间的一致性,并把 bound_device_virtual_no 回填到 CardRow。 diff --git a/scripts/migration/lib/legacy_query.py b/scripts/migration/lib/legacy_query.py index 83c85ac..0acacf0 100644 --- a/scripts/migration/lib/legacy_query.py +++ b/scripts/migration/lib/legacy_query.py @@ -43,23 +43,43 @@ def _import_pymysql(): @dataclass(frozen=True) class LegacyPackage: - """奇成当前生效套餐快照(取自 tbl_card_life)。""" + """奇成当前生效正式套餐快照(取自 tbl_card_life)。""" iccid: str meal_id: str meal_name: str + meal_type: str start_date: Optional[str] expire_date: Optional[str] flow_size_mb: int # 套餐总量,单位 MB status: int +@dataclass(frozen=True) +class LegacyPackageLifecycle: + """奇成正式套餐生命周期记录。""" + + iccid: str + source_table: str + life_id: str + meal_id: str + meal_name: str + meal_type: str + status: int + start_date: Optional[str] + expire_date: Optional[str] + flow_size_mb: int + stable_sort_key: str + migration_status: str + skip_reason: str = "" + + @dataclass(frozen=True) class LegacyCardUsage: """奇成卡累计流量快照。 奇成 ``tbl_card.total_bytes_cnt`` 存的是经过当前套餐 ``flow_add_discount`` - 加成后的"虚用量",我们这里同时拉取当前生效套餐(``tbl_card_life`` status=1 + 加成后的"虚用量",我们这里同时拉取当前生效正式套餐(``tbl_card_life`` status=1,type!=1 且 ``expire_date`` 最大)的 ``flow_add_discount``,反算回新系统期望的"真用量"。 精度说明:跨多个套餐周期累计时,``total_bytes_cnt`` 是按各时段对应套餐 @@ -70,7 +90,7 @@ class LegacyCardUsage: 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 中找到当前生效套餐 + has_active_package: bool # 是否在 tbl_card_life 中找到当前生效正式套餐 @property def real_used_mb(self) -> Decimal: @@ -135,8 +155,9 @@ class LegacyCardMeta: 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_id: str # 当前生效正式套餐 ID(可空,取自 tbl_card_life status=1,type!=1,expire_date 最大) current_meal_name: str + current_meal_type: str # tbl_card_life.type,1=叠加包/加油包,2=月卡,3=季卡,4=半年卡,5=年卡 current_expire_date: Optional[str] # ISO 字符串 @@ -217,7 +238,7 @@ def _to_mb_decimal(value) -> Decimal: def fetch_current_packages(conn, iccids: Iterable[str]) -> dict[str, LegacyPackage]: - """查每张卡的当前生效套餐:status=1 且 expire_date 最大。 + """查每张卡的当前生效主套餐:status=1、type!=1 且 expire_date 最大。 返回: iccid → LegacyPackage(找不到的 iccid 不在结果中) """ @@ -231,6 +252,7 @@ def fetch_current_packages(conn, iccids: Iterable[str]) -> dict[str, LegacyPacka iccid_mark, meal_id, meal_name, + COALESCE(CAST(type AS CHAR), '') AS meal_type, start_date, expire_date, COALESCE(flow_size, 0) AS flow_size, @@ -238,6 +260,7 @@ def fetch_current_packages(conn, iccids: Iterable[str]) -> dict[str, LegacyPacka FROM tbl_card_life WHERE iccid_mark IN %s AND status = 1 + AND COALESCE(CAST(type AS CHAR), '') <> '1' ORDER BY iccid_mark ASC, expire_date DESC """ seen: set[str] = set() @@ -253,6 +276,7 @@ def fetch_current_packages(conn, iccids: Iterable[str]) -> dict[str, LegacyPacka iccid=iccid, meal_id=row.get("meal_id") or "", meal_name=row.get("meal_name") or "", + meal_type=str(row.get("meal_type") 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), @@ -261,11 +285,172 @@ def fetch_current_packages(conn, iccids: Iterable[str]) -> dict[str, LegacyPacka return out +def fetch_package_lifecycles(conn, iccid_pairs: Iterable[tuple]) -> dict[str, list[LegacyPackageLifecycle]]: + """查当前生效和未生效待生效正式套餐生命周期。 + + 只返回可迁移范围(active/pending)和需要在审核文件中说明的跳过记录。 + 加油包(type=1)和已过期记录不生成套餐 SQL,但保留 skip_reason 供审核产物输出。 + """ + pairs = _normalize_iccid_pairs(iccid_pairs) + out: dict[str, list[LegacyPackageLifecycle]] = {} + if not pairs: + return out + + query_to_full: dict[str, str] = {} + for i19, i20, allow_i19_lookup in pairs: + card_key = i20 or i19 + query_to_full[card_key] = card_key + if i20 and allow_i19_lookup: + query_to_full[i19] = card_key + if not i20: + query_to_full[i19] = card_key + iccid_list = sorted(query_to_full.keys()) + + sql = """ + SELECT + 'tbl_card_life' AS source_table, + COALESCE(CAST(id AS CHAR), '') AS life_id, + iccid_mark, + COALESCE(meal_id, '') AS meal_id, + COALESCE(meal_name, '') AS meal_name, + COALESCE(CAST(type AS CHAR), '') AS meal_type, + status, + start_date, + expire_date, + COALESCE(flow_size, 0) AS flow_size + FROM tbl_card_life + WHERE iccid_mark IN %s + ORDER BY + iccid_mark ASC, + COALESCE(start_date, '1970-01-01') ASC, + COALESCE(expire_date, '9999-12-31') ASC, + id ASC + """ + with conn.cursor() as cur: + for batch in _chunks(iccid_list): + cur.execute(sql, (tuple(batch),)) + for row in cur.fetchall(): + legacy_iccid = row.get("iccid_mark") or "" + if not legacy_iccid: + continue + iccid = query_to_full.get(legacy_iccid) + if not iccid: + continue + start = row.get("start_date") + expire = row.get("expire_date") + start_str = start.strftime("%Y-%m-%d %H:%M:%S") if start else None + expire_str = expire.strftime("%Y-%m-%d %H:%M:%S") if expire else None + meal_type = str(row.get("meal_type") or "") + raw_status = int(row.get("status") or 0) + migration_status, skip_reason = _classify_package_lifecycle(meal_type, raw_status, start, expire) + life_id = str(row.get("life_id") or "") + stable_sort_key = "|".join([start_str or "", expire_str or "", life_id]) + item = LegacyPackageLifecycle( + iccid=iccid, + source_table=row.get("source_table") or "tbl_card_life", + life_id=life_id, + meal_id=row.get("meal_id") or "", + meal_name=row.get("meal_name") or "", + meal_type=meal_type, + status=raw_status, + start_date=start_str, + expire_date=expire_str, + flow_size_mb=int(row.get("flow_size") or 0), + stable_sort_key=stable_sort_key, + migration_status=migration_status, + skip_reason=skip_reason, + ) + out.setdefault(iccid, []).append(item) + + next_sql = """ + SELECT + 'tbl_next_month_card_life' AS source_table, + COALESCE(CAST(id AS CHAR), '') AS life_id, + iccid_mark, + COALESCE(meal_id, '') AS meal_id, + COALESCE(meal_name, '') AS meal_name, + COALESCE(CAST(order_type AS CHAR), '') AS meal_type, + status, + effect_begin_time AS start_date, + effect_complete_time AS expire_date, + COALESCE(flow_size, 0) AS flow_size + FROM tbl_next_month_card_life + WHERE iccid_mark IN %s + ORDER BY + iccid_mark ASC, + COALESCE(effect_begin_time, '1970-01-01') ASC, + COALESCE(effect_complete_time, '9999-12-31') ASC, + id ASC + """ + with conn.cursor() as cur: + for batch in _chunks(iccid_list): + cur.execute(next_sql, (tuple(batch),)) + for row in cur.fetchall(): + legacy_iccid = row.get("iccid_mark") or "" + if not legacy_iccid: + continue + iccid = query_to_full.get(legacy_iccid) + if not iccid: + continue + start = row.get("start_date") + expire = row.get("expire_date") + start_str = start.strftime("%Y-%m-%d %H:%M:%S") if start else None + expire_str = expire.strftime("%Y-%m-%d %H:%M:%S") if expire else None + raw_status = int(row.get("status") or 0) + migration_status, skip_reason = _classify_next_month_lifecycle(raw_status) + life_id = str(row.get("life_id") or "") + stable_sort_key = "|".join([start_str or "", expire_str or "", f"next:{life_id}"]) + out.setdefault(iccid, []).append(LegacyPackageLifecycle( + iccid=iccid, + source_table=row.get("source_table") or "tbl_next_month_card_life", + life_id=life_id, + meal_id=row.get("meal_id") or "", + meal_name=row.get("meal_name") or "", + meal_type=str(row.get("meal_type") or ""), + status=raw_status, + start_date=start_str, + expire_date=expire_str, + flow_size_mb=int(row.get("flow_size") or 0), + stable_sort_key=stable_sort_key, + migration_status=migration_status, + skip_reason=skip_reason, + )) + return out + + +def _classify_package_lifecycle(meal_type: str, status: int, start_date, expire_date) -> tuple[str, str]: + """把奇成生命周期记录分类为 active/pending/skipped。""" + if meal_type == "1": + return "skipped", "加油包不迁移" + + from datetime import datetime + + now = datetime.now() + if expire_date and expire_date <= now: + return "skipped", "已过期套餐不迁移" + if status == 1: + if start_date and start_date > now: + return "pending", "" + return "active", "" + if start_date and start_date > now: + return "pending", "" + return "skipped", f"奇成状态 {status} 不在迁移范围" + + +def _classify_next_month_lifecycle(status: int) -> tuple[str, str]: + """把奇成次月待生效套餐记录分类为 pending/skipped。""" + if status == 1: + return "pending", "" + if status == 2: + return "skipped", "次月待生效记录已插入 tbl_card_life,避免重复迁移" + return "skipped", f"奇成次月待生效状态 {status} 不在迁移范围" + + def fetch_card_usage(conn, iccid_pairs: Iterable[tuple]) -> dict[str, LegacyCardUsage]: - """查每张卡的累计已用流量,并同步取当前生效套餐的 flow_add_discount。 + """查每张卡的累计已用流量,并同步取当前生效正式套餐的 flow_add_discount。 奇成 ``tbl_card.total_bytes_cnt`` 是虚用量(已被套餐虚比加成)。 - 这里 JOIN 当前生效套餐(``tbl_card_life`` status=1 且 ``expire_date`` 最大) + 这里 JOIN 当前生效正式套餐(``tbl_card_life`` status=1,type!=1 且 ``expire_date`` 最大) + ``tbl_set_meal`` 取 ``flow_add_discount``,放进 ``LegacyCardUsage``; 上层通过 ``real_used_mb_floor`` 反算真用量写入新系统。 @@ -312,12 +497,15 @@ def fetch_card_usage(conn, iccid_pairs: Iterable[tuple]) -> dict[str, LegacyCard INNER JOIN ( SELECT iccid_mark, MAX(expire_date) AS max_expire FROM tbl_card_life - WHERE status = 1 AND iccid_mark IN %s + WHERE status = 1 + AND COALESCE(CAST(type AS CHAR), '') <> '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 + AND COALESCE(CAST(cl1.type AS CHAR), '') <> '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) @@ -440,6 +628,7 @@ def fetch_card_meta( COALESCE(vn.vcode, '') AS vcode, COALESCE(cl.meal_id, '') AS meal_id, COALESCE(cl.meal_name, '') AS meal_name, + COALESCE(CAST(cl.meal_type AS CHAR), '') AS meal_type, cl.expire_date AS expire_date FROM tbl_card c LEFT JOIN tbl_vendor_category vc ON vc.category = c.category @@ -448,17 +637,20 @@ def fetch_card_meta( 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 + SELECT cl1.iccid_mark, cl1.meal_id, cl1.meal_name, cl1.type AS meal_type, 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 + WHERE status = 1 + AND COALESCE(CAST(type AS CHAR), '') <> '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 + AND COALESCE(CAST(cl1.type AS CHAR), '') <> '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) @@ -511,6 +703,7 @@ def fetch_card_meta( virtual_no=row.get("vcode") or "", current_meal_id=row.get("meal_id") or "", current_meal_name=row.get("meal_name") or "", + current_meal_type=str(row.get("meal_type") or ""), current_expire_date=expire_str, ) return metas, duplicates diff --git a/scripts/migration/lib/mapping_loader.py b/scripts/migration/lib/mapping_loader.py index b1fff0e..757c553 100644 --- a/scripts/migration/lib/mapping_loader.py +++ b/scripts/migration/lib/mapping_loader.py @@ -7,6 +7,8 @@ """ from __future__ import annotations +import json +import re from dataclasses import dataclass, field from pathlib import Path from typing import Any, Optional @@ -27,6 +29,69 @@ class MissingTargetError(Exception): ) +SHOP_CODE_RE = re.compile(r"^[A-Za-z0-9_-]{2,64}$") +ALLOWED_OWNERSHIP_MODES = {"default_shop", "legacy_agent", "none"} +ALLOWED_PACKAGE_STATUSES = {"active", "pending"} + + +@dataclass(frozen=True) +class OwnershipDeviceRule: + """设备批量归属规则。""" + + mode: str = "none" + current_slot: Optional[int] = None + package_source_slot: Optional[int] = None + + +@dataclass(frozen=True) +class OwnershipStandaloneCardRule: + """独立卡批量归属规则。""" + + mode: str = "none" + + +@dataclass(frozen=True) +class OwnershipRules: + """资产归属决策规则。""" + + default_target_shop_code: Optional[str] = None + device: OwnershipDeviceRule = field(default_factory=OwnershipDeviceRule) + standalone_card: OwnershipStandaloneCardRule = field(default_factory=OwnershipStandaloneCardRule) + + +@dataclass(frozen=True) +class PackageRules: + """套餐迁移规则。""" + + migrate_statuses: tuple[str, ...] = ("active", "pending") + + +@dataclass(frozen=True) +class DeviceOverride: + """设备级覆盖项。""" + + virtual_no: str + target_shop_code: Optional[str] = None + current_slot: Optional[int] = None + package_source_slot: Optional[int] = None + + +@dataclass(frozen=True) +class CardOverride: + """卡级覆盖项。""" + + iccid: str + target_shop_code: Optional[str] = None + + +@dataclass(frozen=True) +class Overrides: + """少量资产级覆盖项。""" + + devices: dict[str, DeviceOverride] = field(default_factory=dict) + cards: dict[str, CardOverride] = field(default_factory=dict) + + @dataclass(frozen=True) class CarrierMapping: legacy_account_id: str @@ -73,7 +138,7 @@ class SeriesMapping: class AgentMapping: legacy_agent_id: str legacy_agent_name: str - target_shop_code: Optional[str] # 允许 null,表示该代理的卡进平台库存 + target_shop_code: Optional[str] # legacy_agent 模式下允许 null,表示该代理资产进平台库存 @dataclass @@ -82,6 +147,9 @@ class Mapping: migration_user_id: int = 0 migration_batch_no: str = "MIGRATION-QICHENG" + ownership_rules: OwnershipRules = field(default_factory=OwnershipRules) + package_rules: PackageRules = field(default_factory=PackageRules) + overrides: Overrides = field(default_factory=Overrides) carriers: dict[str, CarrierMapping] = field(default_factory=dict) packages: dict[str, PackageMapping] = field(default_factory=dict) series: dict[str, SeriesMapping] = field(default_factory=dict) @@ -108,6 +176,18 @@ class Mapping: code = (agent.target_shop_code or "").strip() return code or None + def lookup_device_override(self, virtual_no: str) -> Optional[DeviceOverride]: + """按设备号查覆盖项。""" + return self.overrides.devices.get(str(virtual_no or "").strip()) + + def lookup_card_override(self, *iccids: str) -> Optional[CardOverride]: + """按完整 ICCID 或 19 位 ICCID 查覆盖项。""" + for iccid in iccids: + item = self.overrides.cards.get(str(iccid or "").strip()) + if item is not None: + return item + return None + # ---------------- 加载 ---------------- @@ -125,10 +205,18 @@ def load_mapping(config_dir: Path) -> Mapping: raw = yaml.safe_load(f) or {} if not isinstance(raw, dict): raise ValueError(f"{path} 顶层必须是 mapping 类型") + if "ownership_rules" not in raw: + raise ValueError( + f"{path} 缺少 ownership_rules。新迁移流程必须显式声明资产归属模式,请参考 " + "config/mapping.yaml.example 添加 ownership_rules、package_rules 和 overrides 后重跑。" + ) m = Mapping( migration_user_id=int(raw.get("migration_user_id", 0)), migration_batch_no=_opt_str(raw.get("migration_batch_no")) or "MIGRATION-QICHENG", + ownership_rules=_load_ownership_rules(raw.get("ownership_rules") or {}), + package_rules=_load_package_rules(raw.get("package_rules") or {}), + overrides=_load_overrides(raw.get("overrides") or {}), ) for item in raw.get("carriers") or []: @@ -147,7 +235,11 @@ def load_mapping(config_dir: Path) -> Mapping: ) m.carriers[c.legacy_account_id] = c - for item in raw.get("packages") or []: + package_items = list(raw.get("packages") or []) + package_rules_raw = raw.get("package_rules") or {} + if isinstance(package_rules_raw, dict): + package_items.extend(package_rules_raw.get("packages") or []) + for item in package_items: p = PackageMapping( legacy_meal_id=str(item["legacy_meal_id"]).strip(), legacy_meal_name=str(item.get("legacy_meal_name") or ""), @@ -167,13 +259,105 @@ def load_mapping(config_dir: Path) -> Mapping: a = AgentMapping( legacy_agent_id=str(item["legacy_agent_id"]).strip(), legacy_agent_name=str(item.get("legacy_agent_name") or ""), - target_shop_code=_opt_str(item.get("target_shop_code")), + target_shop_code=_opt_shop_code(item.get("target_shop_code"), "agents[].target_shop_code"), ) m.agents[a.legacy_agent_id] = a return m +def _load_ownership_rules(raw: dict) -> OwnershipRules: + """加载归属规则配置。""" + if not isinstance(raw, dict): + raise ValueError("mapping.yaml.ownership_rules 必须是 mapping 类型") + + default_shop_code = _opt_shop_code(raw.get("default_target_shop_code"), "ownership_rules.default_target_shop_code") + device_raw = raw.get("device") or {} + standalone_raw = raw.get("standalone_card") or {} + if not isinstance(device_raw, dict): + raise ValueError("mapping.yaml.ownership_rules.device 必须是 mapping 类型") + if not isinstance(standalone_raw, dict): + raise ValueError("mapping.yaml.ownership_rules.standalone_card 必须是 mapping 类型") + + device_mode = _mode(device_raw.get("mode"), "ownership_rules.device.mode") + standalone_mode = _mode(standalone_raw.get("mode"), "ownership_rules.standalone_card.mode") + if (device_mode == "default_shop" or standalone_mode == "default_shop") and not default_shop_code: + raise ValueError("mapping.yaml.ownership_rules.default_target_shop_code 不能为空:default_shop 模式必须指定目标店铺码") + + return OwnershipRules( + default_target_shop_code=default_shop_code, + device=OwnershipDeviceRule( + mode=device_mode, + current_slot=_slot_or_none(device_raw.get("current_slot"), "ownership_rules.device.current_slot"), + package_source_slot=_slot_or_none( + device_raw.get("package_source_slot"), "ownership_rules.device.package_source_slot" + ), + ), + standalone_card=OwnershipStandaloneCardRule(mode=standalone_mode), + ) + + +def _load_package_rules(raw: dict) -> PackageRules: + """加载套餐迁移规则。""" + if not isinstance(raw, dict): + raise ValueError("mapping.yaml.package_rules 必须是 mapping 类型") + statuses_raw = raw.get("migrate_statuses") + if statuses_raw is None: + statuses = ("active", "pending") + elif not isinstance(statuses_raw, list): + raise ValueError("mapping.yaml.package_rules.migrate_statuses 必须是列表") + else: + statuses = tuple(str(x).strip() for x in statuses_raw if str(x).strip()) + invalid = [x for x in statuses if x not in ALLOWED_PACKAGE_STATUSES] + if invalid: + raise ValueError(f"mapping.yaml.package_rules.migrate_statuses 仅支持 active/pending,无效值: {invalid}") + if not statuses: + raise ValueError("mapping.yaml.package_rules.migrate_statuses 至少要包含 active 或 pending") + return PackageRules(migrate_statuses=statuses) + + +def _load_overrides(raw: dict) -> Overrides: + """加载资产覆盖项。""" + if not isinstance(raw, dict): + raise ValueError("mapping.yaml.overrides 必须是 mapping 类型") + + devices: dict[str, DeviceOverride] = {} + for idx, item in enumerate(raw.get("devices") or [], start=1): + if not isinstance(item, dict): + raise ValueError(f"mapping.yaml.overrides.devices[{idx}] 必须是 mapping 类型") + virtual_no = _opt_str(item.get("virtual_no")) + if not virtual_no: + raise ValueError(f"mapping.yaml.overrides.devices[{idx}].virtual_no 不能为空") + if virtual_no in devices: + raise ValueError(f"mapping.yaml.overrides.devices virtual_no 重复: {virtual_no}") + devices[virtual_no] = DeviceOverride( + virtual_no=virtual_no, + target_shop_code=_opt_shop_code(item.get("target_shop_code"), f"overrides.devices[{idx}].target_shop_code"), + current_slot=_slot_or_none(item.get("current_slot"), f"overrides.devices[{idx}].current_slot"), + package_source_slot=_slot_or_none( + item.get("package_source_slot"), f"overrides.devices[{idx}].package_source_slot" + ), + ) + + cards: dict[str, CardOverride] = {} + for idx, item in enumerate(raw.get("cards") or [], start=1): + if not isinstance(item, dict): + raise ValueError(f"mapping.yaml.overrides.cards[{idx}] 必须是 mapping 类型") + iccid = _opt_str(item.get("iccid")) + if not iccid: + raise ValueError(f"mapping.yaml.overrides.cards[{idx}].iccid 不能为空") + if len(iccid) not in (19, 20): + raise ValueError(f"mapping.yaml.overrides.cards[{idx}].iccid 必须是 19 或 20 位") + if iccid in cards: + raise ValueError(f"mapping.yaml.overrides.cards iccid 重复: {iccid}") + cards[iccid] = CardOverride( + iccid=iccid, + target_shop_code=_opt_shop_code(item.get("target_shop_code"), f"overrides.cards[{idx}].target_shop_code"), + ) + + return Overrides(devices=devices, cards=cards) + + def merge_and_save(config_dir: Path, mapping: Mapping) -> Path: """把 mapping 写回 config/mapping.yaml,保留人工已填的 target_* 字段。 @@ -181,49 +365,233 @@ def merge_and_save(config_dir: Path, mapping: Mapping) -> Path: """ config_dir.mkdir(parents=True, exist_ok=True) path = config_dir / MAPPING_FILE - data = { - "migration_user_id": mapping.migration_user_id, - "migration_batch_no": mapping.migration_batch_no, - "carriers": [ + path.write_text(_render_commented_mapping_yaml(mapping), encoding="utf-8") + return path + + +def _render_commented_mapping_yaml(mapping: Mapping) -> str: + """渲染带中文注释的 mapping.yaml。 + + PyYAML 写回时会丢弃注释,这里用固定模板承载说明,动态列表仍按扫描结果增量输出。 + """ + lines: list[str] = [ + "# 奇成迁移映射配置(由 scan_legacy.py 生成/增量补全)", + "#", + "# 使用方式:", + "# 1. cards.csv 必填;devices.csv 可选。没有 devices.csv 时就是纯卡迁移。", + "# 2. legacy_* 字段来自奇成扫描,一般不要手工修改。", + "# 3. target_* 字段需要人工填写,否则生成 SQL 时会写入 errors.csv 并阻断对应资产。", + "# 4. 资产归属先看 overrides,再看 ownership_rules;只有显式 legacy_agent 模式才读取 agents。", + "", + "# ============== 全局配置 ==============", + "# 迁移操作者账号 ID,写入 creator/updater/operator_id。线上建议使用后台超级管理员 ID。", + f"migration_user_id: {_yaml_scalar(mapping.migration_user_id)}", + "# 本次迁移批次号。每批线上迁移保持唯一,方便收尾、验收和回滚定位。", + f"migration_batch_no: {_yaml_scalar(mapping.migration_batch_no)}", + "", + "# ============== 资产归属与槽位规则 ==============", + "# default_shop 模式下的批量目标店铺编码;mode=legacy_agent/none 时可保持 null。", + "ownership_rules:", + f" default_target_shop_code: {_yaml_scalar(mapping.ownership_rules.default_target_shop_code)}", + " # 设备规则只在存在 resources/devices.csv 时生效。", + " device:", + " # default_shop=默认店铺;legacy_agent=按旧套餐读取槽位绑定卡的奇成 agent_id 映射店铺;none=平台库存。", + f" mode: {_yaml_scalar(mapping.ownership_rules.device.mode)}", + " # 当前卡槽,写入 tb_device_sim_binding.is_current;纯卡迁移可保持 null。", + f" current_slot: {_yaml_scalar(mapping.ownership_rules.device.current_slot)}", + " # 设备旧套餐读取槽位。新系统套餐挂设备,这里只决定从奇成哪张绑定卡读套餐生命周期;纯卡迁移可保持 null。", + f" package_source_slot: {_yaml_scalar(mapping.ownership_rules.device.package_source_slot)}", + " standalone_card:", + " # default_shop=默认店铺;legacy_agent=按奇成 agent_id 映射店铺;none=平台库存。", + f" mode: {_yaml_scalar(mapping.ownership_rules.standalone_card.mode)}", + "", + "# ============== 套餐迁移规则 ==============", + "package_rules:", + " # active=当前生效正式套餐;pending=未生效待生效正式套餐。", + " migrate_statuses:", + ] + for status in mapping.package_rules.migrate_statuses: + lines.append(f" - {_yaml_scalar(status)}") + + lines.extend([ + "", + "# ============== 少量资产级覆盖 ==============", + "# 覆盖优先级:devices.csv 行级槽位 > overrides.devices 槽位 > ownership_rules.device 默认槽位。", + "overrides:", + " # 设备级特殊处理。virtual_no 对应 devices.csv 的 virtual_no/imei 兜底值。", + ]) + _append_device_overrides(lines, mapping.overrides) + lines.append(" # 独立卡特殊处理。iccid 可填 19 位或 20 位。") + _append_card_overrides(lines, mapping.overrides) + + lines.extend([ + "", + "# ============== 运营商映射 ==============", + "# scan_legacy 按奇成 tbl_card.account_id/account_name 分组写入;只需要人工填写 target_*。", + "# target_carrier_type 只能填 CMCC / CUCC / CTCC / CBN。", + ]) + _append_carriers(lines, mapping) + + lines.extend([ + "", + "# ============== 套餐映射 ==============", + "# scan_legacy 按奇成 tbl_card_life.meal_id 分组写入;target_package_id 必须指向新库正式套餐。", + ]) + _append_packages(lines, mapping) + + lines.extend([ + "", + "# ============== 套餐系列快照(可选) ==============", + "# 仅保留奇成套餐系列供人工参考;资产 series_id 会从 target_package_id 对应的 tb_package.series_id 反查。", + ]) + _append_series(lines, mapping) + + lines.extend([ + "", + "# ============== 代理-店铺映射(legacy_agent 归属 + 代理钱包) ==============", + "# standalone_card/device mode=legacy_agent 时,资产按这里的 target_shop_code 分配。", + "# 不使用 legacy_agent 时,这里只影响 migrate_runtime.py 的代理钱包初始化。", + ]) + _append_agents(lines, mapping) + + return "\n".join(lines) + "\n" + + +def _append_device_overrides(lines: list[str], overrides: Overrides) -> None: + if not overrides.devices: + lines.append(" devices: []") + return + lines.append(" devices:") + for item in sorted(overrides.devices.values(), key=lambda x: x.virtual_no): + lines.extend([ + f" - virtual_no: {_yaml_scalar(item.virtual_no)}", + f" target_shop_code: {_yaml_scalar(item.target_shop_code)}", + f" current_slot: {_yaml_scalar(item.current_slot)}", + f" package_source_slot: {_yaml_scalar(item.package_source_slot)}", + ]) + + +def _append_card_overrides(lines: list[str], overrides: Overrides) -> None: + if not overrides.cards: + lines.append(" cards: []") + return + lines.append(" cards:") + for item in sorted(overrides.cards.values(), key=lambda x: x.iccid): + lines.extend([ + f" - iccid: {_yaml_scalar(item.iccid)}", + f" target_shop_code: {_yaml_scalar(item.target_shop_code)}", + ]) + + +def _append_carriers(lines: list[str], mapping: Mapping) -> None: + if not mapping.carriers: + lines.append("carriers: []") + return + lines.append("carriers:") + for item in sorted(mapping.carriers.values(), key=lambda x: (x.legacy_account_name, x.legacy_account_id)): + lines.extend([ + f" - legacy_account_id: {_yaml_scalar(item.legacy_account_id)}", + f" legacy_account_name: {_yaml_scalar(item.legacy_account_name)}", + f" legacy_category: {_yaml_scalar(item.legacy_category)}", + f" legacy_category_name: {_yaml_scalar(item.legacy_category_name)}", + f" target_carrier_id: {_yaml_scalar(item.target_carrier_id)}", + f" target_carrier_type: {_yaml_scalar(item.target_carrier_type)}", + f" target_carrier_name: {_yaml_scalar(item.target_carrier_name)}", + ]) + + +def _append_packages(lines: list[str], mapping: Mapping) -> None: + if not mapping.packages: + lines.append("packages: []") + return + lines.append("packages:") + for item in sorted(mapping.packages.values(), key=lambda x: x.legacy_meal_id): + lines.extend([ + f" - legacy_meal_id: {_yaml_scalar(item.legacy_meal_id)}", + f" legacy_meal_name: {_yaml_scalar(item.legacy_meal_name)}", + f" target_package_id: {_yaml_scalar(item.target_package_id)}", + ]) + + +def _append_series(lines: list[str], mapping: Mapping) -> None: + if not mapping.series: + lines.append("series: []") + return + lines.append("series:") + for item in sorted(mapping.series.values(), key=lambda x: x.legacy_series_id): + lines.extend([ + f" - legacy_series_id: {_yaml_scalar(item.legacy_series_id)}", + f" legacy_series_name: {_yaml_scalar(item.legacy_series_name)}", + f" target_series_id: {_yaml_scalar(item.target_series_id)}", + ]) + + +def _append_agents(lines: list[str], mapping: Mapping) -> None: + if not mapping.agents: + lines.append("agents: []") + return + lines.append("agents:") + for item in sorted(mapping.agents.values(), key=lambda x: x.legacy_agent_id): + lines.extend([ + f" - legacy_agent_id: {_yaml_scalar(item.legacy_agent_id)}", + f" legacy_agent_name: {_yaml_scalar(item.legacy_agent_name)}", + f" target_shop_code: {_yaml_scalar(item.target_shop_code)}", + ]) + + +def _yaml_scalar(value: Any) -> str: + """把 Python 标量转成稳定 YAML 标量。""" + if value is None: + return "null" + if isinstance(value, bool): + return "true" if value else "false" + if isinstance(value, int): + return str(value) + return json.dumps(str(value), ensure_ascii=False) + + +def _dump_ownership_rules(rules: OwnershipRules) -> dict: + """序列化归属规则。""" + return { + "default_target_shop_code": rules.default_target_shop_code, + "device": { + "mode": rules.device.mode, + "current_slot": rules.device.current_slot, + "package_source_slot": rules.device.package_source_slot, + }, + "standalone_card": { + "mode": rules.standalone_card.mode, + }, + } + + +def _dump_package_rules(rules: PackageRules) -> dict: + """序列化套餐规则。""" + return { + "migrate_statuses": list(rules.migrate_statuses), + } + + +def _dump_overrides(overrides: Overrides) -> dict: + """序列化资产覆盖项。""" + return { + "devices": [ { - "legacy_account_id": c.legacy_account_id, - "legacy_account_name": c.legacy_account_name, - "legacy_category": c.legacy_category, - "legacy_category_name": c.legacy_category_name, - "target_carrier_id": c.target_carrier_id, - "target_carrier_type": c.target_carrier_type, - "target_carrier_name": c.target_carrier_name, + "virtual_no": x.virtual_no, + "target_shop_code": x.target_shop_code, + "current_slot": x.current_slot, + "package_source_slot": x.package_source_slot, } - for c in sorted(mapping.carriers.values(), key=lambda x: (x.legacy_account_name, x.legacy_account_id)) + for x in sorted(overrides.devices.values(), key=lambda item: item.virtual_no) ], - "packages": [ + "cards": [ { - "legacy_meal_id": p.legacy_meal_id, - "legacy_meal_name": p.legacy_meal_name, - "target_package_id": p.target_package_id, + "iccid": x.iccid, + "target_shop_code": x.target_shop_code, } - for p in sorted(mapping.packages.values(), key=lambda x: x.legacy_meal_id) - ], - "series": [ - { - "legacy_series_id": s.legacy_series_id, - "legacy_series_name": s.legacy_series_name, - "target_series_id": s.target_series_id, - } - for s in sorted(mapping.series.values(), key=lambda x: x.legacy_series_id) - ], - "agents": [ - { - "legacy_agent_id": a.legacy_agent_id, - "legacy_agent_name": a.legacy_agent_name, - "target_shop_code": a.target_shop_code, - } - for a in sorted(mapping.agents.values(), key=lambda x: x.legacy_agent_id) + for x in sorted(overrides.cards.values(), key=lambda item: item.iccid) ], } - with path.open("w", encoding="utf-8") as f: - yaml.safe_dump(data, f, allow_unicode=True, sort_keys=False, default_flow_style=False) - return path def load_legacy_dsn(config_dir: Path) -> dict: @@ -253,3 +621,31 @@ def _opt_str(v: Any) -> Optional[str]: return None s = str(v).strip() return s or None + + +def _opt_shop_code(v: Any, field_name: str) -> Optional[str]: + code = _opt_str(v) + if code is None: + return None + if not SHOP_CODE_RE.match(code): + raise ValueError(f"mapping.yaml.{field_name} 店铺码格式无效: {code!r},仅允许 2-64 位字母/数字/_/-") + return code + + +def _mode(v: Any, field_name: str) -> str: + mode = _opt_str(v) or "none" + if mode not in ALLOWED_OWNERSHIP_MODES: + raise ValueError(f"mapping.yaml.{field_name} 仅支持 default_shop/legacy_agent/none,无效值: {mode!r}") + return mode + + +def _slot_or_none(v: Any, field_name: str) -> Optional[int]: + if v is None or v == "": + return None + try: + slot = int(v) + except (TypeError, ValueError) as exc: + raise ValueError(f"mapping.yaml.{field_name} 必须是 1-4 的整数") from exc + if slot < 1 or slot > 4: + raise ValueError(f"mapping.yaml.{field_name} 必须在 1-4 之间") + return slot diff --git a/scripts/migration/lib/sql_builder.py b/scripts/migration/lib/sql_builder.py index 84f5de4..9e73516 100644 --- a/scripts/migration/lib/sql_builder.py +++ b/scripts/migration/lib/sql_builder.py @@ -19,12 +19,13 @@ - 所有 INSERT 通过 SELECT 子查询动态取 id,不硬编码新库 id - 使用 ON CONFLICT (...) WHERE ... DO NOTHING 保证幂等(PG 14+ 支持 partial index 推断) - 字符串 ' 转义为 '' - - 卡的元数据(carrier/agent_id/msisdn/virtual_no/当前套餐)从奇成 fetch_card_meta 取, - cards.csv 只承载 is_industry 覆盖,店铺归属完全由 mapping.agents 推导 + - 卡的运营商、号码等历史事实从奇成 fetch_card_meta 取 + - 资产归属、当前槽位、旧套餐读取槽位由 mapping.yaml 的 ownership_rules / overrides 决定 """ from __future__ import annotations import csv +from dataclasses import dataclass from datetime import datetime from decimal import Decimal from pathlib import Path @@ -37,16 +38,54 @@ from .legacy_query import ( LegacyCardMeta, LegacyCardUsage, LegacyCommissionAccount, + LegacyPackageLifecycle, ) from .mapping_loader import Mapping, MissingTargetError # 写死的默认值(对应原 migration.yaml,实际不需要业务侧关心) -ORDER_NO_PREFIX_CARD = "MIG-CARD-" ALLOCATION_NO_PREFIX = "MIG-ALLOC-" METADATA_SOURCE = "qicheng" STOP_REASON_NOT_REALNAME = "not_realname" STOP_REASON_NO_PACKAGE = "no_package" +@dataclass(frozen=True) +class ResolvedPackage: + """已解析的迁移套餐。""" + + source_iccid: str + asset_type: str + asset_identifier: str + legacy_life_id: str + legacy_meal_id: str + legacy_meal_name: str + legacy_meal_type: str + legacy_status: int + target_package_id: int + migration_status: str + package_usage_status: int + priority: int + start_date: str + expire_date: str + stable_sort_key: str + + +# 完整 ICCID → 多条可迁移套餐 +ResolvedPackageMap = dict[str, list[ResolvedPackage]] + + +@dataclass(frozen=True) +class OwnershipResolution: + """资产归属解析结果。""" + + asset_type: str + asset_identifier: str + source_iccid: str + target_shop_code: str + decision_source: str + should_allocate: bool + reason: str + error_code: str = "" + # ---------------- 通用工具 ---------------- @@ -92,15 +131,105 @@ def _file_header(title: str) -> str: ) -def _resolve_card_shop_code(card: CardRow, meta: LegacyCardMeta, mapping: Mapping) -> str: - """决定卡最终 shop_code:设备上的卡→空(跟设备走);否则按 agent 映射;查不到→平台库存。""" +def _resolve_card_ownership( + card: CardRow, + mapping: Mapping, + card_metas: Optional[dict[str, LegacyCardMeta]] = None, + devices_by_virtual_no: Optional[dict[str, DeviceRow]] = None, +) -> OwnershipResolution: + """解析卡归属。 + + 默认不读取奇成 agent_id;只有 ownership_rules.*.mode 显式配置为 legacy_agent 时, + 才按 mapping.agents[].target_shop_code 做批量多店铺分配。 + """ if card.bound_device_virtual_no: - return "" - if meta and meta.agent_id: - shop = mapping.lookup_agent_shop_code(meta.agent_id) - if shop: - return shop - return "" + device = (devices_by_virtual_no or {}).get(card.bound_device_virtual_no) + if device is None: + return OwnershipResolution( + asset_type="iot_card", + asset_identifier=card.iccid_full, + source_iccid=card.iccid_full, + target_shop_code="", + decision_source="device_missing", + should_allocate=False, + reason="绑定设备不存在,卡不单独分配", + error_code="bound_device_missing", + ) + device_resolution = _resolve_device_ownership(device, mapping, card_metas) + return OwnershipResolution( + asset_type="iot_card", + asset_identifier=card.iccid_full, + source_iccid=card.iccid_full, + target_shop_code=device_resolution.target_shop_code, + decision_source=f"follow_device:{device_resolution.decision_source}", + should_allocate=False, + reason=f"设备绑定卡跟随设备 {device.virtual_no}", + error_code=device_resolution.error_code, + ) + + override = mapping.lookup_card_override(card.iccid_full, card.iccid_19) + if override is not None: + shop_code = override.target_shop_code or "" + return OwnershipResolution( + asset_type="iot_card", + asset_identifier=card.iccid_full, + source_iccid=card.iccid_full, + target_shop_code=shop_code, + decision_source="override", + should_allocate=bool(shop_code), + reason="卡覆盖项指定目标店铺" if shop_code else "卡覆盖项指定进入平台库存", + ) + + rule = mapping.ownership_rules.standalone_card + if rule.mode == "legacy_agent": + meta = (card_metas or {}).get(card.iccid_full) + if meta is None or not meta.agent_id: + return OwnershipResolution( + asset_type="iot_card", + asset_identifier=card.iccid_full, + source_iccid=card.iccid_full, + target_shop_code="", + decision_source="legacy_agent", + should_allocate=False, + reason="奇成卡没有 agent_id,进入平台库存", + ) + shop_code = mapping.lookup_agent_shop_code(meta.agent_id) or "" + return OwnershipResolution( + asset_type="iot_card", + asset_identifier=card.iccid_full, + source_iccid=card.iccid_full, + target_shop_code=shop_code, + decision_source=f"legacy_agent:{meta.agent_id}", + should_allocate=bool(shop_code), + reason="按奇成 agent_id 映射目标店铺" if shop_code else "agent 未配置目标店铺,进入平台库存", + ) + + if rule.mode == "default_shop": + shop_code = mapping.ownership_rules.default_target_shop_code or "" + return OwnershipResolution( + asset_type="iot_card", + asset_identifier=card.iccid_full, + source_iccid=card.iccid_full, + target_shop_code=shop_code, + decision_source="default_shop", + should_allocate=bool(shop_code), + reason="独立卡使用批量默认店铺", + ) + + return OwnershipResolution( + asset_type="iot_card", + asset_identifier=card.iccid_full, + source_iccid=card.iccid_full, + target_shop_code="", + decision_source="none", + should_allocate=False, + reason="独立卡规则指定进入平台库存", + ) + + +def _resolve_card_shop_code(card: CardRow, meta: LegacyCardMeta, mapping: Mapping) -> str: + """兼容旧调用:返回卡最终 shop_code。""" + return _resolve_card_ownership(card, mapping, {card.iccid_full: meta}).target_shop_code def _resolve_device_shop_code( @@ -108,14 +237,75 @@ def _resolve_device_shop_code( card_metas: dict[str, LegacyCardMeta], mapping: Mapping, ) -> str: - """决定设备最终 shop_code:按主卡(slot 1)的 agent 映射;无主卡或映射不到→平台库存。""" - primary_iccid = device.sim_iccids.get(1) or next(iter(device.sim_iccids.values()), "") - if not primary_iccid: - return "" - meta = card_metas.get(primary_iccid) - if not meta or not meta.agent_id: - return "" - return mapping.lookup_agent_shop_code(meta.agent_id) or "" + """兼容旧调用:返回设备最终 shop_code。""" + return _resolve_device_ownership(device, mapping, card_metas).target_shop_code + + +def _resolve_device_ownership( + device: DeviceRow, + mapping: Mapping, + card_metas: Optional[dict[str, LegacyCardMeta]] = None, +) -> OwnershipResolution: + """解析设备归属。""" + override = mapping.lookup_device_override(device.virtual_no) + source_iccid = device.sim_iccids.get(device.package_source_slot) or "" + if override is not None and override.target_shop_code is not None: + shop_code = override.target_shop_code or "" + return OwnershipResolution( + asset_type="device", + asset_identifier=device.virtual_no, + source_iccid=source_iccid, + target_shop_code=shop_code, + decision_source="override", + should_allocate=bool(shop_code), + reason="设备覆盖项指定目标店铺" if shop_code else "设备覆盖项指定进入平台库存", + ) + + rule = mapping.ownership_rules.device + if rule.mode == "legacy_agent": + meta = (card_metas or {}).get(source_iccid) + if meta is None or not meta.agent_id: + return OwnershipResolution( + asset_type="device", + asset_identifier=device.virtual_no, + source_iccid=source_iccid, + target_shop_code="", + decision_source="legacy_agent", + should_allocate=False, + reason="设备旧套餐读取槽位绑定卡没有 agent_id,设备进入平台库存", + ) + shop_code = mapping.lookup_agent_shop_code(meta.agent_id) or "" + return OwnershipResolution( + asset_type="device", + asset_identifier=device.virtual_no, + source_iccid=source_iccid, + target_shop_code=shop_code, + decision_source=f"legacy_agent:{meta.agent_id}", + should_allocate=bool(shop_code), + reason="按设备旧套餐读取槽位绑定卡的奇成 agent_id 映射目标店铺" if shop_code else "agent 未配置目标店铺,设备进入平台库存", + ) + + if rule.mode == "default_shop": + shop_code = mapping.ownership_rules.default_target_shop_code or "" + return OwnershipResolution( + asset_type="device", + asset_identifier=device.virtual_no, + source_iccid=source_iccid, + target_shop_code=shop_code, + decision_source="default_shop", + should_allocate=bool(shop_code), + reason="设备使用批量默认店铺", + ) + + return OwnershipResolution( + asset_type="device", + asset_identifier=device.virtual_no, + source_iccid=source_iccid, + target_shop_code="", + decision_source="none", + should_allocate=False, + reason="设备规则指定进入平台库存", + ) def _initial_polling_stop_reason(card: CardRow) -> str: @@ -130,18 +320,14 @@ def _initial_polling_stop_reason(card: CardRow) -> str: def _device_package_source_iccid(device: DeviceRow) -> str: - """返回设备级套餐迁移使用的来源卡。 + """返回设备级套餐迁移读取奇成旧套餐的绑定卡。 设备套餐在新系统挂到 device_id；奇成套餐仍来自卡生命周期表。 - 优先取当前槽位，缺失时回退 slot1，再回退任意已绑定卡。 + 旧套餐读取槽位由 devices.csv、设备覆盖项或 ownership_rules.device.package_source_slot 决定。 """ if not device.sim_iccids: return "" - return ( - device.sim_iccids.get(device.current_slot) - or device.sim_iccids.get(1) - or next(iter(device.sim_iccids.values()), "") - ) + return device.sim_iccids.get(device.package_source_slot, "") def _target_series_id_expr_for_iccid( @@ -163,10 +349,64 @@ def _target_series_id_expr_for_iccid( return "NULL" return ( "(SELECT series_id FROM tb_package " - f"WHERE id = {pkg_map.target_package_id} AND deleted_at IS NULL LIMIT 1)" + f"WHERE id = {pkg_map.target_package_id} AND package_type = 'formal' AND deleted_at IS NULL LIMIT 1)" ) +def _write_formal_package_guard(f, target_pkg_id: int, legacy_meal_name: str) -> None: + """写目标套餐校验,防止把加油包按主套餐迁入。 + + 迁移脚本绕过 Service 层,必须在 SQL 文件里兜底校验 package_type。 + """ + f.write( + "DO $$\n" + "DECLARE\n" + " v_package_type text;\n" + "BEGIN\n" + f" SELECT package_type INTO v_package_type FROM tb_package WHERE id = {target_pkg_id} AND deleted_at IS NULL;\n" + " IF v_package_type IS NULL THEN\n" + " RAISE EXCEPTION USING MESSAGE = " + f"'迁移套餐映射无效: target_package_id={target_pkg_id} 不存在或已删除';\n" + " END IF;\n" + " IF v_package_type <> 'formal' THEN\n" + " RAISE EXCEPTION USING MESSAGE = " + f"'迁移套餐映射禁止指向加油包: legacy_meal_name=' || {_sql_str(legacy_meal_name)} || " + f"', target_package_id={target_pkg_id}, package_type=' || v_package_type;\n" + " END IF;\n" + "END $$;\n\n" + ) + + +def _write_shop_guards(f, shop_codes: set[str]) -> None: + """写目标店铺存在性校验,防止 shop_code 缺失时静默写入平台库存。""" + for shop_code in sorted(x for x in shop_codes if x): + f.write( + "DO $$\n" + "BEGIN\n" + " IF NOT EXISTS (\n" + " SELECT 1 FROM tb_shop\n" + f" WHERE shop_code = {_sql_str(shop_code)} AND deleted_at IS NULL\n" + " ) THEN\n" + " RAISE EXCEPTION USING MESSAGE = " + f"'迁移目标店铺不存在或已删除: shop_code={shop_code}';\n" + " END IF;\n" + "END $$;\n\n" + ) + + +def _required_shop_codes_for_step1( + cards: list[CardRow], + devices: list[DeviceRow], + mapping: Mapping, + card_metas: Optional[dict[str, LegacyCardMeta]] = None, +) -> set[str]: + """收集资产导入需要校验的目标店铺码。""" + devices_by_virtual_no = _device_by_virtual_no(devices) + codes = {_resolve_device_ownership(d, mapping, card_metas).target_shop_code for d in devices} + codes.update(_resolve_card_ownership(c, mapping, card_metas, devices_by_virtual_no).target_shop_code for c in cards) + return {code for code in codes if code} + + def _device_by_virtual_no(devices: list[DeviceRow]) -> dict[str, DeviceRow]: return {d.virtual_no: d for d in devices} @@ -232,20 +472,22 @@ def write_step1( # 预检 carrier 映射 + 运营商一致性 + 重复命中,把异常卡剔除 valid_cards = _filter_cards_by_carrier(cards, card_metas, duplicate_iccids, mapping, errors) + valid_devices = _filter_devices_by_slots(devices, errors) cards_by_iccid = {c.iccid_full: c for c in valid_cards} - _write_iot_cards(output_dir / "step1_01_iot_cards.sql", valid_cards, devices, card_metas, mapping, user_id) - _write_devices(output_dir / "step1_02_devices.sql", devices, card_metas, mapping, user_id) - _write_asset_identifiers(output_dir / "step1_03_asset_identifiers.sql", valid_cards, devices, card_metas) - _write_asset_wallets(output_dir / "step1_04_asset_wallets.sql", valid_cards, devices) - _write_sim_bindings(output_dir / "step1_05_sim_bindings.sql", devices, cards_by_iccid) + _write_iot_cards(output_dir / "step1_01_iot_cards.sql", valid_cards, valid_devices, card_metas, mapping, user_id) + _write_devices(output_dir / "step1_02_devices.sql", valid_devices, card_metas, mapping, user_id) + _write_asset_identifiers(output_dir / "step1_03_asset_identifiers.sql", valid_cards, valid_devices, card_metas) + _write_asset_wallets(output_dir / "step1_04_asset_wallets.sql", valid_cards, valid_devices) + _write_sim_bindings(output_dir / "step1_05_sim_bindings.sql", valid_devices, cards_by_iccid) _write_shop_allocations( - output_dir / "step1_06_shop_allocations.sql", valid_cards, devices, card_metas, mapping, user_id + output_dir / "step1_06_shop_allocations.sql", valid_cards, valid_devices, card_metas, mapping, user_id ) + _write_ownership_resolution(output_dir / "ownership_resolution.csv", valid_cards, valid_devices, mapping, card_metas) _write_errors(output_dir / "errors.csv", errors) - _write_summary_step1(output_dir / "summary.txt", valid_cards, devices, errors) + _write_summary_step1(output_dir / "summary.txt", valid_cards, valid_devices, errors, mapping, card_metas) def _filter_cards_by_carrier( @@ -306,6 +548,30 @@ def _filter_cards_by_carrier( return valid +def _filter_devices_by_slots(devices: list[DeviceRow], errors: list[ErrorRow]) -> list[DeviceRow]: + """筛掉当前槽位或旧套餐读取槽位无卡的设备。""" + valid: list[DeviceRow] = [] + for d in devices: + blocked = False + if d.current_slot not in d.sim_iccids: + errors.append(ErrorRow( + "devices.csv", d.line_no, "current_slot", str(d.current_slot), + "current_slot_empty", f"current_slot={d.current_slot} 对应 sim_iccid_{d.current_slot} 为空", + )) + blocked = True + if d.package_source_slot not in d.sim_iccids: + errors.append(ErrorRow( + "devices.csv", d.line_no, "package_source_slot", str(d.package_source_slot), + "package_source_slot_empty", + f"package_source_slot={d.package_source_slot} 对应 sim_iccid_{d.package_source_slot} 为空", + )) + blocked = True + if blocked: + continue + valid.append(d) + return valid + + def _write_iot_cards( path: Path, cards: list[CardRow], @@ -317,8 +583,10 @@ def _write_iot_cards( """写 tb_iot_card。元数据从奇成 card_metas + mapping 推导,iccid_19/iccid_20 用业务方两列。""" device_virtual_nos = _device_virtual_no_set(devices) devices_by_virtual_no = _device_by_virtual_no(devices) + shop_codes = _required_shop_codes_for_step1(cards, devices, mapping, card_metas) with path.open("w", encoding="utf-8") as f: f.write(_file_header("step1_01 卡资产导入 (tb_iot_card)")) + _write_shop_guards(f, shop_codes) for c in cards: meta = card_metas[c.iccid_full] carrier = mapping.lookup_carrier(meta.account_id) @@ -328,10 +596,7 @@ def _write_iot_cards( iccid_20_expr = _sql_str(c.iccid_20) if c.iccid_20 else "NULL" card_category = "industry" if c.is_industry else "normal" - shop_code = _resolve_card_shop_code(c, meta, mapping) - if c.bound_device_virtual_no: - device = devices_by_virtual_no.get(c.bound_device_virtual_no) - shop_code = _resolve_device_shop_code(device, card_metas, mapping) if device else "" + shop_code = _resolve_card_ownership(c, mapping, card_metas, devices_by_virtual_no).target_shop_code series_id_expr = _target_series_id_expr_for_iccid(c.iccid_full, card_metas, mapping) card_virtual_no = _card_virtual_no_for_import(meta, device_virtual_nos) is_standalone = "FALSE" if c.bound_device_virtual_no else "TRUE" @@ -403,6 +668,7 @@ def _write_devices( """ with path.open("w", encoding="utf-8") as f: f.write(_file_header("step1_02 设备资产导入 (tb_device)")) + _write_shop_guards(f, _required_shop_codes_for_step1([], devices, mapping, card_metas)) for d in devices: shop_code = _resolve_device_shop_code(d, card_metas, mapping) device_status = 2 if shop_code else 1 @@ -627,13 +893,13 @@ def _write_shop_allocations( """只为带店铺归属的资产生成分配记录;设备上的卡跟随设备分配。""" with path.open("w", encoding="utf-8") as f: f.write(_file_header("step1_06 店铺分配记录 (tb_asset_allocation_record)")) + _write_shop_guards(f, _required_shop_codes_for_step1(cards, devices, mapping, card_metas)) no_ts = _now_for_no() for c in cards: if c.bound_device_virtual_no: continue - meta = card_metas[c.iccid_full] - shop_code = _resolve_card_shop_code(c, meta, mapping) + shop_code = _resolve_card_ownership(c, mapping, card_metas, _device_by_virtual_no(devices)).target_shop_code if not shop_code: continue allocation_no = f"{ALLOCATION_NO_PREFIX}CARD-{no_ts}-{c.line_no}" @@ -685,13 +951,81 @@ def _write_errors(path: Path, errors: list[ErrorRow]) -> None: writer.writerow(e.to_csv_row()) +def _write_ownership_resolution( + path: Path, + cards: list[CardRow], + devices: list[DeviceRow], + mapping: Mapping, + card_metas: dict[str, LegacyCardMeta], +) -> None: + """写资产归属审核文件。""" + devices_by_virtual_no = _device_by_virtual_no(devices) + with path.open("w", encoding="utf-8-sig", newline="") as f: + writer = csv.writer(f) + writer.writerow([ + "asset_type", + "asset_identifier", + "source_iccid", + "target_shop_code", + "decision_source", + "should_allocate", + "current_slot", + "current_slot_source", + "package_source_slot", + "package_source_slot_source", + "reason", + ]) + for d in devices: + item = _resolve_device_ownership(d, mapping, card_metas) + writer.writerow([ + item.asset_type, + item.asset_identifier, + item.source_iccid, + item.target_shop_code, + item.decision_source, + "true" if item.should_allocate else "false", + d.current_slot, + d.current_slot_source, + d.package_source_slot, + d.package_source_slot_source, + item.reason, + ]) + for c in cards: + item = _resolve_card_ownership(c, mapping, card_metas, devices_by_virtual_no) + writer.writerow([ + item.asset_type, + item.asset_identifier, + item.source_iccid, + item.target_shop_code, + item.decision_source, + "true" if item.should_allocate else "false", + "", + "", + "", + "", + item.reason, + ]) + + def write_runtime_input_errors(output_dir: Path, errors: list[ErrorRow]) -> None: """写运行态迁移的输入错误。""" _write_errors(output_dir / "errors.csv", errors) -def _write_summary_step1(path: Path, cards: list[CardRow], devices: list[DeviceRow], errors: list[ErrorRow]) -> None: +def _write_summary_step1( + path: Path, + cards: list[CardRow], + devices: list[DeviceRow], + errors: list[ErrorRow], + mapping: Mapping, + card_metas: dict[str, LegacyCardMeta], +) -> None: bound = sum(1 for c in cards if c.bound_device_virtual_no) + devices_by_virtual_no = _device_by_virtual_no(devices) + allocated_cards = sum( + 1 for c in cards if _resolve_card_ownership(c, mapping, card_metas, devices_by_virtual_no).target_shop_code + ) + allocated_devices = sum(1 for d in devices if _resolve_device_ownership(d, mapping, card_metas).target_shop_code) lines = [ "奇成迁移 - 脚本1 资产导入摘要", "=" * 48, @@ -700,9 +1034,11 @@ def _write_summary_step1(path: Path, cards: list[CardRow], devices: list[DeviceR f" 绑设备 : {bound}", f" 独立卡 : {len(cards) - bound}", f"设备总数 : {len(devices)}", + f"分配卡数 : {allocated_cards}", + f"分配设备数 : {allocated_devices}", f"异常行 : {len(errors)} (详见 errors.csv)", "", - "店铺归属:由 mapping.agents 自动推导(详见 step1_01/step1_02/step1_06 内的 shop_id 子查询)", + "店铺归属:由 mapping.yaml 的 overrides / ownership_rules / agents 决定,详见 ownership_resolution.csv", ] path.write_text("\n".join(lines) + "\n", encoding="utf-8") @@ -720,32 +1056,26 @@ def write_step2( mapping: Mapping, card_metas: dict[str, LegacyCardMeta], usage_snapshots: dict[str, LegacyCardUsage], + package_lifecycles: dict[str, list[LegacyPackageLifecycle]], commission_snapshots: dict[str, LegacyCommissionAccount], agent_balances: dict[str, LegacyAgentBalance], ) -> list[str]: """生成 step2_* SQL + warnings.csv + summary.txt。""" user_id = mapping.migration_user_id warnings: list[tuple[str, str, str]] = [] + errors: list[ErrorRow] = [] devices_by_virtual_no = _device_by_virtual_no(devices) - # 卡 → package(从 card_meta.current_meal_id 经 mapping.packages 查 target) - resolved_packages: dict[str, tuple[int, str]] = {} # 完整 ICCID → (target_package_id, expire_date) - for c in cards: - meta = card_metas.get(c.iccid_full) - if meta is None or not meta.current_meal_id: - continue - pkg_map = mapping.lookup_package(meta.current_meal_id) - if pkg_map is None: - warnings.append(("package_not_in_mapping", c.iccid_full, - f"奇成 meal_id={meta.current_meal_id} ({meta.current_meal_name}) 未在 mapping.yaml.packages")) - continue - try: - pkg_map.require_target() - except MissingTargetError as exc: - warnings.append(("package_target_missing", c.iccid_full, str(exc))) - continue - resolved_packages[c.iccid_full] = (pkg_map.target_package_id, meta.current_expire_date or "") + package_resolution_rows: list[list[str]] = [] + resolved_packages = _resolve_runtime_packages( + cards=cards, + devices_by_virtual_no=devices_by_virtual_no, + mapping=mapping, + package_lifecycles=package_lifecycles, + errors=errors, + resolution_rows=package_resolution_rows, + ) - # 用量预检:虚用量 > 0 但没找到当前生效套餐的卡,无法精确反算真用量 + # 用量预检:虚用量 > 0 但没找到当前生效正式套餐的卡,无法精确反算真用量 # (按 flow_add_discount=0 处理,即真=虚,可能高估真用量;不阻断) for c in cards: snap = usage_snapshots.get(c.iccid_full) @@ -754,8 +1084,8 @@ def write_step2( if not snap.has_active_package: warnings.append(( "usage_without_active_package", c.iccid_full, - f"奇成虚用量 {snap.virtual_used_mb}MB,但 tbl_card_life 中没有当前生效套餐" - "(status=1),无法精确反算真用量,按 flow_add_discount=0 处理(真=虚)", + f"奇成虚用量 {snap.virtual_used_mb}MB,但 tbl_card_life 中没有当前生效正式套餐" + "(status=1 且 type!=1),无法精确反算真用量,按 flow_add_discount=0 处理(真=虚)", )) _write_migration_orders(output_dir / "step2_01_migration_orders.sql", cards, resolved_packages, devices_by_virtual_no, user_id) @@ -776,11 +1106,15 @@ def write_step2( eligible_agents, agent_balances, commission_snapshots, user_id, ) + _write_errors(output_dir / "errors.csv", errors) _write_warnings(output_dir / "warnings.csv", warnings) + _write_package_resolution(output_dir / "package_resolution.csv", package_resolution_rows) _write_summary_step2( output_dir / "summary.txt", cards=cards, resolved=resolved_packages, + errors=errors, + package_resolution_rows=package_resolution_rows, devices_by_virtual_no=devices_by_virtual_no, agents=eligible_agents, warnings=warnings, @@ -788,19 +1122,164 @@ def write_step2( return [w[2] for w in warnings] -def _make_card_order_no(iccid_full: str) -> str: +def _resolve_runtime_packages( + *, + cards: list[CardRow], + devices_by_virtual_no: dict[str, DeviceRow], + mapping: Mapping, + package_lifecycles: dict[str, list[LegacyPackageLifecycle]], + errors: list[ErrorRow], + resolution_rows: list[list[str]], +) -> ResolvedPackageMap: + """解析运行态套餐队列。""" + resolved: ResolvedPackageMap = {} + source_cards: list[tuple[CardRow, str, str]] = [] + for card in cards: + asset = _runtime_asset_for_card(card, devices_by_virtual_no) + if asset is None: + if card.bound_device_virtual_no: + resolution_rows.append([ + "iot_card", card.iccid_full, card.iccid_full, "", "", "", "", "", "", "", + "skipped", "", "skipped", "非设备旧套餐读取槽位,不读取该卡旧套餐", "", "", "", + ]) + continue + asset_type, asset_identifier = asset + source_cards.append((card, asset_type, asset_identifier)) + + for card, asset_type, asset_identifier in source_cards: + lifecycles = package_lifecycles.get(card.iccid_full, []) + migratable = [ + item for item in lifecycles + if item.migration_status in mapping.package_rules.migrate_statuses + ] + skipped = [item for item in lifecycles if item.migration_status == "skipped"] + for item in skipped: + resolution_rows.append(_package_resolution_row( + asset_type, asset_identifier, card.iccid_full, item, "", "", "skipped", item.skip_reason + )) + + active_items = [item for item in migratable if item.migration_status == "active"] + if len(active_items) > 1: + msg = "同一资产存在多个当前生效正式套餐,无法唯一裁决" + errors.append(ErrorRow( + "tbl_card_life", 0, "iccid", card.iccid_full, + "multiple_active_packages", f"{msg}: {card.iccid_full}", + )) + for item in migratable: + resolution_rows.append(_package_resolution_row( + asset_type, asset_identifier, card.iccid_full, item, "", "", "blocked", msg + )) + continue + + pending_items = sorted( + [item for item in migratable if item.migration_status == "pending"], + key=lambda item: item.stable_sort_key, + ) + ordered_items = active_items + pending_items + for item in ordered_items: + pkg_map = mapping.lookup_package(item.meal_id) + if pkg_map is None: + msg = f"mapping.yaml.packages 未配置 legacy_meal_id={item.meal_id} ({item.meal_name})" + errors.append(ErrorRow("tbl_card_life", 0, "meal_id", item.meal_id, "package_not_in_mapping", msg)) + resolution_rows.append(_package_resolution_row( + asset_type, asset_identifier, card.iccid_full, item, "", "", "blocked", msg + )) + continue + try: + pkg_map.require_target() + except MissingTargetError as exc: + errors.append(ErrorRow( + "tbl_card_life", 0, "meal_id", item.meal_id, "package_target_missing", str(exc) + )) + resolution_rows.append(_package_resolution_row( + asset_type, asset_identifier, card.iccid_full, item, "", "", "blocked", str(exc) + )) + continue + + priority = 1 if item.migration_status == "active" else len(active_items) + pending_items.index(item) + 1 + usage_status = 1 if item.migration_status == "active" else 0 + resolved_item = ResolvedPackage( + source_iccid=card.iccid_full, + asset_type=asset_type, + asset_identifier=asset_identifier, + legacy_life_id=item.life_id, + legacy_meal_id=item.meal_id, + legacy_meal_name=item.meal_name or pkg_map.legacy_meal_name, + legacy_meal_type=item.meal_type, + legacy_status=item.status, + target_package_id=pkg_map.target_package_id or 0, + migration_status=item.migration_status, + package_usage_status=usage_status, + priority=priority, + start_date=item.start_date or "", + expire_date=item.expire_date or "", + stable_sort_key=item.stable_sort_key, + ) + resolved.setdefault(card.iccid_full, []).append(resolved_item) + resolution_rows.append(_package_resolution_row( + asset_type, + asset_identifier, + card.iccid_full, + item, + str(resolved_item.target_package_id), + str(priority), + item.migration_status, + "已生成迁移套餐", + )) + + if not ordered_items and not skipped: + resolution_rows.append([ + asset_type, asset_identifier, card.iccid_full, "", "", "", "", "", "", "", + "skipped", "", "skipped", "未查询到正式套餐生命周期", "", "", "", + ]) + + return resolved + + +def _package_resolution_row( + asset_type: str, + asset_identifier: str, + source_iccid: str, + item: LegacyPackageLifecycle, + target_package_id: str, + priority: str, + decision: str, + reason: str, +) -> list[str]: + return [ + asset_type, + asset_identifier, + source_iccid, + item.source_table, + item.life_id, + item.meal_id, + item.meal_name, + item.meal_type, + str(item.status), + target_package_id, + item.migration_status, + priority, + decision, + reason, + item.start_date or "", + item.expire_date or "", + item.stable_sort_key, + ] + + +def _make_card_order_no(iccid_full: str, priority: int = 1) -> str: """生成稳定迁移订单号。 - 新库 order_no 限制为 varchar(30);MIG-CARD- + 20 位 ICCID 总长 29。 - 固定使用 ICCID 还能保证 step2_01/step2_02 分开重跑时仍能匹配同一订单。 + 新库 order_no 限制为 varchar(30)。多套餐队列使用 MIG--, + 20 位 ICCID + 3 位以内 priority 总长仍小于 30。 """ - return f"{ORDER_NO_PREFIX_CARD}{iccid_full}" + return f"MIG-{iccid_full}-{priority}" def _write_migration_orders( path: Path, cards: list[CardRow], - resolved_packages: dict[str, tuple[int, str]], + resolved_packages: ResolvedPackageMap, devices_by_virtual_no: dict[str, DeviceRow], user_id: int, ) -> None: @@ -810,18 +1289,45 @@ def _write_migration_orders( with path.open("w", encoding="utf-8") as f: f.write(_file_header("step2_01 迁移伪订单 (tb_order)")) for c in cards: - if c.iccid_full not in resolved_packages: + packages = resolved_packages.get(c.iccid_full, []) + if not packages: continue asset = _runtime_asset_for_card(c, devices_by_virtual_no) if asset is None: continue asset_type, asset_identifier = asset - order_no = _make_card_order_no(c.iccid_full) - if asset_type == "device": + for pkg in packages: + order_no = _make_card_order_no(c.iccid_full, pkg.priority) + _write_formal_package_guard(f, pkg.target_package_id, pkg.legacy_meal_name) + if asset_type == "device": + f.write( + "INSERT INTO tb_order (\n" + " order_no, order_type, buyer_type, buyer_id,\n" + " device_id, asset_identifier,\n" + " total_amount, actual_paid_amount,\n" + " payment_method, payment_status, paid_at,\n" + " commission_status, commission_result, commission_config_version,\n" + " source, generation,\n" + " operator_account_type, operator_account_name,\n" + " creator, updater, created_at, updated_at\n" + ")\n" + f"SELECT {_sql_str(order_no)}, 'device', 'personal', 0,\n" + f" d.id, {_sql_str(asset_identifier)},\n" + " 0, 0,\n" + " 'offline', 2, NOW(),\n" + " 2, 2, 0,\n" + " 'admin', 1,\n" + " 'platform', '奇成数据迁移',\n" + f" {user_id}, {user_id}, NOW(), NOW()\n" + "FROM tb_device d\n" + f"WHERE d.virtual_no = {_sql_str(asset_identifier)} AND d.deleted_at IS NULL\n" + "ON CONFLICT (order_no) WHERE deleted_at IS NULL DO NOTHING;\n\n" + ) + continue f.write( "INSERT INTO tb_order (\n" " order_no, order_type, buyer_type, buyer_id,\n" - " device_id, asset_identifier,\n" + " iot_card_id, asset_identifier,\n" " total_amount, actual_paid_amount,\n" " payment_method, payment_status, paid_at,\n" " commission_status, commission_result, commission_config_version,\n" @@ -829,48 +1335,24 @@ def _write_migration_orders( " operator_account_type, operator_account_name,\n" " creator, updater, created_at, updated_at\n" ")\n" - f"SELECT {_sql_str(order_no)}, 'device', 'personal', 0,\n" - f" d.id, {_sql_str(asset_identifier)},\n" + f"SELECT {_sql_str(order_no)}, 'single_card', 'personal', 0,\n" + f" c.id, {_sql_str(c.iccid_full)},\n" " 0, 0,\n" " 'offline', 2, NOW(),\n" " 2, 2, 0,\n" " 'admin', 1,\n" " 'platform', '奇成数据迁移',\n" f" {user_id}, {user_id}, NOW(), NOW()\n" - "FROM tb_device d\n" - f"WHERE d.virtual_no = {_sql_str(asset_identifier)} AND d.deleted_at IS NULL\n" + "FROM tb_iot_card c\n" + f"WHERE c.iccid = {_sql_str(c.iccid_full)} AND c.deleted_at IS NULL\n" "ON CONFLICT (order_no) WHERE deleted_at IS NULL DO NOTHING;\n\n" ) - continue - f.write( - "INSERT INTO tb_order (\n" - " order_no, order_type, buyer_type, buyer_id,\n" - " iot_card_id, asset_identifier,\n" - " total_amount, actual_paid_amount,\n" - " payment_method, payment_status, paid_at,\n" - " commission_status, commission_result, commission_config_version,\n" - " source, generation,\n" - " operator_account_type, operator_account_name,\n" - " creator, updater, created_at, updated_at\n" - ")\n" - f"SELECT {_sql_str(order_no)}, 'single_card', 'personal', 0,\n" - f" c.id, {_sql_str(c.iccid_full)},\n" - " 0, 0,\n" - " 'offline', 2, NOW(),\n" - " 2, 2, 0,\n" - " 'admin', 1,\n" - " 'platform', '奇成数据迁移',\n" - f" {user_id}, {user_id}, NOW(), NOW()\n" - "FROM tb_iot_card c\n" - f"WHERE c.iccid = {_sql_str(c.iccid_full)} AND c.deleted_at IS NULL\n" - "ON CONFLICT (order_no) WHERE deleted_at IS NULL DO NOTHING;\n\n" - ) def _write_package_usages( path: Path, cards: list[CardRow], - resolved_packages: dict[str, tuple[int, str]], + resolved_packages: ResolvedPackageMap, devices_by_virtual_no: dict[str, DeviceRow], usage_snapshots: dict[str, LegacyCardUsage], user_id: int, @@ -878,55 +1360,59 @@ def _write_package_usages( with path.open("w", encoding="utf-8") as f: f.write(_file_header("step2_02 套餐使用记录 (tb_package_usage)")) for c in cards: - if c.iccid_full not in resolved_packages: + packages = resolved_packages.get(c.iccid_full, []) + if not packages: continue asset = _runtime_asset_for_card(c, devices_by_virtual_no) if asset is None: continue asset_type, _asset_identifier = asset - target_pkg_id, expire = resolved_packages[c.iccid_full] - order_no = _make_card_order_no(c.iccid_full) - expires_expr = f"{_sql_str(expire)}::timestamp" if expire else "NULL" - snap = usage_snapshots.get(c.iccid_full) - used_mb = snap.real_used_mb_floor if snap is not None else 0 - effective_limit_expr = ( - "CASE WHEN p.enable_virtual_data AND p.virtual_data_mb > 0 " - "THEN p.virtual_data_mb ELSE p.real_data_mb END" - ) - usage_expr = f"LEAST({used_mb}, {effective_limit_expr})" - carrier_column = "device_id" if asset_type == "device" else "iot_card_id" - carrier_select = "o.device_id" if asset_type == "device" else "o.iot_card_id" - usage_type = "device" if asset_type == "device" else "single_card" - f.write( - "INSERT INTO tb_package_usage (\n" - " order_id, order_no, package_id, usage_type,\n" - f" {carrier_column},\n" - " data_limit_mb, data_usage_mb,\n" - " virtual_total_mb_snapshot, display_gain_ratio_snapshot, enable_virtual_data_snapshot,\n" - " status, priority, generation,\n" - " activated_at, expires_at,\n" - " package_name,\n" - " package_price_config_status, package_is_gift,\n" - " data_reset_cycle,\n" - " creator, updater, created_at, updated_at\n" - ")\n" - f"SELECT o.id, o.order_no, p.id, {_sql_str(usage_type)},\n" - f" {carrier_select},\n" - f" p.real_data_mb, {usage_expr},\n" - f" {effective_limit_expr},\n" - " CASE WHEN p.enable_virtual_data AND p.virtual_data_mb > 0 THEN p.virtual_ratio ELSE 1.0 END,\n" - " CASE WHEN p.enable_virtual_data AND p.virtual_data_mb > 0 THEN TRUE ELSE FALSE END,\n" - f" CASE WHEN {effective_limit_expr} > 0 AND {usage_expr} >= {effective_limit_expr} THEN 2 ELSE 1 END, 1, 1,\n" - f" NULL, {expires_expr},\n" - " p.package_name,\n" - " p.price_config_status, p.is_gift,\n" - " p.data_reset_cycle,\n" - f" {user_id}, {user_id}, NOW(), NOW()\n" - "FROM tb_order o\n" - f"JOIN tb_package p ON p.id = {target_pkg_id} AND p.deleted_at IS NULL\n" - f"WHERE o.order_no = {_sql_str(order_no)} AND o.deleted_at IS NULL\n" - "ON CONFLICT (order_id, package_id) WHERE deleted_at IS NULL DO NOTHING;\n\n" - ) + for pkg in packages: + order_no = _make_card_order_no(c.iccid_full, pkg.priority) + _write_formal_package_guard(f, pkg.target_package_id, pkg.legacy_meal_name) + activated_expr = f"{_sql_str(pkg.start_date)}::timestamp" if pkg.migration_status == "active" and pkg.start_date else "NULL" + expires_expr = f"{_sql_str(pkg.expire_date)}::timestamp" if pkg.expire_date else "NULL" + snap = usage_snapshots.get(c.iccid_full) + active_used_mb = snap.real_used_mb_floor if snap is not None else 0 + used_mb = active_used_mb if pkg.migration_status == "active" else 0 + effective_limit_expr = ( + "CASE WHEN p.enable_virtual_data AND p.virtual_data_mb > 0 " + "THEN p.virtual_data_mb ELSE p.real_data_mb END" + ) + usage_expr = f"LEAST({used_mb}, {effective_limit_expr})" + carrier_column = "device_id" if asset_type == "device" else "iot_card_id" + carrier_select = "o.device_id" if asset_type == "device" else "o.iot_card_id" + usage_type = "device" if asset_type == "device" else "single_card" + f.write( + "INSERT INTO tb_package_usage (\n" + " order_id, order_no, package_id, usage_type,\n" + f" {carrier_column},\n" + " data_limit_mb, data_usage_mb,\n" + " virtual_total_mb_snapshot, display_gain_ratio_snapshot, enable_virtual_data_snapshot,\n" + " status, priority, generation,\n" + " activated_at, expires_at,\n" + " package_name,\n" + " package_price_config_status, package_is_gift,\n" + " data_reset_cycle,\n" + " creator, updater, created_at, updated_at\n" + ")\n" + f"SELECT o.id, o.order_no, p.id, {_sql_str(usage_type)},\n" + f" {carrier_select},\n" + f" p.real_data_mb, {usage_expr},\n" + f" {effective_limit_expr},\n" + " CASE WHEN p.enable_virtual_data AND p.virtual_data_mb > 0 THEN p.virtual_ratio ELSE 1.0 END,\n" + " CASE WHEN p.enable_virtual_data AND p.virtual_data_mb > 0 THEN TRUE ELSE FALSE END,\n" + f" {pkg.package_usage_status}, {pkg.priority}, 1,\n" + f" {activated_expr}, {expires_expr},\n" + " p.package_name,\n" + " p.price_config_status, p.is_gift,\n" + " p.data_reset_cycle,\n" + f" {user_id}, {user_id}, NOW(), NOW()\n" + "FROM tb_order o\n" + f"JOIN tb_package p ON p.id = {pkg.target_package_id} AND p.package_type = 'formal' AND p.deleted_at IS NULL\n" + f"WHERE o.order_no = {_sql_str(order_no)} AND o.deleted_at IS NULL\n" + "ON CONFLICT (order_id, package_id) WHERE deleted_at IS NULL DO NOTHING;\n\n" + ) def _write_card_data_usage(path: Path, cards: list[CardRow], usage_snapshots: dict[str, LegacyCardUsage]) -> None: @@ -962,7 +1448,7 @@ def _write_card_data_usage(path: Path, cards: list[CardRow], usage_snapshots: di def _write_asset_series_update( path: Path, cards: list[CardRow], - resolved_packages: dict[str, tuple[int, str]], + resolved_packages: ResolvedPackageMap, devices_by_virtual_no: dict[str, DeviceRow], ) -> None: """按迁移套餐回填资产套餐系列。 @@ -972,16 +1458,19 @@ def _write_asset_series_update( with path.open("w", encoding="utf-8") as f: f.write(_file_header("step2_06 资产套餐系列回填 (tb_iot_card / tb_device)")) for c in cards: - if c.iccid_full not in resolved_packages: + packages = resolved_packages.get(c.iccid_full, []) + if not packages: continue - target_pkg_id, _expire = resolved_packages[c.iccid_full] + primary_pkg = packages[0] + target_pkg_id = primary_pkg.target_package_id + _write_formal_package_guard(f, target_pkg_id, primary_pkg.legacy_meal_name) f.write( "UPDATE tb_iot_card c\n" "SET series_id = p.series_id,\n" " updated_at = NOW()\n" "FROM tb_package p\n" f"WHERE c.iccid = {_sql_str(c.iccid_full)} AND c.deleted_at IS NULL\n" - f" AND p.id = {target_pkg_id} AND p.deleted_at IS NULL\n" + f" AND p.id = {target_pkg_id} AND p.package_type = 'formal' AND p.deleted_at IS NULL\n" " AND p.series_id IS NOT NULL\n" " AND c.series_id IS DISTINCT FROM p.series_id;\n\n" ) @@ -997,7 +1486,7 @@ def _write_asset_series_update( " updated_at = NOW()\n" "FROM tb_package p\n" f"WHERE d.virtual_no = {_sql_str(asset_identifier)} AND d.deleted_at IS NULL\n" - f" AND p.id = {target_pkg_id} AND p.deleted_at IS NULL\n" + f" AND p.id = {target_pkg_id} AND p.package_type = 'formal' AND p.deleted_at IS NULL\n" " AND p.series_id IS NOT NULL\n" " AND d.series_id IS DISTINCT FROM p.series_id;\n\n" ) @@ -1125,18 +1614,51 @@ def _write_warnings(path: Path, warnings: list[tuple[str, str, str]]) -> None: writer.writerow([kind, key, message]) -def _write_summary_step2(path: Path, *, cards, resolved, devices_by_virtual_no, agents, warnings) -> None: +def _write_package_resolution(path: Path, rows: list[list[str]]) -> None: + """写套餐解析审核文件。""" + with path.open("w", encoding="utf-8-sig", newline="") as f: + writer = csv.writer(f) + writer.writerow([ + "asset_type", + "asset_identifier", + "source_iccid", + "source_table", + "legacy_life_id", + "legacy_meal_id", + "legacy_meal_name", + "legacy_meal_type", + "legacy_status", + "target_package_id", + "target_status", + "priority", + "decision", + "reason", + "start_date", + "expire_date", + "stable_sort_key", + ]) + writer.writerows(rows) + + +def _write_summary_step2(path: Path, *, cards, resolved, errors, package_resolution_rows, devices_by_virtual_no, agents, warnings) -> None: generated_count = sum( - 1 + len(resolved.get(c.iccid_full, [])) for c in cards - if c.iccid_full in resolved and _runtime_asset_for_card(c, devices_by_virtual_no) is not None + if _runtime_asset_for_card(c, devices_by_virtual_no) is not None ) + active_count = sum(1 for items in resolved.values() for item in items if item.migration_status == "active") + pending_count = sum(1 for items in resolved.values() for item in items if item.migration_status == "pending") + skipped_count = sum(1 for row in package_resolution_rows if len(row) > 11 and row[11] == "skipped") lines = [ "奇成迁移 - 脚本2 关联数据导入摘要", "=" * 48, f"生成时间 : {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}", f"卡总数 : {len(cards)}", f" 生成伪订单+套餐 : {generated_count}", + f" active 套餐 : {active_count}", + f" pending 套餐 : {pending_count}", + f" 跳过套餐 : {skipped_count}", + f"错误条目 : {len(errors)} (详见 errors.csv)", f"代理钱包初始化 : {len(agents)}", f"警告条目 : {len(warnings)} (详见 warnings.csv)", ] diff --git a/scripts/migration/migrate_assets.py b/scripts/migration/migrate_assets.py index 73e80f2..fb95ed4 100644 --- a/scripts/migration/migrate_assets.py +++ b/scripts/migration/migrate_assets.py @@ -40,7 +40,7 @@ def main() -> int: output_dir.mkdir(parents=True, exist_ok=True) mapping = mapping_loader.load_mapping(config_dir) - cards, devices, errors = csv_loader.load_assets(resources_dir) + cards, devices, errors = csv_loader.load_assets(resources_dir, mapping) # 连奇成查卡元数据(运营商分类、agent_id、msisdn、virtual_no、当前生效套餐) iccid_pairs = [(c.iccid_19, c.iccid_20, c.allow_iccid_19_lookup) for c in cards] diff --git a/scripts/migration/migrate_runtime.py b/scripts/migration/migrate_runtime.py index c140784..d9879cc 100644 --- a/scripts/migration/migrate_runtime.py +++ b/scripts/migration/migrate_runtime.py @@ -37,7 +37,7 @@ def main() -> int: output_dir.mkdir(parents=True, exist_ok=True) mapping = mapping_loader.load_mapping(config_dir) - cards, devices, errors = csv_loader.load_assets(resources_dir) + cards, devices, errors = csv_loader.load_assets(resources_dir, mapping) if errors: sql_builder.write_runtime_input_errors(output_dir, errors) print(f"cards.csv/devices.csv 存在 {len(errors)} 个基础错误,已写入 {output_dir / 'errors.csv'}", file=sys.stderr) @@ -50,6 +50,7 @@ def main() -> int: with legacy_query.connect_readonly(dsn) as conn: card_metas, _dup_iccids = legacy_query.fetch_card_meta(conn, iccid_pairs) usage_snapshots = legacy_query.fetch_card_usage(conn, iccid_pairs) + package_lifecycles = legacy_query.fetch_package_lifecycles(conn, iccid_pairs) commission_snapshots = legacy_query.fetch_commission_accounts(conn, agent_ids) if agent_ids else {} agent_balances = legacy_query.fetch_agent_balances(conn, agent_ids) if agent_ids else {} @@ -60,6 +61,7 @@ def main() -> int: mapping=mapping, card_metas=card_metas, usage_snapshots=usage_snapshots, + package_lifecycles=package_lifecycles, commission_snapshots=commission_snapshots, agent_balances=agent_balances, ) diff --git a/scripts/migration/resources/README.md b/scripts/migration/resources/README.md index 33c3969..728f3d6 100644 --- a/scripts/migration/resources/README.md +++ b/scripts/migration/resources/README.md @@ -2,8 +2,13 @@ 业务方按以下规则把数据放到本目录,**重命名去掉 `.example` 后缀**,即可被脚本读取。 -> 设计原则:**业务方只填"是哪张卡"和"在哪台设备上",归属、运营商、套餐等元数据全部由 -> 脚本从奇成自动获取,通过 `config/mapping.yaml` 决定如何落到新系统。** +> 设计原则:**业务方只填"是哪张卡"、"在哪台设备上"和少量行级槽位覆盖**。 +> 归属、批量默认槽位、套餐迁移范围由 `config/mapping.yaml` 决定;运营商、套餐、号码等历史事实从奇成只读查询。 + +`cards.csv` 必填。`devices.csv` 可选: + +- 只有卡迁移:只放 `cards.csv`,不要放 `devices.csv`。 +- 卡 + 设备迁移:同时放 `cards.csv` 和 `devices.csv`,设备槽位引用的卡必须出现在 `cards.csv`。 ## cards.csv 卡资产清单 @@ -24,13 +29,14 @@ - 非 CTCC 卡 `iccid_20` 留空 → 同上阻断 - 同一完整 ICCID 在奇成 `tbl_card` 命中多条记录 → `errors.csv: legacy_duplicate` 阻断,请业务方人工确认 -**店铺归属不在 CSV 里填**,由 `mapping.yaml.agents[].target_shop_code` 通过奇成 -`tbl_card.agent_id` 自动推导。如果该卡在奇成没有 agent_id 或代理映射未填 shop_code, -卡进平台库存。 +**店铺归属不在 CSV 里填**。独立卡由 `mapping.yaml.ownership_rules.standalone_card` +和 `overrides.cards` 决定。多店铺批次使用 `standalone_card.mode: legacy_agent`, +再按奇成 `agent_id` 填 `mapping.yaml.agents[].target_shop_code`。设备上的卡跟随设备归属, +不单独生成分配记录。 -## devices.csv 设备资产清单 +## devices.csv 设备资产清单(可选) -每台要迁移的设备一行。 +每台要迁移的设备一行。没有设备时不要创建这个文件,脚本会把所有卡按独立卡处理。 | 列名 | 必填 | 说明 | |------|------|------| @@ -41,12 +47,22 @@ | `device_type` | 否 | 设备类型,写入 `tb_device.device_type` | | `max_sim_slots` | 否 | 最大卡槽数,范围 1-4。不填时按非空 `sim_iccid_*` 的最大槽位推导 | | `sim_iccid_1` ~ `sim_iccid_4` | 否 | 各插槽 ICCID,引用 `cards.csv` 的卡。可填 19 或 20 位;20 位精确匹配完整 ICCID,19 位仅在该前缀唯一对应一张卡时允许(至少要有 1 个) | +| `current_slot` | 否 | 当前槽位,范围 1-4。写入 `tb_device_sim_binding.is_current`;空值用批量规则或覆盖项 | +| `package_source_slot` | 否 | 设备旧套餐读取槽位,范围 1-4。新系统套餐仍归属设备;该字段只决定从奇成哪张绑定卡读取正式套餐生命周期 | -**店铺归属规则**:用 `sim_iccid_1`(主卡)的 agent_id 走 `mapping.agents` -推导,与卡归属同口径。 +**槽位优先级**: +1. `devices.csv` 行级 `current_slot` / `package_source_slot` +2. `mapping.yaml.overrides.devices` 中同设备号覆盖项 +3. `mapping.yaml.ownership_rules.device` 批量默认槽位 + +若 `current_slot` 或 `package_source_slot` 对应的 `sim_iccid_N` 为空,脚本会写入 +`errors.csv` 并阻断该设备相关 SQL。非 `package_source_slot` 的设备绑定卡只生成 +基础卡和绑定 SQL,不会再读取它的奇成套餐,避免同一设备重复迁入多份主套餐。 **其他字段自动取默认值**:`manufacturer` 留空,`max_sim_slots` 按已填写卡槽推导, -`current_slot=1`(以 `sim_iccid_1` 为当前生效卡)。 +设备归属由 `mapping.yaml.ownership_rules.device` 和 `overrides.devices` 决定。设备使用 +`device.mode: legacy_agent` 时,脚本会按 `package_source_slot` 对应绑定卡的奇成 `agent_id` +查 `mapping.yaml.agents[].target_shop_code`。 ## 一致性约束 diff --git a/scripts/migration/scan_legacy.py b/scripts/migration/scan_legacy.py index 6837c9f..29ec9e6 100644 --- a/scripts/migration/scan_legacy.py +++ b/scripts/migration/scan_legacy.py @@ -17,6 +17,7 @@ from __future__ import annotations import argparse import csv import sys +from dataclasses import dataclass from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) @@ -31,15 +32,30 @@ from lib.mapping_loader import ( # noqa: E402 ) -def _read_iccid_pairs(cards_csv: Path) -> list[tuple[str, str, bool]]: +@dataclass(frozen=True) +class ICCIDInput: + """业务方提供的单行 ICCID 输入。""" + + line_no: int + iccid_19: str + iccid_20: str + allow_iccid_19_lookup: bool + + @property + def card_key(self) -> str: + """奇成扫描结果使用的业务方完整键。""" + return self.iccid_20 or self.iccid_19 + + +def _read_iccid_pairs(cards_csv: Path) -> list[ICCIDInput]: """从 cards.csv 读 iccid_19 + iccid_20 列(只关心这两列,其他列忽略)。 - 返回 (iccid_19, iccid_20, allow_iccid_19_lookup) 列表。 + 返回带行号的 ICCID 输入列表。 只填 iccid_20 时,iccid_19 从前 19 位派生,但不用于奇成主表兜底查询。 """ if not cards_csv.exists(): raise FileNotFoundError(f"找不到 {cards_csv},请按 resources/README.md 准备") - pairs: dict[str, tuple[str, str, bool]] = {} + pairs: dict[str, ICCIDInput] = {} first_seen_line: dict[str, int] = {} errors: list[str] = [] with cards_csv.open("r", encoding="utf-8-sig", newline="") as f: @@ -75,7 +91,12 @@ def _read_iccid_pairs(cards_csv: Path) -> list[tuple[str, str, bool]]: errors.append(f"第 {line_no} 行 {'iccid_20' if i20 else 'iccid_19'} 与第 {first_seen_line[dedup_key]} 行重复: {dedup_key!r}") continue first_seen_line[dedup_key] = line_no - pairs[dedup_key] = (i19, i20, allow_i19_lookup) + pairs[dedup_key] = ICCIDInput( + line_no=line_no, + iccid_19=i19, + iccid_20=i20, + allow_iccid_19_lookup=allow_i19_lookup, + ) if errors: preview = "\n".join(f" - {msg}" for msg in errors[:20]) if len(errors) > 20: @@ -106,7 +127,10 @@ def main() -> int: dsn = mapping_loader.load_legacy_dsn(config_dir) with legacy_query.connect_readonly(dsn) as conn: - card_metas, dup_iccids = legacy_query.fetch_card_meta(conn, iccid_pairs) + card_metas, dup_iccids = legacy_query.fetch_card_meta( + conn, + [(x.iccid_19, x.iccid_20, x.allow_iccid_19_lookup) for x in iccid_pairs], + ) meal_ids = {m.current_meal_id for m in card_metas.values() if m.current_meal_id} meal_metas = legacy_query.fetch_meal_meta(conn, meal_ids) @@ -120,8 +144,11 @@ def main() -> int: merged = _merge(existing, card_metas, meal_metas) out_path = mapping_loader.merge_and_save(config_dir, merged) todo = _collect_todo(merged) + unmatched = _collect_unmatched_iccids(iccid_pairs, card_metas, dup_iccids) _write_todo(config_dir / "mapping_todo.txt", todo) - _print_summary(out_path, card_metas, meal_metas, todo) + unmatched_path = config_dir / "unmatched_iccids.csv" + _write_unmatched_iccids(unmatched_path, unmatched) + _print_summary(out_path, card_metas, meal_metas, todo, unmatched_path, len(unmatched)) return 0 @@ -134,6 +161,9 @@ def _merge( merged = Mapping( migration_user_id=existing.migration_user_id, migration_batch_no=existing.migration_batch_no, + ownership_rules=existing.ownership_rules, + package_rules=existing.package_rules, + overrides=existing.overrides, carriers=dict(existing.carriers), packages=dict(existing.packages), series=dict(existing.series), @@ -210,6 +240,19 @@ def _collect_todo(mapping: Mapping) -> dict[str, list[str]]: return todo +def _collect_unmatched_iccids( + iccid_pairs: list[ICCIDInput], + card_metas: dict[str, "legacy_query.LegacyCardMeta"], + duplicate_iccids: set[str], +) -> list[ICCIDInput]: + """收集未在奇成 tbl_card 命中的 ICCID。 + + 重复命中的卡属于脏数据异常,不是未命中,因此从未命中清单中排除。 + """ + resolved_keys = set(card_metas) | set(duplicate_iccids) + return [item for item in iccid_pairs if item.card_key not in resolved_keys] + + def _write_todo(path: Path, todo: dict[str, list[str]]) -> None: lines = ["奇成迁移映射待填写清单", "=" * 40, ""] for kind in ("carriers", "packages", "agents"): @@ -220,13 +263,47 @@ def _write_todo(path: Path, todo: dict[str, list[str]]) -> None: path.write_text("\n".join(lines), encoding="utf-8") -def _print_summary(out_path: Path, card_metas, meal_metas, todo: dict[str, list[str]]) -> None: +def _write_unmatched_iccids(path: Path, unmatched: list[ICCIDInput]) -> None: + """把未命中的 ICCID 写成 CSV,便于转给业务方核对。""" + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("w", encoding="utf-8-sig", newline="") as f: + writer = csv.DictWriter( + f, + fieldnames=["line_no", "iccid_19", "iccid_20", "lookup_keys", "note"], + ) + writer.writeheader() + for item in unmatched: + lookup_keys = [item.iccid_20 or item.iccid_19] + if item.iccid_20 and item.allow_iccid_19_lookup: + lookup_keys.append(item.iccid_19) + writer.writerow( + { + "line_no": item.line_no, + "iccid_19": item.iccid_19, + "iccid_20": item.iccid_20, + "lookup_keys": ";".join(lookup_keys), + "note": "奇成 tbl_card 未命中", + } + ) + + +def _print_summary( + out_path: Path, + card_metas, + meal_metas, + todo: dict[str, list[str]], + unmatched_path: Path, + unmatched_count: int, +) -> None: print(f"扫描完成,mapping 写入: {out_path}") print(f" 奇成卡命中 : {len(card_metas)}") + print(f" 奇成卡未命中 : {unmatched_count}") print(f" 奇成套餐命中 : {len(meal_metas)}") print(f" 待填 carriers : {len(todo['carriers'])}") print(f" 待填 packages : {len(todo['packages'])}") print(f" 待填 agents : {len(todo['agents'])}") + if unmatched_count: + print(f"未命中清单: {unmatched_path}") if any(todo.values()): print(f"详见: {out_path.parent / 'mapping_todo.txt'}")