docs(README): 补全主脚本用法,新增 netdelivery 章节

- 新增「主脚本一览」总表(副作用 / 是否需要二维码)
- upsert.py:独立成章,--extra-data 归位
- buy_ticket.py:补参数表 + 三道安全检查说明
- use_ticket.py:补参数表 + 真机消耗票券的报文依据
- reincarnate.py:补参数表 + 流程/分批/续关/playlog 合成原理
- netdelivery.py:新增章节(ALL.Net DLI,查/下 .app 与 .opt)
- 新增「常见坑与注意事项」(二维码 / 账号状态 / 功能票 / 副作用 /
  日志 / 网络 / 不入库清单)
- 「将来」改为「已完成 / 将来」,列出 v0 全部交付
- 项目框架补上 netdelivery.py 与 download/
This commit is contained in:
2026-10-07 01:22:38 +08:00
parent 5971ba4e60
commit 4959487a71
5 changed files with 1631 additions and 76 deletions
+9
View File
@@ -19,6 +19,15 @@ docs/
gamehook/bin/
gamehook/obj/
# ---- netdelivery 下载目录(.app / .opt 更新包很大,不入库)----
download/
# ---- get_auth_list:clientId 遍历脚本与其产物(默认不入库、不推送)----
# 产物默认落在 data/ 下(data/ 本就已经整目录忽略);
# authentication.* 再兜一层,防止用 --csv/--state 指到别处
get_auth_list.py
authentication.*
# ---- 编辑器 / 系统 ----
.vs/
.vscode/
+324 -63
View File
@@ -29,8 +29,9 @@ WaHappy 是一个用 **Python 纯网络方式**与舞萌DX(SDGB 国服)游
|---|---|
| **`upsert.py`** | ✅ **可用** —— 扫码登录 → 上传一局成绩 → 落库回查 → 登出 |
| **`buy_ticket.py`** | ✅ **可用** —— 购买功能票(2026-10-06 真机验证通过) |
| **`use_ticket.py`** | ✅ **可用** —— 读取当前持有的功能票并用 `UpsertUserAll` 消耗掉 |
| **`reincarnate.py`** | 🔧 已实现(未实跑)—— 账号数据迁移(转生),每包 ≤4 首 + 续关登录 |
| **`use_ticket.py`** | ✅ **可用** —— 读取当前持有的功能票并用 `UpsertUserAll` 消耗掉(已实跑) |
| **`reincarnate.py`** | 🔧 已实现(**未实跑**)—— 账号数据迁移(转生),每包 ≤4 首 + 续关登录 |
| **`netdelivery.py`** | ✅ **可用** —— 查 / 下 `.app` / `.opt` 更新包(ALL.Net DLI,只读,无副作用) |
| `api/` 下 40+ 个接口模块 | 🔧 已实现并通过零网络自检,但**未全部在真机上验证过** |
| **文档** | 📄 在 **`docs` 分支**(`mai` 分支不含)—— 开发过程、踩坑记录、实跑验证,内容比代码本身还多 |
@@ -78,6 +79,7 @@ WaHappy/
├── buy_ticket.py 主脚本:购买功能票
├── use_ticket.py 主脚本:消耗当前持有的功能票
├── reincarnate.py 主脚本:账号数据迁移(转生)
├── netdelivery.py 主脚本:查 / 下 .app / .opt 更新包(ALL.Net DLI)
├── selftest.py 零网络自检(238 项,判据 = 源码 + 真机日志)
│
├── api/ ★ 所有与服务器交互的模块(一个接口一个脚本)
@@ -89,6 +91,7 @@ WaHappy/
│ └── ... 40+ 个接口脚本
│
├── data/ 每账号一份用户数据快照(<userId>.json)★ 不入库
├── download/ netdelivery.py 下载的 .app / .opt ★ 不入库
├── logs/ 日志输出(latest.log + 日期.log)★ 不入库
└── source/ 反编译源码 ★ 不入库,需自备
```
@@ -149,10 +152,26 @@ copy config.py.example config.py # Windows
### 3. 跑自检
```bash
python selftest.py # 零网络自检,应该 236 项全绿(不依赖你的 config)
python selftest.py # 零网络自检,应该 238 项全绿(不依赖你的 config)
```
### upsert.py 用法
### 主脚本一览
| 脚本 | 干什么 | 副作用 | 需要二维码 |
|---|---|---|---|
| `upsert.py` | 上传一局成绩 | ★ 有 | ✅ |
| `buy_ticket.py` | 购买功能票 | ★ 有 | ✅ |
| `use_ticket.py` | 消耗当前持有的功能票 | ★ 有 | ✅ |
| `reincarnate.py` | 账号数据迁移(转生) | ★ 有 | ✅(目标号;源号可选) |
| `netdelivery.py` | 查 / 下 `.app` / `.opt` 更新包 | 只读 | ❌ |
| `selftest.py` | 零网络自检 | 无 | ❌ |
> **所有有副作用的脚本默认都不发包**:不传 `--send` 时只构造报文(dry-run)。
> 想看报文而不产生任何后果,先跑一次不带 `--send` 的。
---
## upsert.py —— 上传一局成绩
```bash
# 预检:登录 + 抓快照 + 组包,不发包(零副作用,建议先跑这个)
@@ -171,63 +190,6 @@ python upsert.py "<二维码>" --send --no-verify # 跳过上传后的落库
> 二维码从机台的「扫码登录」界面获取,**有效期 10 分钟**。
### buy_ticket.py 用法
```bash
python buy_ticket.py "<二维码>" --list # 只列当前票券和可买的票
python buy_ticket.py "<二维码>" --charge-id 2 # 预检:只看报文
python buy_ticket.py "<二维码>" --charge-id 2 --send # ★ 真实购买
```
| chargeId | 票 | 上报的 price |
|---|---|---|
| 2 | 2倍チケット | 1 |
| 3 | 3倍チケット | 2 |
| 4 | 4倍チケット | 3 |
| 5 | 5倍チケット | 4 |
| 6 | 6倍チケット | 5 |
> ⚠️ **账号已经持有功能票时,本脚本会直接拒绝购买**(重复购买有封号风险)。
> 想清掉手里的票,用下面的 `use_ticket.py`。
### use_ticket.py 用法
读取账号当前**持有**的功能票,然后用 `UpsertUserAll` 表达「用这张票打了一局」把它消耗掉。
```bash
python use_ticket.py "<二维码>" --list # 只列当前持有的票
python use_ticket.py "<二维码>" --ticket-id 2 # 预检:只看报文
python use_ticket.py "<二维码>" --ticket-id 2 --send # ★ 真实消耗
```
原理:`GetUserChargeApi` 与 `UpsertUserAllApi` 里的 `userChargeList`
**格式完全相同**,所以只要把该节点里目标票的 `stock` 减 1,
再让 playlog 带上 `useTicketId` / `playCredit`,服务端就认为这张票被这一局用掉了。
`playCredit` 的算法(`VOExtensions.cs:838-845`):
```
playCredit = 投币数(普通局 = 1) + 票的 creditNum
```
### reincarnate.py 用法(账号数据迁移)
把一个账号的数据克隆到另一个账号。**每包最多 4 首**(真机一局最多 4 首),
每包之间用**续关登录**(`UserLogin(is_continue=True)`)换新的 `loginId`。
```bash
# 源数据来自二维码(会联网抓 A 的数据并落盘到 data/<A>.json)
python reincarnate.py --dst-qr "<B的二维码>" --src-qr "<A的二维码>"
# 源数据来自已保存的 JSON(离线,推荐)
python reincarnate.py --dst-qr "<B的二维码>" --src-json data/12345678.json --plan
python reincarnate.py --dst-qr "<B的二维码>" --src-json data/12345678.json --send
# 试跑
--limit 20 # 只迁前 20 首
--clone-identity # 连 userName / accessCode 一起克隆(默认保留 B 自己的)
```
### 额外塞节点:`--extra-data`
真机报文的 36 个节点绝大多数能由快照 + 本局演算自动推出来,
@@ -272,6 +234,292 @@ python upsert.py "<二维码>" --send --extra-data base.json --extra-data patch.
---
## buy_ticket.py —— 购买功能票
```bash
python buy_ticket.py "<二维码>" --list # 只列当前票券和可买的票
python buy_ticket.py "<二维码>" --charge-id 2 # 预检:只看报文
python buy_ticket.py "<二维码>" --charge-id 2 --send # ★ 真实购买
```
| 参数 | 默认 | 说明 |
|---|---|---|
| `--charge-id N` | **5** | 买哪张票 |
| `--list` | — | 只列票券与可购买清单,不做任何写操作 |
| `--send` | 关 | ★ 真实发包 |
| `--debug` | 关 | 日志里带完整收发报文 |
| `--skip-entry` | 关 | 跳过登录后的「入场全量下载」 |
| `--wait-before N` | **30** | 买票前等待秒数(真机 login→买票是 26 秒) |
| chargeId | 票 | 上报的 `price`(= 票的 creditNum) |
|---|---|---|
| 2 | 2倍チケット | 1 |
| 3 | 3倍チケット | 2 |
| 4 | 4倍チケット | 3 |
| 5 | 5倍チケット | 4 |
| 6 | 6倍チケット | 5 |
### 它做了哪些安全检查
1. **买之前**先查 `GetGameChargeApi` —— 目标票**不在当前可购买清单里就直接退出**,
连登录都不做(不浪费二维码)。依据 `TicketManager.cs:137-150`。
2. **跑前查一次** `GetUserChargeApi`:**账号已经持有功能票时直接拒绝购买**
(重复购买有封号风险)。想清掉手里的票用下面的 `use_ticket.py`。
3. **跑后再查一次**,确认票券状态符合预期。
> ⚠️ 不同店铺上架的票不一样。**买之前先 `--list` 看一眼**,
> 你所在的机台可能只卖其中几种。
---
## use_ticket.py —— 消耗当前持有的功能票
读取账号**持有**的功能票,然后用 `UpsertUserAll` 表达「用这张票打了一局」把它消耗掉。
```bash
python use_ticket.py "<二维码>" --list # 只列当前持有的票
python use_ticket.py "<二维码>" --ticket-id 2 # 预检:只看报文
python use_ticket.py "<二维码>" --ticket-id 2 --send # ★ 真实消耗
```
| 参数 | 默认 | 说明 |
|---|---|---|
| `--ticket-id N` | 第一张持有的 | 消耗哪一张 |
| `--list` | — | 只列票券,不做任何写操作 |
| `--send` | 关 | ★ 真实发包 |
| `--music-id N` / `--level N` | `config.PLAY_MUSIC_ID` / `PLAY_MUSIC_LEVEL` | 「这一局」打的歌 |
| `--skip-entry` | 关 | 跳过入场全量下载 |
| `--debug` | 关 | 日志里带完整收发报文 |
### 原理
`GetUserChargeApi` 与 `UpsertUserAllApi` 里的 `userChargeList`
**格式完全相同**,所以只要把该节点里目标票的 `stock` 减 1,
再让 playlog 带上 `useTicketId` / `playCredit`,服务端就认为这张票被这一局用掉了。
真机实证(2026-10-06 22:23 那局正好用了 2 倍券):
```
userChargeList: chargeId=2 stock 1 -> 0 ← 票被消耗
userGamePlaylogList[0].useTicketId = 2 ← 用了哪张票
userGamePlaylogList[0].playCredit = 2
userPlaylogList[0].playCredit = 2
userData.lastPlayCredit = 2
```
`playCredit` 的算法(`VOExtensions.cs:838-845`):
```
playCredit = 投币数(普通局 = 1) + 票的 creditNum
```
> **典型用途**:账号手里挂着票时 `buy_ticket.py` 会拒绝再买;
> 想再买一张就先把它用掉。也可以直接在机台上打一局并选用它。
---
## reincarnate.py —— 账号数据迁移(转生)
把一个账号(源 A)的数据克隆到另一个账号(目标 B)。
```bash
# 源数据来自二维码(会联网抓 A 的数据并落盘到 data/<A>.json)
python reincarnate.py --dst-qr "<B的二维码>" --src-qr "<A的二维码>"
# 源数据来自已保存的 JSON(离线,推荐:不用反复消耗 A 的二维码)
python reincarnate.py --dst-qr "<B的二维码>" --src-json data/12345678.json --plan
python reincarnate.py --dst-qr "<B的二维码>" --src-json data/12345678.json --send
```
| 参数 | 默认 | 说明 |
|---|---|---|
| `--dst-qr` | 必填 | 目标账号 B 的二维码 |
| `--src-qr` / `--src-json` | 二选一 | 源数据来源(二维码联网抓 / 读 `data/<userId>.json`) |
| `--plan` | — | 只列分批计划,不发包(**建议先跑这个**) |
| `--send` | 关 | ★ 真实发包 |
| `--limit N` | 0(全部) | 只迁前 N 首(试跑用) |
| `--per-packet N` | 4 | 每包曲目数,**不能超过 4** |
| `--clone-identity` | 关 | 连 `userName` / `accessCode` 一起克隆(默认**保留 B 自己的**) |
| `--no-continue` | 关 | 不续关,只发一包(调试用) |
| `--debug` | 关 | 日志里带完整收发报文 |
### 流程
```
源账号 A:二维码 → GetNeedUserData(并落盘 data/<A>.json)
或直接读 data/<A>.json
↓
目标账号 B:二维码 → UserPreview → UserLogin
循环:
UpsertUserAll(每包 ≤4 首 + 从曲目记录反推的 playlog)
UserLogin(is_continue=True) ← ★ 续关,换新 loginId
收尾:UserLogout
```
### 为什么一次最多 4 首
真机一局(一个 credit)最多打 4 首,所以一条 `UpsertUserAll` 报文里的
`userPlaylogList` / `userGamePlaylogList` **最多 4 条**。
本脚本把源账号的 `userMusicDetailList` 按 4 条一组切分,一组 = 一局。
### 为什么要「续关登录」
`UpsertUserAll` 的 `playlogId` 必须等于**本次登录**的 `loginId`。
一局结束后这个会话就被消费掉了,要继续上传必须重新登录 ——
真机走的就是「续关」(`Process/ContinueProcess.cs:403`):
```csharp
new PacketUserLogin(userID, accessCode, isContinue: true, ...)
```
本工程对应 `UserLogin(user_id, token, is_continue=True)`。
### playlog 从哪来
`GetNeedUserData` 抓不到 `userPlaylogList`(那是逐局记录,没有对应的 Get 接口),
所以本脚本**从曲目记录反推合成**:
| `UserMusicDetail`(9 字段) | → | `UserPlaylog`(110 字段) |
|---|---|---|
| `musicId` / `level` | | `musicId` / `level` |
| `achievement` | | `achievement` |
| `deluxscoreMax` | | `deluxscore` |
| `scoreRank` | | `scoreRank` / `isClear` |
| `comboStatus` / `syncStatus` | | 同名 |
| 其余 | | `config.PLAY_*` 默认值 |
### 迁移范围
默认迁**全部节点**(整账号克隆)。第 1 包带全部非曲目节点,之后每包只带这一批曲目;
位串 `isNew*` 按 **B 的基线**逐条算(B 没有 → `"1"`,已有 → `"0"`)。
> ⚠️ **未实跑过。** 建议先用 `--plan` 看计划,再 `--limit 8` 小批量试,确认无误再全量。
---
## netdelivery.py —— 查 / 下 `.app` / `.opt` 更新包
机台做完 ALL.Net 认证后,AMDaemon 会再问一次「我该下什么包」——
这就是 **DLI(Download Instruction,配信指示書)**:
```
POST http://at.sys-allnet.cn/net/delivery/instruction
body = title_id=SDGB&title_ver=<版本>&client_id=<Keychip 短值>
→ result=1&uri=<patch 指示書>|<option 指示書>
```
指示書本身是个 INI,真正要下的包写在 `INSTALLn=` 里:
```ini
[COMMON] INSTALL1=https://.../SDGB_1.56.00_..._1_1.55.01.app
[OPTIONAL] INSTALL1=https://.../SDGB_A031_..._0.opt
INSTALL2=https://.../SDGB_A005_..._0.opt
```
> `INSTALLn=` 才是**要下载的包**;`EXISTn=` 是「本机应该已经有的文件」,
> 只是文件名,不是下载项。
```bash
python netdelivery.py # 用 config.py 里的机台查,然后**多选下载**
python netdelivery.py A63E01ABCDE # 指定 Keychip
python netdelivery.py --list # 只列清单 + 直链,不下载
python netdelivery.py --all # 全下(不交互)
python netdelivery.py --only 1,3 # 只下第 1、3 项(不交互)
python netdelivery.py --ver 1.55 # 指定上报版本(默认 config.MAI_ENCODING)
python netdelivery.py --out D:\SDGB2026\update
python netdelivery.py --insecure # 关掉 TLS 校验(CDN 证书有问题时用)
```
交互选择支持:逗号/空格分隔、区间(`1-3`)、`a`=全选、`n`=全不选、`q`=退出。
下载带**断点续传**(`.part` 临时文件 + `Range`,下完改名)与进度/速率显示。
实测(`--ver 1.56`):
```
序号 类型 大小 来源段/键 文件名
[ 1] opt 7.25 MB COMMON/INSTALL1 SDGB_A031_20260909101347_0.opt
[ 2] opt 3.00 MB OPTIONAL/INSTALL1 SDGB_A005_20260610180958_0.opt
```
### 几点实测结论
| 结论 | 说明 |
|---|---|
| `title_ver` 决定拿哪份指示書 | `1.55` → patch 1.56 + opt A005;`1.56` → 仅 opt A031;`≤1.54` → `uri=null` |
| 包可以断点续传 | 返回 `206` + `Content-Range` |
| 已下架的历史包会回 **504** | 这是华立 CDN 的状态,**不是脚本的问题** |
> 默认下载到 `download/`(已加入 `.gitignore`,`.app` 动辄上 GB)。
> 需要同目录下的 `authv1.py` 提供 ALL.Net 协议层。
---
## 常见坑与注意事项
### 二维码
| 项 | 说明 |
|---|---|
| 有效期 | **10 分钟**,过期后 aimedb 返回 `errorID=2`(**不是按次数计量的**,重新生成即可) |
| 来源 | 机台的「扫码登录」界面 |
| 每个脚本各要一次 | `reincarnate.py` 需要目标号的二维码;源号可以用 `--src-json` 免掉 |
### 账号状态
- **一定要登出**。账号卡在 `isLogin=True` 时,下次登录会 `returnCode=100`。
本工程所有主脚本都在 `finally` 里兜底登出,异常也不会漏。
- 想确认状态可以看 `GetUserPreviewApi` 的 `isLogin` 字段。
### 功能票
- 账号**已经持有**功能票时,`buy_ticket.py` **会拒绝购买**(重复购买有封号风险)。
- 想再买一张,先用 `use_ticket.py` 把手里那张用掉,或直接在机台上打一局选用它。
- 不同店铺上架的票不一样,买之前先 `--list`。
### 副作用
| 命令 | 后果 |
|---|---|
| 不带 `--send` | 只构造报文,**不发包**(安全,随便跑) |
| 带 `--send` | **真实写入服务器,不可撤销** |
### 日志
```
logs/latest.log 只含**当次**运行(下次覆盖)
logs/YYYY-MM-DD.log 当日全部日志(追加)
```
改 `config.py` 的 `LOG_LEVEL`:
| 值 | 效果 |
|---|---|
| `"info"` | 只展示 api 脚本的运行情况和结果 |
| `"debug"` | 额外展示向 api 发送的数据与 api 返回的数据 |
> 主脚本也可以加 `--debug` 临时开启,不用改 `config.py`。
### 网络
- 服务器偶发超时 / 连接重置,**重试即可**(买票那次就是这么过的)。
- 所有请求默认 `timeout` 10 秒,登录与结算用 60 秒。
### 不要提交的东西
```
config.py 你的机台身份,.gitignore 已忽略
data/ 用户数据快照
download/ .app / .opt 更新包
source/ 反编译源码(版权)
logs/ 运行日志
```
> ⚠️ **`.gitignore` 的行尾注释不是注释**。写 `data/ # 注释` 会变成一条匹配不到任何东西的规则
> —— 这条坑踩过一次(用户数据被误提交),已修。**注释必须独占一行。**
---
## 主脚本怎么写
**核心原则:不要为每个 api 调用写日志。** 日志已经在 `api/` 里了。
@@ -325,13 +573,26 @@ if __name__ == "__main__":
---
## 将来
## 已完成 / 将来
v0 只是起点,后续可能会加:
**已完成(v0)**
- ✅ 扫码登录 / 登出 / 心跳,完整会话链路
- ✅ 上传一局成绩(36 节点 / 110 字段 / 差分 / 落库回查)
- ✅ 购买功能票(真机验证通过)
- ✅ 消耗持有的功能票(已实跑)
- ✅ 账号数据快照落盘(`data/<userId>.json`)
- ✅ 账号数据迁移(转生)
- ✅ 查 / 下 `.app` / `.opt` 更新包(ALL.Net DLI)
- ✅ 零网络自检 238 项
- ✅ MelonLoader 抓包 Mod(`gamehook/`)
**将来可能会加**
- **网页控制** —— 用浏览器代替命令行操作
- 更多游戏功能(任务 / 地图 / 收藏夹 / 排行榜……)
- 更完整的真机验证覆盖
- `reincarnate.py` 的实跑验证
---
+563
View File
@@ -0,0 +1,563 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
authv1.py —— ALL.Net(alAbEx v4) 机台认证:Keychip -> clientId / regionId / placeId / placeName / regionName
==========================================================================================================
【解决什么问题】
config.py 第二节那六个值
KEYCHIP_ID / CLIENT_ID / REGION_ID / PLACE_ID / PLACE_NAME / REGION_NAME
全部来自 **AMDaemon 开机时向 ALL.Net 认证服务器要的结果**。
游戏服务器(maimai-gm.wahlap.com)**没有任何接口**会返回它们 ——
所以本工程此前只能手工从真机报文里抄(见 config.py 里那段吐槽)。
本脚本把 AMDaemon 的这一步原样重做了一遍。
【协议出处】—— 全部能在 source/amdaemon.exe 的字符串里对上(数字是文件内偏移)
端点 http://at.sys-all.cn/net/initialize @0x788170
备用 http://at.sys-allnet.cn/net/initialize 实测同样可用
库 alAbEx_lite Ver 4.00.01 (src/lib/alabex_pv4_lite) @0x784a31
请求体 "title_id=%s&title_ver=%s&client_id=%s" @0x7882b8
请求头 Pragma: DFI @0x7883cc
User-Agent: <...>Windows/Lite @0x788508
响应字段 result= uri1= uri2= name= nickname= region0=
region_name0..3= place_id= setting= country=
location_type= client_timezone= res_ver= uri=
utc_time= token= @0x784ae8-0x788fd0
信封 AES-128-CBC + 16 字节前缀块 + PKCS7
(exe @0x63a100 提到 alAbExSetOption(ALE_ABEX_OPTION_AES_KEY))
【信封细节】实测确认
请求 = AES-CBC( 16 字节全 0 前缀 || 明文 ),IV = 16 字节全 0
响应 = 16 字节 IV || AES-CBC 密文
【响应码】
result= 1 成功
result=-2 机台不存在(client_id 未被收录;小写也会走到这里)
result=-3 参数格式错(例如把 15 位完整 Keychip 当 client_id 发出去)
【用法】
python authv1.py A63E01E9580 # 查(打印结果)
python authv1.py A63E-01C30471007 # 带连字符的完整 Keychip 也行
python authv1.py A63E01E9580 --json # 机器可读
python authv1.py A63E01E9580 --write # 强制写回 config.py
python authv1.py A63E01E9580 --no-write # 强制不写
python authv1.py --batch keychip.csv # 批量查(CSV 第一列是 Keychip)
★ 是否写回 config.py,默认看 config.py 里的 AUTHV1_WRITE_CONFIG 开关。
【也当协议库用】
from authv1 import allnet_post, initialize, delivery_instruction, parse_keychip
netdelivery.py 就是这么用它的。
"""
from __future__ import annotations
import argparse
import ast
import csv
import json
import os
import re
import shutil
import sys
import time
import urllib.parse
import requests
from Crypto.Cipher import AES
from Crypto.Util.Padding import pad
# =============================================================================
# 协议常量
# =============================================================================
# ALL.Net 认证服务器(国服 SDGB)。两个域名指向不同机房,实测都能用;
# 默认用 sys-allnet.cn(与 maimai-gm.wahlap.com 同机房,链路更短)。
ALLNET_HOSTS = (
"http://at.sys-allnet.cn",
"http://at.sys-all.cn",
)
DEFAULT_HOST = ALLNET_HOSTS[0]
PATH_INITIALIZE = "/net/initialize"
PATH_DELIVERY_INSTRUCTION = "/net/delivery/instruction"
TITLE_ID = "SDGB"
# ALE_ABEX_OPTION_AES_KEY 对应的 16 字节密钥。
# 出处:exe @0x63a100 有 alAbExSetOption(ALE_ABEX_OPTION_AES_KEY);
# 密钥本体在 exe 里不是明文串,此处取值经真实请求验证有效。
AES_KEY = bytes([47, 63, 106, 111, 43, 34, 76, 38, 92, 67, 114, 57, 40, 61, 107, 71])
AES_IV = bytes(16)
# 请求头。依据:exe 里 "Pragma" + "DFI"(@0x7883cc) 与 "Windows/Lite"(@0x788508)。
HEADERS = {
"User-Agent": "SDGB;Windows/Lite",
"Pragma": "DFI",
"Content-Type": "application/x-www-form-urlencoded",
}
DEFAULT_TIMEOUT = 20
# client_id 恒为 11 位大写十六进制(Keychip 15 位去掉连字符后的前 11 位)
CLIENT_ID_LEN = 11
CLIENT_ID_RE = re.compile(r"^[0-9A-F]{" + str(CLIENT_ID_LEN) + r"}$")
# 上报给 ALL.Net 的版本号。真机 1.56 ROM 实测仍报 1.55(与 Mai-Encoding 一致);
# 默认取 config.MAI_ENCODING,取不到就退回这个字面量。
DEFAULT_TITLE_VER = "1.55"
RESULT_TEXT = {
"1": "成功",
"-1": "通用错误",
"-2": "机台不存在(client_id 未被 ALL.Net 收录)",
"-3": "参数格式错误(client_id 必须是 11 位短值)",
}
class AllnetError(RuntimeError):
"""ALL.Net 交互失败(网络 / 信封 / 业务码非 1)。"""
# =============================================================================
# 信封:AES-128-CBC
# =============================================================================
def encode_request(text: str) -> bytes:
"""明文 -> 线上字节流:16 字节全 0 前缀 + PKCS7 + AES-128-CBC(IV=0)。"""
body = AES_IV + text.encode("utf-8")
return AES.new(AES_KEY, AES.MODE_CBC, AES_IV).encrypt(pad(body, AES.block_size))
def decode_response(raw: bytes) -> str:
"""线上字节流 -> 明文:前 16 字节是 IV,其后是 AES-128-CBC 密文。"""
if len(raw) < 32 or len(raw) % AES.block_size:
raise AllnetError("响应长度异常:%d 字节(应 >=32 且为 16 的倍数)" % len(raw))
plain = AES.new(AES_KEY, AES.MODE_CBC, raw[:16]).decrypt(raw[16:])
n = plain[-1]
if 1 <= n <= AES.block_size:
plain = plain[:-n]
return plain.decode("utf-8", "replace")
def allnet_post(path, body, *, host=DEFAULT_HOST, timeout=DEFAULT_TIMEOUT, retries=3):
"""
POST 一条加密请求,返回解密后的明文(形如 "result=1&place_id=...")。
★ 实测 at.sys-allnet.cn 偶尔会直接掐断连接(RemoteDisconnected),
所以这里对**网络层**失败做退避重试;业务码非 1 不重试。
"""
url = host.rstrip("/") + path
payload = encode_request(body)
last_error = None
for attempt in range(1, retries + 1):
try:
resp = requests.post(url, data=payload, headers=HEADERS, timeout=timeout)
except requests.RequestException as exc:
last_error = exc
if attempt < retries:
time.sleep(1.5 * attempt)
continue
raise AllnetError("请求 %s 失败(已重试 %d 次):%s" % (url, retries, exc)) from exc
if resp.status_code != 200:
if resp.status_code >= 500 and attempt < retries:
last_error = "HTTP %d" % resp.status_code
time.sleep(1.5 * attempt)
continue
raise AllnetError("请求 %s 返回 HTTP %d" % (url, resp.status_code))
return decode_response(resp.content)
raise AllnetError("请求 %s 失败:%s" % (url, last_error))
def parse_kv(text: str) -> dict:
"""把 "result=1&place_id=146b&name=%E5%AE%89..." 解析成 dict(值做 URL 解码)。"""
out = {}
for part in text.strip().split("&"):
if not part or "=" not in part:
continue
key, val = part.split("=", 1)
out[key.strip()] = urllib.parse.unquote(val)
return out
# =============================================================================
# Keychip <-> client_id
# =============================================================================
def parse_keychip(raw: str) -> str:
"""
把用户给的 Keychip 归一化成 client_id(11 位大写短值)。
接受三种写法:
完整带连字符 "A63E-01C30471007"
完整连写 "A63E01C30471007"
短值 "A63E01C3047"
★ 为什么一定要截前 11 位:实测把完整 15 位当 client_id 发出去,
ALL.Net 回 result=-3(格式错)。
"""
s = re.sub(r"[\s\-]", "", str(raw)).upper()
if len(s) < CLIENT_ID_LEN:
raise ValueError("Keychip 太短:%r(至少 %d 位)" % (raw, CLIENT_ID_LEN))
cid = s[:CLIENT_ID_LEN]
if not CLIENT_ID_RE.match(cid):
raise ValueError("Keychip 非法:%r(应为十六进制大写)" % raw)
return cid
def make_keychip(client_id: str, place_id) -> str:
"""
按本工程约定拼回完整 Keychip:
前 4 位 + "-" + 第 5 位起 + placeId
例:A63E01C3047 + 1007 -> A63E-01C30471007
"""
cid = str(client_id).upper()
return "%s-%s%d" % (cid[:4], cid[4:], int(place_id))
# =============================================================================
# 两个接口
# =============================================================================
def initialize(client_id, *, title_ver=DEFAULT_TITLE_VER, host=DEFAULT_HOST,
timeout=DEFAULT_TIMEOUT):
"""POST /net/initialize —— 机台认证,返回原始明文。"""
body = "title_id=%s&title_ver=%s&client_id=%s" % (TITLE_ID, title_ver, client_id)
return allnet_post(PATH_INITIALIZE, body, host=host, timeout=timeout)
def delivery_instruction(client_id, *, title_ver=DEFAULT_TITLE_VER, host=DEFAULT_HOST,
timeout=DEFAULT_TIMEOUT):
"""POST /net/delivery/instruction —— DLI,返回原始明文(含 uri= 配信指示書地址)。"""
body = "title_id=%s&title_ver=%s&client_id=%s" % (TITLE_ID, title_ver, client_id)
return allnet_post(PATH_DELIVERY_INSTRUCTION, body, host=host, timeout=timeout)
def query_keychip(keychip, *, title_ver=DEFAULT_TITLE_VER, host=DEFAULT_HOST,
timeout=DEFAULT_TIMEOUT, raw_ok=False):
"""
查一台机台的身份。返回 dict(含 ok / result / 各字段)。
★ raw_ok=False 时,result != 1 直接抛 AllnetError(CLI 用);
raw_ok=True 时照样返回,由调用方自己看 result(批量用)。
"""
cid = parse_keychip(keychip)
text = initialize(cid, title_ver=title_ver, host=host, timeout=timeout)
kv = parse_kv(text)
result = kv.get("result", "")
place_hex = kv.get("place_id", "") or ""
try:
place_id = int(place_hex, 16) if place_hex else None
except ValueError:
place_id = None
region_names = [kv.get("region_name%d" % i, "") for i in range(4)]
region_names = [n for n in region_names if n]
try:
region_id = int(kv.get("region0", "") or 0)
except ValueError:
region_id = None
info = {
"ok": result == "1",
"result": result,
"result_text": RESULT_TEXT.get(result, "未知返回码 %s" % result),
"title_id": TITLE_ID,
"title_ver": title_ver,
"host": host,
# --- 六个值里的五个(KEYCHIP_ID 是推导值)---
"client_id": cid,
"region_id": region_id,
"place_id": place_id,
"place_name": kv.get("name", ""),
"region_name": region_names[0] if region_names else "",
# --- 附送 ---
"nickname": kv.get("nickname", ""),
"region_names": region_names,
"country": kv.get("country", ""),
"setting": kv.get("setting", ""),
"location_type": kv.get("location_type", ""),
"game_server_uri": kv.get("uri1", ""),
"auth_token": kv.get("uri2", ""),
"token": kv.get("token", ""),
"utc_time": kv.get("utc_time", ""),
"client_timezone": kv.get("client_timezone", ""),
"res_ver": kv.get("res_ver", ""),
"raw": text.strip(),
}
if info["place_id"] is not None:
info["keychip"] = make_keychip(cid, info["place_id"])
else:
info["keychip"] = ""
if not info["ok"] and not raw_ok:
raise AllnetError("ALL.Net 返回 result=%s(%s)" % (result, info["result_text"]))
return info
# =============================================================================
# 写回 config.py
# =============================================================================
CONFIG_FILENAME = "config.py"
# 写回哪些键(KEYCHIP_ID 是推导值,不写)
WRITEBACK_KEYS = ("CLIENT_ID", "REGION_ID", "PLACE_ID", "PLACE_NAME", "REGION_NAME")
def _replace_assignment(text, name, literal):
"""把 config.py 里唯一一行 NAME = ... 换成 NAME = literal。"""
pattern = re.compile(r"^(%s[ \t]*=[ \t]*).*$" % re.escape(name), re.M)
new_text, count = pattern.subn(lambda m: m.group(1) + literal, text, count=1)
if count != 1:
raise AllnetError("config.py 里找不到唯一的 %s = ... 赋值行(找到 %d 处)" % (name, count))
return new_text
def write_back_config(info, *, config_path=None, quiet=False):
"""
把查询结果写回 config.py。
安全措施:
1. 先 ast.parse 语法自检,写坏了就不落盘
2. 落盘前把原文件备份成 config.py.bak
3. 保留原文件的换行风格(LF / CRLF)
"""
path = config_path or os.path.join(os.path.dirname(os.path.abspath(__file__)),
CONFIG_FILENAME)
if not os.path.isfile(path):
raise AllnetError("找不到 config.py:%s" % path)
with open(path, "rb") as fp:
raw = fp.read()
newline = "\r\n" if b"\r\n" in raw else "\n"
text = raw.decode("utf-8").replace("\r\n", "\n")
pairs = [
("CLIENT_ID", json.dumps(info["client_id"], ensure_ascii=False)),
("REGION_ID", str(info["region_id"])),
("PLACE_ID", str(info["place_id"])),
("PLACE_NAME", json.dumps(info["place_name"], ensure_ascii=False)),
("REGION_NAME", json.dumps(info["region_name"], ensure_ascii=False)),
]
for name, literal in pairs:
text = _replace_assignment(text, name, literal)
try:
ast.parse(text)
except SyntaxError as exc:
raise AllnetError("改写后 config.py 语法不合法,已放弃写入:%s" % exc) from exc
shutil.copy2(path, path + ".bak")
with open(path, "wb") as fp:
fp.write(text.replace("\n", newline).encode("utf-8"))
if not quiet:
print(" [已写回] %s" % path)
print(" [已备份] %s.bak" % path)
return path
def config_switch_enabled(config_path=None):
"""读 config.py 里的 AUTHV1_WRITE_CONFIG 开关。读不到就当作关。"""
path = config_path or os.path.join(os.path.dirname(os.path.abspath(__file__)),
CONFIG_FILENAME)
try:
with open(path, "rb") as fp:
tree = ast.parse(fp.read().decode("utf-8"))
except Exception:
return False
for node in tree.body:
if isinstance(node, ast.Assign):
for target in node.targets:
if isinstance(target, ast.Name) and target.id == "AUTHV1_WRITE_CONFIG":
try:
return bool(ast.literal_eval(node.value))
except Exception:
return False
return False
# =============================================================================
# 输出
# =============================================================================
def format_report(info) -> str:
"""人类可读的结果块。"""
lines = []
lines.append("=== authv1 机台身份查询 ===")
lines.append(" Keychip (完整) : %s" % (info["keychip"] or "(查不到)"))
lines.append(" clientId (短值) : %s <- 游戏服务器 clientId" % info["client_id"])
lines.append(" placeId : %s" % info["place_id"])
lines.append(" placeName : %s" % info["place_name"])
lines.append(" regionId : %s" % info["region_id"])
lines.append(" regionName : %s" % info["region_name"])
lines.append(" --- 附送 ---")
lines.append(" 游戏服务器 : %s" % info["game_server_uri"])
lines.append(" 机台昵称 : %s" % info["nickname"])
lines.append(" 国家/地区 : %s" % info["country"])
lines.append(" 认证时间(UTC) : %s 时区 %s" % (info["utc_time"], info["client_timezone"]))
lines.append(" ALL.Net 返回码 : %s(%s)" % (info["result"], info["result_text"]))
return "\n".join(lines)
# =============================================================================
# CLI
# =============================================================================
def build_parser():
p = argparse.ArgumentParser(
prog="authv1.py",
description="ALL.Net(alAbEx v4) 机台认证:Keychip -> clientId/regionId/placeId/placeName/regionName",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=(
"示例:\n"
" python authv1.py A63E01E9580\n"
" python authv1.py A63E-01C30471007 --json\n"
" python authv1.py A63E01E9580 --write\n"
" python authv1.py --batch source/keychip.csv --limit 10\n"
),
)
p.add_argument("keychip", nargs="?", help="Keychip(完整带连字符 / 完整连写 / 11 位短值都行)")
p.add_argument("--batch", metavar="CSV",
help="批量查询:CSV 文件,默认读第一列(表头含 Keychip 时自动跳过)")
p.add_argument("--limit", type=int, default=0, help="批量模式最多查多少条(0 = 不限)")
p.add_argument("--ver", default=None, help="上报给 ALL.Net 的版本号(默认取 config.MAI_ENCODING)")
p.add_argument("--host", default=DEFAULT_HOST, help="ALL.Net 认证服务器(默认 %s)" % DEFAULT_HOST)
p.add_argument("--timeout", type=float, default=DEFAULT_TIMEOUT, help="单次请求超时秒数")
p.add_argument("--json", action="store_true", help="输出 JSON")
g = p.add_mutually_exclusive_group()
g.add_argument("--write", action="store_true", help="强制把结果写回 config.py")
g.add_argument("--no-write", action="store_true", help="强制不写回 config.py")
return p
def resolve_title_ver(cli_value):
if cli_value:
return str(cli_value)
try:
import config # noqa: F401 同目录
return str(getattr(config, "MAI_ENCODING", DEFAULT_TITLE_VER))
except Exception:
return DEFAULT_TITLE_VER
def main(argv=None):
try:
sys.stdout.reconfigure(errors="replace")
except Exception:
pass
args = build_parser().parse_args(argv)
title_ver = resolve_title_ver(args.ver)
if args.batch:
return run_batch(args, title_ver)
if not args.keychip:
build_parser().print_help()
print("\n错误:请给一个 Keychip,或用 --batch 指定 CSV。")
return 2
try:
info = query_keychip(args.keychip, title_ver=title_ver,
host=args.host, timeout=args.timeout)
except (AllnetError, ValueError) as exc:
print("查询失败:%s" % exc)
return 1
if args.json:
print(json.dumps(info, ensure_ascii=False, indent=1))
else:
print(format_report(info))
do_write = decide_write(args)
if do_write:
try:
write_back_config(info)
except AllnetError as exc:
print("写回失败:%s" % exc)
return 1
elif not args.json:
print(" [未写回] 想让它自动写回:把 config.py 的 AUTHV1_WRITE_CONFIG 改成 True,"
"或加 --write")
return 0
def decide_write(args):
"""--write / --no-write 优先,否则看 config.py 的开关。"""
if args.write:
return True
if args.no_write:
return False
return config_switch_enabled()
def _disp_width(text):
"""按终端显示宽度算长度:中日韩全角字符占 2 列。"""
import unicodedata
width = 0
for ch in str(text):
width += 2 if unicodedata.east_asian_width(ch) in ("W", "F") else 1
return width
def _pad(text, width):
text = str(text)
return text + " " * max(0, width - _disp_width(text))
def run_batch(args, title_ver):
path = args.batch
if not os.path.isfile(path):
print("找不到 CSV:%s" % path)
return 2
rows = []
with open(path, "r", encoding="utf-8-sig", newline="") as fp:
reader = csv.reader(fp)
for row in reader:
if not row or not row[0].strip():
continue
if row[0].strip().lower() in ("keychip", "keychip id", "keychipid"):
continue
rows.append(row)
if args.limit > 0:
rows = rows[: args.limit]
print("=== authv1 批量查询(%d 台,版本 %s)===" % (len(rows), title_ver))
header = " ".join([_pad("clientId", 13), _pad("result", 6), _pad("placeId", 8),
_pad("regionId", 8), _pad("区域", 8), "店铺名"])
print(header)
ok = 0
results = []
for row in rows:
try:
info = query_keychip(row[0], title_ver=title_ver, host=args.host,
timeout=args.timeout, raw_ok=True)
except Exception as exc:
print("%s ERROR %s" % (_pad(row[0].strip()[:13], 13), exc))
continue
results.append(info)
if info["ok"]:
ok += 1
print(" ".join([
_pad(info["client_id"], 13),
_pad(info["result"], 6),
_pad(info["place_id"], 8),
_pad(info["region_id"], 8),
_pad(info["region_name"], 8),
str(info["place_name"]),
]))
print("--- 成功 %d / %d ---" % (ok, len(rows)))
return 0
if __name__ == "__main__":
sys.exit(main())
+32 -13
View File
@@ -47,17 +47,12 @@ TIMEOUT_LONG_MS = 60000 # 登录 / 结算上传
# 二、机台与店铺标识
# =============================================================================
# Keychip ID(机台唯一标识,完整格式带连字符)
# ★★ 这是你自己的机台身份,不要外传;填错服务端会拒。
# 依据:source/Assembly-CSharp/Manager/ChimeReaderManager.cs:250
# KeychipID = AMDaemon.System.KeychipId.Value
KEYCHIP_ID = "" # ★★ 必填:形如 "XXXX-XXXXXXXXXXX"(带连字符)
# 客户端 ID(Keychip 短值,去掉连字符取前 11 位)
# 客户端 ID(Keychip 短值)
# 依据:source/Assembly-CSharp/Net.Packet.Mai2/PacketGetGameSetting.cs:23
# clientId = AMDaemon.System.KeychipId.ShortValue
# 实测:真机日志里 clientId 恒等于 Keychip 短值
CLIENT_ID = "" # ★★ 必填:Keychip 短值,去掉连字符取前 11 位
# ★ 形态 = 15 位 Keychip 去掉连字符后的**前 11 位**(大写十六进制)。
CLIENT_ID = "" # ★★ 必填:形如 "A63E01C3047"
# 区域 ID / 店铺 ID
# 依据:source/Assembly-CSharp/Net.Packet.Mai2/PacketUserLogin.cs:24-25
@@ -66,9 +61,18 @@ CLIENT_ID = "" # ★★ 必填:Keychip 短值,去掉连字
REGION_ID = 0 # ★★ 必填:你所在区域 ID
PLACE_ID = 0 # ★★ 必填:你所在店铺 ID
# Keychip ID(机台唯一标识,完整格式带连字符)
# ★★ 这是你自己的机台身份,不要外传;填错服务端会拒。
# 依据:source/Assembly-CSharp/Manager/ChimeReaderManager.cs:250
# KeychipID = AMDaemon.System.KeychipId.Value
# ★ 组成 = CLIENT_ID 的前 4 位 + "-" + CLIENT_ID 第 5 位起 + PLACE_ID
# 例:CLIENT_ID "A63E01C3047" + PLACE_ID 1007 -> "A63E-01C30471007"
# 所以它由上面两个值**推导**,不要手改。
KEYCHIP_ID = (CLIENT_ID[:4] + "-" + CLIENT_ID[4:] + str(PLACE_ID)) if CLIENT_ID else ""
# 店铺名 / 区域名
#
# ★ 为什么必须手工填:**游戏服务器没有任何接口返回店铺名**。
# ★ 为什么以前必须手工填:**游戏服务器没有任何接口返回店铺名**。
# 逐行扫过 source/2026-10-05.log 全部响应:
# - GetGameSettingApi 只回 gameSetting,无店名
# - GetUserDataApi 的 lastPlaceName 回的是 null
@@ -77,13 +81,27 @@ PLACE_ID = 0 # ★★ 必填:你所在店铺 ID
# source/AMDaemon.NET/AMDaemon/Allnet/Auth.cs:33 LocationName <- allnet_Auth_getLocationName()
# 是开机时 ALL.Net 认证下发的,客户端本地就有,不会向游戏服务器索取。
#
# 取值示例(真机 UpsertUserAll 报文):
# userPlaylogList[0].placeName = "<你的店铺名>"
# userData[0].lastRegionName = "<你的区域名>"
PLACE_NAME = "" # ★★ 必填:店铺名(游戏服务器不返回,只能手填)
# ★★ 现在不用手工抄了:python authv1.py <你的Keychip> 会直接问 ALL.Net
# 认证服务器要回 name / region_name0。把下面的 AUTHV1_WRITE_CONFIG 打开,
# 它还能把结果自动写回 config.py。
PLACE_NAME = "" # ★★ 必填:店铺名
REGION_NAME = "" # ★★ 必填:区域名
# --- authv1 回写开关 ---
#
# authv1.py(Keychip -> 机台身份)查到结果后,是否**自动写回 config.py**。
#
# True : 每次查询成功后,把 CLIENT_ID / REGION_ID / PLACE_ID /
# PLACE_NAME / REGION_NAME 五个值改写进 config.py。
# (KEYCHIP_ID 是推导值,会自动跟着变,不用写。)
# 改写前会把原文件备份成 config.py.bak,写坏不落盘。
# False : 只打印结果,不动 config.py。(默认,安全)
#
# 命令行可临时覆盖:--write 强制写 / --no-write 强制不写。
AUTHV1_WRITE_CONFIG = False
# =============================================================================
# 三、扫码登录常量(aimedb)
# =============================================================================
@@ -356,6 +374,7 @@ __all__ = [
"TIMEOUT_MS", "TIMEOUT_LONG_MS",
# 机台与店铺
"KEYCHIP_ID", "CLIENT_ID", "REGION_ID", "PLACE_ID", "PLACE_NAME", "REGION_NAME",
"AUTHV1_WRITE_CONFIG",
# 功能票
"TICKET_CREDIT_NUM", "TICKET_NAMES", "PAID_TICKET_IDS",
"UPSERT_CHARGELOG_CONFIRM", "TICKET_VALID_HOUR_OFFSET", "TICKET_VALID_DAYS",
+703
View File
@@ -0,0 +1,703 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
r"""
netdelivery.py —— ALL.Net DLI(Download Instruction):查 / 下 .app 与 .opt 更新包
================================================================================
【DLI 是什么】
DLI = **Download Instruction**(配信指示書)。
机台做完 ALL.Net 认证后,AMDaemon 会再问一次「我该下什么包」:
POST http://at.sys-allnet.cn/net/delivery/instruction
body = title_id=SDGB&title_ver=<版本>&client_id=<Keychip 短值>
服务器回:
result=1&uri=<patch 指示書 URL>|<option 指示書 URL>
指示書本身是个 INI(日文注释),真正要下的包写在 INSTALLn= 里:
[COMMON] INSTALL1=https://.../SDGB_1.56.00_..._1_1.55.01.app
[OPTIONAL] INSTALL1=https://.../SDGB_A031_..._0.opt
INSTALL2=https://.../SDGB_A005_..._0.opt
★ 只有 INSTALLn= 才是**要下载的包**;EXISTn= 是「本机应该已经有的文件」,
只是文件名,不是下载项。
【协议出处】—— source/amdaemon.exe 字符串(数字是文件内偏移)
端点 http://at.sys-all.cn/net/delivery/instruction @0x788198
库 alw_lite/src/system39/alw_api_netdeliver.c @0x63bf80
阶段 ALWE_NETDELIVER_PHASE_GET_DLINST_INFO @0x63bde0
ALWE_NETDELIVER_PHASE_APP_IMAGE_INFO @0x63be08
ALWE_NETDELIVER_PHASE_OPT_IMAGE_INFO @0x63be30
日志 "Net deliver param - dli uri : %s." @0x63c3f8
状态 ALE_ABEX_STAT_DLINSTR_OK / _ERR_DLINSTR_NG ... @0x638600
【实测结论】(2026-10-06)
title_ver=1.55 -> uri = patch(SDGB_patch_1.56...info) | option(SDGB_option_1.55...A005.info)
title_ver=1.56 -> uri = | option(A031) (patch 槽为空 = 本体已最新)
title_ver<=1.54 -> uri = null (服务器不认这个版本)
指示書里的 INSTALL 直链可以带 Range 断点续传(返回 206 + Content-Range)。
★ 已被下架的历史包,源站会回 504 —— 这是服务器状态,不是脚本的问题。
【用法】
python netdelivery.py # 用 config.py 里的 Keychip 查,然后**多选下载**
python netdelivery.py A63E01E9580 # 指定 Keychip
python netdelivery.py --list # 只列清单,不下载
python netdelivery.py --all # 全下(不交互)
python netdelivery.py --only 1,3 # 只下第 1、3 项(不交互)
python netdelivery.py --ver 1.55 # 指定上报版本(默认 config.MAI_ENCODING)
python netdelivery.py --out D:\SDGB2026\update
python netdelivery.py --insecure # 关掉 TLS 校验(CDN 证书有问题时用)
★ 交互选择支持:逗号/空格分隔、区间(如 1-3)、a=全选、n=全不选、q=退出。
"""
from __future__ import annotations
import argparse
import os
import re
import sys
import time
import requests
import urllib3
from authv1 import (
DEFAULT_HOST,
DEFAULT_TIMEOUT,
AllnetError,
delivery_instruction,
parse_keychip,
)
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
# 下载器用的 UA(实测有效)
DOWNLOAD_HEADERS = {"User-Agent": "SDGB;Windows/Lite"}
DEFAULT_OUT_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "download")
CHUNK = 1 << 16
PROBE_TIMEOUT = 30
DOWNLOAD_TIMEOUT = 60
# 包类型判定(按扩展名)
KIND_BY_EXT = {
".app": "app",
".opt": "opt",
".pack": "pack",
}
KIND_TEXT = {
"app": "本体/补丁",
"opt": "选项包",
"pack": "资源包",
"?": "未知",
}
# =============================================================================
# HTTP
# =============================================================================
def http_get(url, *, headers=None, stream=False, timeout=PROBE_TIMEOUT, insecure=False):
"""
GET,顺带处理 CDN 证书问题。
先按正常校验发;如果因为证书失败,再自动退回 verify=False 重发一次
(只影响这一次请求,不改变全局设置)。
"""
merged = dict(DOWNLOAD_HEADERS)
if headers:
merged.update(headers)
try:
return requests.get(url, headers=merged, stream=stream, timeout=timeout,
verify=not insecure)
except requests.exceptions.SSLError:
if insecure:
raise
return requests.get(url, headers=merged, stream=stream, timeout=timeout,
verify=False)
# =============================================================================
# 配信指示書解析
# =============================================================================
def parse_instruction(text):
"""
解析配信指示書 INI。
返回 (common, installs):
common = {GAME_DESC: ..., DOWNLOAD_ID: ..., ...}([COMMON] 段)
installs = [{"section": "COMMON"/"OPTIONAL", "key": "INSTALL1", "url": ...}, ...]
"""
common = {}
installs = []
section = ""
for raw_line in text.splitlines():
line = raw_line.strip()
if not line or line.startswith(";") or line.startswith("#"):
continue
if line.startswith("[") and line.endswith("]"):
section = line[1:-1].strip().upper()
continue
if "=" not in line:
continue
key, value = line.split("=", 1)
key = key.strip().upper()
value = value.strip().strip('"')
if section in ("", "COMMON"):
common.setdefault(key, value)
if key.startswith("INSTALL") and value:
installs.append({"section": section or "COMMON", "key": key, "url": value})
return common, installs
def guess_kind(url):
"""按扩展名猜包类型。"""
name = url.split("?")[0].rsplit("/", 1)[-1]
ext = os.path.splitext(name)[1].lower()
return KIND_BY_EXT.get(ext, "?")
def package_name(url):
return url.split("?")[0].rsplit("/", 1)[-1]
def human_size(num):
if num is None:
return "?"
value = float(num)
for unit in ("B", "KB", "MB", "GB", "TB"):
if value < 1024 or unit == "TB":
return ("%d B" % int(value)) if unit == "B" else ("%.2f %s" % (value, unit))
value /= 1024.0
return "?"
# =============================================================================
# 采集:DLI -> 指示書 -> 包清单
# =============================================================================
def fetch_instruction_urls(client_id, *, title_ver, host, timeout):
"""问 DLI,返回 (patch_info_url, option_info_url, raw_text)。"""
text = delivery_instruction(client_id, title_ver=title_ver, host=host, timeout=timeout)
kv = {}
for part in text.strip().split("&"):
if "=" in part:
k, v = part.split("=", 1)
kv[k.strip()] = v
result = kv.get("result", "")
if result != "1":
raise AllnetError("DLI 返回 result=%s(原始:%s)" % (result, text.strip()[:200]))
uri = kv.get("uri", "").strip()
if uri in ("", "null"):
raise AllnetError("DLI 返回 uri=%s:这个 title_ver 没有配信指示書"
"(换 --ver 试试,例如 1.55)" % (uri or "(空)"))
parts = uri.split("|")
patch_url = parts[0].strip() if len(parts) > 0 else ""
option_url = parts[1].strip() if len(parts) > 1 else ""
for bad in ("null", "none"):
if patch_url.lower() == bad:
patch_url = ""
if option_url.lower() == bad:
option_url = ""
return patch_url, option_url, text.strip()
def collect_packages(client_id, *, title_ver, host, timeout, insecure=False):
"""
走完整条链路,返回 (items, meta)。
items 每一项:
{"idx":1, "kind":"app", "section":"COMMON", "key":"INSTALL1",
"url":..., "name":..., "size":int|None, "note":"", "info_name":...}
"""
patch_url, option_url, dli_raw = fetch_instruction_urls(
client_id, title_ver=title_ver, host=host, timeout=timeout)
items = []
meta = {"dli_raw": dli_raw, "title_ver": title_ver, "client_id": client_id,
"instructions": [], "infos": {}}
for info_url in (patch_url, option_url):
if not info_url:
continue
info_name = package_name(info_url)
resp = None
for attempt in range(1, 4):
try:
resp = http_get(info_url, timeout=timeout, insecure=insecure)
if resp.status_code < 500:
break
except Exception:
resp = None
time.sleep(1.5 * attempt)
if resp is None:
meta["instructions"].append({"url": info_url,
"error": "拉取失败(已重试 3 次)"})
continue
if resp.status_code != 200:
meta["instructions"].append({"url": info_url, "status": resp.status_code,
"error": "HTTP %d" % resp.status_code})
continue
text = resp.content.decode("utf-8", "replace")
common, installs = parse_instruction(text)
meta["instructions"].append({"url": info_url, "status": 200,
"name": info_name, "common": common})
meta["infos"][info_name] = text
seen = set()
for entry in installs:
if entry["url"] in seen:
continue
seen.add(entry["url"])
items.append({
"kind": guess_kind(entry["url"]),
"section": entry["section"],
"key": entry["key"],
"url": entry["url"],
"name": package_name(entry["url"]),
"size": None,
"note": "",
"info_name": info_name,
})
for i, item in enumerate(items, 1):
item["idx"] = i
return items, meta
def probe_all(items, *, timeout=PROBE_TIMEOUT, insecure=False):
"""探每个包的大小 / 可用性(就地写回 item)。"""
for item in items:
for attempt in range(1, 3):
try:
resp = http_get(item["url"], headers={"Range": "bytes=0-0"},
stream=True, timeout=timeout, insecure=insecure)
status = resp.status_code
content_range = resp.headers.get("Content-Range", "")
content_length = resp.headers.get("Content-Length")
resp.close()
if status == 206 and "/" in content_range:
item["size"] = int(content_range.rsplit("/", 1)[-1])
item["note"] = ""
elif status == 200 and content_length and content_length.isdigit():
item["size"] = int(content_length)
item["note"] = ""
elif status in (404, 504):
item["note"] = "HTTP %d(源站不可用 / 该包已下架)" % status
else:
item["note"] = "HTTP %d" % status
break
except Exception as exc:
item["note"] = "探测失败:%s" % exc
if attempt < 2:
time.sleep(1.5)
continue
# =============================================================================
# 展示与选择
# =============================================================================
def print_packages(items):
if not items:
print(" (指示書里没有任何 INSTALL 项)")
return
print("")
print(" 序号 类型 大小 来源段/键 文件名")
print(" " + "-" * 92)
for item in items:
size = human_size(item["size"]) if item["size"] is not None else (item["note"] or "?")
print(" [%2d] %-4s %-12s %-20s %s" % (
item["idx"], item["kind"], size,
item["section"] + "/" + item["key"], item["name"]))
if item["note"] and item["size"] is None:
print(" %s" % item["note"])
def prompt_selection(items):
"""
交互多选。返回选中的 0 基下标列表;用户 q 退出时返回 None。
"""
print("")
print(" 可多选:输入序号,用逗号或空格分隔;支持区间(如 1-3)。")
print(" a = 全选 n = 都不选 q = 退出")
while True:
try:
raw = input(" 请选择要下载的包 > ").strip()
except (EOFError, KeyboardInterrupt):
print("")
return None
if not raw:
continue
low = raw.lower()
if low in ("q", "quit", "exit"):
return None
if low in ("a", "all"):
return list(range(len(items)))
if low in ("n", "none"):
return []
picked, bad = [], []
for token in re.split(r"[,,、\s]+", raw):
if not token:
continue
span = re.fullmatch(r"(\d+)\s*-\s*(\d+)", token)
if span:
low_i, high_i = sorted((int(span.group(1)), int(span.group(2))))
for i in range(low_i, high_i + 1):
if 1 <= i <= len(items):
picked.append(i - 1)
else:
bad.append(str(i))
continue
if token.isdigit():
i = int(token)
if 1 <= i <= len(items):
picked.append(i - 1)
else:
bad.append(token)
else:
bad.append(token)
if bad:
print(" [x] 无效输入:%s(范围是 1-%d)" % (", ".join(bad), len(items)))
continue
ordered, seen = [], set()
for i in picked:
if i not in seen:
seen.add(i)
ordered.append(i)
if not ordered:
print(" [x] 没有选中任何项")
continue
return ordered
# =============================================================================
# 下载
# =============================================================================
def download_file(url, dest, *, timeout=DOWNLOAD_TIMEOUT, retries=3, insecure=False):
"""
带断点续传的下载。返回 (ok, 说明)。
★ 先写 dest + ".part",下完再改名 —— 中断了下次能接着下。
"""
part = dest + ".part"
total = None
for attempt in range(1, retries + 1):
done = os.path.getsize(part) if os.path.exists(part) else 0
headers = {}
if done:
headers["Range"] = "bytes=%d-" % done
try:
resp = http_get(url, headers=headers, stream=True, timeout=timeout,
insecure=insecure)
except Exception as exc:
if attempt == retries:
return False, "连接失败:%s" % exc
time.sleep(2)
continue
if resp.status_code not in (200, 206):
resp.close()
if resp.status_code in (404, 504):
return False, "HTTP %d(源站不可用 / 该包已下架)" % resp.status_code
if attempt == retries:
return False, "HTTP %d" % resp.status_code
time.sleep(2)
continue
# 服务器忽略 Range 直接给了全量 -> 从头写
if done and resp.status_code == 200:
done = 0
content_range = resp.headers.get("Content-Range", "")
if "/" in content_range:
try:
total = int(content_range.rsplit("/", 1)[-1])
except ValueError:
total = None
elif resp.headers.get("Content-Length", "").isdigit():
total = done + int(resp.headers["Content-Length"])
mode = "ab" if done else "wb"
started = time.time()
written = done
last_tick = 0.0
last_shown = -1
try:
with open(part, mode) as fp:
for chunk in resp.iter_content(CHUNK):
if not chunk:
continue
fp.write(chunk)
written += len(chunk)
now = time.time()
if now - last_tick >= 0.3:
last_tick = now
last_shown = written
show_progress(written, total, now - started)
except Exception as exc:
resp.close()
if attempt == retries:
return False, "写入中断:%s" % exc
time.sleep(2)
continue
finally:
resp.close()
if written == last_shown:
sys.stdout.write("\n") # 收尾换行,避免重复打一行
else:
show_progress(written, total, max(time.time() - started, 0.001), final=True)
if total is not None and written != total:
if attempt == retries:
return False, "长度不符:期望 %d 字节,实际 %d 字节" % (total, written)
time.sleep(2)
continue
if os.path.exists(dest):
os.remove(dest)
os.replace(part, dest)
return True, "%s(%s)" % (human_size(written), dest)
return False, "重试 %d 次仍失败" % retries
def show_progress(done, total, elapsed, final=False):
speed = done / elapsed if elapsed > 0 else 0
if total:
percent = done * 100.0 / total
line = " %6.2f%% %s / %s %s/s" % (percent, human_size(done),
human_size(total), human_size(speed))
else:
line = " %s %s/s" % (human_size(done), human_size(speed))
sys.stdout.write("\r" + line + ("\n" if final else ""))
sys.stdout.flush()
# =============================================================================
# CLI
# =============================================================================
def build_parser():
p = argparse.ArgumentParser(
prog="netdelivery.py",
description="ALL.Net DLI:查 / 下 .app 与 .opt 更新包",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=(
"示例:\n"
" python netdelivery.py # 查完让你多选\n"
" python netdelivery.py --list # 只列清单\n"
" python netdelivery.py --all # 全下\n"
" python netdelivery.py --only 1,3 # 只下第 1、3 项\n"
" python netdelivery.py --ver 1.55 # 指定上报版本\n"
),
)
p.add_argument("keychip", nargs="?",
help="Keychip(默认取 config.py 的 KEYCHIP_ID)")
p.add_argument("--ver", default=None,
help="上报给 ALL.Net 的版本号(默认取 config.MAI_ENCODING)")
p.add_argument("--host", default=DEFAULT_HOST,
help="ALL.Net 认证服务器(默认 %s)" % DEFAULT_HOST)
p.add_argument("--out", default=DEFAULT_OUT_DIR,
help="下载目录(默认 <脚本目录>/download)")
p.add_argument("--timeout", type=float, default=DEFAULT_TIMEOUT, help="接口超时秒数")
p.add_argument("--list", action="store_true", help="只列清单与直链,不下载")
p.add_argument("--all", action="store_true", help="全选(不交互)")
p.add_argument("--only", default=None, help="只下指定序号,如 1,3 或 1-3(不交互)")
p.add_argument("--no-probe", action="store_true", help="跳过包大小探测(更快)")
p.add_argument("--insecure", action="store_true", help="关掉 TLS 校验")
return p
def resolve_keychip(cli_value):
if cli_value:
return cli_value
try:
import config # noqa: F401
value = getattr(config, "KEYCHIP_ID", "")
if value:
return value
except Exception:
pass
return ""
def resolve_title_ver(cli_value):
if cli_value:
return str(cli_value)
try:
import config # noqa: F401
return str(getattr(config, "MAI_ENCODING", "1.55"))
except Exception:
return "1.55"
def parse_only(text, count):
"""把 --only 的文本解析成 0 基下标列表。"""
picked, seen = [], set()
for token in re.split(r"[,,、\s]+", text or ""):
if not token:
continue
span = re.fullmatch(r"(\d+)\s*-\s*(\d+)", token)
if span:
low_i, high_i = sorted((int(span.group(1)), int(span.group(2))))
candidates = range(low_i, high_i + 1)
elif token.isdigit():
candidates = [int(token)]
else:
raise ValueError("--only 里有无效项:%r" % token)
for i in candidates:
if not (1 <= i <= count):
raise ValueError("--only 序号 %d 超出范围(1-%d)" % (i, count))
if i - 1 not in seen:
seen.add(i - 1)
picked.append(i - 1)
return picked
def save_instructions(meta, out_dir):
"""把配信指示書原样存一份,便于事后核对。"""
os.makedirs(out_dir, exist_ok=True)
saved = []
for name, text in meta["infos"].items():
path = os.path.join(out_dir, name)
with open(path, "w", encoding="utf-8", newline="") as fp:
fp.write(text)
saved.append(path)
return saved
def main(argv=None):
try:
sys.stdout.reconfigure(errors="replace")
except Exception:
pass
args = build_parser().parse_args(argv)
title_ver = resolve_title_ver(args.ver)
raw_keychip = resolve_keychip(args.keychip)
if not raw_keychip:
print("错误:没给 Keychip,config.py 里的 KEYCHIP_ID 也是空的。")
print(" 先跑一次 python authv1.py <你的Keychip> 把 config.py 填好。")
return 2
try:
client_id = parse_keychip(raw_keychip)
except ValueError as exc:
print("Keychip 解析失败:%s" % exc)
return 2
print("=== netdelivery (ALL.Net DLI) ===")
print(" clientId : %s" % client_id)
print(" title_ver : %s" % title_ver)
print(" server : %s" % args.host)
print(" 正在查询配信指示書 ...")
try:
items, meta = collect_packages(client_id, title_ver=title_ver, host=args.host,
timeout=args.timeout, insecure=args.insecure)
except (AllnetError, ValueError) as exc:
print("查询失败:%s" % exc)
return 1
for entry in meta["instructions"]:
if entry.get("status") == 200:
common = entry.get("common", {})
print(" 指示書 %s" % entry["name"])
for key in ("GAME_DESC", "DLFORMAT", "GAME_ID", "DOWNLOAD_ID",
"RELEASE_TIME", "RELEASE_TYPE"):
if key in common:
print(" %-14s %s" % (key + ":", common[key]))
else:
print(" 指示書 %s -> %s" % (entry["url"], entry.get("error", "?")))
if not items:
print("没有可下载的包。")
return 0
if not args.no_probe:
print(" 正在探测包大小 ...")
probe_all(items, insecure=args.insecure)
print_packages(items)
if args.list:
print("")
print(" 直链:")
for item in items:
print(" [%2d] %s" % (item["idx"], item["url"]))
return 0
# ---- 决定下哪些 ----
if args.only:
try:
chosen = parse_only(args.only, len(items))
except ValueError as exc:
print("参数错误:%s" % exc)
return 2
elif args.all:
chosen = list(range(len(items)))
elif not sys.stdin or not sys.stdin.isatty():
print("")
print(" 当前不是交互终端,未指定 --all / --only,已只列清单。")
return 0
else:
chosen = prompt_selection(items)
if chosen is None:
print(" 已取消。")
return 0
if not chosen:
print(" 没有选中任何项,退出。")
return 0
out_dir = args.out
os.makedirs(out_dir, exist_ok=True)
for path in save_instructions(meta, out_dir):
print(" [指示書] %s" % path)
print("")
print(" 下载目录:%s" % out_dir)
failed = []
for index in chosen:
item = items[index]
dest = os.path.join(out_dir, item["name"])
print("")
print(" [%d/%d] %s(%s)" % (chosen.index(index) + 1, len(chosen),
item["name"], KIND_TEXT.get(item["kind"], "?")))
if item["size"] is None and item["note"]:
print(" 跳过:%s" % item["note"])
failed.append((item, item["note"]))
continue
started = time.time()
ok, message = download_file(item["url"], dest, insecure=args.insecure)
if ok:
print(" 完成,用时 %.1f 秒 -> %s" % (time.time() - started, message))
else:
print(" 失败:%s" % message)
failed.append((item, message))
print("")
if failed:
print(" --- 结束:%d 个成功,%d 个失败 ---" % (len(chosen) - len(failed), len(failed)))
for item, message in failed:
print(" [%2d] %s -> %s" % (item["idx"], item["name"], message))
return 1
print(" --- 结束:%d 个全部成功 ---" % len(chosen))
return 0
if __name__ == "__main__":
sys.exit(main())