chore: 移除剩余迁移验证脚本并说明汇总任务依赖

2026-11-04 14:49:34 +08:00
parent 297e80827d
commit 7764b03906
20 changed files with 40 additions and 1640 deletions
@@ -50,6 +50,6 @@

 ## 文档与脚本清理

-104 份阶段报告、重复 JSON、旧目录审查和独立菜单 SQL，以及 26 个旧脚本已逐文件复制到仓库外并校验 SHA-256 后移除。保留统一迁移 SQL、业务口径、综合验收索引和 14 个当前验证工具，补充维护入口与脚本说明。
+首次清理将 104 份阶段报告、重复 JSON、旧目录审查和独立菜单 SQL，以及 26 个旧脚本移出项目。用户追加要求后，剩余 14 个验证工具、脚本说明和 Python 缓存也全部移出，`scripts` 目录已删除。所有移除文件均在仓库外归档并校验 SHA-256；统一迁移 SQL、业务口径与综合验收索引保留。

 真实本地配置 `configs/config.yaml`、私有运行数据，以及其他会话的 V2 估算和 outputs 成果不纳入本次提交。未推送远端。
@@ -4,6 +4,6 @@
 - [当前接口审查](API_REVIEW.md)：此次修复、验证范围及保留的技术风险。
 - [后台说明](ADMIN_BACKEND.md)、[支付接入](PAYMENT.md)、[短信接入](SMS_INTEGRATION.md)。
 - [OA 迁移计划](OA_MIGRATION_PLAN.md)：独立 OA 工作的设计材料，不是当前商务迁移的完成证明。
- [验证脚本](../scripts/README.md)：保留工具的用途和私有输入约定。
+- [汇总任务说明](sw6-to-kra/report-build-source-notes.md)：任务读写表及接口依赖；临时验证脚本已移出项目。

 历史阶段记录已归档到仓库外；后续将最新结果更新到对应维护文档，不再每个接口生成一份日期报告。其他会话生成的商务 V2 估算及表格保持原状，本次清理和提交不包含这些文件。
@@ -8,7 +8,7 @@
 2. 在 [统一迁移 SQL](sw_to_kra_migrate.sql) 开头修改 `USE`，当前为 `USE sw9`。用支持 `DELIMITER` 的客户端在同一连接执行，遇错停止。
 3. 启动当前 Go 版本，执行版本迁移，补齐业务锁、密码重置表、附件引用和查询索引。大表首次建索引需要等待完成。
 4. 配置主库和 `lx`、`hw`、`yq`、`oa` 数据源。配置和凭据保存在仓库外；数据库连接信息不写入本目录。
-5. 对照 [验收记录](validation.md) 与 [验证脚本说明](../../scripts/README.md) 复核接口和任务。迁入计划默认停用，启用前应确认业务所需运行时段。
+5. 对照 [验收记录](validation.md) 与 [汇总任务说明](report-build-source-notes.md) 复核接口和任务。迁入计划默认停用，启用前应确认业务所需运行时段。

 SQL 自动区分首次迁移、已完成迁移的菜单调整和异常中间状态；不会对完成迁移的库重新导入账号。重复执行及旧表已归档/清理的路径均有演练记录。

@@ -25,4 +25,4 @@ SQL 自动区分首次迁移、已完成迁移的菜单调整和异常中间状

 SQL 生成器位于旧仓库 `script/sql/sw6_to_kra/build_sw7_migration.py`。修改生成器后同步本目录统一 SQL，并重新核对散列及迁移结果；不要恢复分阶段 SQL 补丁。

-阶段报告、重复 JSON、旧 sw7 冒烟脚本已归档到本机仓库外 `D:/Users/Yvan/backups/sw9/20261104_rehearsal/cleanup-archive/`。该目录以及相邻的运行配置、原始响应、分区备份不纳入 Git。历史索引中 `reports` 项的文件名可在归档内查找。
+阶段报告、重复 JSON、全部临时验证脚本已归档到本机仓库外 `D:/Users/Yvan/backups/sw9/20261104_rehearsal/cleanup-archive/`，最后保留的工具位于其中的 `scripts-final/`。项目不再保留 `scripts` 目录，Go 任务运行不依赖这些脚本。归档及相邻的运行配置、原始响应、分区备份不纳入 Git。历史索引中 `reports` 项的文件名可在归档内查找。
@@ -25,3 +25,38 @@
 领域与存储测试覆盖身份/账号规则、重复构建、自然日筛选和身份明细写失败后的原子回滚。真实 TCP HTTP 使用隔离数据库和 Kra 原生任务创建、手动触发、执行日志，核对两种报表实际落库且任务仍为禁用。

 当前验证状态统一见 [验收与性能记录](validation.md)，不再维护阶段性运行报告。
+
+## 任务、表与接口的对应关系
+
+这些任务是 Go 内部注册的方法，与已删除的临时验证脚本无关。它们从当前主库读取已同步的原始数据，再生成统计结果；不负责将 SDK 平台数据同步进主库。当前运行服务主库为 sw9。
+
+| 任务方法 | 日期范围 | 写入的统计表 |
+|---|---|---|
+| swReportDayTodayTask | 上海时区今天 | tab_sw_report_day、tab_sw_report_day_key |
+| swReportDayYesterdayTask | 上海时区昨天 | 同上 |
+| swReportDayBackfillTask | startDate 至 endDate，含两端 | 同上 |
+| swReportServerDayTodayTask | 上海时区今天 | tab_sw_report_server_day、tab_sw_report_server_day_key |
+| swReportServerDayYesterdayTask | 上海时区昨天 | 同上 |
+| swReportServerDayBackfillTask | startDate 至 endDate，含两端 | 同上 |
+
+商务日报读取 `tab_user`（主账号注册及身份资料）、`tab_user_day_login`（登录）、`tab_spend`（成功充值）、`tab_game`（游戏名）。区服日报读取 `tab_user_play_info`（创角 create_time、活跃 play_time）、`tab_spend`、`tab_user`、`tab_game`。
+
+`tab_sw_report_day` 按日期、平台、渠道、游戏、去重模式保存注册数、活跃人数、付费人数、新增付费人数和充值金额。`tab_sw_report_server_day` 在相同维度上增加区服，注册数代表创角玩家数，不包含新增付费人数。
+
+两个 `_key` 表保存参与统计的身份键，供跨日期、渠道或游戏聚合时重新 `COUNT(DISTINCT identity_key)`。例如同一玩家连续两天活跃，查询这两天的独立活跃人数应为 1，不能直接将两个日活人数相加。`dedup_mode=0` 使用账号 ID，`dedup_mode=1` 使用规范化身份证；空身份证不计入模式 1。`metric` 为注册 1、活跃 2、付费 3，商务日报另有新增付费 4；金额查询只读模式 0，避免两种模式重复累加。
+
+每次重建会在一个事务内锁定“报表类型＋日期”，读取原始数据，删除统计表及对应 key 表该 `stat_date` 的旧行，批量写入新结果。读取或写入失败、取消时回滚；每个日期单独提交，回跑中途失败不会撤销此前已完成日期。除统计表外，会使用 `sys_business_report_build_locks` 保存互斥锁键，任务执行框架将结果记入 `sys_timed_task_logs`。原始用户、充值和登录记录不会被这些任务改写。
+
+| 使用方 | 接口 | 依赖 |
+|---|---|---|
+| 渠道数据统计 | GET /sw/channelData/list | 商务日报两表 |
+| 游戏数据 | GET /sw/gameData/list | 商务日报两表 |
+| 三类排行榜 | GET /sw/rank/channel、/business、/game | 商务日报两表 |
+| 仪表盘/工作台 | GET /sw/workbench/summary、/registerTop、/payRank | 商务日报两表；汇总中的新增渠道另查渠道表 |
+| 商务画像 | GET /sw/business/:userId/portrait、/sw/business/bindPlatform/portrait | 商务日报两表 |
+| 渠道画像及游戏分布 | GET /sw/promote/portrait、/sw/promote/game-distribution | 商务日报两表；付费分层还会读取原始充值 |
+| 区服分析 | GET /sw/serverAnalysis/list | 区服日报两表 |
+
+普通渠道列表、注册明细、充值明细、渠道激活、工作台未登录榜，以及游戏/区服下拉选项不直接使用这四张汇总表。预警检测 `alertTriggerTask` 另行读取原始注册/充值数据，不由这些汇总任务生成预警。
+
+今天任务用于当天持续刷新，昨天任务用于补算昨日完整数据，backfill 用于补历史或纠正迟到数据。2026-11-04 核实时六条计划全部停用：不自动刷新汇总，但已有历史结果仍可查询；若原始数据继续同步，依赖汇总的指标会与明细逐渐产生时间差。
@@ -65,4 +65,4 @@

 ## 页面

-渠道激活页不再在发起分页请求时清零 total；排行榜与仪表盘四榜并排、付费率用百分比、头部两项偏好直接勾选；三级商务保留商务列表/绑定列表切换。完整页面检查脚本仍保留，运行截图和带业务数据的响应仅存仓库外。
+渠道激活页不再在发起分页请求时清零 total；排行榜与仪表盘四榜并排、付费率用百分比、头部两项偏好直接勾选；三级商务保留商务列表/绑定列表切换。页面检查脚本、运行截图和带业务数据的响应均已保存在仓库外。
@@ -1,49 +0,0 @@
-# 商务验证工具
-
-仅保留仍用于当前 sw8/sw9 验证的工具。旧 sw7、固定端口 18000、重复页面夹具脚本已归档；接口边界回归优先运行相应 Go 测试。
-
-## 环境
-
- Python 3.10+；数据库工具需要 PyMySQL、PyYAML，Excel 对比需要 openpyxl。
- 页面检查使用 Node 和 Playwright；通过 `SW_PLAYWRIGHT_MODULE` 指定已有模块路径。
- `--runtime` 必须指向仓库外的私有目录。配置、令牌、实际响应、导出文件和 SQL 日志不得提交。
- 旧 Java 连接 sw8，新 Go 连接 sw9。运行命令时明确传入端口，避免误连日常服务。任务保持停用。
-
-通用输入：`legacy-<identity>.token`、`new-<identity>.token`、`legacy-client.json`、`id-maps.json`；写入测试还需要专门创建的三级测试身份及对应凭据。不要把业务正常账号当作可随意清理的测试身份。
-
-## 工具用途
-
-| 工具 | 用途与写入范围 |
-|---|---|
-| start-sw-rehearsal.ps1 | 从私有配置启动本机 Go 服务，临时注入 SDK 兼容参数，日志留在 runtime |
-| sw_database_rehearsal.py | 数据库盘点、备份、迁移及验证；同时提供其他脚本依赖的 connect/query |
-| sw_api_compare.py | 新旧读取接口逐字段对比；输入 case JSON 明确路径、参数、允许的协议转换 |
-| sw_write_verify.py | 规则与预警状态流程测试；提供其他写入工具的 Rehearsal 支持 |
-| sw_account_create_verify.py | 创建临时下级、验证层级/角色并清理；不删除预先存在的同名账号 |
-| sw_account_write_verify.py | 测试账号资料/状态并恢复原值 |
-| sw_preference_compare.py | 四种包含下级/身份证去重组合，结束恢复偏好 |
-| sw_export_compare.py | 新旧注册和充值 Excel 逐单元格对比 |
-| sw_sdk_write_verify.py | 独立平台夹具上的绑定、重提、解绑与渠道创建；受控 OA 审核 |
-| sw_report_rehearsal.py | 指定日报分区备份、比较和恢复 |
-| sw_run_report_tasks.py | 要求已有日期匹配的备份；创建停用的临时计划、手工执行并清理计划 |
-| sw_endpoint_benchmark.py | 接口耗时、完整响应摘要及 SQL 采集，只读 |
-| sw_endpoint_parity.py | 优化前后接口及偏好矩阵对比；即使 HTTP 恢复失败也尝试还原数据库偏好 |
-| sw_dashboard_layout_smoke.cjs | 激活分页、偏好和榜单布局检查，截图留在指定私有目录 |
-
-先通过各 Python 工具的 `--help` 查看必需参数。日报、预警及写入工具会改动测试数据；使用其备份/恢复流程，并核对最终表计数与任务状态。
-
-## 读取对比示例
-
-```powershell
-python scripts/sw_api_compare.py --runtime D:/private/sw-review --cases D:/private/sw-review/cases.json --old-url http://127.0.0.1:18080 --new-url http://127.0.0.1:19004 --output D:/private/sw-review/result.json
-```
-
-`--recompare` 只比较已保存响应；`--refresh-target` 使用已保存的同参数旧响应、重新请求新服务。报告会标记 `source_capture_reused`，不能把快照对比描述为双方重新实测。
-
-本地 MySQL 回归只接受 `SW_TEST_MYSQL_DSN` 指向 `127.0.0.1` 的隔离实例：
-
-```powershell
-go test ./internal/data/business -run MySQL -count=1
-```
-
-这些测试创建并清理自己的随机数据库，不在 sw8/sw9 上运行测试建表和删除。
@@ -1,37 +0,0 @@
-param(
-    [Parameter(Mandatory = $true)][string]$RuntimePath,
-    [string]$ConfigName = 'config-next.yaml',
-    [string]$BinaryPath = '.cache/sw89/sw-new-complete.exe'
-)
-
-$ErrorActionPreference = 'Stop'
-$workspaceRoot = (Resolve-Path (Join-Path $PSScriptRoot '..')).Path
-$runtimeRoot = (Resolve-Path -LiteralPath $RuntimePath).Path
-$configPath = (Resolve-Path -LiteralPath (Join-Path $runtimeRoot $ConfigName)).Path
-if (-not [System.IO.Path]::IsPathRooted($BinaryPath)) {
-    $BinaryPath = Join-Path $workspaceRoot $BinaryPath
-}
-$binary = (Resolve-Path -LiteralPath $BinaryPath).Path
-$compat = Get-Content -LiteralPath (Join-Path $runtimeRoot 'sdk-compatibility.json') -Raw | ConvertFrom-Json
-$previous = @{}
-$settings = @{
-    SW_SDK_AUTH_CODE = $compat.AUTH_CODE
-    SW_SDK_MIGRATE_AUTH_CODE = $compat.MIGRATE_AUTH_CODE
-    SW_SDK_TABLE_PREFIX = $compat.TABLE_PREFIX
-}
-try {
-    foreach ($name in $settings.Keys) {
-        $previous[$name] = [Environment]::GetEnvironmentVariable($name, 'Process')
-        [Environment]::SetEnvironmentVariable($name, $settings[$name], 'Process')
-    }
-    $process = Start-Process -FilePath $binary -ArgumentList @('-conf', ('"' + $configPath + '"')) `
-        -WorkingDirectory $workspaceRoot -WindowStyle Hidden `
-        -RedirectStandardOutput (Join-Path $runtimeRoot 'rehearsal.stdout.log') `
-        -RedirectStandardError (Join-Path $runtimeRoot 'rehearsal.stderr.log') -PassThru
-    $process.Id | Set-Content -LiteralPath (Join-Path $runtimeRoot 'rehearsal.pid')
-    Write-Output "Rehearsal process: $($process.Id)"
-} finally {
-    foreach ($name in $previous.Keys) {
-        [Environment]::SetEnvironmentVariable($name, $previous[$name], 'Process')
-    }
-}
@@ -1,73 +0,0 @@
-"""在 sw8/sw9 用专用二级账号创建临时三级账号，核对层级和角色后清理。"""
-
-import argparse
-from pathlib import Path
-
-from sw_write_verify import Rehearsal
-
-
-def main():
-    parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument("--runtime", type=Path, required=True)
-    parser.add_argument("--config", type=Path, required=True)
-    parser.add_argument("--old-url", default="http://127.0.0.1:18080")
-    parser.add_argument("--new-url", default="http://127.0.0.1:19002")
-    parser.add_argument("--output", type=Path, required=True)
-    args = parser.parse_args()
-    if args.runtime.resolve().is_relative_to(Path(__file__).resolve().parents[1]):
-        raise ValueError("测试凭据必须位于仓库外")
-    run = Rehearsal(args)
-    username = "mig_recheck_" + run.suffix
-    password = (args.runtime / "synthetic-test-password.txt").read_text(encoding="utf-8").strip()
-    source_roles = run.query("SELECT role_id FROM sw8.sys_user_role WHERE user_id=%s", (run.owner("legacy", 3),))
-    expected_roles = run.query("SELECT sys_authority_authority_id id FROM sw9.sys_user_authority WHERE sys_user_id=%s", (run.owner("new", 3),))
-    created_sides = set()
-    try:
-        run.check("测试角色存在", len(source_roles) == 1 and len(expected_roles) == 1)
-        for side, database, table, column in (("legacy", "sw8", "sys_user", "user_name"), ("new", "sw9", "sys_users", "username")):
-            run.check("临时账号不存在 " + side, not run.query(f"SELECT 1 FROM {database}.{table} WHERE {column}=%s", (username,)))
-        payload = {"userName": username, "password": password, "nickName": "迁移复核临时账号", "realName": "迁移复核临时账号"}
-        run.call("legacy", "testlevel2", "POST", "/system/user", {**payload, "roleIds": [str(source_roles[0]["role_id"])]})
-        created_sides.add("sw8")
-        run.call("new", "testlevel2", "POST", "/sw/business", {**payload, "parentId": 0, "level": 1, "authorityId": 888})
-        created_sides.add("sw9")
-        old = run.query("SELECT user_id,level,parent_id,id_path,status FROM sw8.sys_user WHERE user_name=%s", (username,))[0]
-        new = run.query("SELECT id,level,parent_id,id_path,business_status,authority_id,enable FROM sw9.sys_users WHERE username=%s", (username,))[0]
-        run.check("新旧创建三级且挂靠当前二级", old["level"] == new["level"] == 3 and str(old["parent_id"]) == str(run.owner("legacy", 2)) and new["parent_id"] == run.owner("new", 2))
-        path = ",".join(run.mapping.get(part, part) for part in str(old["id_path"]).split(","))
-        run.check("祖先路径按用户映射一致", path == new["id_path"])
-        run.check("默认状态正常且角色不能由请求提权", str(old["status"]) == str(new["business_status"]) == "0" and new["enable"] == 1 and new["authority_id"] == expected_roles[0]["id"] and new["authority_id"] != 888)
-        old_roles = run.query("SELECT role_id FROM sw8.sys_user_role WHERE user_id=%s", (old["user_id"],))
-        new_roles = run.query("SELECT sys_authority_authority_id id FROM sw9.sys_user_authority WHERE sys_user_id=%s", (new["id"],))
-        run.check("关联角色正确", old_roles == source_roles and new_roles == expected_roles)
-        values = []
-        for database, user_id in (("sw8", old["user_id"]), ("sw9", new["id"])):
-            values.append(run.query(f"SELECT include_sub_promote,idcard_dedup FROM {database}.sys_user_preference WHERE user_id=%s", (user_id,)))
-        run.check("默认统计偏好一致", values[0] == values[1] == [{"include_sub_promote": 1, "idcard_dedup": 1}])
-    finally:
-        # 只清理本次唯一名称创建的临时账号及其角色/偏好，不触碰原有测试链。
-        for database, table, id_col, name_col in (("sw8", "sys_user", "user_id", "user_name"), ("sw9", "sys_users", "id", "username")):
-            if database not in created_sides:
-                continue
-            rows = run.query(f"SELECT {id_col} id FROM {database}.{table} WHERE {name_col}=%s", (username,))
-            for row in rows:
-                user_id = row["id"]
-                run.db.begin()
-                try:
-                    run.query(f"DELETE FROM {database}.sys_user_preference WHERE user_id=%s", (user_id,))
-                    if database == "sw8":
-                        for relation in ("sys_user_role", "sys_user_post"):
-                            run.query(f"DELETE FROM sw8.{relation} WHERE user_id=%s", (user_id,))
-                    else:
-                        run.query("DELETE FROM sw9.sys_user_authority WHERE sys_user_id=%s", (user_id,))
-                    run.query(f"DELETE FROM {database}.{table} WHERE {id_col}=%s AND {name_col}=%s", (user_id, username))
-                    run.db.commit()
-                except Exception:
-                    run.db.rollback()
-                    raise
-            run.check("临时账号清理完成 " + database, not run.query(f"SELECT 1 FROM {database}.{table} WHERE {name_col}=%s", (username,)))
-        run.db.close()
-
-
-if __name__ == "__main__":
-    main()
@@ -1,75 +0,0 @@
-"""核对专用二级测试账号的资料更新、四种商务状态及恢复，不修改真实商务账号。"""
-
-import argparse
-import json
-import time
-from pathlib import Path
-
-from sw_write_verify import Rehearsal
-
-
-def main():
-    parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument("--runtime", type=Path, required=True)
-    parser.add_argument("--config", type=Path, default=Path("configs/config.yaml"))
-    parser.add_argument("--old-url", default="http://127.0.0.1:18080")
-    parser.add_argument("--new-url", default="http://127.0.0.1:19001")
-    parser.add_argument("--output", type=Path, required=True)
-    args = parser.parse_args()
-    run = Rehearsal(args)
-    account = run.users[1]
-    if not account["username"].startswith("mig_verify_"):
-        raise ValueError("仅允许修改本轮创建的专用测试账号")
-    originals = {
-        "legacy": run.query("SELECT real_name,status,reject_reason,update_time FROM sw8.sys_user WHERE user_id=%s AND user_name=%s", (account["source_id"], account["username"]))[0],
-        "new": run.query("SELECT real_name,business_status,enable,reject_reason,updated_at FROM sw9.sys_users WHERE id=%s AND username=%s", (account["target_id"], account["username"]))[0],
-    }
-    (args.runtime / "test-account-before-write.json").write_text(json.dumps(originals, default=str, indent=2), encoding="utf-8")
-
-    def update(side, status, name, reason):
-        payload = {"userId": account["source_id" if side == "legacy" else "target_id"], "userName": account["username"], "realName": name,
-                   "status": str(status), "rejectReason": reason or "", "roleIds": ["2087435436573081601"]}
-        return run.call(side, "admin", "PUT", "/system/user" if side == "legacy" else "/sw/business", payload)
-
-    try:
-        started = time.monotonic()
-        for index, status in enumerate((0, 3, 2, 1, 0)):
-            if index == 4:
-                # 旧控制器对相同请求体有五秒重复提交限制。
-                time.sleep(max(0, 5.2 - (time.monotonic() - started)))
-            name = "迁移核对二级资料更新"
-            reason = "迁移核对驳回" if status == 2 else ""
-            for side in ("legacy", "new"):
-                update(side, status, name, reason)
-                path = "/system/user/" + str(account["source_id"]) if side == "legacy" else "/sw/business/" + str(account["target_id"])
-                detail = run.call(side, "admin", "GET", path)["data"]
-                if side == "legacy":
-                    detail = detail["user"]
-                run.check(f"资料及状态 {status} {side}", detail["realName"] == name and str(detail["status"]) == str(status) and detail["level"] == 2)
-            enabled = run.query("SELECT enable FROM sw9.sys_users WHERE id=%s", (account["target_id"],))[0]["enable"]
-            run.check(f"商务状态 {status} 与启停一致", enabled == (1 if status == 0 else 0))
-            replies = {side: run.call(side, "testlevel2", "GET", run.prefix(side) + "/user/preference", success=False) for side in ("legacy", "new")}
-            if status != 0:
-                run.check(f"状态 {status} 拒绝旧会话", replies["new"]["code"] != 0,
-                          legacy_session_allowed=replies["legacy"]["code"] == 200, accepted_change="新系统每次请求核对商务状态，阻止停用/待审/驳回账号沿用旧会话")
-            else:
-                run.check("正常账号会话可用", replies["new"]["code"] == 0 and replies["legacy"]["code"] == 200)
-    finally:
-        for side in ("legacy", "new"):
-            original = originals[side]
-            update(side, original["status" if side == "legacy" else "business_status"], original["real_name"], original["reject_reason"])
-        old, new = originals["legacy"], originals["new"]
-        run.query("UPDATE sw8.sys_user SET real_name=%s,status=%s,reject_reason=%s,update_time=%s WHERE user_id=%s AND user_name=%s",
-                  (old["real_name"], old["status"], old["reject_reason"], old["update_time"], account["source_id"], account["username"]))
-        run.query("UPDATE sw9.sys_users SET real_name=%s,business_status=%s,enable=%s,reject_reason=%s,updated_at=%s WHERE id=%s AND username=%s",
-                  (new["real_name"], new["business_status"], new["enable"], new["reject_reason"], new["updated_at"], account["target_id"], account["username"]))
-        actual = {
-            "legacy": run.query("SELECT real_name,status,reject_reason,update_time FROM sw8.sys_user WHERE user_id=%s", (account["source_id"],))[0],
-            "new": run.query("SELECT real_name,business_status,enable,reject_reason,updated_at FROM sw9.sys_users WHERE id=%s", (account["target_id"],))[0],
-        }
-        run.check("测试账号原始资料和状态已恢复", actual == originals)
-        run.db.close()
-
-
-if __name__ == "__main__":
-    main()
@@ -1,201 +0,0 @@
-"""请求隔离的新旧商务实例；原始响应和令牌留在仓库外，报告只记录校验结果。"""
-
-import argparse
-import concurrent.futures
-import decimal
-import hashlib
-import json
-import re
-import time
-import urllib.error
-import urllib.parse
-import urllib.request
-from pathlib import Path
-
-
-def fetch(base, token, path, params, legacy, client_id="", timeout=120):
-    url = base.rstrip("/") + path
-    if params:
-        url += "?" + urllib.parse.urlencode(params, doseq=True)
-    headers = {"Authorization": "Bearer " + token} if legacy else {"x-token": token}
-    if legacy:
-        headers["clientid"] = client_id
-    request = urllib.request.Request(url, headers=headers)
-    started = time.monotonic()
-    try:
-        with urllib.request.urlopen(request, timeout=timeout) as response:
-            body = json.load(response)
-            return {"http": response.status, "elapsed_ms": round((time.monotonic() - started) * 1000), "body": body}
-    except (urllib.error.URLError, TimeoutError, ValueError) as error:
-        return {"http": None, "elapsed_ms": round((time.monotonic() - started) * 1000), "error": type(error).__name__}
-
-
-def data_of(reply):
-    body = reply.get("body", {})
-    if "rows" in body:
-        return {key: value for key, value in body.items() if key not in ("code", "msg")}
-    return body.get("data")
-
-
-def shape(value):
-    if isinstance(value, dict):
-        return {key: (len(item) if isinstance(item, list) else type(item).__name__) for key, item in value.items()}
-    return {"type": type(value).__name__, "length": len(value) if isinstance(value, list) else None}
-
-
-def expected_relative_pay_rate(row):
-    """用户确认的新口径：由旧的本期付费率及百分点差推导相对环比。"""
-    if row.get("momValue") is None:
-        return row
-    current = decimal.Decimal(str(row["metricValue"]))
-    difference = decimal.Decimal(str(row["momValue"]))
-    previous = current - difference
-    result = dict(row)
-    if previous <= 0:
-        result.update(momValue=None, momStatus=None)
-    else:
-        change = (difference / previous * 100).quantize(decimal.Decimal("0.01"), rounding=decimal.ROUND_HALF_UP)
-        result.update(momValue=format(abs(change) if change == 0 else change, ".2f"), momStatus=0 if change == 0 else 1 if change > 0 else -1)
-    return result
-
-
-def compare_data(source, target, case, users):
-    differences = []
-    omitted = {"source": case.get("source_omit_fields", []), "target": case.get("target_omit_fields", [])}
-    # 两边通过 HTTP 分别创建的专用测试账号，其创建时刻天然不同。
-    # 只忽略显式列出的夹具账号时间，真实迁移账号的所有时间字段仍须相等。
-    generated = {str(value) for value in case.get("generated_user_ids", [])}
-    generated_ids = {"source": generated, "target": {users.get(value, value) for value in generated}}
-    user_fields = set(case.get("user_fields", ["userId", "parentId", "creatorId", "businessId", "operatorId", "parentUserId"]))
-
-    def normalize(value, side, field="", path=""):
-        if isinstance(value, dict):
-            if side == "source" and field == "payRateRankList" and case.get("relative_pay_rate"):
-                value = expected_relative_pay_rate(value)
-            if side == "target" and value.get("available") is False:
-                differences.append({"path": path + ".available", "kind": "unavailable"})
-            if side == "source" and case.get("normalize_root_path") and value.get("level") in (0, 1, "0", "1") and value.get("idPath") is None:
-                value = {**value, "idPath": "0"}
-            excluded = set(omitted[side])
-            if str(value.get("userId")) in generated_ids[side]:
-                excluded.add("createTime")
-            return {key: normalize(item, side, key, path + "." + key) for key, item in value.items() if key not in excluded}
-        if isinstance(value, list):
-            rows = [normalize(item, side, field, path) for item in value]
-            keys = case.get("row_key")
-            if keys and rows and all(isinstance(row, dict) and all(key in row for key in keys) for row in rows):
-                rows.sort(key=lambda row: tuple(str(row[key]) for key in keys))
-            return rows
-        if value is None and field in case.get("null_empty_fields", []):
-            return ""
-        if side == "source" and case.get("map_users", True) and value is not None:
-            if field in user_fields:
-                value = users.get(str(value), str(value))
-            elif field == "idPath":
-                value = ",".join(users.get(item, item) for item in str(value).split(","))
-        return value
-
-    def walk(left, right, path="data"):
-        # 只兼容数值与 JSON 数字字符串；账号、手机号、区服编号等字符串保留前导零。
-        numeric_field = path.rsplit(".", 1)[-1] in case.get("numeric_fields", [])
-        if not isinstance(left, bool) and not isinstance(right, bool) and (numeric_field or isinstance(left, (int, float)) or isinstance(right, (int, float))):
-            try:
-                if decimal.Decimal(str(left)) == decimal.Decimal(str(right)):
-                    return
-            except decimal.InvalidOperation:
-                pass
-        if type(left) is not type(right):
-            differences.append({"path": path, "kind": "type", "source": type(left).__name__, "target": type(right).__name__})
-        elif isinstance(left, dict):
-            for key in sorted(left.keys() | right.keys()):
-                if key not in left or key not in right:
-                    differences.append({"path": path + "." + key, "kind": "missing_field", "side": "source" if key not in left else "target"})
-                else:
-                    walk(left[key], right[key], path + "." + key)
-        elif isinstance(left, list):
-            if len(left) != len(right):
-                differences.append({"path": path, "kind": "length", "source": len(left), "target": len(right)})
-            for i, (a, b) in enumerate(zip(left, right)):
-                walk(a, b, path + f"[{i}]")
-        elif left != right:
-            # 数据值只留在仓库外；公开报告保留字段路径和摘要供复核。
-            differences.append({"path": path, "kind": "value", "source_sha256": hashlib.sha256(str(left).encode()).hexdigest(), "target_sha256": hashlib.sha256(str(right).encode()).hexdigest()})
-
-    walk(normalize(source, "source"), normalize(target, "target"))
-    return {"semantic_equal": not differences, "difference_count": len(differences), "differences": differences[:30],
-            "normalization": {"user_ids_mapped": case.get("map_users", True), "numeric_representation": True,
-                              "user_fields": sorted(user_fields), "numeric_string_fields": case.get("numeric_fields", []),
-                              "normalize_root_path": case.get("normalize_root_path", False),
-                              "expected_relative_pay_rate": case.get("relative_pay_rate", False),
-                              "generated_user_creation_times": len(generated),
-                              "row_key": case.get("row_key"), "omitted_fields": omitted, "null_empty_fields": case.get("null_empty_fields", [])}}
-
-
-def main():
-    parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument("--runtime", type=Path, required=True)
-    parser.add_argument("--cases", type=Path, required=True)
-    parser.add_argument("--identity", default="admin")
-    parser.add_argument("--old-url", default="http://127.0.0.1:18080")
-    parser.add_argument("--new-url", default="http://127.0.0.1:19000")
-    parser.add_argument("--output", type=Path, required=True)
-    mode = parser.add_mutually_exclusive_group()
-    mode.add_argument("--recompare", action="store_true", help="使用已保存的同参数 HTTP 响应重新核对，不重复访问数据库")
-    mode.add_argument("--refresh-target", action="store_true", help="保留已采集的同参数旧响应，重新请求优化后的新实现；新增用例仍请求两边")
-    args = parser.parse_args()
-    root = Path(__file__).resolve().parents[1]
-    if args.runtime.resolve().is_relative_to(root):
-        raise ValueError("原始业务响应必须保存在仓库外")
-    tokens = [(args.runtime / (prefix + "-" + args.identity + ".token")).read_text(encoding="utf-8").strip() for prefix in ("legacy", "new")]
-    client_id = json.loads((args.runtime / "legacy-client.json").read_text(encoding="utf-8"))["client_id"]
-    users = {str(row["old_id"]): str(row["new_id"]) for row in json.loads((args.runtime / "id-maps.json").read_text(encoding="utf-8")) if row["entity"] == "user"}
-    cases = json.loads(args.cases.read_text(encoding="utf-8"))
-    reports = []
-    folder = args.runtime / ("responses-" + args.identity)
-    folder.mkdir(exist_ok=True)
-    with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
-        for case in cases:
-            capture = folder / (case["name"] + ".json")
-            reused_source = args.refresh_target and capture.exists()
-            if args.recompare or reused_source:
-                saved = json.loads(capture.read_text(encoding="utf-8"))
-                if any(saved["case"].get(key) != case.get(key) for key in ("source", "target", "params", "source_params", "target_params")):
-                    raise ValueError("已有 HTTP 响应的请求条件不符：" + case["name"])
-                replies = [saved["source"], saved["target"]]
-                if args.refresh_target:
-                    replies[1] = fetch(args.new_url, tokens[1], case["target"], case.get("target_params", case.get("params", {})), False, timeout=case.get("timeout_seconds", 120))
-                    capture.write_text(json.dumps({"case": case, "source": replies[0], "target": replies[1]}, ensure_ascii=False, indent=2), encoding="utf-8")
-            else:
-                pairs = [pool.submit(fetch, base, tokens[i], case[side], case.get(side + "_params", case.get("params", {})), i == 0, client_id, case.get("timeout_seconds", 120))
-                         for i, (base, side) in enumerate(((args.old_url, "source"), (args.new_url, "target")))]
-                replies = [job.result() for job in pairs]
-                capture.write_text(json.dumps({"case": case, "source": replies[0], "target": replies[1]}, ensure_ascii=False, indent=2), encoding="utf-8")
-            values = [data_of(reply) for reply in replies]
-            for i, side in enumerate(("source", "target")):
-                for key in case.get(side + "_data_path", []):
-                    values[i] = values[i][key]
-            codes = [reply.get("body", {}).get("code") for reply in replies]
-            report = {"case": case["name"], "source": case["source"], "target": case["target"], "codes": codes,
-                      "source_capture_reused": reused_source,
-                      "http_passed": codes == [200, 0], "elapsed_ms": [reply["elapsed_ms"] for reply in replies],
-                      "shapes": [shape(value) for value in values],
-                      "exact_data_equal": values[0] == values[1]}
-            if all(isinstance(value, dict) and "total" in value for value in values):
-                report["totals"] = [value["total"] for value in values]
-                report["totals_equal"] = str(values[0]["total"]) == str(values[1]["total"])
-            reports.append(report)
-            report.update(compare_data(values[0], values[1], case, users))
-            report["passed"] = report["http_passed"] and report["semantic_equal"]
-            if case.get("expected_error_codes"):
-                messages = [reply.get("body", {}).get("msg") for reply in replies]
-                report["rejection_equal"] = codes == case["expected_error_codes"] and messages[0] == messages[1]
-                report["expected_error_codes"] = case["expected_error_codes"]
-                report["passed"] = report["rejection_equal"]
-            print(json.dumps(report, ensure_ascii=False), flush=True)
-            args.output.write_text(json.dumps({"identity": args.identity, "checks": reports}, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
-    if not all(report["passed"] for report in reports):
-        raise SystemExit(1)
-
-
-if __name__ == "__main__":
-    main()
@@ -1,218 +0,0 @@
-// 真实开发服务：检查 OA 数据、统计口径持久化与响应式布局。
-// 临时修改当前测试账号的统计口径，结束时恢复；截图只写入指定的私有目录。
-const fs = require('node:fs/promises')
-const path = require('node:path')
-const assert = require('node:assert/strict')
-const { chromium } = require(process.env.SW_PLAYWRIGHT_MODULE || 'playwright')
-
-;(async () => {
-  const base = process.env.SW_UI_BASE_URL || 'http://127.0.0.1:8080'
-  const output = process.env.SW_UI_OUTPUT_DIR
-  assert.ok(output && process.env.SW_TEST_TOKEN_FILE, 'Set SW_UI_OUTPUT_DIR and SW_TEST_TOKEN_FILE')
-  const token = (await fs.readFile(process.env.SW_TEST_TOKEN_FILE, 'utf8')).trim()
-  const browser = await chromium.launch({ channel: 'chrome', headless: true })
-  const context = await browser.newContext({ viewport: { width: 1600, height: 1000 } })
-  const headers = { 'x-token': token }
-  const api = async (method, endpoint, data) => {
-    const response = await context.request.fetch(`${base}/api${endpoint}`, { method, headers, data })
-    assert.ok(response.ok())
-    const body = await response.json()
-    assert.equal(body.code, 0)
-    return body.data
-  }
-  let original
-  const report = { checks: [], pageErrors: [] }
-  try {
-    await fs.mkdir(output, { recursive: true })
-    original = await api('GET', '/sw/user/preference')
-    await fs.writeFile(path.join(output, 'original-preference.json'), JSON.stringify(original))
-    await context.addInitScript(value => localStorage.setItem('token', value), token)
-    const page = await context.newPage()
-    page.on('pageerror', e => report.pageErrors.push(e.message))
-    const activationPath = '/sw/promoteActivePool/bindList'
-    await page.goto(`${base}/#/layout/sw-BusMgmt/ChannelActivationList`)
-    await page.locator('.el-table__body-wrapper tr').first().waitFor()
-    const activation = await api('GET', `${activationPath}?pageNum=1&pageSize=10&platformNames=lx,hw,yq`)
-    assert.equal(activation.available, true)
-    assert.ok(activation.total > 0)
-    assert.equal(await page.locator('.el-table__body-wrapper tr').count(), activation.rows.length)
-    assert.equal(await page.locator('.el-alert--warning').count(), 0)
-    report.checks.push({ name: 'actual OA activation rows', total: activation.total })
-    await page.screenshot({ path: path.join(output, 'activation.png'), animations: 'disabled' })
-
-    const pagination = page.locator('.kra-pagination')
-    const pageRequests = []
-    const trackPage = request => {
-      if (request.url().includes(activationPath)) pageRequests.push(Number(new URL(request.url()).searchParams.get('pageNum')))
-    }
-    page.on('request', trackPage)
-    const movePage = async (target, action) => {
-      pageRequests.length = 0
-      const [response] = await Promise.all([
-        page.waitForResponse(r => r.url().includes(activationPath) && Number(new URL(r.url()).searchParams.get('pageNum')) === target),
-        action()
-      ])
-      const expected = (await response.json()).data
-      await page.waitForFunction(() => !document.querySelector('.el-loading-mask'))
-      assert.equal((await pagination.locator('.el-pager .is-active').innerText()).trim(), String(target))
-      assert.equal(await page.locator('.el-table__body-wrapper tr').count(), expected.rows.length)
-      const names = await page.locator('.el-table__body-wrapper tr td:nth-child(2) .cell').allTextContents()
-      assert.deepEqual(names.map(name => name.trim()), expected.rows.map(row => row.promoteAccount))
-      assert.deepEqual(pageRequests, [target], 'pagination must not trigger a second request back to page 1')
-    }
-    assert.ok(activation.total > 10, 'Pagination check needs more than one page')
-    await movePage(2, () => pagination.locator('.btn-next').click())
-    const lastPage = Math.ceil(activation.total / 10)
-    if (lastPage > 2) await movePage(lastPage, () => pagination.locator('.el-pager .number').filter({ hasText: new RegExp(`^${lastPage}$`) }).click())
-    await page.screenshot({ path: path.join(output, 'activation-last-page.png'), animations: 'disabled' })
-    await movePage(lastPage - 1, () => pagination.locator('.btn-prev').click())
-    if (lastPage - 1 !== 1) await movePage(1, () => pagination.locator('.el-pager .number').filter({ hasText: /^\s*1\s*$/ }).click())
-    page.off('request', trackPage)
-    report.checks.push({ name: 'activation pagination: next, last, previous, first; response rows match and no page reset', total: activation.total, lastPage })
-
-    const preference = page.getByRole('group', { name: '统计口径', exact: true })
-    for (const [key, label] of [['includeSubPromote', '包含下级渠道'], ['idcardDedup', '身份证去重']]) {
-      const checkbox = preference.getByRole('checkbox', { name: label, exact: true })
-      await page.waitForFunction(label => [...document.querySelectorAll('.business-preferences input')].some(el => el.closest('label')?.textContent.includes(label) && !el.disabled), label)
-      assert.equal(await checkbox.isChecked(), original[key])
-      const [saved] = await Promise.all([
-        page.waitForResponse(r => r.url().includes('/sw/user/preference') && r.request().method() === 'PUT'),
-        page.waitForResponse(r => r.url().includes(activationPath)),
-        preference.getByText(label, { exact: true }).click()
-      ])
-      assert.equal((await saved.json()).code, 0)
-      assert.equal((await api('GET', '/sw/user/preference'))[key], !original[key])
-      await page.reload()
-      await preference.waitFor()
-      await page.waitForFunction(() => [...document.querySelectorAll('.business-preferences input')].every(el => !el.disabled))
-      assert.equal(await preference.getByRole('checkbox', { name: label, exact: true }).isChecked(), !original[key])
-      report.checks.push({ name: `${key}: visible, persisted, page refreshed` })
-    }
-    await api('PUT', '/sw/user/preference', original)
-
-    // 保存失败不能让勾选状态与服务端失去同步。
-    await page.reload()
-    await page.waitForFunction(() => document.querySelectorAll('.business-preferences input').length === 2 && [...document.querySelectorAll('.business-preferences input')].every(el => !el.disabled))
-    const rejectSave = route => route.request().method() === 'PUT'
-      ? route.fulfill({ json: { code: 7, msg: '测试保存失败', data: {} } }) : route.continue()
-    await page.route('**/api/sw/user/preference', rejectSave)
-    await Promise.all([
-      page.waitForResponse(r => r.url().includes('/sw/user/preference') && r.request().method() === 'PUT'),
-      preference.getByText('包含下级渠道', { exact: true }).click()
-    ])
-    await page.waitForFunction(() => [...document.querySelectorAll('.business-preferences input')].every(el => !el.disabled))
-    assert.equal(await preference.getByRole('checkbox', { name: '包含下级渠道', exact: true }).isChecked(), original.includeSubPromote)
-    await page.unroute('**/api/sw/user/preference', rejectSave)
-    report.checks.push({ name: 'failed preference save preserves previous value' })
-
-    const search = () => page.locator('.kra-search-box').getByRole('button', { name: '查询', exact: true }).click()
-    const activationPattern = '**/api/sw/promoteActivePool/bindList?*'
-    await page.route(activationPattern, route => route.fulfill({ json: { code: 7, msg: '测试查询失败', data: {} } }))
-    await search()
-    await page.getByText('查询失败，请重试', { exact: true }).waitFor()
-    assert.equal(await page.locator('.el-table__body-wrapper tr').count(), 0)
-    await page.unroute(activationPattern)
-    await page.route(activationPattern, route => route.fulfill({ json: { code: 0, data: { available: false, rows: [], total: 0 } } }))
-    await search()
-    await page.getByText('OA 数据源暂不可用', { exact: true }).waitFor()
-    await page.unroute(activationPattern)
-    await search()
-    await page.locator('.el-table__body-wrapper tr').first().waitFor()
-    report.checks.push({ name: 'activation error and unavailable states differ from empty results; retry recovers' })
-
-    for (const [name, route, grid] of [
-      ['ranking', 'sw-DataAnalysis/RankingAnalysis', '.ranking-panels'],
-      ['dashboard', 'dashboard', '.business-ranks']
-    ]) {
-      await page.goto(`${base}/#/layout/${route}`)
-      await page.locator(grid).waitFor()
-      await page.waitForFunction(() => !document.querySelector('.el-loading-mask'))
-      await page.locator('.el-message').evaluateAll(nodes => nodes.forEach(node => node.querySelector('.el-message__closeBtn')?.click()))
-      if (name === 'ranking') {
-        for (const panel of await page.locator('.ranking-panel').all()) {
-          assert.match(await panel.innerText(), /上期环比/)
-          assert.doesNotMatch(await panel.innerText(), /百分点|\bpp\b/)
-        }
-      } else {
-        assert.equal(await page.getByText('快捷功能', { exact: true }).count(), 0)
-        assert.equal(await page.getByText('公告', { exact: true }).count(), 0)
-        assert.equal(await page.getByText('商务数据', { exact: true }).count(), 0)
-        assert.ok(await page.getByText('数据平台', { exact: true }).isVisible())
-        for (const index of [0, 1]) {
-          const panel = page.locator('.business-ranks > .el-card').nth(index)
-          await panel.locator('.business-rank-period').click()
-          const optionsId = await panel.getByRole('combobox').getAttribute('aria-controls')
-          await Promise.all([
-            page.waitForResponse(r => /\/sw\/workbench\/(registerTop|payRank)/.test(r.url()) && r.url().includes('period=last7')),
-            page.locator(`[id="${optionsId}"]`).getByRole('option', { name: '近7天', exact: true }).click()
-          ])
-          await panel.locator('.ranking-table tbody tr').first().waitFor()
-          assert.equal(await panel.locator('.ranking-empty').count(), 0)
-        }
-        report.checks.push({ name: 'dashboard period filters refresh real data in compact tables' })
-      }
-      for (const [width, columns] of [[1920, 4], [1600, 4], [1366, 4], [1024, 2], [390, 1]]) {
-        await page.setViewportSize({ width, height: 1000 })
-        if (width === 390) await page.getByRole('button', { name: '打开菜单', exact: true }).waitFor()
-        await page.waitForFunction(({grid, columns}) => getComputedStyle(document.querySelector(grid)).gridTemplateColumns.split(' ').length === columns, { grid, columns })
-        const boxes = await page.locator(`${grid} > *`).evaluateAll(nodes => nodes.map(n => { const r = n.getBoundingClientRect(); return { x: r.x, y: r.y, right: r.right } }))
-        if (columns === 4) assert.ok(boxes.every(box => Math.abs(box.y - boxes[0].y) < 1))
-        assert.ok(boxes.every(box => box.x >= 0 && box.right <= width + 1), `panel overflow: ${name}, ${width}`)
-        const overflow = await page.locator(`${grid} .el-table__body-wrapper, ${grid} .ranking-table-wrap`).evaluateAll(nodes => nodes.some(n => n.querySelector('table')?.clientWidth > n.clientWidth + 1))
-        assert.equal(overflow, false, `table horizontal overflow: ${name}, ${width}`)
-        if (name === 'ranking') {
-          const tallRows = await page.locator('.ranking-table tr').evaluateAll(nodes => nodes.some(n => n.getBoundingClientRect().height > 40))
-          assert.equal(tallRows, false, `ranking rows should stay compact: ${width}`)
-        }
-        const clippedNumbers = await page.locator(`${grid} tbody .ranking-numeric, ${grid} .ranking-number`).evaluateAll(nodes => nodes.some(n => n.scrollWidth > n.clientWidth + 1))
-        assert.equal(clippedNumbers, false, `numeric values must be fully visible: ${name}, ${width}`)
-        const prefBox = await preference.boundingBox()
-        assert.ok(prefBox && prefBox.x >= 0 && prefBox.x + prefBox.width <= width, `preference overflow at ${width}`)
-        const toolbarOverflow = await page.getByRole('button', { name: '打开菜单', exact: true }).count() ? await page.locator('.business-preferences').evaluate(el => {
-          const header = el.closest('.fixed')
-          return header.scrollWidth > header.clientWidth + 1
-        }) : false
-        assert.equal(toolbarOverflow, false, `mobile header overflow: ${width}`)
-        await page.screenshot({ path: path.join(output, `${name}-${width}.png`), animations: 'disabled' })
-        report.checks.push({ name: `${name} layout`, width, columns, noOverflow: true })
-      }
-      await page.setViewportSize({ width: 1600, height: 1000 })
-    }
-    await page.goto(`${base}/#/layout/sw-BusMgmt/BusinessList`)
-    await page.getByRole('button', { name: '添加子商务', exact: true }).waitFor()
-    assert.equal(await page.locator('.business-view-switch').count(), 0)
-    assert.equal(await page.getByRole('tab', { name: '商务列表', exact: true }).count(), 0)
-    await page.screenshot({ path: path.join(output, 'business-list-admin.png'), animations: 'disabled' })
-    // 只替换展示身份核对三级专属切换器；读取接口仍用授权测试身份，不做业务写入。
-    await page.route('**/api/user/getUserInfo', async route => {
-      const response = await route.fetch()
-      const body = await response.json()
-      Object.assign(body.data.userInfo, { level: 3, authorityId: 903 })
-      await route.fulfill({ response, json: body })
-    })
-    await page.reload()
-    await page.getByRole('radiogroup', { name: '商务管理视图', exact: true }).waitFor()
-    assert.equal(await page.getByRole('button', { name: '添加子商务', exact: true }).count(), 0)
-    const viewSwitch = page.getByRole('radiogroup', { name: '商务管理视图', exact: true })
-    await Promise.all([
-      page.waitForResponse(r => r.url().includes('/sw/business/bindPlatform')),
-      viewSwitch.getByText('绑定列表', { exact: true }).click()
-    ])
-    assert.equal(await page.getByPlaceholder('请输入账号', { exact: true }).isVisible(), false)
-    await viewSwitch.getByText('商务列表', { exact: true }).click()
-    assert.ok(await page.getByPlaceholder('请输入账号', { exact: true }).isVisible())
-    await page.screenshot({ path: path.join(output, 'business-list-level3.png'), animations: 'disabled' })
-    report.checks.push({ name: 'business list: no redundant single tab; controlled level-3 switch works' })
-    assert.deepEqual(report.pageErrors, [])
-    report.passed = true
-  } catch (e) {
-    const page = context.pages()[0]
-    if (page) await page.screenshot({ path: path.join(output, 'failure.png'), animations: 'disabled' }).catch(() => {})
-    throw e
-  } finally {
-    if (original) await api('PUT', '/sw/user/preference', original)
-    await fs.writeFile(path.join(output, 'ui-checks.json'), JSON.stringify(report, null, 2) + '\n')
-    await browser.close()
-  }
-  console.log(JSON.stringify(report))
-})().catch(e => { console.error(e.stack); process.exitCode = 1 })
@@ -1,188 +0,0 @@
-"""在同一测试服务器核对 SW 源副本、备份目标库并执行迁移；凭据只读取本地配置。"""
-
-import argparse
-import datetime as dt
-import gzip
-import hashlib
-import json
-import re
-import shutil
-import subprocess
-import tempfile
-from pathlib import Path
-
-import pymysql
-import yaml
-
-
-def identifier(value):
-    if not re.fullmatch(r"[A-Za-z0-9_]+", value):
-        raise ValueError("数据库和表名只能包含字母、数字、下划线")
-    return "`" + value + "`"
-
-
-def write_json(path, data):
-    path.write_text(json.dumps(data, ensure_ascii=False, indent=2, default=str) + "\n", encoding="utf-8")
-
-
-def connect(config):
-    return pymysql.connect(host=config["host"], port=int(config["port"]), user=config["user"],
-                           password=config["password"], charset="utf8mb4", connect_timeout=10,
-                           read_timeout=180, autocommit=True, cursorclass=pymysql.cursors.DictCursor)
-
-
-def query(conn, sql, args=None):
-    with conn.cursor() as cursor:
-        cursor.execute(sql, args)
-        return cursor.fetchall()
-
-
-def source_tables(sql):
-    tables = re.findall(r"CREATE TABLE `bak_sw2kra_20261103_([^`]+)` LIKE", sql)
-    if len(tables) != 26 or len(set(tables)) != 26:
-        raise ValueError("迁移源表清单发生变化，请先复核备份范围")
-    return tables
-
-
-def fingerprint(conn, database, table):
-    columns = query(conn, "SELECT column_name,data_type FROM information_schema.columns WHERE table_schema=%s AND table_name=%s ORDER BY ordinal_position", (database, table))
-    keys = query(conn, "SELECT column_name FROM information_schema.statistics WHERE table_schema=%s AND table_name=%s AND index_name='PRIMARY' ORDER BY seq_in_index", (database, table))
-    if not keys:
-        raise ValueError("备份核对要求稳定主键：" + table)
-    order = ",".join(identifier(item["column_name"]) for item in keys)
-    digest = hashlib.sha256()
-    rows = query(conn, f"SELECT * FROM {identifier(database)}.{identifier(table)} ORDER BY {order}")
-    json_columns = [item["column_name"] for item in columns if item["data_type"] == "json"]
-    for row in rows:
-        # JSON 的内部二进制布局在导入后可能不同，比较实际业务值而非 CHECKSUM 的存储字节。
-        for name in json_columns:
-            if row[name] is not None:
-                row[name] = json.loads(row[name])
-        digest.update(json.dumps(row, ensure_ascii=False, sort_keys=True, default=str, separators=(",", ":")).encode())
-        digest.update(b"\n")
-    return {"rows": len(rows), "sha256": digest.hexdigest()}
-
-
-def client_options(config, folder):
-    # 使用独立选项文件，避免密码出现在进程命令行或输出中。
-    def escape(value):
-        return str(value).replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n").replace("\r", "\\r")
-    handle = tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", suffix=".cnf", dir=folder, delete=False)
-    with handle:
-        handle.write("[client]\n")
-        for key in ("host", "port", "user", "password"):
-            handle.write(f'{key}="{escape(config[key])}"\n')
-        handle.write("default-character-set=utf8mb4\n")
-    return Path(handle.name)
-
-
-def prepare(args, config, sql):
-    if (args.state / (args.target + "-migration-source.sql.gz")).exists():
-        raise ValueError("目标备份已存在；请使用新的状态目录，保留原始备份")
-    tables = source_tables(sql)
-    manifest = {"source": args.source, "target": args.target, "created_at": dt.datetime.now(dt.timezone.utc),
-                "sql_sha256": hashlib.sha256(sql.encode()).hexdigest(), "checks": []}
-    with connect(config) as conn:
-        manifest["server"] = query(conn, "SELECT @@hostname hostname,@@version version")[0]
-        if query(conn, "SELECT 1 FROM information_schema.tables WHERE table_schema=%s AND table_name='sys_users'", (args.target,)):
-            raise ValueError("目标已迁移，不能覆盖原始基线")
-        for table in tables:
-            checksums = query(conn, f"CHECKSUM TABLE {identifier(args.source)}.{identifier(table)}, {identifier(args.target)}.{identifier(table)} EXTENDED")
-            values = [row["Checksum"] for row in checksums]
-            fingerprints = [fingerprint(conn, db, table) for db in (args.source, args.target)]
-            item = {"table": table, "rows": [item["rows"] for item in fingerprints], "checksums": values,
-                    "fingerprints": fingerprints, "passed": fingerprints[0] == fingerprints[1]}
-            manifest["checks"].append(item)
-            print(json.dumps(item, ensure_ascii=False), flush=True)
-        if not all(item["passed"] for item in manifest["checks"]):
-            write_json(args.state / "baseline.json", manifest)
-            raise RuntimeError("源库与目标库框架数据不一致，未执行迁移")
-        manifest["tables_before"] = query(conn, "SELECT table_name,engine FROM information_schema.tables WHERE table_schema=%s ORDER BY table_name", (args.target,))
-        schemas = {}
-        for row in manifest["tables_before"]:
-            table = row["table_name"]
-            schemas[table] = query(conn, f"SHOW CREATE TABLE {identifier(args.target)}.{identifier(table)}")[0]["Create Table"]
-        write_json(args.state / "schema-before.json", schemas)
-    dump = Path(shutil.which("mysqldump") or Path(shutil.which("mysql")).with_name("mysqldump.exe"))
-    options = client_options(config, args.state)
-    backup = args.state / (args.target + "-migration-source.sql.gz")
-    try:
-        command = [str(dump), "--defaults-extra-file=" + str(options), "--single-transaction", "--quick", "--hex-blob",
-                   "--set-gtid-purged=OFF", "--skip-lock-tables", args.target, *tables]
-        with (args.state / "backup.stderr.log").open("wb") as errors:
-            process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=errors)
-            with gzip.open(backup, "wb") as output:
-                shutil.copyfileobj(process.stdout, output)
-            if process.wait() != 0:
-                raise RuntimeError("备份失败，详情仅保存在本地日志")
-        # 完整解压检查，避免把未写完的压缩文件当作成功备份。
-        with gzip.open(backup, "rb") as stream:
-            while stream.read(1024 * 1024):
-                pass
-        manifest["backup"] = {"path": str(backup.resolve()), "bytes": backup.stat().st_size,
-                              "sha256": hashlib.sha256(backup.read_bytes()).hexdigest()}
-        write_json(args.state / "baseline.json", manifest)
-        print(json.dumps({"backup": manifest["backup"], "passed": True}), flush=True)
-    finally:
-        options.unlink(missing_ok=True)
-
-
-def migrate(args, config, sql):
-    manifest = json.loads((args.state / "baseline.json").read_text(encoding="utf-8"))
-    if (manifest["source"], manifest["target"]) != (args.source, args.target):
-        raise ValueError("备份对应的数据库与本次选择不符")
-    backup = Path(manifest["backup"]["path"])
-    if hashlib.sha256(backup.read_bytes()).hexdigest() != manifest["backup"]["sha256"]:
-        raise ValueError("备份文件校验失败")
-    statements = re.findall(r"(?m)^USE `[^`]+`;", sql)
-    if len(statements) != 1:
-        raise ValueError("SQL 必须只有一个显式 USE 入口")
-    prepared = sql.replace(statements[0], "USE " + identifier(args.target) + ";")
-    (args.state / "executed-migration.sql").write_text(prepared, encoding="utf-8")
-    with connect(config) as conn:
-        actual_server = query(conn, "SELECT @@hostname hostname,@@version version")[0]
-        if actual_server != manifest["server"]:
-            raise ValueError("连接服务器与备份时不同")
-        # 仅首次执行需要核对原始基线；重复执行由迁移 SQL 的完成账本保护。
-        migrated = query(conn, "SELECT 1 FROM information_schema.tables WHERE table_schema=%s AND table_name='sys_users'", (args.target,))
-        if not migrated:
-            for item in manifest["checks"]:
-                if fingerprint(conn, args.target, item["table"]) != item["fingerprints"][1]:
-                    raise ValueError("目标数据在备份后变化：" + item["table"])
-    options = client_options(config, args.state)
-    try:
-        with (args.state / "migration.stdout.log").open("wb") as output, (args.state / "migration.stderr.log").open("wb") as errors:
-            result = subprocess.run([shutil.which("mysql"), "--defaults-extra-file=" + str(options), "--batch", "--binary-mode"],
-                                    input=prepared.encode("utf-8"), stdout=output, stderr=errors)
-        print(json.dumps({"target": args.target, "exit_code": result.returncode,
-                          "sql_sha256": hashlib.sha256(prepared.encode()).hexdigest()}), flush=True)
-        if result.returncode:
-            raise RuntimeError("迁移 SQL 失败，请检查本地日志；不会自动覆盖部分迁移状态")
-    finally:
-        options.unlink(missing_ok=True)
-
-
-def main():
-    parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument("action", choices=("prepare", "migrate"))
-    parser.add_argument("--config", type=Path, default=Path("configs/config.yaml"))
-    parser.add_argument("--sql", type=Path, default=Path("docs/sw6-to-kra/sw_to_kra_migrate.sql"))
-    parser.add_argument("--source", default="sw8")
-    parser.add_argument("--target", default="sw9")
-    parser.add_argument("--state", type=Path, required=True)
-    args = parser.parse_args()
-    identifier(args.source)
-    identifier(args.target)
-    if args.source == args.target or args.target in ("sw", "sw6"):
-        raise ValueError("演练目标必须为独立测试副本")
-    args.state = args.state.resolve()
-    if args.state.is_relative_to(Path(__file__).resolve().parents[1]):
-        raise ValueError("含账号数据的备份必须保存在仓库外")
-    args.state.mkdir(parents=True, exist_ok=True)
-    config = yaml.safe_load(args.config.read_text(encoding="utf-8"))["data"]["database"]
-    sql = args.sql.read_text(encoding="utf-8")
-    {"prepare": prepare, "migrate": migrate}[args.action](args, config, sql)
-
-
-if __name__ == "__main__":
-    main()
@@ -1,76 +0,0 @@
-"""只读请求商务接口，记录耗时、完整数据摘要及 SQL；敏感内容必须存放仓库外。"""
-
-import argparse
-import hashlib
-import json
-import re
-import statistics
-import time
-import urllib.parse
-import urllib.request
-from pathlib import Path
-
-
-def log_offsets(directory):
-    return {p: p.stat().st_size for p in directory.rglob("*.log")}
-
-
-def read_queries(directory, offsets):
-    queries = []
-    files = list(directory.rglob("*.log"))
-    files = [file for file in files if file.parent.name == "sql"] or files
-    for file in files:
-        with file.open("rb") as stream:
-            stream.seek(offsets.get(file, 0))
-            for line in stream:
-                try:
-                    row = json.loads(line)
-                except (ValueError, UnicodeDecodeError):
-                    continue
-                if row.get("gorm_logger") and "sql" in row:
-                    queries.append(row)
-    return queries
-
-
-def main():
-    parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument("--runtime", type=Path, required=True)
-    parser.add_argument("--base-url", default="http://127.0.0.1:19004")
-    parser.add_argument("--phase", required=True)
-    parser.add_argument("--repeat", type=int, default=2)
-    parser.add_argument("--cases", type=Path)
-    args = parser.parse_args()
-    if args.runtime.resolve().is_relative_to(Path(__file__).resolve().parents[1]):
-        raise ValueError("完整接口响应和 SQL 必须留在仓库外")
-    output = args.runtime / args.phase
-    output.mkdir(exist_ok=True)
-    cases = json.loads((args.cases or args.runtime / "cases.json").read_text(encoding="utf-8"))
-    token = (args.runtime / "new-admin.token").read_text(encoding="utf-8").strip()
-    report = []
-    for case in cases:
-        elapsed = []
-        for round_number in range(args.repeat):
-            offsets = log_offsets(args.runtime / "sql-logs")
-            url = args.base_url + case["path"] + "?" + urllib.parse.urlencode(case["params"], doseq=True)
-            request = urllib.request.Request(url, headers={"x-token": token, "X-Request-Id": f"perf-{args.phase}-{case['name']}-{round_number}"})
-            started = time.monotonic()
-            with urllib.request.urlopen(request, timeout=300) as response:
-                body = json.load(response)
-            elapsed.append(round((time.monotonic() - started) * 1000, 2))
-            if body.get("code") != 0:
-                raise RuntimeError("请求失败：" + case["name"])
-            queries = read_queries(args.runtime / "sql-logs", offsets)
-            data = body["data"]
-            capture = {"case": case, "data": data, "elapsedMs": elapsed[-1], "queries": queries}
-            (output / f"{case['name']}-{round_number}.json").write_text(json.dumps(capture, ensure_ascii=False, indent=2), encoding="utf-8")
-            digest = hashlib.sha256(json.dumps(data, sort_keys=True, ensure_ascii=False).encode()).hexdigest()
-            slow = sorted(queries, key=lambda row: row.get("elapsed_ms", 0), reverse=True)[:5]
-            item = {"case": case["name"], "round": round_number, "elapsedMs": elapsed[-1], "dataSha256": digest, "queryCount": len(queries), "slowQueries": [{"elapsedMs": q.get("elapsed_ms"), "rows": q.get("rows"), "tables": list(dict.fromkeys(re.findall(r"(?:FROM|JOIN)\s+`?([\w.]+)", q["sql"], re.I)))} for q in slow]}
-            report.append(item)
-            (output / "report.json").write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8")
-            print(json.dumps(item), flush=True)
-        print(json.dumps({"case": case["name"], "medianMs": statistics.median(elapsed)}), flush=True)
-
-
-if __name__ == "__main__":
-    main()
@@ -1,79 +0,0 @@
-"""在同一个 sw9 快照上核对优化前后接口，涵盖分页/排序及统计偏好。"""
-
-import argparse
-import json
-import urllib.parse
-import urllib.request
-from pathlib import Path
-
-import yaml
-
-from sw_database_rehearsal import connect, query
-
-
-def main():
-    parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument("--runtime", type=Path, required=True)
-    parser.add_argument("--before-url", default="http://127.0.0.1:19002")
-    parser.add_argument("--after-url", default="http://127.0.0.1:19004")
-    parser.add_argument("--output", type=Path, required=True)
-    args = parser.parse_args()
-    if args.runtime.resolve().is_relative_to(Path(__file__).resolve().parents[1]):
-        raise ValueError("原始响应必须保存在仓库外")
-    config = yaml.safe_load((args.runtime / "config.yaml").read_text(encoding="utf-8"))["data"]["database"]
-    if config["name"] != "sw9":
-        raise ValueError("统计偏好测试仅允许 sw9")
-    folder = args.runtime / "parity"
-    folder.mkdir(exist_ok=True)
-    reports = []
-
-    def api(base, token, path, params=None, payload=None):
-        url = base + path + ("?" + urllib.parse.urlencode(params, doseq=True) if params else "")
-        request = urllib.request.Request(url, headers={"x-token": token, "Content-Type": "application/json"}, data=json.dumps(payload).encode() if payload is not None else None, method="PUT" if payload is not None else "GET")
-        with urllib.request.urlopen(request, timeout=300) as response:
-            data = json.load(response)
-        if data.get("code") != 0:
-            raise RuntimeError("接口请求失败：" + path)
-        return data["data"]
-
-    def compare(case, token, group):
-        values = [api(base, token, case["path"], case["params"]) for base in (args.before_url, args.after_url)]
-        name = group + "-" + case["name"]
-        (folder / (name + ".json")).write_text(json.dumps({"case": case, "before": values[0], "after": values[1]}, ensure_ascii=False, indent=2), encoding="utf-8")
-        row = {"case": name, "passed": values[0] == values[1]}
-        reports.append(row)
-        args.output.write_text(json.dumps({"checks": reports, "passed": all(row["passed"] for row in reports)}, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
-        print(json.dumps(row), flush=True)
-        if not row["passed"]:
-            raise AssertionError("接口数据不一致：" + name)
-
-    admin = (args.runtime / "new-admin.token").read_text(encoding="utf-8").strip()
-    for case in json.loads((args.runtime / "parity-cases.json").read_text(encoding="utf-8")):
-        compare(case, admin, "admin")
-
-    token = (args.runtime / "new-level3.token").read_text(encoding="utf-8").strip()
-    user = api(args.before_url, token, "/user/getUserInfo")["userInfo"]["ID"]
-    original_preference = api(args.before_url, token, "/sw/user/preference")
-    cases = json.loads((args.runtime / "cases.json").read_text(encoding="utf-8"))
-    with connect(config) as db:
-        original = query(db, "SELECT * FROM sw9.sys_user_preference WHERE user_id=%s", (user,))
-        try:
-            for include, dedup in ((True, True), (True, False), (False, True), (False, False)):
-                api(args.before_url, token, "/sw/user/preference", payload={"includeSubPromote": include, "idcardDedup": dedup})
-                for case in cases:
-                    compare(case, token, f"level3-sub{int(include)}-dedup{int(dedup)}")
-        finally:
-            try:
-                api(args.before_url, token, "/sw/user/preference", payload=original_preference)
-            finally:
-                # Restore the row even if the service stopped during verification.
-                if original:
-                    row = original[0]
-                    query(db, "UPDATE sw9.sys_user_preference SET include_sub_promote=%s,idcard_dedup=%s,update_time=%s WHERE user_id=%s", (row["include_sub_promote"], row["idcard_dedup"], row["update_time"], user))
-                else:
-                    query(db, "DELETE FROM sw9.sys_user_preference WHERE user_id=%s", (user,))
-                assert query(db, "SELECT * FROM sw9.sys_user_preference WHERE user_id=%s", (user,)) == original
-
-
-if __name__ == "__main__":
-    main()
@@ -1,76 +0,0 @@
-"""以同一主账号的实际数据请求新旧明细导出，并逐单元格核对，文件保留在仓库外。"""
-
-import argparse
-import decimal
-import io
-import json
-import time
-import urllib.parse
-import urllib.request
-from pathlib import Path
-
-import openpyxl
-
-
-def main():
-    parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument("--runtime", type=Path, required=True)
-    parser.add_argument("--old-url", default="http://127.0.0.1:18080")
-    parser.add_argument("--new-url", default="http://127.0.0.1:19001")
-    parser.add_argument("--output", type=Path, required=True)
-    args = parser.parse_args()
-    if args.runtime.resolve().is_relative_to(Path(__file__).resolve().parents[1]):
-        raise ValueError("导出文件必须保存在仓库外")
-    client = json.loads((args.runtime / "legacy-client.json").read_text(encoding="utf-8"))["client_id"]
-    headers = [{"Authorization": "Bearer " + (args.runtime / "legacy-admin.token").read_text().strip(), "clientid": client},
-               {"x-token": (args.runtime / "new-admin.token").read_text().strip()}]
-    reports = []
-    for kind, time_key in (("register", "Register"), ("spend", "Pay")):
-        params = {"sourceDb": "lx", "pageSize": 10, "begin" + time_key + "Time": "2026-11-03", "end" + time_key + "Time": "2026-11-03"}
-        request = urllib.request.Request(args.new_url + "/sw/" + kind + "/list?" + urllib.parse.urlencode(params), headers=headers[1])
-        with urllib.request.urlopen(request, timeout=60) as response:
-            body = json.load(response)
-        if body.get("code") != 0:
-            raise RuntimeError("读取导出样本失败：" + kind)
-        sample = next(row for row in body["data"]["rows"] if row.get("userId") is not None)
-        params["userId"] = sample["userId"]
-        sheets, durations = [], []
-        for i, base in enumerate((args.old_url, args.new_url)):
-            payload = urllib.parse.urlencode(params).encode()
-            request = urllib.request.Request(base + ("/system/" if i == 0 else "/sw/") + kind + "/export", data=payload,
-                                             headers={**headers[i], "Content-Type": "application/x-www-form-urlencoded"}, method="POST")
-            started = time.monotonic()
-            with urllib.request.urlopen(request, timeout=120) as response:
-                raw = response.read()
-            durations.append(round((time.monotonic() - started) * 1000))
-            if not raw.startswith(b"PK"):
-                raise RuntimeError("导出未返回工作簿：" + kind + "/" + str(i))
-            (args.runtime / (("legacy-" if i == 0 else "new-") + kind + "-export.xlsx")).write_bytes(raw)
-            workbook = openpyxl.load_workbook(io.BytesIO(raw), read_only=True, data_only=True)
-            sheets.append(list(workbook.worksheets[0].values))
-            workbook.close()
-        differences = []
-        for row_number, (old, new) in enumerate(zip(*sheets), 1):
-            if len(old) != len(new):
-                differences.append({"row": row_number, "kind": "columns"})
-            for col, (a, b) in enumerate(zip(old, new), 1):
-                if a == b or (a in (None, "") and b in (None, "")):
-                    continue
-                if isinstance(a, (int, float)) or isinstance(b, (int, float)):
-                    try:
-                        if decimal.Decimal(str(a)) == decimal.Decimal(str(b)):
-                            continue
-                    except decimal.InvalidOperation:
-                        pass
-                differences.append({"row": row_number, "column": col, "kind": "value"})
-        rows = [len(sheet) - 1 for sheet in sheets]
-        reports.append({"kind": kind, "rows": rows, "columns": [len(sheet[0]) for sheet in sheets], "elapsed_ms": durations,
-                        "differences": differences, "passed": rows[0] > 0 and rows[0] == rows[1] and not differences})
-        print(json.dumps(reports[-1], ensure_ascii=False), flush=True)
-        args.output.write_text(json.dumps({"checks": reports, "passed": all(row["passed"] for row in reports)}, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
-    if not all(row["passed"] for row in reports):
-        raise SystemExit(1)
-
-
-if __name__ == "__main__":
-    main()
@@ -1,61 +0,0 @@
-"""用已有真实绑定的三级账号逐项切换统计偏好，对比报表后恢复原始偏好行。"""
-
-import argparse
-import json
-import subprocess
-import sys
-from pathlib import Path
-
-from sw_write_verify import Rehearsal
-
-
-def main():
-    parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument("--runtime", type=Path, required=True)
-    parser.add_argument("--config", type=Path, default=Path("configs/config.yaml"))
-    parser.add_argument("--old-url", default="http://127.0.0.1:18080")
-    parser.add_argument("--new-url", default="http://127.0.0.1:19001")
-    parser.add_argument("--output", type=Path, required=True)
-    args = parser.parse_args()
-    run = Rehearsal(args)
-    identity = json.loads((args.runtime / "test-business-identities.json").read_text())["level3"]
-    ids = {"legacy": identity["user_id"], "new": run.mapping[str(identity["user_id"])]}
-    originals, preferences = {}, {}
-    for side, db in (("legacy", "sw8"), ("new", "sw9")):
-        originals[side] = run.query(f"SELECT * FROM {db}.sys_user_preference WHERE user_id=%s", (ids[side],))
-        preferences[side] = run.call(side, "level3", "GET", run.prefix(side) + "/user/preference")["data"]
-    (args.runtime / "preferences-before-matrix.json").write_text(json.dumps(originals, ensure_ascii=False, indent=2, default=str), encoding="utf-8")
-    templates = json.loads((args.runtime / "level3-business-cases.json").read_text())
-    templates = [row for row in templates if row["name"] in ("channelData-list", "gameData-list", "rank-game", "workbench-summary")]
-    try:
-        for include, dedup in ((True, True), (True, False), (False, True), (False, False)):
-            mode = f"sub{int(include)}-dedup{int(dedup)}"
-            value = {"includeSubPromote": include, "idcardDedup": dedup}
-            for side in ("legacy", "new"):
-                run.call(side, "level3", "PUT", run.prefix(side) + "/user/preference", value)
-                actual = run.call(side, "level3", "GET", run.prefix(side) + "/user/preference")["data"]
-                run.check("偏好写入 " + mode + " " + side, actual == value)
-            cases = [{**row, "name": mode + "-" + row["name"]} for row in templates]
-            cases_file = args.runtime / ("preference-cases-" + mode + ".json")
-            cases_file.write_text(json.dumps(cases), encoding="utf-8")
-            output = args.output.with_name(args.output.stem + "-" + mode + args.output.suffix)
-            with (args.runtime / ("preference-" + mode + ".log")).open("wb") as log:
-                completed = subprocess.run([sys.executable, str(Path(__file__).with_name("sw_api_compare.py")), "--runtime", str(args.runtime), "--cases", str(cases_file),
-                                            "--identity", "level3", "--old-url", args.old_url, "--new-url", args.new_url, "--output", str(output)], stdout=log, stderr=subprocess.STDOUT)
-            run.check("偏好模式报表一致 " + mode, completed.returncode == 0, report=output.name)
-    finally:
-        for side, db in (("legacy", "sw8"), ("new", "sw9")):
-            run.call(side, "level3", "PUT", run.prefix(side) + "/user/preference", preferences[side])
-            if not originals[side]:
-                run.query(f"DELETE FROM {db}.sys_user_preference WHERE user_id=%s", (ids[side],))
-            else:
-                row = originals[side][0]
-                run.query(f"UPDATE {db}.sys_user_preference SET include_sub_promote=%s,idcard_dedup=%s,update_time=%s WHERE user_id=%s",
-                          (row["include_sub_promote"], row["idcard_dedup"], row["update_time"], ids[side]))
-            actual = run.query(f"SELECT * FROM {db}.sys_user_preference WHERE user_id=%s", (ids[side],))
-            run.check("偏好原始行已恢复 " + side, actual == originals[side])
-        run.db.close()
-
-
-if __name__ == "__main__":
-    main()
@@ -1,119 +0,0 @@
-"""备份/比较/恢复 sw8 与 sw9 指定自然日的四张报表，不备份或改写原始业务流水。"""
-
-import argparse
-import datetime as dt
-import gzip
-import hashlib
-import json
-import shutil
-import subprocess
-from pathlib import Path
-
-import yaml
-
-from sw_database_rehearsal import client_options, connect, query, write_json
-
-TABLES = ("tab_sw_report_day", "tab_sw_report_day_key", "tab_sw_report_server_day", "tab_sw_report_server_day_key")
-
-
-def snapshot(conn, database, day, logical=False):
-    result = {}
-    for table in TABLES:
-        rows = query(conn, f"SELECT * FROM `{database}`.`{table}` WHERE stat_date=%s ORDER BY id", (day,))
-        if logical:
-            rows = [{key: value for key, value in row.items() if key not in ("id", "created_at", "updated_at")} for row in rows]
-        serialized = [json.dumps(row, ensure_ascii=False, sort_keys=True, separators=(",", ":"), default=str) for row in rows]
-        if logical:
-            serialized.sort()
-        result[table] = {"rows": len(rows), "sha256": hashlib.sha256("\n".join(serialized).encode()).hexdigest()}
-    return result
-
-
-def main():
-    parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument("action", choices=("backup", "verify", "restore"))
-    parser.add_argument("--config", type=Path, default=Path("configs/config.yaml"))
-    parser.add_argument("--date", default="2026-11-03")
-    parser.add_argument("--state", type=Path, required=True)
-    parser.add_argument("--output", type=Path)
-    args = parser.parse_args()
-    dt.date.fromisoformat(args.date)
-    args.state = args.state.resolve()
-    if args.state.is_relative_to(Path(__file__).resolve().parents[1]):
-        raise ValueError("报表备份和身份键必须保存在仓库外")
-    args.state.mkdir(parents=True, exist_ok=True)
-    config = yaml.safe_load(args.config.read_text(encoding="utf-8"))["data"]["database"]
-    manifest_path = args.state / "report-baseline.json"
-    if args.action == "backup":
-        if manifest_path.exists():
-            raise ValueError("已有基线备份，禁止覆盖")
-        manifest = {"date": args.date, "databases": {}}
-        with connect(config) as conn:
-            manifest["server"] = query(conn, "SELECT @@hostname hostname,@@version version")[0]
-            for database in ("sw8", "sw9"):
-                baseline = snapshot(conn, database, args.date)
-                options = client_options(config, args.state)
-                file = args.state / (database + "-report-partition.sql.gz")
-                try:
-                    with (args.state / (database + "-dump.stderr.log")).open("wb") as log:
-                        process = subprocess.Popen([shutil.which("mysqldump"), "--defaults-extra-file=" + str(options), "--single-transaction", "--quick", "--hex-blob",
-                                                    "--no-create-info", "--skip-triggers", "--skip-add-locks", "--skip-disable-keys", "--complete-insert", "--set-gtid-purged=OFF",
-                                                    "--where=stat_date='" + args.date + "'", database, *TABLES], stdout=subprocess.PIPE, stderr=log)
-                        with gzip.open(file, "wb") as target:
-                            shutil.copyfileobj(process.stdout, target)
-                        if process.wait() != 0:
-                            raise RuntimeError("报表备份失败")
-                    with gzip.open(file, "rb") as stream:
-                        while stream.read(1024 * 1024):
-                            pass
-                    manifest["databases"][database] = {"tables": baseline, "backup": str(file), "sha256": hashlib.sha256(file.read_bytes()).hexdigest()}
-                finally:
-                    options.unlink(missing_ok=True)
-        write_json(manifest_path, manifest)
-        print(json.dumps({"date": args.date, "tables": {db: value["tables"] for db, value in manifest["databases"].items()}}, ensure_ascii=False), flush=True)
-        return
-    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
-    if manifest["date"] != args.date:
-        raise ValueError("日期与备份不符")
-    with connect(config) as conn:
-        if query(conn, "SELECT @@hostname hostname,@@version version")[0] != manifest["server"]:
-            raise ValueError("数据库服务器与备份不符")
-    if args.action == "restore":
-        for database, value in manifest["databases"].items():
-            if database not in ("sw8", "sw9"):
-                raise ValueError("未知数据库")
-            file = Path(value["backup"])
-            if hashlib.sha256(file.read_bytes()).hexdigest() != value["sha256"]:
-                raise ValueError("备份校验失败")
-            content = gzip.decompress(file.read_bytes())
-            if b"LOCK TABLES" in content or b"ALTER TABLE" in content or b"CREATE TABLE" in content:
-                raise ValueError("分区恢复文件不允许包含 DDL 或隐式提交")
-            deletes = "\n".join(f"DELETE FROM `{table}` WHERE stat_date='{args.date}';" for table in TABLES)
-            script = ("START TRANSACTION;\n" + deletes + "\n").encode() + content + b"\nCOMMIT;\n"
-            options = client_options(config, args.state)
-            try:
-                completed = subprocess.run([shutil.which("mysql"), "--defaults-extra-file=" + str(options), "--batch", database], input=script, capture_output=True)
-                (args.state / (database + "-restore.stderr.log")).write_bytes(completed.stderr)
-                if completed.returncode:
-                    raise RuntimeError("分区恢复失败")
-            finally:
-                options.unlink(missing_ok=True)
-        with connect(config) as conn:
-            checks = [{"database": db, "passed": snapshot(conn, db, args.date) == value["tables"]} for db, value in manifest["databases"].items()]
-    else:
-        with connect(config) as conn:
-            source, target = [snapshot(conn, db, args.date, logical=True) for db in ("sw8", "sw9")]
-            checks = [{"table": table, "source": source[table], "target": target[table], "passed": source[table] == target[table]} for table in TABLES]
-            for db in ("sw8", "sw9"):
-                for table in TABLES:
-                    rows = query(conn, f"SELECT * FROM `{db}`.`{table}` WHERE stat_date=%s ORDER BY id", (args.date,))
-                    write_json(args.state / (db + "-" + table + "-rebuilt.json"), rows)
-    result = {"action": args.action, "date": args.date, "checks": checks, "passed": all(item["passed"] for item in checks)}
-    write_json(args.output or args.state / (args.action + ".json"), result)
-    print(json.dumps(result, ensure_ascii=False), flush=True)
-    if not result["passed"]:
-        raise SystemExit(1)
-
-
-if __name__ == "__main__":
-    main()
@@ -1,58 +0,0 @@
-"""通过新商务原生任务 HTTP 重跑指定日期，要求已有分区备份并保持任务禁用。"""
-
-import argparse
-import json
-import time
-from pathlib import Path
-
-from sw_write_verify import Rehearsal
-
-
-def main():
-    parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument("--runtime", type=Path, required=True)
-    parser.add_argument("--config", type=Path, default=Path("configs/config.yaml"))
-    parser.add_argument("--old-url", default="http://127.0.0.1:18080")
-    parser.add_argument("--new-url", default="http://127.0.0.1:19001")
-    parser.add_argument("--output", type=Path, required=True)
-    parser.add_argument("--date", default="2026-11-03")
-    parser.add_argument("--backup", type=Path, required=True)
-    parser.add_argument("--methods", nargs="+", choices=("swReportDayBackfillTask", "swReportServerDayBackfillTask", "alertTriggerTask"),
-                        default=["swReportDayBackfillTask", "swReportServerDayBackfillTask"])
-    args = parser.parse_args()
-    backup = json.loads(args.backup.read_text(encoding="utf-8"))
-    if backup["date"] != args.date:
-        raise ValueError("已有备份的日期与任务不符")
-    run = Rehearsal(args)
-    ids = []
-    try:
-        for method in args.methods:
-            name = "迁移核对-" + method + "-" + str(int(time.time()))
-            run.call("new", "admin", "POST", "/timedTask/createTimedTask", {"name": name, "spec": "@daily", "executorType": "method", "methodName": method,
-                     "params": {"startDate": args.date, "endDate": args.date, "_execution": {"timeoutSeconds": 1800}}, "enabled": False})
-            task = run.query("SELECT id,enabled FROM sw9.sys_timed_tasks WHERE name=%s AND deleted_at IS NULL", (name,))[0]
-            ids.append(task["id"])
-            (args.runtime / "report-rehearsal-task-ids.json").write_text(json.dumps(ids), encoding="utf-8")
-            run.check(method + " 创建且保持禁用", task["enabled"] == 0)
-            started = time.monotonic()
-            run.call("new", "admin", "POST", "/timedTask/triggerTimedTask", {"id": task["id"]})
-            while time.monotonic() - started < 1830:
-                value = run.call("new", "admin", "GET", f"/timedTask/getTimedTaskLogList?taskId={task['id']}&page=1&pageSize=1")["data"]
-                logs = value.get("list", [])
-                if logs and logs[0].get("status") not in ("running", "pending"):
-                    (args.runtime / (method + "-execution.json")).write_text(json.dumps(logs[0], ensure_ascii=False, indent=2), encoding="utf-8")
-                    run.check(method + " 真实执行完成", logs[0]["status"] == "success", status=logs[0]["status"], elapsed_seconds=round(time.monotonic() - started, 2))
-                    break
-                time.sleep(2)
-            else:
-                run.check(method + " 等待执行完成", False, reason="timeout")
-    except Exception as error:
-        if all(check["passed"] for check in run.checks):
-            run.check("报表重跑执行异常", False, error_type=type(error).__name__)
-        raise
-    finally:
-        run.db.close()
-
-
-if __name__ == "__main__":
-    main()
@@ -1,118 +0,0 @@
-"""主库使用 sw8/sw9，平台写入使用独立本地 SDK 夹具；核对验密、重提、解绑与新增渠道。"""
-
-import argparse
-import json
-import time
-from pathlib import Path
-
-import pymysql
-
-from sw_api_compare import compare_data
-from sw_write_verify import Rehearsal
-
-
-def main():
-    parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument("--runtime", type=Path, required=True)
-    parser.add_argument("--config", type=Path, default=Path("configs/config.yaml"))
-    parser.add_argument("--old-url", default="http://127.0.0.1:18081")
-    parser.add_argument("--new-url", default="http://127.0.0.1:19002")
-    parser.add_argument("--output", type=Path, required=True)
-    args = parser.parse_args()
-    run = Rehearsal(args)
-    fixture = json.loads((args.runtime / "sdk-fixture/fixture.json").read_text())
-    owner = fixture["owner_id"]
-    local = pymysql.connect(host="127.0.0.1", port=13309, user="root", autocommit=True, cursorclass=pymysql.cursors.DictCursor)
-    channel = "migchannel" + str(int(time.time()))
-
-    def query(sql, params=()):
-        with local.cursor() as cursor:
-            cursor.execute(sql, params)
-            return cursor.fetchall()
-
-    def api_path(side, path):
-        return ("/system/user" if side == "legacy" else "/sw/business") + path
-
-    def bind(side, password=None, success=True, identity="testlevel3"):
-        return run.call(side, identity, "POST", api_path(side, "/bindPlatform"),
-                        {"sourceDb": "lx", "account": fixture["account"], "password": password or fixture["password"]}, success=success)
-
-    def snapshot(label, status):
-        values = []
-        for side, database in (("legacy", "sw8"), ("new", "sw9")):
-            rows = run.query(f"SELECT user_id AS userId,source_db AS sourceDb,platform_busier_id AS platformBusierId,platform_busier_account AS platformBusierAccount,status,reject_reason AS rejectReason,review_account AS reviewAccount FROM {database}.sys_user_bind_platform WHERE source_db='lx' AND platform_busier_id=%s", (owner,))
-            run.check(label + " 行数和状态 " + side, len(rows) == 1 and rows[0]["status"] == status)
-            values.append(rows)
-        compared = compare_data(*values, {}, run.mapping)
-        run.check(label + " 字段一致", compared["semantic_equal"], differences=compared["differences"])
-
-    try:
-        for database in ("sw_test_sdk_old", "sw_test_sdk_new"):
-            for table in ("tab_promote", "tab_promote_settlement_time"):
-                run.check("本地平台写入夹具为空 " + database + "." + table, query(f"SELECT COUNT(*) n FROM {database}.{table}")[0]["n"] == 0)
-                query(f"ALTER TABLE {database}.{table} AUTO_INCREMENT=1")
-        for side, database in (("legacy", "sw8"), ("new", "sw9")):
-            run.check("夹具绑定键未占用 " + side, not run.query(f"SELECT user_id FROM {database}.sys_user_bind_platform WHERE source_db='lx' AND platform_busier_id=%s", (owner,)))
-            denied = bind(side, password="invalid-password", success=False)
-            run.check("错误密码不允许绑定 " + side, denied["code"] != (200 if side == "legacy" else 0))
-            denied = bind(side, success=False, identity="testlevel1")
-            run.check("仅三级允许绑定 " + side, denied["code"] != (200 if side == "legacy" else 0))
-        started = time.monotonic()
-        for side in ("legacy", "new"):
-            bind(side)
-        snapshot("首次绑定待审核", "3")
-        for side in ("legacy", "new"):
-            denied = bind(side, success=False)
-            run.check("待审核重复提交被拒绝 " + side, denied["code"] != (200 if side == "legacy" else 0))
-        # 仅在主库夹具模拟 OA 的审核结果，不调用外部 OA 写接口。
-        for database in ("sw8", "sw9"):
-            run.query(f"UPDATE {database}.sys_user_bind_platform SET status='2',reject_reason='测试驳回',review_account='migration-test',review_time=NOW() WHERE source_db='lx' AND platform_busier_id=%s", (owner,))
-        time.sleep(max(0, 5.2 - (time.monotonic() - started)))
-        for side in ("legacy", "new"):
-            bind(side)
-        snapshot("驳回后重提并清空旧审核信息", "3")
-        for database in ("sw8", "sw9"):
-            run.query(f"UPDATE {database}.sys_user_bind_platform SET status='0',review_account='migration-test',review_time=NOW() WHERE source_db='lx' AND platform_busier_id=%s", (owner,))
-        snapshot("审核通过绑定", "0")
-        payload = {"sourceDb": "lx", "platformBusierId": owner, "account": channel, "password": "Channel168", "registerType": 0}
-        for side in ("legacy", "new"):
-            denied = run.call(side, "testlevel1", "POST", run.prefix(side) + "/promote", payload, success=False)
-            run.check("保留原一级角色的新增渠道权限限制 " + side, denied["code"] != (200 if side == "legacy" else 0))
-            run.call(side, "admin", "POST", run.prefix(side) + "/promote", payload)
-        saved, settlements = [], []
-        for side, database in (("legacy", "sw_test_sdk_old"), ("new", "sw_test_sdk_new")):
-            rows = query(f"SELECT id,account,password,status,busier_id,register_type,promote_level,parent_id,top_promote_id,game_ids2 FROM {database}.tab_promote WHERE account=%s", (channel,))
-            run.check("源平台渠道只创建一条 " + side, len(rows) == 1)
-            saved.append(rows[0])
-            settlement = query(f"SELECT * FROM {database}.tab_promote_settlement_time WHERE promote_id=%s", (rows[0]["id"],))
-            run.check("七日结算及时间关系 " + side, len(settlement) == 1 and settlement[0]["day_period"] == 7 and settlement[0]["time_type"] == 0 and settlement[0]["next_count_time"] - settlement[0]["create_time"] == 7 * 86400)
-            relation = query(f"SELECT promote_ids FROM {database}.tab_promote_business WHERE id=%s", (owner,))[0]["promote_ids"]
-            run.check("平台商务渠道关系已追加 " + side, str(rows[0]["id"]) in (relation or "").split(","))
-        run.check("新渠道属性及兼容密码完全一致", saved[0] == saved[1] and saved[0]["status"] == -3 and saved[0]["promote_level"] == 1)
-        for side, database in (("legacy", "sw8"), ("new", "sw9")):
-            denied = run.call(side, "admin", "POST", run.prefix(side) + "/promote", payload, success=False)
-            run.check("重复渠道账号被拒绝 " + side, denied["code"] != (200 if side == "legacy" else 0))
-            run.call(side, "testlevel3", "DELETE", api_path(side, "/bindPlatform"), {"sourceDb": "lx", "platformBusierId": owner})
-            run.check("解绑只删除当前绑定 " + side, not run.query(f"SELECT user_id FROM {database}.sys_user_bind_platform WHERE user_id=%s AND source_db='lx' AND platform_busier_id=%s", (run.owner(side, 3), owner)))
-    except Exception as error:
-        if all(check["passed"] for check in run.checks):
-            run.check("SDK 验证执行异常", False, error_type=type(error).__name__)
-        raise
-    finally:
-        for side, database in (("legacy", "sw8"), ("new", "sw9")):
-            run.query(f"DELETE FROM {database}.sys_user_bind_platform WHERE user_id=%s AND source_db='lx' AND platform_busier_id=%s AND platform_busier_account=%s", (run.owner(side, 3), owner, fixture["account"]))
-        run.query("DELETE FROM sw9.sys_business_binding_locks WHERE source_db='lx' AND platform_busier_id=%s", (owner,))
-        for database in ("sw_test_sdk_old", "sw_test_sdk_new"):
-            rows = query(f"SELECT id FROM {database}.tab_promote WHERE account=%s", (channel,))
-            for row in rows:
-                query(f"DELETE FROM {database}.tab_promote_settlement_time WHERE promote_id=%s", (row["id"],))
-            query(f"DELETE FROM {database}.tab_promote WHERE account=%s", (channel,))
-            query(f"UPDATE {database}.tab_promote_business SET promote_ids='' WHERE id=%s AND account=%s", (owner, fixture["account"]))
-        remaining = sum(run.query(f"SELECT COUNT(*) n FROM {db}.sys_user_bind_platform WHERE source_db='lx' AND platform_busier_id=%s", (owner,))[0]["n"] for db in ("sw8", "sw9"))
-        run.check("主库 SDK 绑定夹具已清理", remaining == 0)
-        local.close()
-        run.db.close()
-
-
-if __name__ == "__main__":
-    main()
@@ -1,207 +0,0 @@
-"""在 sw8/sw9 专用测试账号下核对写流程；临时渠道、绑定、预警及规则在 finally 中清理。"""
-
-import argparse
-import json
-import time
-import urllib.error
-import urllib.parse
-import urllib.request
-from pathlib import Path
-
-import pymysql
-import yaml
-
-from sw_api_compare import compare_data
-
-
-class Rehearsal:
-    def __init__(self, args):
-        self.args = args
-        self.checks = []
-        args.output.write_text(json.dumps({"checks": [], "passed": False, "status": "running"}) + "\n", encoding="utf-8")
-        self.root = args.runtime
-        self.users = json.loads((self.root / "synthetic-test-users.json").read_text(encoding="utf-8"))
-        self.tokens = {(side, identity): (self.root / f"{side}-{identity}.token").read_text().strip()
-                       for side in ("legacy", "new") for identity in ("admin", "level3", "testlevel1", "testlevel2", "testlevel3")}
-        self.client = json.loads((self.root / "legacy-client.json").read_text())["client_id"]
-        maps = json.loads((self.root / "id-maps.json").read_text())
-        self.mapping = {str(row["old_id"]): str(row["new_id"]) for row in maps if row["entity"] == "user"}
-        config = yaml.safe_load(args.config.read_text(encoding="utf-8"))["data"]["database"]
-        self.db = pymysql.connect(host=config["host"], port=int(config["port"]), user=config["user"], password=config["password"],
-                                  autocommit=True, charset="utf8mb4", read_timeout=60, cursorclass=pymysql.cursors.DictCursor)
-        self.rule_ids = {}
-        self.suffix = str(int(time.time()))
-        self.rule_type = "mig_verify_" + self.suffix
-        self.account = "mig_verify_" + self.suffix
-        self.promote_id = 2000000000 + int(self.suffix[-7:])
-        self.alert_id = 9007199254740993 + int(self.suffix[-7:])
-        self.fixtures = []
-
-    def query(self, sql, params=()):
-        with self.db.cursor() as cursor:
-            cursor.execute(sql, params)
-            return cursor.fetchall()
-
-    def check(self, name, passed, **details):
-        self.checks.append({"check": name, "passed": bool(passed), **details})
-        self.args.output.write_text(json.dumps({"checks": self.checks, "passed": all(item["passed"] for item in self.checks)}, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
-        print(json.dumps(self.checks[-1], ensure_ascii=False), flush=True)
-        if not passed:
-            raise AssertionError(name)
-
-    def call(self, side, identity, method, path, payload=None, success=True):
-        old = side == "legacy"
-        headers = {"Content-Type": "application/json"}
-        headers.update({"Authorization": "Bearer " + self.tokens[side, identity], "clientid": self.client} if old else {"x-token": self.tokens[side, identity]})
-        request = urllib.request.Request((self.args.old_url if old else self.args.new_url) + path,
-                                         data=json.dumps(payload).encode() if payload is not None else None, method=method, headers=headers)
-        try:
-            with urllib.request.urlopen(request, timeout=60) as response:
-                raw = response.read()
-                status = response.status
-        except urllib.error.HTTPError as response:
-            raw = response.read()
-            status = response.code
-        try:
-            reply = json.loads(raw)
-        except (ValueError, UnicodeDecodeError):
-            self.check(path + " " + side, False, http=status, response_bytes=len(raw), error="响应不是 JSON")
-        if success and reply.get("code") != (200 if old else 0):
-            self.check(path + " " + side, False, code=reply.get("code"), message=reply.get("msg"))
-        return reply
-
-    def owner(self, side, level):
-        return self.users[level - 1]["source_id" if side == "legacy" else "target_id"]
-
-    def prefix(self, side):
-        return "/system" if side == "legacy" else "/sw"
-
-    def rule_snapshot(self, name):
-        values = [self.call(side, "testlevel1", "GET", self.prefix(side) + "/alertRule/" + str(self.rule_ids[side]))["data"] for side in ("legacy", "new")]
-        options = {"source_omit_fields": ["id", "createdAt", "updatedAt"], "target_omit_fields": ["id", "createdAt", "updatedAt"],
-                   "null_empty_fields": ["momDirectionLabel"]}
-        result = compare_data(*values, options, self.mapping)
-        self.check(name, result["semantic_equal"], differences=result["differences"])
-        return values
-
-    def rules(self):
-        self.rule = {"ruleName": "迁移核对临时规则", "ruleType": self.rule_type, "monitorMetric": 3, "unloginDays": 2, "monitorScope": 1, "warningColor": "#f56c6c"}
-        for side, database in (("legacy", "sw8"), ("new", "sw9")):
-            self.call(side, "testlevel1", "POST", self.prefix(side) + "/alertRule/add", self.rule)
-            rows = self.query(f"SELECT id FROM {database}.sys_alert_rule WHERE creator_id=%s AND rule_type=%s", (self.owner(side, 1), self.rule_type))
-            self.check("新增规则 " + side, len(rows) == 1)
-            self.rule_ids[side] = rows[0]["id"]
-        self.rule_snapshot("新增规则字段一致")
-        for side in ("legacy", "new"):
-            self.call(side, "testlevel1", "PUT", self.prefix(side) + f"/alertRule/toggle/{self.rule_ids[side]}?enabled=0")
-        values = self.rule_snapshot("规则禁用一致")
-        self.check("禁用状态生效", all(row["enabled"] == 0 for row in values))
-        self.rule.update(monitorMetric=1, statPeriodDays=3, momDirection=2, momThreshold=12.5)
-        self.rule.pop("unloginDays")
-        for side in ("legacy", "new"):
-            self.call(side, "testlevel1", "PUT", self.prefix(side) + "/alertRule/edit", {**self.rule, "id": self.rule_ids[side]})
-        values = self.rule_snapshot("编辑保留启停和可选历史字段")
-        self.check("编辑未重启规则且保留未登录天数", all(row["enabled"] == 0 and row["unloginDays"] == 2 for row in values))
-        for side in ("legacy", "new"):
-            rejected = self.call(side, "testlevel2", "PUT", self.prefix(side) + f"/alertRule/toggle/{self.rule_ids[side]}?enabled=1", success=False)
-            self.check("二级不能修改一级规则 " + side, rejected.get("code") != (200 if side == "legacy" else 0))
-            self.call(side, "testlevel1", "PUT", self.prefix(side) + f"/alertRule/toggle/{self.rule_ids[side]}?enabled=1")
-        self.rule_snapshot("重新启用规则一致")
-
-    def prepare_alert(self):
-        for side, database in (("legacy", "sw8"), ("new", "sw9")):
-            existing = self.query(f"SELECT id FROM {database}.tab_promote WHERE id=%s AND source_db='lx'", (self.promote_id,))
-            self.check("临时渠道 ID 未占用 " + side, not existing)
-            existing = self.query(f"SELECT id FROM {database}.sys_alert_record WHERE id=%s", (self.alert_id,))
-            self.check("临时预警 ID 未占用 " + side, not existing)
-            self.fixtures.append({"side": side, "database": database, "promote_id": self.promote_id, "alert_id": self.alert_id, "account": self.account,
-                                  "owner": self.owner(side, 1), "user": self.owner(side, 3)})
-            (self.root / "write-fixtures.json").write_text(json.dumps(self.fixtures, indent=2), encoding="utf-8")
-            self.query(f"INSERT INTO {database}.tab_promote(id,source_db,account,busier_id,promote_level,status) VALUES(%s,'lx',%s,%s,1,1)", (self.promote_id, self.account, self.promote_id))
-            self.query(f"INSERT INTO {database}.sys_user_bind_platform(user_id,source_db,platform_busier_id,platform_busier_account,status,create_time) VALUES(%s,'lx',%s,%s,'0',NOW())", (self.owner(side, 3), self.promote_id, self.account))
-            self.query(f"INSERT INTO {database}.sys_alert_record(id,platform_code,platform_name,rule_id,rule_name,rule_type,monitor_metric,target_promote_id,target_promote_name,business_id,business_name,alert_content,status,alert_date,created_at,updated_at) VALUES(%s,'lx','龙翔',%s,%s,%s,1,%s,%s,%s,%s,'迁移核对临时预警',1,CURDATE(),NOW(),NOW())",
-                       (self.alert_id, self.rule_ids[side], self.rule["ruleName"], self.rule_type, self.promote_id, self.account, self.owner(side, 1), self.users[0]["username"]))
-
-    def alert_snapshot(self, label, expected_status):
-        values = [self.call(side, "testlevel1", "GET", self.prefix(side) + f"/alertRecord/detail/{self.alert_id}")["data"] for side in ("legacy", "new")]
-        for row in values:
-            row["alertRecord"].pop("ruleId", None)
-        options = {"source_omit_fields": ["id", "createdAt", "updatedAt"], "target_omit_fields": ["id", "createdAt", "updatedAt"],
-                   "null_empty_fields": ["actionDetail", "noteContent"]}
-        # 日志主键独立生成；按原操作序列比较业务字段及操作人映射。
-        result = compare_data(*values, options, self.mapping)
-        self.check(label, result["semantic_equal"] and all(row["alertRecord"]["status"] == expected_status for row in values),
-                   status=expected_status, log_counts=[len(row["operationLogs"]) for row in values], differences=result["differences"])
-
-    def workflow(self):
-        self.prepare_alert()
-        self.alert_snapshot("临时预警初始状态一致", 1)
-        for side in ("legacy", "new"):
-            rejected = self.call(side, "level3", "GET", self.prefix(side) + f"/alertRecord/detail/{self.alert_id}", success=False)
-            self.check("其他商务不可查看临时预警 " + side, rejected.get("code") != (200 if side == "legacy" else 0))
-        steps = [("startProcess", "testlevel3", 2), ("submitAudit", "testlevel3", 3), ("reject", "testlevel1", 4),
-                 ("reProcess", "testlevel3", 2), ("submitAudit", "testlevel3", 3), ("approve", "testlevel1", 5)]
-        for index, (action, identity, status) in enumerate(steps):
-            for side in ("legacy", "new"):
-                self.call(side, identity, "POST", self.prefix(side) + "/alertRecord/" + action, {"alertId": str(self.alert_id), "note": "迁移核对步骤" + str(index)})
-            self.alert_snapshot(action + " 状态及日志一致", status)
-            if index == 1:
-                for side in ("legacy", "new"):
-                    rejected = self.call(side, "testlevel3", "POST", self.prefix(side) + "/alertRecord/approve", {"alertId": str(self.alert_id)}, success=False)
-                    self.check("三级商务不能审核 " + side, rejected.get("code") != (200 if side == "legacy" else 0))
-        for side in ("legacy", "new"):
-            self.call(side, "testlevel3", "POST", self.prefix(side) + "/alertRecord/addNote", {"alertId": str(self.alert_id), "noteContent": "迁移核对独立备注", "attachmentIds": []})
-        self.alert_snapshot("已处理记录追加备注不改变状态", 5)
-
-    def remove_rule(self):
-        for side, database in (("legacy", "sw8"), ("new", "sw9")):
-            self.call(side, "testlevel1", "DELETE", self.prefix(side) + "/alertRule/" + str(self.rule_ids[side]))
-            row = self.query(f"SELECT is_deleted FROM {database}.sys_alert_rule WHERE id=%s", (self.rule_ids[side],))[0]
-            self.check("删除规则保留历史行 " + side, row["is_deleted"] == 1)
-
-    def cleanup(self):
-        for fixture in reversed(self.fixtures):
-            database = fixture["database"]
-            self.query(f"DELETE FROM {database}.sys_alert_operation_log WHERE alert_id=%s", (fixture["alert_id"],))
-            self.query(f"DELETE FROM {database}.sys_alert_record WHERE id=%s AND target_promote_name=%s", (fixture["alert_id"], fixture["account"]))
-            self.query(f"DELETE FROM {database}.sys_user_bind_platform WHERE user_id=%s AND source_db='lx' AND platform_busier_id=%s AND platform_busier_account=%s", (fixture["user"], fixture["promote_id"], fixture["account"]))
-            self.query(f"DELETE FROM {database}.tab_promote WHERE id=%s AND source_db='lx' AND account=%s", (fixture["promote_id"], fixture["account"]))
-        for side, database in (("legacy", "sw8"), ("new", "sw9")):
-            self.query(f"DELETE FROM {database}.sys_alert_rule WHERE creator_id=%s AND rule_type=%s", (self.owner(side, 1), self.rule_type))
-        remaining = 0
-        for fixture in self.fixtures:
-            database = fixture["database"]
-            for table, condition, params in (
-                ("sys_alert_record", "id=%s", (fixture["alert_id"],)),
-                ("sys_alert_operation_log", "alert_id=%s", (fixture["alert_id"],)),
-                ("tab_promote", "id=%s AND source_db='lx'", (fixture["promote_id"],)),
-                ("sys_user_bind_platform", "user_id=%s AND source_db='lx' AND platform_busier_id=%s", (fixture["user"], fixture["promote_id"])),
-            ):
-                remaining += self.query(f"SELECT COUNT(*) n FROM {database}.{table} WHERE {condition}", params)[0]["n"]
-        for side, database in (("legacy", "sw8"), ("new", "sw9")):
-            remaining += self.query(f"SELECT COUNT(*) n FROM {database}.sys_alert_rule WHERE creator_id=%s AND rule_type=%s", (self.owner(side, 1), self.rule_type))[0]["n"]
-        self.check("临时预警渠道绑定和规则已清理", remaining == 0, remaining=remaining)
-        self.db.close()
-
-
-def main():
-    parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument("--runtime", type=Path, required=True)
-    parser.add_argument("--config", type=Path, default=Path("configs/config.yaml"))
-    parser.add_argument("--old-url", default="http://127.0.0.1:18080")
-    parser.add_argument("--new-url", default="http://127.0.0.1:19001")
-    parser.add_argument("--output", type=Path, required=True)
-    args = parser.parse_args()
-    if args.runtime.resolve().is_relative_to(Path(__file__).resolve().parents[1]):
-        raise ValueError("测试状态必须保存在仓库外")
-    run = Rehearsal(args)
-    try:
-        run.rules()
-        run.workflow()
-        run.remove_rule()
-    finally:
-        run.cleanup()
-
-
-if __name__ == "__main__":
-    main()