7 Commits
Author SHA1 Message Date
shouko 5220eb1708 release: v1.0.1
全曲解锁改为一包发完(per 默认 0),不再每包 4 首反复登出重登;失败时自动降级到 200 首 / 50 首重试
2026-10-29 11:52:51 +08:00
shouko 791d11a5c2 更新 README.md 2026-10-28 11:51:20 -04:00
shouko a7c79ff52c add: 发版脚本 release.py
每次发版新开 Release、新版本号,同名 tag 直接报错退出,绝不覆盖旧 asset。
token 从 GITEA_TOKEN 或 tools/.release-token 读,不入库。
2026-10-28 22:59:39 +08:00
shouko a2f4984ef7 fix: 删掉重复定义的旧 checkUpdate
旧版读 \upSrc.value,那个元素早就删了,null.value 抛 TypeError,
点「检查更新」什么都不发生(旧定义覆盖了新定义)。
2026-10-28 22:51:35 +08:00
shouko cbd36445f0 add: 空白配置模板 maipulse-config.json
仓库根和 maipulse/ 各一份,字段全空,UTF-8 无 BOM
2026-10-28 22:40:04 +08:00
shouko 132ff656e1 fix: 更新检查改用浏览器 UA
git.fragrance.moe 走 Cloudflare,非浏览器 UA 会被 403 error 1010
2026-10-28 22:36:51 +08:00
shouko 4223b0cd19 feat: 更新源改到自建 Gitea
- /api/update-check 走 git.fragrance.moe 的 Release API(公开仓库,无需 token)
- 设置页加「检查更新 / 下载安装」,下载走 assets 的 browser_download_url
- asset.rev 由打包脚本生成,覆盖安装必定重建资源
2026-10-28 22:34:14 +08:00
12 changed files with 511 additions and 141 deletions

No files matched your search

+4
View File
@@ -28,3 +28,7 @@ astral/config.py
.DS_Store
Thumbs.db
*.swp
# 发版 token
maipulse/tools/.release-token
tools/.release-token
+11 -1
View File
@@ -2,7 +2,7 @@
舞萌 DX 国服(SDGB)的本地发包台。Flask + 单页前端,安卓壳用 Chaquopy 把 Python 打进 APK。
**这个仓库是脱敏存档**:所有机台身份(标题服务器地址、AES 密钥与 IV、混淆盐)和店铺信息都已经抹掉,
所有机台身份(标题服务器地址、AES 密钥与 IV、混淆盐)和店铺信息都已经抹掉,
字段留空,运行时从配置文件读。想用的话自己填——见下面的「配置」。
## 目录
@@ -33,8 +33,18 @@ android/ Android 壳工程(Gradle + MainActivity)
填完在设置页点「导入配置」加载 JSON,或者直接改这两个文件后重启。
仓库根目录有一份空白模板 `maipulse-config.json`(`maipulse/` 下也放了一份),所有字段都是空的,
照着填就行。文件要存成 UTF-8 无 BOM,否则 JSON 解析会失败。
## 跑起来
先把配置模板复制一份,否则 import 会缺 config:
```bash
cd astral && cp config.py.example config.py
```
```bash
python app.py # http://127.0.0.1:8787
```
+6 -6
View File
@@ -11,17 +11,17 @@ MaimaiNet.py —— 舞萌DX 网络核心层(纯 HTTP 复刻)
1) 服务器基址与 API 路径
source/Assembly-CSharp/Manager/OperationManager.cs:225,338
ServerUri = Auth.GameServerUri + MovieServerUri("<REDACTED>/")
ServerUri = Auth.GameServerUri + MovieServerUri("<OBFUSCATE_PARAM>/")
source/Assembly-CSharp/Net.Packet/Packet.cs:116
Client = NetHttpClient.Create(BaseUrl + Obfuscator(query.Api))
source/Assembly-CSharp/Net.Packet/Packet.cs:485
Obfuscator(s) = md5_hex_lower(s + "<REDACTED>")
Obfuscator(s) = md5_hex_lower(s + "<OBFUSCATE_PARAM>")
source/Assembly-CSharp/Net.VO/NetQuery.cs:14
Api = api + "MaimaiChn"
=> 最终 URL = BASE_URL + md5(逻辑接口名 + "MaimaiChn" + "<REDACTED>")
=> 最终 URL = BASE_URL + md5(逻辑接口名 + "MaimaiChn" + "<OBFUSCATE_PARAM>")
已用真实日志验证:md5("GetGameSettingApiMaimaiChn" + "<REDACTED>")
已用真实日志验证:md5("GetGameSettingApiMaimaiChn" + "<OBFUSCATE_PARAM>")
= 83a5d5a20b062c5c2ad817460b8f7f76
(source/2026-10-05.log:1 的请求就是打到这个尾段)
@@ -121,7 +121,7 @@ from config import CLIENT_ID, MAI_ENCODING, TIMEOUT_LONG_MS, TIMEOUT_MS
#: 游戏服务器(Maimai2Servlet)
#: 依据 source/Assembly-CSharp/Manager/OperationManager.cs:225,338
#: _operationData.ServerUri = Auth.GameServerUri + MovieServerUri("<REDACTED>/")
#: _operationData.ServerUri = Auth.GameServerUri + MovieServerUri("<OBFUSCATE_PARAM>/")
BASE_URL = ""
#: AES 参数,依据 source/Assembly-CSharp/Net/CipherAES.cs:11-12
@@ -162,7 +162,7 @@ def obfuscate_api_name(api_name: str) -> str:
对逻辑接口名做 MD5 混淆,得到 URL 尾段(对应 Packet.Obfuscator)。
逻辑名 "GetUserPreviewApi" -> "GetUserPreviewApiMaimaiChn"
-> md5(上面 + "<REDACTED>")
-> md5(上面 + "<OBFUSCATE_PARAM>")
参数:
api_name: 逻辑接口名,**不要**自己带 MaimaiChn 后缀
+13
View File
@@ -0,0 +1,13 @@
{
"aesIV": "",
"aesKey": "",
"apiVersion": "",
"baseUrl": "",
"clientId": "",
"keychip": "",
"obfuscate": "",
"placeId": "",
"placeName": "",
"regionId": "",
"regionName": ""
}
+1 -1
View File
@@ -39,7 +39,7 @@ EasyMai 的路子是绕的:自建后端 `getUserMusic` + 水鱼 `update_record
| 鉴权 | jwt-token + import-token | 只要 userId + 机台 clientId,**不用登录** |
| 写操作 | 往水鱼导入成绩 | 无,纯只读 |
#### 已验证的接口形状(2026-10-28 实测,账号 <REDACTED>)
#### 已验证的接口形状(2026-10-28 实测,账号 12800621)
GetUserRating -> userRating:
rating : 0(服务端不填,别指望它)
+1 -1
View File
@@ -148,7 +148,7 @@ MaiTechUI 金票包 UpsertUserAllApi.poop() 原样:
它从不在登录后 60 秒内提交结算。服务端对"登录不到一分钟就交成绩"的包一律回
returnCode=1 然后把 userItemList / userMusicDetailList 整块丢掉。
已验证(账号 <REDACTED>):每次发包前补足 60 秒后,四类道具全部真落地。
已验证(账号 12042860):每次发包前补足 60 秒后,四类道具全部真落地。
| kind | 名称 | 发送 id | 库存变化 | waited |
| --- | --- | --- | --- | --- |
+66 -24
View File
@@ -790,8 +790,8 @@ def api_master():
# ---------------------------------------------------------------- 更新
APP_VERSION = "1.0.0"
APP_BUILD = "2026-10-28"
APP_VERSION = "1.0.1"
APP_BUILD = "2026-10-29"
@app.get("/api/version")
@@ -912,36 +912,78 @@ def api_update_gh():
repo=repo)
UA_BROWSER = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/120.0 Safari/537.36")
UPDATE_SITE = "https://git.fragrance.moe"
UPDATE_REPO = "shouko/maipulse"
@app.get("/api/update-check")
@guarded
def api_update_check():
"""拉远端 version.json 与本地比较。远端格式:
{"version": "1.0.1", "url": "下载地址", "notes": "更新说明", "published": "..."}
"""检查更新。默认走自建 Gitea 的 Release(公开仓库,不用 token)。
asset 用 browser_download_url —— 那是 /releases/download/<tag>/<name> 的稳定地址,
比 attachments/uuid 可靠。给了 source 参数则按老的远端 version.json 走。
"""
import urllib.request
src = (request.args.get("source") or "").strip()
if not src:
return fail("还没填更新源地址")
if not re.match(r"^https?://", src):
return fail("更新源必须是 http(s) 地址")
try:
req = urllib.request.Request(src, headers={"User-Agent": "maipulse/" + APP_VERSION})
raw = urllib.request.urlopen(req, timeout=20).read().decode("utf-8")
remote = json.loads(raw)
except Exception as exc: # noqa: BLE001
return fail("拉取更新源失败:%s" % exc)
ver = str(remote.get("version") or "")
if not ver:
return fail("更新源里没有 version 字段")
src = (request.args.get("source") or "").strip()
def _v(text):
return tuple(int(x) for x in re.findall(r"\d+", text)[:3])
nums = [int(x) for x in re.findall(r"\d+", str(text))[:3]]
while len(nums) < 3:
nums.append(0)
return tuple(nums)
newer = _v(ver) > _v(APP_VERSION)
return ok(current=APP_VERSION, remote=ver, newer=newer,
url=remote.get("url"), notes=remote.get("notes"),
published=remote.get("published"))
if src:
if not re.match(r"^https?://", src):
return fail("更新源必须是 http(s) 地址")
try:
req = urllib.request.Request(src, headers={"User-Agent": "maipulse/" + APP_VERSION})
remote = json.loads(urllib.request.urlopen(req, timeout=20).read().decode("utf-8"))
except Exception as exc: # noqa: BLE001
return fail("拉取更新源失败:%s" % exc)
ver = str(remote.get("version") or "")
if not ver:
return fail("更新源里没有 version 字段")
return ok(current=APP_VERSION, remote=ver, newer=_v(ver) > _v(APP_VERSION),
url=remote.get("url"), notes=remote.get("notes"),
published=remote.get("published"), repo="(自填)")
api = "%s/api/v1/repos/%s/releases/latest" % (UPDATE_SITE, UPDATE_REPO)
try:
# Cloudflare 会拦非浏览器 UA(403 error code 1010),必须伪装成普通浏览器
req = urllib.request.Request(api, headers={
"User-Agent": UA_BROWSER,
"Accept": "application/json",
})
rel = json.loads(urllib.request.urlopen(req, timeout=20).read().decode("utf-8"))
except urllib.error.HTTPError as exc:
return fail("Release 接口 HTTP %s(仓库或 Release 不存在)" % exc.code)
except Exception as exc: # noqa: BLE001
return fail("连不上更新仓库:%s" % exc)
tag = str(rel.get("tag_name") or "").strip()
if not tag:
return fail("Release 没有 tag")
ver = re.sub(r"^[vV]", "", tag)
assets = rel.get("assets") or []
apk = next((a for a in assets if str(a.get("name", "")).lower().endswith(".apk")), None)
if apk:
url = apk.get("browser_download_url") or ""
size = apk.get("size") or 0
name = apk.get("name") or "maipulse.apk"
else:
url = "%s/%s/releases/download/%s/maipulse.apk" % (UPDATE_SITE, UPDATE_REPO, tag)
size = 0
name = "maipulse.apk"
return ok(current=APP_VERSION, remote=ver, newer=_v(ver) > _v(APP_VERSION),
url=url, name=name, size=size, tag=tag,
notes=(rel.get("body") or "")[:500],
published=rel.get("published_at"),
page="%s/%s/releases/tag/%s" % (UPDATE_SITE, UPDATE_REPO, tag),
repo="%s/%s" % (UPDATE_REPO.split("/")[0], UPDATE_REPO.split("/")[-1]))
# ---------------------------------------------------------------- 配置
+74 -47
View File
@@ -289,6 +289,60 @@ def _unlock_once(state: dict, ids: list, master: bool, remaster: bool,
return one
def _run_batches(result: dict, state: dict, batches: list, have: set,
master: bool, remaster: bool, who: dict,
snapshot_ok: bool = True) -> bool:
"""跑一批包,返回「是否全部成功」。失败不抛,写进 result。"""
user_id = int(state["user_id"])
snapshot = api.GetNeedUserData(user_id, token=state["token"], save=False)
total_before = result.get("done", 0)
for index, chunk in enumerate(batches, 1):
rows = build_items(chunk, have, master, remaster)
if not rows:
result["done"] = result.get("done", 0) + len(chunk)
continue
try:
session.fresh(state)
play_data, playlog = settle.prepare(state, snapshot)
settle.ensure_play_window(state)
settle.upload(state, playlog)
patch = settle.patch(state, snapshot, play_data)
patch.update({"userItemList": rows, "isNewItemList": "1" * len(rows)})
resp = api.UpsertUserAll(
user_id, state["login_id"],
login_date_time=state["login_date_time"],
auth_date_time=state["date_time"],
token=state["token"],
snapshot=snapshot,
play=play_data,
user_playlog_list=[playlog],
patch=patch,
verify_landing=False,
auto_characters=False,
place_id=who["place_id"], place_name=who["place_name"],
region_id=who["region_id"], region_name=who["region_name"],
client_id=who["client_id"],
cooldown_sec=0.0,
dry_run=False,
confirm=True,
)
code = getattr(resp, "returnCode", None)
if code not in (1, "1", None):
result["failed"] = result.get("failed", 0) + len(chunk)
result["errors"].append("第 %d 包 returnCode=%s" % (index, code))
continue
result["done"] = result.get("done", 0) + len(chunk)
result["sentItems"] = result.get("sentItems", 0) + len(rows)
for row in rows:
have.add((int(row["itemKind"]), int(row["itemId"])))
except Exception as exc: # noqa: BLE001
result["failed"] = result.get("failed", 0) + len(chunk)
result["errors"].append("第 %d 包异常:%s" % (index, exc))
if len(result.get("errors", [])) >= 5:
break
time.sleep(0.3)
return result.get("failed", 0) == 0
def unlock_all(state: dict, *, master: bool = True, remaster: bool = True,
per: int = 0, limit: int = 0, dry_run: bool = False,
progress=None) -> dict:
@@ -338,53 +392,26 @@ def unlock_all(state: dict, *, master: bool = True, remaster: bool = True,
result["sampleItems"] = len(sample)
return result
for index, chunk in enumerate(batches, 1):
rows = build_items(chunk, have, master, remaster)
if not rows:
result["done"] += len(chunk)
continue
try:
session.fresh(state)
play_data, playlog = settle.prepare(state, snapshot)
settle.ensure_play_window(state)
settle.upload(state, playlog)
patch = settle.patch(state, snapshot, play_data)
patch.update({"userItemList": rows, "isNewItemList": "1" * len(rows)})
resp = api.UpsertUserAll(
user_id, state["login_id"],
login_date_time=state["login_date_time"],
auth_date_time=state["date_time"],
token=state["token"],
snapshot=snapshot,
play=play_data,
user_playlog_list=[playlog],
patch=patch,
verify_landing=False,
auto_characters=False,
place_id=who["place_id"], place_name=who["place_name"],
region_id=who["region_id"], region_name=who["region_name"],
client_id=who["client_id"],
cooldown_sec=0.0,
dry_run=False,
confirm=True,
)
code = getattr(resp, "returnCode", None)
if code not in (1, "1", None):
result["failed"] += len(chunk)
result["errors"].append("第 %d 包 returnCode=%s" % (index, code))
else:
result["done"] += len(chunk)
result["sentItems"] += len(rows)
for row in rows:
have.add((int(row["itemKind"]), int(row["itemId"])))
except Exception as exc: # noqa: BLE001
result["failed"] += len(chunk)
result["errors"].append("第 %d 包异常:%s" % (index, exc))
if len(result["errors"]) >= 5:
break
if progress:
progress(index, len(batches), result["done"], result["failed"])
time.sleep(0.3)
# ★ 降级阶梯:一包发失败(异常 / returnCode 非 1 / 一行都没落库)就自动换更小的包
# 重来。大包最省时间,但部分服务端会静默丢弃超大的 upsertUserAll。
retry_sizes = [int(per) if int(per) > 0 else 0, 200, 50]
attempt = 0
while True:
ok = _run_batches(result, state, batches, have, master, remaster, who, snapshot_ok=True)
if ok or attempt + 1 >= len(retry_sizes):
break
attempt += 1
# 只把还没落库的重新切包,已落库的跳过
remaining = [m for m in targets if (5, m) not in have]
size = retry_sizes[attempt]
if not remaining or size <= 0:
break
batches = chunked(remaining, size)
result["failed"] = 0
result["errors"].append(
"第 %d 次降级:改用每包 %d 首重试(剩余 %d 首)" % (attempt, size, len(remaining)))
result["retries"] = attempt
result.pop("landed", None)
have5 = owned(user_id, 5)
result["ownedMusic"] = len(have5)
Binary file not shown.

After

Width:  |  Height:  |  Size: 35 KiB

+38 -61
View File
@@ -500,7 +500,7 @@
<label><input id="mRemaster" type="checkbox" checked> Re:MASTER</label>
</div>
<div class="row">
<div><label>每包曲目数</label><input id="songPer" type="number" value="4" min="1" max="8"></div>
<div><label>每包曲目数(0=一次发完)</label><input id="songPer" type="number" value="0" min="0" max="2000"></div>
<div><label>全量试跑首数(0=全部)</label><input id="songLimit" type="number" value="0" min="0"></div>
</div>
<div class="check">
@@ -657,6 +657,12 @@
<button class="ghost" style="margin:0" onclick="updateMaster()">更新曲库</button>
<button class="ghost" style="margin:0;flex:0 0 auto" onclick="loadVersion()">刷新</button>
</div>
<div class="kv"><b>最新版本</b><span id="verRemote">—</span></div>
<div class="row">
<button class="ghost" style="margin:0" onclick="checkUpdate()">检查更新</button>
<button class="ghost" style="margin:0;flex:0 0 auto;display:none" id="verDl" onclick="downloadUpdate()">下载安装</button>
</div>
<p class="desc" id="verNote" style="display:none"></p>
</section>
@@ -1284,6 +1290,31 @@ function importConf(file){
/* ---------- 更新 ---------- *//* ---------- 更新 ---------- */
const UP_SRC_KEY = 'maipulse.updatesrc';
var _upd = null;
async function checkUpdate(){
var box = document.getElementById('verRemote');
box.textContent = '检查中…';
try{
var r = await (await fetch('/api/update-check')).json();
if(!r.ok){ box.textContent = '检查失败'; log('检查更新:' + r.error, 1); return; }
_upd = r;
box.textContent = r.remote + (r.newer ? '(有新版)' : '(已是最新)');
var note = document.getElementById('verNote');
if(r.notes){ note.style.display = ''; note.textContent = r.notes; }
document.getElementById('verDl').style.display = r.newer ? '' : 'none';
}catch(e){ box.textContent = '检查失败'; log('检查更新:' + e, 1); }
}
function downloadUpdate(){
if(!_upd || !_upd.url){ log('没有可下载的地址', 1); return; }
// 安卓壳里有原生下载器,没有就新开标签
if(window.Maipulse && window.Maipulse.installApk){
window.Maipulse.installApk(_upd.url);
}else{
window.open(_upd.url, '_blank');
}
}
async function loadVersion(){
const r = await get('/api/version');
if(!r.ok) return;
@@ -1303,61 +1334,6 @@ async function updateMaster(){
ticker('曲库已更新:' + r.new + ' 首' + (r.added > 0 ? '(新增 ' + r.added + ')' : ''), 'ok');
}
const GH_KEY = 'maipulse.ghrepo';
const GHT_KEY = 'maipulse.ghtoken';
async function checkGh(){
const repo = ($('ghRepo').value || '').trim();
const token = ($('ghToken').value || '').trim();
if(!repo) return ticker('先填 owner/name', 'err');
try{ localStorage.setItem(GH_KEY, repo); if(token) localStorage.setItem(GHT_KEY, token); }catch(e){}
const out = $('upOut');
out.textContent = '正在问 GitHub…';
let q = '/api/update-gh?repo=' + encodeURIComponent(repo);
if(token) q += '&token=' + encodeURIComponent(token);
const r = await get(q);
if(!r.ok){ out.textContent = r.error; return; }
if(r.newer){
let act = '';
if(r.url){
if(window.Maipulse && window.Maipulse.installApk){
act = '<br><button class="ghost" onclick="Maipulse.installApk(\'' + r.url + '\')">立即下载并安装</button>';
}else{
act = '<br><a href="' + r.url + '" target="_blank">点这里下载</a>';
}
}
out.innerHTML = '<b style="color:#c0392b">GitHub 上有新版本 ' + r.remote + '</b>(当前 ' + r.current + ')'
+ (r.notes ? '<br>' + r.notes.slice(0, 300) : '') + act;
}else{
out.innerHTML = 'GitHub 上已经是最新的(' + r.current + ',远端 ' + r.remote + ')';
}
}
async function checkUpdate(){
const src = ($('upSrc').value || '').trim();
if(!src) return ticker('先填更新源地址', 'err');
try{ localStorage.setItem(UP_SRC_KEY, src); }catch(e){}
const out = $('upOut');
out.textContent = '正在检查…';
const r = await get('/api/update-check?source=' + encodeURIComponent(src));
if(!r.ok){ out.textContent = r.error; return; }
if(r.newer){
let act = '';
if(r.url){
// Android 壳里直接走系统下载器 + 安装器;网页版给个普通链接
if(window.Maipulse && window.Maipulse.installApk){
act = '<br><button class="ghost" onclick="Maipulse.installApk(\'' + r.url + '\')">立即下载并安装</button>';
}else{
act = '<br><a href="' + r.url + '" target="_blank">点这里下载</a>';
}
}
out.innerHTML = '<b style="color:#c0392b">有新版本 ' + r.remote + '</b>(当前 ' + r.current + ')'
+ (r.notes ? '<br>' + r.notes : '') + act;
}else{
out.innerHTML = '已经是最新的(' + r.current + ',远端 ' + r.remote + ')';
}
}
/* ---------- 旅行伙伴 ---------- */
let charPicked = [];
async function loadCharacters(){
@@ -1883,17 +1859,19 @@ async function doUnlock(){
const body = {master:$('mMaster').checked, remaster:$('mRemaster').checked, dryRun:dry};
if(all){
body.all = true;
body.per = parseInt($('songPer').value || '4', 10);
body.per = parseInt($('songPer').value || '0', 10);
body.limit = parseInt($('songLimit').value || '0', 10);
log(`全量解锁(每包 ${body.per} 首,试跑 ${body.limit||'全部'})${dry?',预检':''}`);
log(`全量解锁(${body.per>0?'每包 '+body.per+' 首':'一次发完'},试跑 ${body.limit||'全部'})${dry?',预检':''}`);
}else{
body.ids = ids;
log(`选中解锁 ${ids.length} 首${dry?',预检':''}`);
}
const r = await postLong('/api/unlock', body, all ? '正在批量解锁歌曲'
: '正在解锁歌曲',
all ? `每包 ${body.per} 首、每包都要等满 60 秒,包数多时请耐心等。` : WAIT_SUB,
all ? 60 : 60);
all ? (body.per>0
? `每包 ${body.per} 首,每包都要登出重登换一局,包数多会慢。`
: '一次把整个曲库的解锁道具发上去,只需一局,通常十几秒。') : WAIT_SUB,
all ? 300 : 60);
if(!r.ok){ log('失败:'+r.error, 'err'); if(r.trace) log(r.trace,'dim'); return; }
if(all){
log(`完成 ${r.done} · 失败 ${r.failed} · 共 ${r.targets} 首 / ${r.batches} 包 · 现有 ${r.ownedMusic}`,
@@ -1939,7 +1917,6 @@ log('就绪。先扫码 → 登录 → 选功能。', 'dim');
applyLook();
loadVersion();
confSummary();
try{ const _u = localStorage.getItem(UP_SRC_KEY); if(_u && $('upSrc')) $('upSrc').value = _u; const _g = localStorage.getItem(GH_KEY); if(_g && $('ghRepo')) $('ghRepo').value = _g; const _t = localStorage.getItem(GHT_KEY); if(_t && $('ghToken')) $('ghToken').value = _t; }catch(e){}
loadAccounts();
// URL 带 ?uid= 时自动查这个账号的 B50 与推分(方便直接看别人的分)
const uidParam = new URLSearchParams(location.search).get('uid');
+13
View File
@@ -0,0 +1,13 @@
{
"aesIV": "",
"aesKey": "",
"apiVersion": "",
"baseUrl": "",
"clientId": "",
"keychip": "",
"obfuscate": "",
"placeId": "",
"placeName": "",
"regionId": "",
"regionName": ""
}
+284
View File
@@ -0,0 +1,284 @@
# -*- coding: utf-8 -*-
"""发版:改版本号 -> 打包 -> 推仓库 -> 新开一个 Release -> 上传 APK。
★ 每次发版都新建 Release,绝不覆盖旧的那个:
旧版本留着,别人手上的旧包点「检查更新」还能按 tag 找到自己的历史版本,
也能回滚;覆盖 asset 会让同一个 tag 指向不同内容,hash 对不上,排查时非常难缠。
用法:
python tools/release.py 1.0.1 "修了什么"
python tools/release.py 1.0.1 "修了什么" --dry 只打印计划
python tools/release.py 1.0.1 --no-build 复用桌面上已有的 APK
token 从环境变量 GITEA_TOKEN 读,读不到再看 tools/.release-token(该文件不入库)。
"""
from __future__ import annotations
import io
import json
import os
import re
import subprocess
import sys
import urllib.request
APP_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
APP_PY = os.path.join(APP_DIR, "app.py")
REPO_DIR = r"D:\DSH\maimai\_repo"
APK = r"C:\Users\20558\Desktop\maipulse-debug.apk"
BUILD_SCRIPT = os.path.join(APP_DIR, "tools", "build_apk.ps1")
SITE = "https://git.fragrance.moe"
OWNER = "shouko"
REPO = "maipulse"
# Cloudflare 会拦非浏览器 UA
UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/120.0 Safari/537.36")
# 要同步进仓库的源码面(相对 APP_DIR)。只列「改了必须让用户拿到」的,
# 探针脚本、dump 工具、快照数据一律不进库。
SYNC_DIRS = ["core", "html", "tools", "data"]
SYNC_FILES = ["app.py", "PLAN.md", "PLAN-APP.md", "PLAN-EXT.md"]
# 数据目录里只带运行必需的资产,其余(用户数据 / 快照 / 缓存)排除
DATA_KEEP = ["master.json", "master", "icons"]
SKIP_NAMES = {"__pycache__", ".release-token", "accounts.json", "shops.json",
"shop_current.json", "config.filled.json", "covers", "_backup",
"logs", "data"}
SKIP_SUFFIX = (".pyc", ".pyo")
def _force_copy(src, dst):
"""覆盖式复制:仓库里旧文件可能带只读属性(html/assets 下就有只读的 mp3/png),
shutil.copy2 会直接 PermissionError。先摘掉只读再写。"""
import shutil
import stat
if os.path.exists(dst):
try:
os.chmod(dst, stat.S_IWRITE | stat.S_IREAD)
except OSError:
pass
shutil.copy2(src, dst)
def sync_source(verbose=True):
"""把源码面同步进 REPO_DIR/maipulse。返回同步的文件数。"""
import shutil
dst_root = os.path.join(REPO_DIR, "maipulse")
os.makedirs(dst_root, exist_ok=True)
count = 0
for name in SYNC_FILES:
src = os.path.join(APP_DIR, name)
if os.path.isfile(src):
_force_copy(src, os.path.join(dst_root, name))
count += 1
for d in ("core", "html", "tools"):
src_dir = os.path.join(APP_DIR, d)
if not os.path.isdir(src_dir):
continue
dst_dir = os.path.join(dst_root, d)
os.makedirs(dst_dir, exist_ok=True)
for root, dirs, files in os.walk(src_dir):
dirs[:] = [x for x in dirs if x not in SKIP_NAMES]
rel = os.path.relpath(root, src_dir)
target = dst_dir if rel == "." else os.path.join(dst_dir, rel)
os.makedirs(target, exist_ok=True)
for f in files:
if f in SKIP_NAMES or f.endswith(SKIP_SUFFIX):
continue
# tools 里的探针/dump 脚本不进库,只带运行与发版必需的
if d == "tools" and f not in {
"build_apk.ps1", "blank_secrets.py", "apply_update.py",
"fetch_df.py", "dump_master.py", "release.py", "restart.ps1"}:
continue
_force_copy(os.path.join(root, f), os.path.join(target, f))
count += 1
# data:只带曲库 / 图标这些运行资产
data_src = os.path.join(APP_DIR, "data")
data_dst = os.path.join(dst_root, "data")
for name in DATA_KEEP:
s_item = os.path.join(data_src, name)
if os.path.isfile(s_item):
os.makedirs(data_dst, exist_ok=True)
_force_copy(s_item, os.path.join(data_dst, name))
count += 1
elif os.path.isdir(s_item):
for root, dirs, files in os.walk(s_item):
dirs[:] = [x for x in dirs if x not in SKIP_NAMES]
rel = os.path.relpath(root, data_src)
target = os.path.join(data_dst, rel)
os.makedirs(target, exist_ok=True)
for f in files:
if f in SKIP_NAMES or f.endswith(SKIP_SUFFIX):
continue
_force_copy(os.path.join(root, f), os.path.join(target, f))
count += 1
if verbose:
print(" 源码同步 %d 个文件 -> %s" % (count, dst_root))
return count
def token() -> str:
t = (os.environ.get("GITEA_TOKEN") or "").strip()
if t:
return t
p = os.path.join(APP_DIR, "tools", ".release-token")
if os.path.isfile(p):
return io.open(p, encoding="utf-8").read().strip()
raise SystemExit("没有 token:设置环境变量 GITEA_TOKEN,或写 tools/.release-token")
def api(path: str, method: str = "GET", body=None, raw: bytes = None, ctype: str = "application/json"):
url = "%s/api/v1/repos/%s/%s%s" % (SITE, OWNER, REPO, path)
data = raw if raw is not None else (json.dumps(body, ensure_ascii=False).encode("utf-8") if body else None)
headers = {"Authorization": "token " + token(), "User-Agent": UA, "Accept": "application/json"}
if data is not None:
headers["Content-Type"] = ctype
req = urllib.request.Request(url, data=data, method=method, headers=headers)
try:
resp = urllib.request.urlopen(req, timeout=60)
txt = resp.read().decode("utf-8")
return json.loads(txt) if txt.strip() else {}
except urllib.error.HTTPError as exc:
raise SystemExit("%s %s -> HTTP %s %s" % (method, path, exc.code, exc.read()[:300]))
def tag_exists(tag: str) -> bool:
"""只问存在性,不抛异常。"""
url = "%s/api/v1/repos/%s/%s/releases/tags/%s" % (SITE, OWNER, REPO, tag)
req = urllib.request.Request(url, headers={
"Authorization": "token " + token(), "User-Agent": UA, "Accept": "application/json"})
try:
urllib.request.urlopen(req, timeout=20)
return True
except urllib.error.HTTPError as exc:
return exc.code != 404
except Exception:
return False
def read_version() -> str:
s = io.open(APP_PY, encoding="utf-8").read()
m = re.search(r'^APP_VERSION\s*=\s*"([^"]+)"', s, re.M)
return m.group(1) if m else "0.0.0"
def set_version(ver: str, build: str):
s = io.open(APP_PY, encoding="utf-8").read()
s = re.sub(r'^APP_VERSION\s*=\s*"[^"]*"', 'APP_VERSION = "%s"' % ver, s, flags=re.M)
s = re.sub(r'^APP_BUILD\s*=\s*"[^"]*"', 'APP_BUILD = "%s"' % build, s, flags=re.M)
io.open(APP_PY, "w", encoding="utf-8", newline="\n").write(s)
def bump(ver: str) -> str:
parts = [int(x) for x in re.findall(r"\d+", ver)[:3]]
while len(parts) < 3:
parts.append(0)
parts[2] += 1
return "%d.%d.%d" % tuple(parts)
def run(cmd, cwd=None, check=True):
print(" $", " ".join(cmd) if isinstance(cmd, list) else cmd)
r = subprocess.run(cmd, cwd=cwd, shell=isinstance(cmd, str),
capture_output=True, text=True, encoding="utf-8", errors="replace")
if check and r.returncode != 0:
raise SystemExit("命令失败:%s\n%s\n%s" % (cmd, r.stdout[-800:], r.stderr[-800:]))
return r
def main():
args = [a for a in sys.argv[1:] if not a.startswith("--")]
dry = "--dry" in sys.argv
no_build = "--no-build" in sys.argv
cur = read_version()
new_ver = args[0].lstrip("vV") if args else bump(cur)
notes = args[1] if len(args) > 1 else ""
import time
build = time.strftime("%Y-%m-%d")
print("当前版本:%s" % cur)
print("新版版本:%s" % new_ver)
print("构建日期:%s" % build)
print("更新说明:%s" % (notes or "(空)"))
# 已有同名 tag 就直接停,绝不复用(404 = 不存在,正常继续)
if tag_exists("v" + new_ver):
raise SystemExit("tag v%s 已存在 —— 每次发版要用新版本号,不能覆盖旧的" % new_ver)
if dry:
print("\n--dry:不写文件、不打包、不推送")
return 0
# 1) 改版本号
print("\n[1/5] 写版本号")
set_version(new_ver, build)
print(" app.py -> %s / %s" % (new_ver, build))
# 2) 打包
if no_build:
print("\n[2/5] 跳过打包,用现有 APK")
else:
print("\n[2/5] 打包")
run(["powershell", "-NoProfile", "-ExecutionPolicy", "Bypass", "-File", BUILD_SCRIPT])
if not os.path.isfile(APK):
raise SystemExit("没有 APK:%s" % APK)
size = os.path.getsize(APK)
print(" %s %.2f MB" % (APK, size / 1048576))
# 3) 同步进仓库并推送
print("\n[3/5] 推仓库")
sync_source()
run(["git", "add", "-A"], cwd=REPO_DIR)
run(["git", "-c", "user.name=shouko", "-c", "[email protected]",
"commit", "-m", "release: v%s\n\n%s" % (new_ver, notes or "-")], cwd=REPO_DIR, check=False)
run(["git", "push", "origin", "main"], cwd=REPO_DIR)
# 4) 新开 Release(不是复用)
print("\n[4/5] 新建 Release v%s" % new_ver)
rel = api("/releases", "POST", {
"tag_name": "v" + new_ver,
"target_commitish": "main",
"name": "v" + new_ver,
"body": (notes or "更新到 v%s" % new_ver) + "\n\n构建日期 " + build,
"draft": False,
"prerelease": False,
})
print(" id=%s %s" % (rel.get("id"), rel.get("html_url")))
# 5) 上传 APK 到这个新 Release
print("\n[5/5] 上传 APK")
boundary = "----maipulse" + os.urandom(8).hex()
with open(APK, "rb") as fp:
blob = fp.read()
body_parts = []
body_parts.append(("--%s\r\nContent-Disposition: form-data; name=\"attachment\"; "
"filename=\"maipulse.apk\"\r\nContent-Type: application/vnd.android.package-archive\r\n\r\n"
% boundary).encode("utf-8"))
body_parts.append(blob)
body_parts.append(("\r\n--%s--\r\n" % boundary).encode("utf-8"))
payload = b"".join(body_parts)
url = "%s/api/v1/repos/%s/%s/releases/%s/assets?name=maipulse.apk" % (SITE, OWNER, REPO, rel.get("id"))
req = urllib.request.Request(url, data=payload, method="POST", headers={
"Authorization": "token " + token(), "User-Agent": UA,
"Content-Type": "multipart/form-data; boundary=" + boundary,
})
asset = json.loads(urllib.request.urlopen(req, timeout=600).read().decode("utf-8"))
print(" %s %d B" % (asset.get("name"), asset.get("size")))
print("\n发版完成")
print(" Release : %s" % rel.get("html_url"))
print(" 下载 : %s/%s/%s/releases/download/v%s/maipulse.apk" % (SITE, OWNER, REPO, new_ver))
print(" 旧版本 : 都保留着,没有被覆盖")
return 0
if __name__ == "__main__":
sys.exit(main())