shouko
8421898626
- Python 后端 + 单页前端 + Android(Chaquopy) 壳 - 曲库 1343 首 / 称号 2800 个 - 机台身份与店铺信息已抹除，字段留空 - 打包脚本含 16KB 页对齐与自动脱敏
1153 lines
42 KiB
Python
1153 lines
42 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""maipulse —— 舞萌DX 国服 发包工具（本地 Web 版）
|
||
|
||
六项功能：
|
||
1. 发票（2/3/5 倍功能票，走 UpsertUserChargelogApi）
|
||
2. 解禁小黑屋（账号卡登录 -> 登录后 Quit 登出释放会话）
|
||
3. 解锁伙伴（userItemList itemKind=10）
|
||
4. 发礼物（itemKind=4）
|
||
5. 发收藏品（itemKind=1/2/3/9/11）
|
||
6. 解锁歌曲（itemKind=5/6/7，一键或选中）
|
||
|
||
协议实现来自 Astral（D:\\DSH\\maimai\\Astral），本文件只做 HTTP 封装与编排。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
import threading
|
||
import time
|
||
import traceback
|
||
|
||
APP_DIR = os.path.dirname(os.path.abspath(__file__))
|
||
LOG_DIR = os.path.join(APP_DIR, "logs")
|
||
os.makedirs(LOG_DIR, exist_ok=True)
|
||
LOG_PATH = os.path.join(LOG_DIR, "app.log")
|
||
ASTRAL = os.environ.get("ASTRAL_DIR", r"D:\DSH\maimai\Astral")
|
||
for path in (APP_DIR, ASTRAL):
|
||
if path not in sys.path:
|
||
sys.path.insert(0, path)
|
||
if os.path.isdir(ASTRAL):
|
||
os.chdir(ASTRAL)
|
||
|
||
from flask import Flask, jsonify, request, send_file # noqa: E402
|
||
|
||
import api # noqa: E402
|
||
from core import accounts, charge, items, repair, session, shop, unlock, b50, profile # noqa: E402
|
||
from core import characters, mileage, history, recommend as recommend_mod # noqa: E402
|
||
app = Flask(__name__, static_folder=None)
|
||
|
||
LOCK = threading.Lock()
|
||
S = {
|
||
"user_id": 0,
|
||
"token": "",
|
||
"login_id": 0,
|
||
"login_date_time": 0,
|
||
"date_time": 0,
|
||
"name": "",
|
||
"rating": 0,
|
||
"ban_state": 0,
|
||
"is_login": False,
|
||
"logged_in": False,
|
||
}
|
||
|
||
|
||
def ok(**kw):
|
||
data = {"ok": True}
|
||
data.update(kw)
|
||
return jsonify(data)
|
||
|
||
|
||
def fail(msg, **kw):
|
||
data = {"ok": False, "error": str(msg)}
|
||
data.update(kw)
|
||
return jsonify(data)
|
||
|
||
|
||
def guarded(fn):
|
||
"""所有写操作都串行 + 统一异常格式。"""
|
||
def wrapper(*args, **kwargs):
|
||
if not LOCK.acquire(timeout=600):
|
||
return fail("上一个操作还没结束")
|
||
try:
|
||
return fn(*args, **kwargs)
|
||
except Exception as exc: # noqa: BLE001
|
||
return fail(str(exc), trace=traceback.format_exc()[-900:])
|
||
finally:
|
||
LOCK.release()
|
||
wrapper.__name__ = fn.__name__
|
||
return wrapper
|
||
|
||
|
||
def need_login():
|
||
if not S["logged_in"] or not S["user_id"]:
|
||
raise RuntimeError("还没有登录，先粘二维码并登录")
|
||
return S
|
||
|
||
|
||
@app.after_request
|
||
def _log_api(resp):
|
||
"""把每个 API 的返回记到 logs/app.log，方便排查（二维码请求不记 body）。"""
|
||
if request.path.startswith("/api/"):
|
||
try:
|
||
body = "" if request.path == "/api/qr" else resp.get_data(as_text=True)[:600]
|
||
except Exception: # noqa: BLE001
|
||
body = ""
|
||
try:
|
||
with open(LOG_PATH, "a", encoding="utf-8") as fp:
|
||
fp.write("%s %s %s -> %s %s\n" % (
|
||
time.strftime("%Y-%m-%d %H:%M:%S"), request.method,
|
||
request.path, resp.status_code, body.replace("\n", " ")))
|
||
except OSError:
|
||
pass
|
||
return resp
|
||
|
||
|
||
# ---------------------------------------------------------------- 会话
|
||
|
||
@app.post("/api/qr")
|
||
def api_qr():
|
||
body = request.get_json(silent=True) or {}
|
||
try:
|
||
# chipID 从配置文件现读，绕开模块级快照
|
||
uid, token = session.parse_qr(body.get("qr", ""), chip_id=_read_conf().get("keychip", ""))
|
||
return ok(userId=uid, token=token)
|
||
except Exception as exc: # noqa: BLE001
|
||
return fail(str(exc))
|
||
|
||
|
||
@app.post("/api/login")
|
||
@guarded
|
||
def api_login():
|
||
body = request.get_json(silent=True) or {}
|
||
uid = int(body.get("userId") or S.get("user_id") or 0)
|
||
token = (body.get("token") or S.get("token") or "").strip()
|
||
if not uid or not token:
|
||
return fail("缺少 userId / token，先解析二维码")
|
||
info = session.login(S, uid, token, is_continue=bool(body.get("cont")))
|
||
# 扫码换来的 token 存本机，下次不用再扫（机台重新扫码就会顶掉旧的）
|
||
try:
|
||
accounts.upsert(uid, token, name=info.get("name") or "")
|
||
except Exception: # noqa: BLE001 存不下不影响登录
|
||
pass
|
||
return ok(**info, loginId=S["login_id"], loginDateTime=S["login_date_time"])
|
||
|
||
|
||
@app.get("/api/accounts")
|
||
@guarded
|
||
def api_accounts():
|
||
"""本机保存过的账号（只有打码 id / 昵称 / 时间，token 不外传）。"""
|
||
try:
|
||
return ok(accounts=accounts.list_accounts(S.get("user_id")))
|
||
except Exception as exc: # noqa: BLE001
|
||
return fail(str(exc))
|
||
|
||
|
||
@app.post("/api/accounts")
|
||
@guarded
|
||
def api_account_action():
|
||
"""action=switch 用存的 token 重新登录 / rename 改名 / delete 删掉。"""
|
||
body = request.get_json(silent=True) or {}
|
||
action = str(body.get("action") or "switch").lower()
|
||
row = accounts.get_by_ref(body.get("ref"))
|
||
if row is None:
|
||
return fail("没有保存这个账号，先扫码登录一次")
|
||
uid = row["userId"]
|
||
|
||
if action == "delete":
|
||
accounts.remove(uid)
|
||
return ok(accounts=accounts.list_accounts(S.get("user_id")))
|
||
|
||
if action == "rename":
|
||
accounts.rename(uid, body.get("name") or "")
|
||
return ok(accounts=accounts.list_accounts(S.get("user_id")))
|
||
|
||
# switch
|
||
# 当前若还挂着别的账号，先登出，否则服务端会用 returnCode=100 回绝
|
||
try:
|
||
if S.get("logged_in") and int(S.get("user_id") or 0) != uid:
|
||
session.logout(S)
|
||
except Exception: # noqa: BLE001
|
||
S["logged_in"] = False
|
||
try:
|
||
info = session.login(S, uid, row["token"])
|
||
except Exception as exc: # noqa: BLE001
|
||
return fail("这个账号的凭证用不了了（多半是在机台上重新扫过码），"
|
||
"重新扫码登录一次就会刷新：" + str(exc))
|
||
accounts.touch(uid)
|
||
return ok(**info, loginId=S["login_id"], loginDateTime=S["login_date_time"])
|
||
|
||
|
||
def _friendly_music_error(exc) -> str:
|
||
"""把 Astral 那种一屏长的错误压成一句能照着做的提示。
|
||
|
||
★ 实测：GetUserMusicApi 与 GetUserRating 不一样 —— 后者免登录就能查，
|
||
前者必须挂在真实登录会话上，否则服务端回一个"空明文"（HTTP 200 但解密后为空）。
|
||
"""
|
||
msg = str(exc)
|
||
if "返回空明文" in msg:
|
||
return ("全曲成绩要登录才能拉：GetUserMusicApi 不接受免登录查询。"
|
||
"先扫码登录，再点一次。")
|
||
if "还没有登录" in msg:
|
||
return "还没有登录，先扫码登录"
|
||
return msg
|
||
|
||
|
||
@app.get("/api/history")
|
||
@guarded
|
||
def api_history():
|
||
"""全曲成绩（只读）。默认只回前 limit 条，full=1 要全部。"""
|
||
need_login()
|
||
try:
|
||
data = history.fetch(S)
|
||
limit = max(1, min(3000, int(request.args.get("limit") or 120)))
|
||
out = dict(data)
|
||
# ?level=3,4 只看某些难度（3=Master 4=Re:Master 2=Expert 1=Advanced 0=Basic）
|
||
level_arg = (request.args.get("level") or "").strip()
|
||
items = data["items"]
|
||
if level_arg and level_arg != "all":
|
||
wanted = {int(x) for x in level_arg.split(",") if x.strip().isdigit()}
|
||
if wanted:
|
||
items = [r for r in items if r.get("level") in wanted]
|
||
out["levelFilter"] = sorted(wanted)
|
||
if not request.args.get("full"):
|
||
out["items"] = items[:limit]
|
||
out["truncated"] = len(items) > limit
|
||
else:
|
||
out["items"] = items
|
||
out["shownTotal"] = len(items)
|
||
return ok(**out)
|
||
except Exception as exc: # noqa: BLE001
|
||
return fail(_friendly_music_error(exc))
|
||
|
||
|
||
@app.get("/api/recommend")
|
||
@guarded
|
||
def api_recommend():
|
||
"""推分建议：拿全曲成绩 + 当前 B35 算「再打哪首、到多少、能涨几分」。
|
||
|
||
★ 降级逻辑：全曲成绩要登录才能拉（GetUserMusicApi 不认免登录查询），
|
||
但 B50 的 50 条本身也是「我打过的曲子」，拿来当候选池照样能算推分 ——
|
||
只是池子小一圈（50 条 vs 几百条），结果依然成立。这时返回 degraded=true。
|
||
"""
|
||
uid = 0
|
||
try:
|
||
uid = int(request.args.get("userId") or 0)
|
||
except (TypeError, ValueError):
|
||
uid = 0
|
||
if not uid and S.get("logged_in"):
|
||
uid = int(S.get("user_id") or 0)
|
||
if not uid:
|
||
return fail("填账号 ID 再点，或者先扫码登录")
|
||
|
||
try:
|
||
board = b50.fetch(uid)
|
||
hist = None
|
||
if S.get("logged_in") and int(S.get("user_id") or 0) == uid:
|
||
try:
|
||
hist = history.fetch(S)
|
||
except Exception: # noqa: BLE001 登录态在但拉不到成绩，退回 B50 那 50 条
|
||
hist = None
|
||
degraded = hist is None
|
||
if degraded:
|
||
hist = {"items": list(board["b35"]["items"]) + list(board["b15"]["items"])}
|
||
result = recommend_mod.suggest(hist, board["b35"])
|
||
result["degraded"] = degraded
|
||
result["songCount"] = int(hist.get("total") or len(hist["items"]))
|
||
result["unrated"] = int(hist.get("unrated")
|
||
if hist.get("unrated") is not None
|
||
else sum(1 for r in hist["items"] if r.get("unrated")))
|
||
result["totalRating"] = board.get("totalRating")
|
||
return ok(**result)
|
||
except Exception as exc: # noqa: BLE001
|
||
return fail(_friendly_music_error(exc))
|
||
|
||
|
||
@app.get("/api/selftest")
|
||
@guarded
|
||
def api_selftest():
|
||
"""全曲成绩一键自检：登录后点一下就知道这条链路通不通、分对不对。
|
||
|
||
做四件事：
|
||
1. 拉全量成绩（GetUserMusicApi，分页）
|
||
2. 拉 B50（GetUserRating）
|
||
3. 对比 全曲里的最高分 与 B50 里的分（应当一致）
|
||
4. 对比 本地算的 B50 合计 与 客户端 rating（应当一致）
|
||
"""
|
||
need_login()
|
||
checks = []
|
||
try:
|
||
hist = history.fetch(S, use_cache=False)
|
||
except Exception as exc: # noqa: BLE001
|
||
return fail(_friendly_music_error(exc))
|
||
|
||
checks.append({"name": "全曲成绩条数", "ok": hist["total"] > 0,
|
||
"detail": "%d 条（每曲每个难度一条），涉及 %d 首曲子"
|
||
% (hist["total"], len({r["musicId"] for r in hist["items"]}))})
|
||
checks.append({"name": "定数未收录", "ok": hist["unrated"] == 0,
|
||
"detail": ("全部收录" if not hist["unrated"]
|
||
else "%d 条查不到定数（ra 会是 0）" % hist["unrated"])})
|
||
checks.append({"name": "达成率非零", "ok": hist["avgAchievement"] > 0,
|
||
"detail": "平均 %s%%，最高 ra %s"
|
||
% (round(hist["avgAchievement"], 4), hist["bestRa"])})
|
||
|
||
board = b50.fetch(int(S["user_id"]))
|
||
prof = profile.fetch(S)
|
||
client = int(prof.get("playerRating") or 0)
|
||
local = int(board.get("totalRating") or 0)
|
||
checks.append({"name": "B35+B15 合计", "ok": client == local,
|
||
"detail": "本地 %d / 客户端 %d%s"
|
||
% (local, client,
|
||
"" if client == local
|
||
else "（差 %d）" % (client - local))})
|
||
# 全曲里的最高分应该跟 B50 里的第一名对得上
|
||
# ★ 宴/協/息谱不参与 B35 排名（真机实测：協ラグトレイン 175% 算到 ra 292，
|
||
# 比 B50 第一名还高，但服务端没把它算进 ratingList），比较时要排除
|
||
normal = [r for r in hist["items"] if int(r.get("level") or 0) < 5]
|
||
top_hist = normal[0] if normal else {}
|
||
top_b35 = (board["b35"]["items"] or [{}])[0]
|
||
checks.append({"name": "全曲最高分 vs B50 第一",
|
||
"ok": bool(top_hist) and top_hist.get("ra") == top_b35.get("ra"),
|
||
"detail": "%s（%s %s）ra %s / B50 %s ra %s"
|
||
% (top_hist.get("name") or "?", top_hist.get("levelName"),
|
||
top_hist.get("ds"), top_hist.get("ra"),
|
||
top_b35.get("name") or "?", top_b35.get("ra"))})
|
||
return ok(checks=checks,
|
||
passed=sum(1 for c in checks if c["ok"]),
|
||
total=len(checks),
|
||
songCount=hist["total"],
|
||
ratingLocal=local,
|
||
ratingClient=client)
|
||
|
||
|
||
@app.get("/api/verify")
|
||
@guarded
|
||
def api_verify():
|
||
"""对分：客户端显示的 rating vs 服务端 GetUserRating vs 本地按定数算的合计。
|
||
|
||
本地偏低通常就是「定数未收录」那几条 —— 补完再看这里应该对平。
|
||
"""
|
||
need_login()
|
||
try:
|
||
board = b50.fetch(int(S["user_id"]))
|
||
prof = profile.fetch(S)
|
||
client = int(prof.get("playerRating") or 0)
|
||
server = int(board.get("serverRating") or 0)
|
||
local = int(board.get("totalRating") or 0)
|
||
return ok(
|
||
clientRating=client,
|
||
serverRating=server,
|
||
localRating=local,
|
||
diffClientLocal=client - local,
|
||
diffServerClient=server - client,
|
||
missing=int(board.get("missing") or 0),
|
||
b35=board["b35"]["rating"],
|
||
b15=board["b15"]["rating"],
|
||
b35Missing=board["b35"]["missing"],
|
||
b15Missing=board["b15"]["missing"],
|
||
)
|
||
except Exception as exc: # noqa: BLE001
|
||
return fail(str(exc))
|
||
|
||
|
||
@app.post("/api/ds")
|
||
@guarded
|
||
def api_ds():
|
||
"""手工补定数：{musicId: {level: ds}} —— ds<=0 表示删掉这条。"""
|
||
body = request.get_json(silent=True) or {}
|
||
rows = body.get("rows") or {}
|
||
if not isinstance(rows, dict):
|
||
return fail("rows 要是 {musicId: {level: ds}} 这种结构")
|
||
try:
|
||
saved = history.save_override(rows)
|
||
return ok(count=sum(len(v) for v in saved.values()), ds=saved)
|
||
except Exception as exc: # noqa: BLE001
|
||
return fail("写定数覆盖表失败：" + str(exc))
|
||
|
||
|
||
@app.post("/api/logout")
|
||
@guarded
|
||
def api_logout():
|
||
if not S["logged_in"]:
|
||
return fail("未登录")
|
||
resp = session.logout(S)
|
||
return ok(returnCode=getattr(resp, "returnCode", None))
|
||
|
||
|
||
@app.post("/api/shutdown")
|
||
@guarded
|
||
def api_shutdown():
|
||
"""先登出，再退出进程 —— 重启前必须走这里。
|
||
|
||
★ 直接 kill 掉进程会把服务端会话留在 isLogin=True，账号就卡小黑屋了
|
||
（扫码登录返回 returnCode=100），而且那次 loginId 只存在内存里，
|
||
之后想发 UserLogout 都拿不到正确的 loginDateTime。
|
||
"""
|
||
info = {}
|
||
try:
|
||
if S.get("logged_in"):
|
||
resp = session.logout(S)
|
||
info["logout"] = getattr(resp, "returnCode", None)
|
||
else:
|
||
info["logout"] = "未登录，跳过"
|
||
except Exception as exc: # noqa: BLE001
|
||
info["logoutError"] = str(exc)
|
||
threading.Timer(0.8, lambda: os._exit(0)).start()
|
||
return ok(**info)
|
||
|
||
|
||
@app.post("/api/reload")
|
||
@guarded
|
||
def api_reload():
|
||
"""热重载 core 下的模块，用来避免「改代码就得重启 → 丢登录态」。
|
||
|
||
★ 重启会清空内存里的 S（loginId / loginDateTime），而且如果没先登出，
|
||
服务端会话就卡在 isLogin=True。改完 core/*.py 调这个接口即可，
|
||
登录态和会话都保住。只重载 core 包内的纯逻辑模块，不碰 api / Flask。
|
||
"""
|
||
import importlib
|
||
names = ["core.settle", "core.play", "core.items", "core.unlock",
|
||
"core.charge", "core.repair", "core.session", "core.shop",
|
||
"core.b50", "core.profile", "core.characters", "core.mileage"]
|
||
done = []
|
||
for name in names:
|
||
mod = sys.modules.get(name)
|
||
if mod is None:
|
||
continue
|
||
try:
|
||
importlib.reload(mod)
|
||
done.append(name)
|
||
except Exception as exc: # noqa: BLE001
|
||
return fail("重载 %s 失败：%s" % (name, exc))
|
||
return ok(reloaded=done, logged=bool(S.get("logged_in")))
|
||
|
||
|
||
@app.get("/api/state")
|
||
def api_state():
|
||
cur = shop.current()
|
||
# 登录后 60 秒冷却：这段时间内发的 UpsertUserAll 会被服务端静默丢弃，
|
||
# 前端拿这个值画冷却条（金票 LogoutTimeLimit=60000 同口径）。
|
||
cool = 0.0
|
||
if S.get("logged_in"):
|
||
started = float(S.get("login_at") or 0)
|
||
if started > 0:
|
||
cool = max(0.0, 60.0 - (__import__("time").monotonic() - started))
|
||
return jsonify({
|
||
"ok": True,
|
||
"userId": S["user_id"],
|
||
"name": S["name"],
|
||
"rating": S["rating"],
|
||
"banState": S["ban_state"],
|
||
"isLogin": S["is_login"],
|
||
"logged": S["logged_in"],
|
||
"shop": cur,
|
||
"coolRemain": round(cool, 1),
|
||
})
|
||
|
||
|
||
# ---------------------------------------------------------------- 店铺
|
||
|
||
@app.get("/api/shop")
|
||
def api_shop_get():
|
||
return ok(shop=shop.current())
|
||
|
||
|
||
@app.get("/api/shops")
|
||
def api_shop_list():
|
||
keyword = (request.args.get("q") or "").strip().lower()
|
||
try:
|
||
table = shop.load_shops()
|
||
except Exception as exc: # noqa: BLE001
|
||
return fail("店铺表读取失败：" + str(exc))
|
||
rows = table["shops"]
|
||
if keyword:
|
||
rows = [r for r in rows
|
||
if keyword in r["name"].lower()
|
||
or keyword in str(r["place_id"])
|
||
or keyword in r["client_id"].lower()]
|
||
limit = min(int(request.args.get("limit", 60) or 60), 500)
|
||
return ok(total=len(rows), count=table["count"], items=rows[:limit])
|
||
|
||
|
||
@app.post("/api/shop")
|
||
@guarded
|
||
def api_shop_set():
|
||
body = request.get_json(silent=True) or {}
|
||
if not body.get("client_id") or not body.get("place_id"):
|
||
return fail("缺少 client_id / place_id")
|
||
saved = shop.set_current(body)
|
||
return ok(shop=saved)
|
||
|
||
|
||
# ---------------------------------------------------------------- 查询
|
||
|
||
@app.get("/api/items")
|
||
@guarded
|
||
def api_items():
|
||
need_login()
|
||
kind = int(request.args.get("kind", 12))
|
||
rows = items.fetch(S["user_id"], kind)
|
||
return ok(kind=kind, kindName=items.KIND_NAMES.get(kind, str(kind)),
|
||
count=len(rows), items=rows)
|
||
|
||
|
||
@app.get("/api/charge")
|
||
@guarded
|
||
def api_charge_list():
|
||
need_login()
|
||
rows = charge.list_charge(S["user_id"])
|
||
return ok(count=len(rows), items=rows)
|
||
|
||
|
||
@app.get("/api/music")
|
||
@guarded
|
||
def api_music():
|
||
need_login()
|
||
have = sorted(unlock.owned(S["user_id"], 5))
|
||
return ok(count=len(have), musicIds=have)
|
||
|
||
|
||
@app.get("/api/newitems")
|
||
@guarded
|
||
def api_newitems():
|
||
"""查「待领取道具」（GetUserNewItemListApi）。
|
||
|
||
服务端有时把新道具先挂进 newItem 队列，游戏里领取后才进 userItemList。
|
||
发放后 userItemList 没变化时，这个接口是判断到底写没写的关键。
|
||
"""
|
||
need_login()
|
||
from core import settle
|
||
|
||
who = session.ids()
|
||
play, playlog = settle.prepare(S, {})
|
||
ud = api.GetUserData(S["user_id"], client_id=who["client_id"])
|
||
data = ud.get("userData") if isinstance(ud, dict) else None
|
||
rows = data if isinstance(data, list) else ([data] if data else [])
|
||
resp = api.GetUserNewItemList(
|
||
S["user_id"], user_data=rows, user_playlog_list=[playlog],
|
||
client_id=who["client_id"],
|
||
)
|
||
return ok(raw=resp if isinstance(resp, (dict, list)) else str(resp))
|
||
|
||
|
||
# ---------------------------------------------------------------- 发票（功能票）
|
||
|
||
@app.post("/api/ticket")
|
||
@guarded
|
||
def api_ticket():
|
||
need_login()
|
||
body = request.get_json(silent=True) or {}
|
||
result = charge.buy(S, int(body.get("chargeId", 5)),
|
||
dry_run=bool(body.get("dryRun")),
|
||
wait=float(body.get("wait", charge.DEFAULT_WAIT)))
|
||
return ok(**result)
|
||
|
||
|
||
# ---------------------------------------------------------------- 发道具
|
||
|
||
@app.post("/api/grant")
|
||
@guarded
|
||
def api_grant():
|
||
need_login()
|
||
body = request.get_json(silent=True) or {}
|
||
kind = int(body.get("kind", 12))
|
||
ids = body.get("ids") or []
|
||
stock = int(body.get("stock", 1))
|
||
if not ids:
|
||
return fail("没有选择道具 id")
|
||
entries = [(int(i), stock) for i in ids]
|
||
result = items.grant(S, kind, entries,
|
||
dry_run=bool(body.get("dryRun")),
|
||
cooldown=float(body.get("cooldown", 0)),
|
||
wait=float(body.get("wait", 0)))
|
||
return ok(**result)
|
||
|
||
|
||
# ---------------------------------------------------------------- 旅行伙伴
|
||
|
||
@app.get("/api/characters")
|
||
@guarded
|
||
def api_characters():
|
||
"""已拥有角色 + 当前出战编组（只读）。"""
|
||
try:
|
||
return ok(**characters.read(S))
|
||
except Exception as exc: # noqa: BLE001
|
||
return fail(str(exc))
|
||
|
||
|
||
@app.post("/api/characters")
|
||
@guarded
|
||
def api_characters_write():
|
||
"""发放角色 / 改等级 / 编组（写）。默认 dry-run。"""
|
||
need_login()
|
||
body = request.get_json(silent=True) or {}
|
||
# 解析放在 try 里：前端传了非数字时返回 JSON 错误，而不是 Flask 的 500 页面
|
||
try:
|
||
add = [int(x) for x in (body.get("add") or [])]
|
||
levels = {int(k): int(v) for k, v in (body.get("levels") or {}).items()}
|
||
slots = body.get("slots")
|
||
slot_list = [int(x or 0) for x in slots] if isinstance(slots, list) else None
|
||
except (TypeError, ValueError):
|
||
return fail("角色 ID / 等级 / 槽位必须都是数字")
|
||
if not add and slot_list is None:
|
||
return fail("没有要发放的角色，也没给编组")
|
||
try:
|
||
result = characters.write(S, add=add, levels=levels,
|
||
slots=slot_list,
|
||
dry_run=bool(body.get("dryRun", True)))
|
||
return ok(**result)
|
||
except Exception as exc: # noqa: BLE001
|
||
return fail(str(exc))
|
||
|
||
|
||
# ---------------------------------------------------------------- 舞里程
|
||
|
||
@app.get("/api/mileage")
|
||
@guarded
|
||
def api_mileage():
|
||
try:
|
||
return ok(**mileage.read(S))
|
||
except Exception as exc: # noqa: BLE001
|
||
return fail(str(exc))
|
||
|
||
|
||
@app.post("/api/mileage")
|
||
@guarded
|
||
def api_mileage_write():
|
||
"""改舞里程：amount 正负皆可，mode=add 累加 / set 覆盖。默认 dry-run。"""
|
||
need_login()
|
||
body = request.get_json(silent=True) or {}
|
||
try:
|
||
amount = int(body.get("amount", 0))
|
||
except (TypeError, ValueError):
|
||
return fail("数值不合法")
|
||
try:
|
||
result = mileage.write(S, amount, mode=str(body.get("mode", "add")),
|
||
dry_run=bool(body.get("dryRun", True)))
|
||
return ok(**result)
|
||
except Exception as exc: # noqa: BLE001
|
||
return fail(str(exc))
|
||
|
||
|
||
# ---------------------------------------------------------------- 解锁歌曲
|
||
|
||
@app.post("/api/unlock")
|
||
@guarded
|
||
def api_unlock():
|
||
need_login()
|
||
body = request.get_json(silent=True) or {}
|
||
master = bool(body.get("master", True))
|
||
remaster = bool(body.get("remaster", True))
|
||
dry = bool(body.get("dryRun"))
|
||
if body.get("all"):
|
||
result = unlock.unlock_all(S, master=master, remaster=remaster,
|
||
per=int(body.get("per", 0)),
|
||
limit=int(body.get("limit", 0)),
|
||
dry_run=dry)
|
||
else:
|
||
ids = [int(x) for x in (body.get("ids") or [])]
|
||
if not ids:
|
||
return fail("没有选歌，也没勾全量")
|
||
result = unlock.unlock_selected(S, ids, master=master, remaster=remaster,
|
||
dry_run=dry)
|
||
return ok(**result)
|
||
|
||
|
||
# ---------------------------------------------------------------- 小黑屋
|
||
|
||
@app.post("/api/unstick")
|
||
@guarded
|
||
def api_unstick():
|
||
body = request.get_json(silent=True) or {}
|
||
uid = int(body.get("userId") or S.get("user_id") or 0)
|
||
token = (body.get("token") or S.get("token") or "").strip()
|
||
if not uid:
|
||
return fail("缺少 userId（解卡只需要账号 ID，token 可选）")
|
||
S.update({"user_id": uid, "token": token, "logged_in": False})
|
||
return ok(**repair.unstick(S, user_id=uid, token=token))
|
||
|
||
|
||
@app.get("/api/status")
|
||
@guarded
|
||
def api_status():
|
||
uid = int(request.args.get("userId") or S.get("user_id") or 0)
|
||
token = (request.args.get("token") or S.get("token") or "").strip()
|
||
if not uid or not token:
|
||
return fail("缺少 userId / token")
|
||
return ok(**repair.status(uid, token))
|
||
|
||
|
||
# ---------------------------------------------------------------- B50 查分（只读）
|
||
|
||
@app.get("/api/profile")
|
||
def api_profile():
|
||
"""主页资料（只读）：一次 GetNeedUserData 快照填满整页。需要已登录。"""
|
||
try:
|
||
return ok(**profile.fetch(S))
|
||
except Exception as exc: # noqa: BLE001
|
||
return fail(str(exc))
|
||
|
||
|
||
@app.get("/api/b50")
|
||
def api_b50():
|
||
"""只读：GetUserRating + 本地曲库算 ra。不需要登录，给 userId 就行。"""
|
||
uid = int(request.args.get("userId") or S.get("user_id") or 0)
|
||
if not uid:
|
||
return fail("缺少 userId（先登录，或在地址栏带 ?userId=）")
|
||
data = b50.fetch(uid)
|
||
# 昵称只能从已登录的会话里拿（GetUserPreview 需要 token）
|
||
data["name"] = S.get("name") or ""
|
||
return ok(**data)
|
||
|
||
|
||
# ---------------------------------------------------------------- 封面（同源代理，供导出图片用）
|
||
|
||
COVER_DIR = os.path.join(APP_DIR, "data", "covers")
|
||
ICON_DIR = os.path.join(APP_DIR, "data", "icons")
|
||
os.makedirs(COVER_DIR, exist_ok=True)
|
||
os.makedirs(ICON_DIR, exist_ok=True)
|
||
|
||
|
||
def _fetch_png(urls, path):
|
||
"""按候选顺序拉一张图并落盘；全失败返回错误信息。"""
|
||
import urllib.request
|
||
last = ""
|
||
for url in urls:
|
||
try:
|
||
req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
|
||
blob = urllib.request.urlopen(req, timeout=15).read()
|
||
if blob[:4] == b"\x89PNG":
|
||
with open(path, "wb") as fp:
|
||
fp.write(blob)
|
||
return None
|
||
last = "%s 返回非 PNG" % url
|
||
except Exception as exc: # noqa: BLE001
|
||
last = "%s: %s" % (url, exc)
|
||
return last
|
||
|
||
|
||
@app.get("/api/cover/<int:mid>")
|
||
def api_cover(mid):
|
||
"""曲绘代理。diving-fish 没有 CORS 头，直连会让 canvas 被污染导致导出失败。
|
||
|
||
★ 顺序照 Ichino：先 lxns（`jacket/{id-10000}`，DX 谱带 10000 偏移），
|
||
404 再回落 diving-fish（5 位补零）。国服新曲在 lxns 可能没有。
|
||
"""
|
||
path = os.path.join(COVER_DIR, "%05d.png" % mid)
|
||
if not os.path.exists(path):
|
||
real = mid - 10000 if mid > 10000 else mid
|
||
err = _fetch_png([
|
||
"https://assets2.lxns.net/maimai/jacket/%d.png" % real,
|
||
"https://www.diving-fish.com/covers/%05d.png" % mid,
|
||
], path)
|
||
if err:
|
||
return fail("封面拉取失败：%s" % err)
|
||
return send_file(path, mimetype="image/png")
|
||
|
||
|
||
@app.get("/api/icon/<int:icon_id>")
|
||
def api_icon(icon_id):
|
||
"""头像代理（lxns）。"""
|
||
path = os.path.join(ICON_DIR, "%d.png" % icon_id)
|
||
if not os.path.exists(path):
|
||
err = _fetch_png(["https://assets2.lxns.net/maimai/icon/%d.png" % icon_id], path)
|
||
if err:
|
||
return fail("头像拉取失败：%s" % err)
|
||
return send_file(path, mimetype="image/png")
|
||
|
||
|
||
@app.get("/assets/<path:name>")
|
||
def assets(name):
|
||
"""海报素材（base.png / 评级图 / FC-FS 图），从 Ichino 的 assets/b50 复制而来。"""
|
||
base = os.path.join(APP_DIR, "html", "assets")
|
||
path = os.path.abspath(os.path.join(base, name))
|
||
if not path.startswith(os.path.abspath(base)) or not os.path.exists(path):
|
||
return fail("没有这个文件")
|
||
return send_file(path)
|
||
|
||
|
||
# ---------------------------------------------------------------- 数据表
|
||
|
||
@app.get("/api/master")
|
||
def api_master():
|
||
path = os.path.join(APP_DIR, "data", "master.json")
|
||
if not os.path.exists(path):
|
||
return fail("master.json 不存在，先跑 tools/dump_master.py")
|
||
with open(path, encoding="utf-8") as fp:
|
||
data = json.load(fp)
|
||
cat = request.args.get("cat")
|
||
if cat:
|
||
if cat not in data:
|
||
return fail("没有这个分类：" + cat)
|
||
return ok(cat=cat, kind=data[cat]["kind"], count=data[cat]["count"],
|
||
items=data[cat]["items"])
|
||
return ok(cats={k: {"kind": v["kind"], "count": v["count"]} for k, v in data.items()})
|
||
|
||
|
||
# ---------------------------------------------------------------- 静态页
|
||
|
||
# ---------------------------------------------------------------- 更新
|
||
|
||
APP_VERSION = "1.0.0"
|
||
APP_BUILD = "2026-10-28"
|
||
|
||
|
||
@app.get("/api/version")
|
||
def api_version():
|
||
"""当前版本 + 曲库状态。更新页拿它和远端 version.json 比。"""
|
||
import platform as _pf
|
||
df = os.path.join(APP_DIR, "data", "master", "music_df.json")
|
||
info = {"version": APP_VERSION, "build": APP_BUILD,
|
||
"python": _pf.python_version(), "os": _pf.system()}
|
||
if os.path.exists(df):
|
||
info["dfSize"] = os.path.getsize(df)
|
||
info["dfMtime"] = time.strftime("%Y-%m-%d %H:%M:%S",
|
||
time.localtime(os.path.getmtime(df)))
|
||
try:
|
||
with open(df, encoding="utf-8") as fp:
|
||
info["dfCount"] = len(json.load(fp))
|
||
except Exception:
|
||
info["dfCount"] = -1
|
||
return ok(**info)
|
||
|
||
|
||
@app.post("/api/update-master")
|
||
@guarded
|
||
def api_update_master():
|
||
"""重新拉水鱼曲库（补新曲名与定数）。落盘后清掉 load_music 的缓存。"""
|
||
tools = os.path.join(APP_DIR, "tools")
|
||
if tools not in sys.path:
|
||
sys.path.insert(0, tools)
|
||
import fetch_df # noqa: E402
|
||
|
||
old = 0
|
||
try:
|
||
old = len(b50._df_rows() or {})
|
||
except Exception:
|
||
pass
|
||
payload = fetch_df.fetch()
|
||
if not payload:
|
||
return fail("水鱼返回了空数据")
|
||
fetch_df.save(payload)
|
||
b50._music_cache = None # 下次 load_music 重新合并
|
||
return ok(old=old, new=len(payload), added=len(payload) - old,
|
||
source=fetch_df.URL)
|
||
|
||
|
||
@app.get("/api/update-gh")
|
||
@guarded
|
||
def api_update_gh():
|
||
"""走 GitHub Release 检查更新。repo=owner/name，token 可选（私有仓库必填）。
|
||
|
||
私有仓库的 asset 下载要带 Authorization + Accept: octet-stream，
|
||
所以这里只负责「问版本」，真正的下载地址由前端交给系统下载器时带 token。
|
||
"""
|
||
import urllib.request
|
||
repo = (request.args.get("repo") or "").strip().strip("/")
|
||
token = (request.args.get("token") or "").strip()
|
||
if not re.match(r"^[\w.-]+/[\w.-]+$", repo):
|
||
return fail("仓库要写成 owner/name")
|
||
api = "https://api.github.com/repos/%s/releases/latest" % repo
|
||
headers = {"User-Agent": "maipulse/" + APP_VERSION,
|
||
"Accept": "application/vnd.github+json"}
|
||
if token:
|
||
headers["Authorization"] = "Bearer " + token
|
||
rel = {}
|
||
api_error = None
|
||
try:
|
||
req = urllib.request.Request(api, headers=headers)
|
||
raw = urllib.request.urlopen(req, timeout=20).read().decode("utf-8")
|
||
rel = json.loads(raw)
|
||
except urllib.error.HTTPError as exc:
|
||
body = ""
|
||
try:
|
||
body = exc.read().decode("utf-8")[:200]
|
||
except Exception:
|
||
pass
|
||
api_error = "HTTP %s %s" % (exc.code, body)
|
||
if exc.code == 404:
|
||
return fail("仓库或 Release 不存在（私有仓库要填 token）")
|
||
except Exception as exc: # noqa: BLE001
|
||
api_error = str(exc)
|
||
|
||
# API 被限流（免 token 每小时 60 次）时回落到网页重定向 —— 不需要 API。
|
||
tag = str(rel.get("tag_name") or rel.get("name") or "").strip()
|
||
if not tag and api_error:
|
||
try:
|
||
page = "https://github.com/%s/releases/latest" % repo
|
||
req2 = urllib.request.Request(page, headers={"User-Agent": "maipulse/" + APP_VERSION})
|
||
resp = urllib.request.urlopen(req2, timeout=20)
|
||
final = resp.geturl() or ""
|
||
if "/releases/tag/" in final:
|
||
tag = final.rsplit("/releases/tag/", 1)[-1].split("?")[0]
|
||
rel = {"tag_name": tag, "_via": "redirect"}
|
||
except Exception as exc: # noqa: BLE001
|
||
return fail("GitHub 接口不可用（%s），重定向也失败：%s" % (api_error, exc))
|
||
if not tag:
|
||
return fail("拿不到版本号：%s" % (api_error or "Release 里没有 tag"))
|
||
ver = re.sub(r"^[vV]", "", tag)
|
||
if not ver:
|
||
return fail("这个 Release 没有 tag_name")
|
||
|
||
apk_url, apk_name = None, None
|
||
for asset in rel.get("assets") or []:
|
||
name = str(asset.get("name") or "")
|
||
if name.lower().endswith(".apk"):
|
||
apk_url, apk_name = asset.get("browser_download_url"), name
|
||
break
|
||
if not apk_url:
|
||
# 约定：Release 里那个 APK 叫 maipulse.apk，这样不用查 assets 也能拼出稳定地址
|
||
apk_name = "maipulse.apk"
|
||
apk_url = "https://github.com/%s/releases/latest/download/%s" % (repo, apk_name)
|
||
|
||
def _v(text):
|
||
return tuple(int(x) for x in re.findall(r"\d+", text)[:3])
|
||
|
||
newer = _v(ver) > _v(APP_VERSION)
|
||
return ok(current=APP_VERSION, remote=ver, newer=newer,
|
||
url=apk_url, name=apk_name, notes=rel.get("body") or "",
|
||
published=rel.get("published_at"), private=bool(token),
|
||
repo=repo)
|
||
|
||
|
||
@app.get("/api/update-check")
|
||
@guarded
|
||
def api_update_check():
|
||
"""拉远端 version.json 与本地比较。远端格式：
|
||
{"version": "1.0.1", "url": "下载地址", "notes": "更新说明", "published": "..."}
|
||
"""
|
||
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 字段")
|
||
|
||
def _v(text):
|
||
return tuple(int(x) for x in re.findall(r"\d+", text)[:3])
|
||
|
||
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"))
|
||
|
||
|
||
# ---------------------------------------------------------------- 配置
|
||
|
||
CONFIG_PY = os.path.join(ASTRAL, "config.py")
|
||
NET_PY = os.path.join(ASTRAL, "api", "MaimaiNet.py")
|
||
|
||
# (key, 文件, 正则, 替换模板, 是否敏感)
|
||
CONF_FIELDS = [
|
||
("baseUrl", "net", r'^(BASE_URL\s*=\s*)"([^"]*)"', '"%s"', True),
|
||
("aesKey", "net", r'^(AES_KEY\s*=\s*)b"([^"]*)"', 'b"%s"', True),
|
||
("aesIV", "net", r'^(AES_IV\s*=\s*)b"([^"]*)"', 'b"%s"', True),
|
||
("obfuscate", "net", r'^(OBFUSCATE_PARAM\s*=\s*)"([^"]*)"', '"%s"', False),
|
||
("clientId", "conf", r'^(CLIENT_ID\s*=\s*)"([^"]*)"', '"%s"', True),
|
||
("regionId", "conf", r'^(REGION_ID\s*=\s*)(\d+)', '%s', False),
|
||
("regionName", "conf", r'^(REGION_NAME\s*=\s*)"([^"]*)"', '"%s"', False),
|
||
("placeId", "conf", r'^(PLACE_ID\s*=\s*)(\d+)', '%s', False),
|
||
("placeName", "conf", r'^(PLACE_NAME\s*=\s*)"([^"]*)"', '"%s"', False),
|
||
("apiVersion", "conf", r'^(MAI_ENCODING\s*=\s*)"([^"]*)"', '"%s"', False),
|
||
]
|
||
CONF_PATH = {"net": NET_PY, "conf": CONFIG_PY}
|
||
|
||
|
||
def _mask(v):
|
||
v = str(v)
|
||
if len(v) <= 8:
|
||
return "***"
|
||
return v[:4] + "***" + v[-2:]
|
||
|
||
|
||
def _reload_astral():
|
||
"""改完 config.py 要按依赖顺序 reload，不能用删 sys.modules 的方式。
|
||
|
||
★ 手机端 errorID=54 的真正原因（两次踩到）：
|
||
1) api.Chime 在 import 时就把 KEYCHIP_ID 求值成 DEFAULT_CHIP_ID；
|
||
2) core/session.py 是 `from . import shop`，早绑定 —— 只删 sys.modules
|
||
再 import，旧的 session/shop 对象仍指向旧 config 模块，拿到的
|
||
chipID 还是空串。
|
||
reload 保持模块对象身份不变，所有持有引用的地方都能看到新值。
|
||
"""
|
||
import importlib
|
||
order = [
|
||
"config", # 常量源头
|
||
"api.MaimaiNet", # BASE_URL / AES_KEY / AES_IV / 混淆盐
|
||
"api.Chime", # DEFAULT_CHIP_ID = KEYCHIP_ID（import 时求值）
|
||
"api", # 包 __init__ 重新 from .Chime import ...
|
||
"core.shop", # import config，取 CLIENT_ID / PLACE_ID
|
||
"core.session", # from . import shop，拿 keychip_id 去扫码
|
||
]
|
||
done = []
|
||
for name in order:
|
||
mod = sys.modules.get(name)
|
||
if mod is None:
|
||
continue
|
||
try:
|
||
importlib.reload(mod)
|
||
done.append(name)
|
||
except Exception as exc: # noqa: BLE001
|
||
print("reload %s 失败：%s" % (name, exc))
|
||
return len(done)
|
||
|
||
|
||
def _read_conf():
|
||
"""按 CONF_FIELDS 从两个源文件里抠出当前值。"""
|
||
out = {}
|
||
for key, which, pat, _tpl, sensitive in CONF_FIELDS:
|
||
path = CONF_PATH[which]
|
||
out[key] = ""
|
||
if not os.path.exists(path):
|
||
continue
|
||
with open(path, encoding="utf-8") as fp:
|
||
text = fp.read()
|
||
m = re.search(pat, text, re.M)
|
||
if m:
|
||
out[key] = m.group(2) if m.lastindex and m.lastindex >= 2 else ""
|
||
# Keychip 是推导值，只读展示
|
||
cid = out.get("clientId") or ""
|
||
pid = out.get("placeId") or ""
|
||
out["keychip"] = ("%s-%s%s" % (cid[:4], cid[4:], pid)) if cid and len(cid) > 4 else ""
|
||
return out
|
||
|
||
|
||
@app.get("/api/config")
|
||
def api_config():
|
||
vals = _read_conf()
|
||
masked = {}
|
||
for key, _w, _p, _t, sensitive in CONF_FIELDS:
|
||
masked[key] = _mask(vals.get(key, "")) if sensitive else vals.get(key, "")
|
||
# 运行时值：api 模块里实际生效的那份（和文件值不一致 = 改了没重载）
|
||
rt = {}
|
||
try:
|
||
import api as _api
|
||
# Chime 在 import 时就把 KEYCHIP_ID 求值成 DEFAULT_CHIP_ID，
|
||
# 所以真正决定登录 chipID 的是它，不是 api.CLIENT_ID。
|
||
_chime = getattr(_api, "Chime", None)
|
||
if _chime is not None:
|
||
_chip = str(getattr(_chime, "DEFAULT_CHIP_ID", "") or "")
|
||
rt["chipId"] = _mask(_chip)
|
||
rt["chipLen"] = len(_chip)
|
||
for name, key in (("CLIENT_ID", "clientId"), ("PLACE_ID", "placeId"),
|
||
("REGION_ID", "regionId"), ("BASE_URL", "baseUrl"),
|
||
("AES_KEY", "aesKey")):
|
||
v = getattr(_api, name, None)
|
||
if v is None:
|
||
continue
|
||
v = v.decode("utf-8", "ignore") if isinstance(v, bytes) else str(v)
|
||
rt[key] = _mask(v) if key in ("clientId", "baseUrl", "aesKey") else v
|
||
# 扫码真正用的：core.shop.current()["keychip_id"]（session.parse_qr 取它）
|
||
from core import shop as _shop
|
||
_kc = str(_shop.current().get("keychip_id") or "")
|
||
rt["shopChip"] = _mask(_kc)
|
||
rt["shopChipLen"] = len(_kc)
|
||
except Exception as exc: # noqa: BLE001
|
||
rt["error"] = str(exc)
|
||
return ok(values=masked, keychip=vals.get("keychip", ""),
|
||
runtime=rt, ready=bool(rt.get("chipLen")),
|
||
rawKeys=[k for k, _w, _p, _t, s in CONF_FIELDS if s],
|
||
files={"config": CONFIG_PY, "net": NET_PY})
|
||
|
||
|
||
@app.post("/api/config")
|
||
@guarded
|
||
def api_config_save():
|
||
"""写入配置。只接受非掩码值 —— 前端把没改过的敏感字段留成 ***xxx*** 时跳过。"""
|
||
body = request.get_json(silent=True) or {}
|
||
changed = []
|
||
for key, which, pat, tpl, sensitive in CONF_FIELDS:
|
||
if key not in body:
|
||
continue
|
||
val = str(body.get(key) or "").strip()
|
||
if not val:
|
||
continue
|
||
if sensitive and "***" in val:
|
||
continue # 没改过的掩码值，不动
|
||
path = CONF_PATH[which]
|
||
if not os.path.exists(path):
|
||
continue
|
||
with open(path, encoding="utf-8") as fp:
|
||
text = fp.read()
|
||
new_text, n = re.subn(pat, lambda m: m.group(1) + (tpl % val), text, count=1, flags=re.M)
|
||
if n:
|
||
with open(path, "w", encoding="utf-8", newline="\n") as fp:
|
||
fp.write(new_text)
|
||
changed.append(key)
|
||
if not changed:
|
||
return fail("没有可写的字段（敏感字段如果显示成 A1b2***xy 说明没改，跳过）")
|
||
reloaded = _reload_astral()
|
||
return ok(changed=changed, values=_read_conf(), reloaded=reloaded)
|
||
|
||
|
||
@app.get("/api/config/export")
|
||
def api_config_export():
|
||
"""导出完整配置（含明文密钥）—— 只给本机用，页面走 download。"""
|
||
vals = _read_conf()
|
||
payload = {k: vals.get(k, "") for k, _w, _p, _t, _s in CONF_FIELDS}
|
||
payload["keychip"] = vals.get("keychip", "")
|
||
payload["_note"] = "maipulse 配置导出，含明文密钥，别外传"
|
||
resp = jsonify(payload)
|
||
resp.headers["Content-Disposition"] = "attachment; filename=maipulse-config.json"
|
||
return resp
|
||
|
||
|
||
@app.post("/api/config/import")
|
||
@guarded
|
||
def api_config_import():
|
||
"""导入配置 JSON。字段名与导出的一致。"""
|
||
body = request.get_json(silent=True) or {}
|
||
changed = []
|
||
for key, which, pat, tpl, sensitive in CONF_FIELDS:
|
||
if key not in body:
|
||
continue
|
||
val = str(body.get(key) or "").strip()
|
||
if not val or "***" in val:
|
||
continue
|
||
path = CONF_PATH[which]
|
||
if not os.path.exists(path):
|
||
continue
|
||
with open(path, encoding="utf-8") as fp:
|
||
text = fp.read()
|
||
new_text, n = re.subn(pat, lambda m: m.group(1) + (tpl % val), text, count=1, flags=re.M)
|
||
if n:
|
||
with open(path, "w", encoding="utf-8", newline="\n") as fp:
|
||
fp.write(new_text)
|
||
changed.append(key)
|
||
# ★ 必须重载：api 模块在 import 时就把 CLIENT_ID / AES_KEY 读进常量了，
|
||
# 只改文件不重载，手机端发出去的 chipID 还是空的 → 服务端 errorID=54。
|
||
reloaded = _reload_astral() if changed else 0
|
||
return ok(changed=changed, reloaded=reloaded, values=_read_conf())
|
||
|
||
|
||
@app.get("/")
|
||
def index():
|
||
return send_file(os.path.join(APP_DIR, "html", "index.html"))
|
||
|
||
|
||
@app.get("/vendor/<path:name>")
|
||
def vendor(name):
|
||
"""前端依赖（html2canvas 等），只允许 html/vendor 下的文件。"""
|
||
base = os.path.join(APP_DIR, "html", "vendor")
|
||
path = os.path.abspath(os.path.join(base, name))
|
||
if not path.startswith(os.path.abspath(base)) or not os.path.exists(path):
|
||
return fail("没有这个文件")
|
||
return send_file(path)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
port = int(os.environ.get("MAIPULSE_PORT", "8787"))
|
||
print("maipulse 已启动： http://127.0.0.1:%d" % port)
|
||
app.run(host="127.0.0.1", port=port, debug=False, threaded=True) |