zijunchiang
87aa979abd
版本号：
- 1.5.11 越界（规则 3 规定 patch ≤ 10）→ 修正为 1.6.0（进位到 minor、patch 归 0）
- versionCode 43 不变：43 从未对外发布，本次修正只是把命名拉回约定
合并支线（原在工作区根目录、App 仓库之外，DESIGN.md 自称"支线"）：
- selfhost-source/server/ 自建歌曲源服务器（纯 Python 标准库、零 pip 依赖）
- selfhost-source/deploy/、DESIGN.md、LIBRARY.md、build_inventory.py
- selfhost-source/inventory/ 小清单（playlists.csv、song-stats.md 等）
- ⚠️ inventory 生成物（songs-*.csv / singers-top.csv，共 84MB）不入库，
已写入 .gitignore；需要时用 build_inventory.py 依据 muse.db 重新生成
同时入库的 App 侧支线改动（此前为工作树未提交状态）：
- Whoami.kt（whoami 设备心跳）、DeviceIdentity.kt（设备身份伪装）
- PushDownload.kt / PushDownloadClient.kt（点歌推送）
- AdVideoFilter 带 query/fragment 的地址闸门；KTV_BRIDGE_API 11 → 13
- 修 ktv_api.js 的 global.KtvBridgeApiVersion（12 → 13，与注释及 Kotlin 两处校验一致）
- 对应单测：WhoamiTest / DeviceIdentityTest / PushDownloadTest / AdVideoFilterTest（共 39 例）
文档：
- DESIGN.md 的"在 App 仓库之外"约定改为"已并入仓库统一版本化"
- AGENTS.md 版本号行更新为 1.6.0/43，并新增支线并入说明
验证：assembleRelease + testReleaseUnitTest 全过（39 例）
407 lines
18 KiB
Python
407 lines
18 KiB
Python
"""取址：**完全按 App 的方式**向上游歌曲源要地址（i.php → token → do.php → 媒体地址）。
|
||
|
||
并复刻 App 的"可信地址"闸门（对应 `ktv_api.js` 的 isDemoUrl / `SongUrlResolver.isValidSongUrl`
|
||
/ `AdVideoFilter.isTrustedSongUrl`）—— 自建服务器取到的地址也必须过同一套规则，
|
||
否则会把广告/测试资源当歌曲下载入库。
|
||
|
||
闸门规则（任一命中即拒收）：
|
||
1. **host 或路径**含 wb66 / demo；
|
||
2. 域名命中 *.ac16.vip、路径含 /ad_files/、my_ad_video；
|
||
3. host 是裸 IP（http://1.2.3.4[:port]/…）→ **必须是域名**；
|
||
4. pub-*.r2.dev/ss/ ；
|
||
5. `/<编号>.ts` 文件名与点歌编号不一致。
|
||
|
||
⚠️ **一律先在「剥离 query/fragment」的 URL 上判定**（`urlsplit` 取 host 与 path）：
|
||
真源地址带 `?token=..&t=..`（2026-10-11 抓包：`.../480p/7200088.ts?token=…&t=…`），
|
||
若拿整串做 `.ts` 比对就会误判；反过来若把整串拿去搜 `wb66|demo`，query 里的随机串
|
||
也可能把**合法地址误杀**（App 侧注释同样提到 `demo` 这类词太宽松）。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import json
|
||
import random
|
||
import re
|
||
import threading
|
||
import time
|
||
import urllib.parse
|
||
import urllib.request
|
||
|
||
APP_ID = "d4eeacc6cec3434fbc8c41608a3056a0"
|
||
APP_KEY = "024210cba40d4385a93e6c2d3249bfb5"
|
||
SDK_KEY = "19042303a8374f67ae3fe1e25c97936f"
|
||
VER = "2.0"
|
||
VN = "4.1.3.03281430"
|
||
|
||
_IP_URL = re.compile(r"^https?://\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}(?::\d+)?/", re.I)
|
||
_R2_DEMO = re.compile(r"pub-[a-z0-9]+\.r2\.dev/ss/", re.I)
|
||
_AC16 = re.compile(r"^https?://(?:[^/]+\.)?ac16\.vip(?::\d+)?/", re.I)
|
||
# `.ts` 文件名：**只在 URL 的 path 上匹配**（path 里没有 `?`/`#`），故按 `[^/]+` 取整段文件名。
|
||
_TS_NAME = re.compile(r"/([^/]+)\.ts(?:/|$)", re.I)
|
||
# 广告/测试资源的路径特征（只搜 path，见模块头注释）
|
||
_AD_PATH = re.compile(r"/ad_files/|my_ad_video", re.I)
|
||
# 测试资源关键字（只搜 host + path，**不搜 query**：query 里是签名 token，误杀风险高）
|
||
_DEMO_WORD = re.compile(r"wb66|demo", re.I)
|
||
|
||
_rate_lock = threading.Lock()
|
||
_last_request = [0.0]
|
||
|
||
|
||
def md5(text: str) -> str:
|
||
return hashlib.md5(text.encode()).hexdigest()
|
||
|
||
|
||
def split_url(url: str) -> tuple[str, str]:
|
||
"""把 URL 拆成 `(host, path)`，**query 与 fragment 一律丢弃**。
|
||
|
||
正常走 `urllib.parse.urlsplit`；对畸形 URL（`urlsplit` 抛异常/取不到 host）
|
||
退化为手工剥离 `?`/`#`，宁可少判也不要拿整串去误判。
|
||
"""
|
||
s = str(url or "")
|
||
try:
|
||
p = urllib.parse.urlsplit(s)
|
||
if p.netloc:
|
||
return (p.hostname or "").lower(), (p.path or "")
|
||
except Exception:
|
||
pass
|
||
head = s.split("#", 1)[0].split("?", 1)[0]
|
||
m = re.match(r"^[a-zA-Z][a-zA-Z0-9+.-]*://([^/]*)(/.*)?$", head)
|
||
if not m:
|
||
return "", head
|
||
host = m.group(1).rsplit("@", 1)[-1] # 去掉 user:pass@
|
||
if host.startswith("["): # IPv6 字面量
|
||
host = host[1:host.find("]")] if "]" in host else host
|
||
elif ":" in host:
|
||
host = host.split(":", 1)[0]
|
||
return host.lower(), (m.group(2) or "")
|
||
|
||
|
||
def ts_basename(url: str) -> str | None:
|
||
"""取 URL **路径**里 `.ts` 的文件名主体（不含扩展名）；路径里没有则返回 None。
|
||
|
||
例：`http://txog.ktvsky.com/…/480p/7200088.ts?token=…&t=…` → `7200088`。
|
||
"""
|
||
_host, path = split_url(url)
|
||
m = _TS_NAME.search(path)
|
||
return m.group(1) if m else None
|
||
|
||
|
||
def _pace(min_interval_ms: int, max_interval_ms: int | None = None) -> float:
|
||
"""按 [min,max] 随机间隔限速（**全局串行**）。返回本次实际等待秒数，便于测试。"""
|
||
lo = max(0, int(min_interval_ms or 0)) / 1000.0
|
||
hi = max(lo, (int(max_interval_ms) if max_interval_ms else int(min_interval_ms or 0)) / 1000.0)
|
||
with _rate_lock:
|
||
gap = random.uniform(lo, hi)
|
||
wait = gap - (time.time() - _last_request[0])
|
||
if wait > 0:
|
||
time.sleep(wait)
|
||
_last_request[0] = time.time()
|
||
return max(0.0, wait)
|
||
|
||
|
||
def _get(url: str, timeout: int = 20, ua: str = "MaidongKTV/1.2",
|
||
min_interval_ms: int = 300, max_interval_ms: int | None = None) -> tuple[int, str]:
|
||
"""干净 GET（App 的原生通道只发 Host+Accept）。
|
||
|
||
**取链限速**：每次请求前等 `[min_interval_ms, max_interval_ms]` 内的一个随机值
|
||
（默认 3~5 秒，随机化避免节奏固定被上游识别）。限速是**全局串行**的 ——
|
||
10 个下载线程共用同一个节奏，因此并发不会叠加成对上游的洪水。
|
||
"""
|
||
_pace(min_interval_ms, max_interval_ms)
|
||
req = urllib.request.Request(url, headers={"Accept": "*/*", "User-Agent": ua})
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
||
return resp.status, resp.read().decode("utf-8", "replace")
|
||
except urllib.error.HTTPError as exc: # 403/5xx：把状态带回去判断
|
||
return exc.code, ""
|
||
except Exception:
|
||
return 0, ""
|
||
|
||
|
||
def is_trusted_url(url: str, tid: int) -> tuple[bool, str]:
|
||
s = str(url or "")
|
||
if not re.match(r"^https?://", s, re.I):
|
||
return False, "非 http(s)"
|
||
# ⚠️ host/path 一律在**剥离 query/fragment** 之后取：真源地址带 `?token=..&t=..`，
|
||
# 拿整串做 `.ts` 比对会误判，拿整串搜 wb66/demo 也可能误杀（见模块头注释）。
|
||
host, path = split_url(s)
|
||
if _DEMO_WORD.search(host + path):
|
||
return False, "命中 wb66/demo"
|
||
if _AC16.match(s) or _AD_PATH.search(path):
|
||
return False, "命中广告域名/路径"
|
||
if _IP_URL.match(s):
|
||
return False, "host 是裸 IP（App 只接受域名）"
|
||
if _R2_DEMO.search(host + path):
|
||
return False, "命中 r2.dev 测试桶"
|
||
name = ts_basename(s)
|
||
if name is not None and name != str(tid):
|
||
return False, f"文件名 {name}.ts != 编号 {tid}"
|
||
return True, "ok"
|
||
|
||
|
||
# ── 取址签名串（**顺序即原版协议**，2026-10-11 抓包逐字节复核通过）──
|
||
# 参数顺序、原始 token 里的 `=`、以及"拼串 + KEY 再 md5"必须与原版 App 完全一致；
|
||
# whoami.py 也用这两个函数复用设备身份口径（各取所需字段）。
|
||
|
||
def token_params(mac: str, sn: str, ts: int | None = None) -> str:
|
||
"""`i.php` 的签名串：appid→mac→sn→time→ver→vn。"""
|
||
return (f"appid={APP_ID}&mac={mac}_{sn}&sn={sn}"
|
||
f"&time={int(ts if ts is not None else time.time())}&ver={VER}&vn={VN}")
|
||
|
||
|
||
def token_sign(params: str) -> str:
|
||
"""`i.php` 签名 = md5(原样拼串 + APP_KEY)。"""
|
||
return md5(params + APP_KEY)
|
||
|
||
|
||
def song_params(tid: int | str, token: str, mac: str, sn: str, resolution: str = "720",
|
||
h265: bool = False, ts: int | None = None) -> str:
|
||
"""`music/do.php` 的签名串：appid→device→ish265→ls→musicno→resolution→sn→time→token。"""
|
||
return (f"appid={APP_ID}&device={mac}_{sn}&ish265={'1' if h265 else '0'}&ls=1"
|
||
f"&musicno={tid}&resolution={resolution}&sn={sn}"
|
||
f"&time={int(ts if ts is not None else time.time())}&token={token}")
|
||
|
||
|
||
def song_sign(params: str) -> str:
|
||
"""`music/do.php` 签名 = md5(原样拼串 + SDK_KEY)。"""
|
||
return md5(params + SDK_KEY)
|
||
|
||
|
||
def fetch_config(host: str, device_id: str, timeout: int = 15, ua: str = "MaidongKTV/1.2") -> dict | None:
|
||
"""POST {host}/api.php action=get_config（原版取址前会调；用于了解试用额度）。"""
|
||
host = host.rstrip("/")
|
||
if not host.startswith("http"):
|
||
host = "http://" + host
|
||
body = urllib.parse.urlencode({"action": "get_config", "device_id": device_id}).encode()
|
||
req = urllib.request.Request(host + "/api.php", data=body, method="POST")
|
||
req.add_header("Content-Type", "application/x-www-form-urlencoded")
|
||
req.add_header("User-Agent", ua)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
||
data = json.loads(resp.read().decode("utf-8", "replace"))
|
||
return data.get("config") or {}
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def get_token(host: str, mac: str, sn: str, timeout: int = 20, ua: str = "MaidongKTV/1.2",
|
||
min_interval_ms: int = 300, max_interval_ms: int | None = None) -> str | None:
|
||
host = host.rstrip("/")
|
||
if not host.startswith("http"):
|
||
host = "http://" + host
|
||
params = token_params(mac, sn)
|
||
status, body = _get(f"{host}/i.php?{params}&sign={token_sign(params)}", timeout, ua,
|
||
min_interval_ms, max_interval_ms)
|
||
if status != 200 or not body:
|
||
return None
|
||
try:
|
||
data = json.loads(body)
|
||
except Exception:
|
||
return None
|
||
if data.get("code") != 200:
|
||
return None
|
||
return data.get("token") or None
|
||
|
||
|
||
def get_song_url(host: str, tid: int, token: str, mac: str, sn: str, resolution: str = "720",
|
||
h265: bool = False, timeout: int = 20, ua: str = "MaidongKTV/1.2",
|
||
min_interval_ms: int = 300, max_interval_ms: int | None = None) -> tuple[str | None, str]:
|
||
"""返回 (url, detail)。detail 用于日志：ok / code=xxx / http=xxx。"""
|
||
host = host.rstrip("/")
|
||
if not host.startswith("http"):
|
||
host = "http://" + host
|
||
params = song_params(tid, token, mac, sn, resolution, h265)
|
||
status, body = _get(f"{host}/music/do.php?{params}&sign={song_sign(params)}", timeout, ua,
|
||
min_interval_ms, max_interval_ms)
|
||
if status != 200 or not body:
|
||
return None, f"http={status}"
|
||
try:
|
||
data = json.loads(body)
|
||
except Exception:
|
||
return None, "非JSON"
|
||
code = data.get("code")
|
||
if code != 200:
|
||
return None, f"code={code} msg={data.get('msg', '')}"
|
||
url = str(data.get("data") or "")
|
||
if not url:
|
||
return None, "无 data"
|
||
ok, why = is_trusted_url(url, tid)
|
||
if not ok:
|
||
return None, f"地址不可信({why})"
|
||
return url, "ok"
|
||
|
||
|
||
class Resolver:
|
||
"""对上游源的取址器：**多个源自动选择**。
|
||
|
||
自动选择策略（需求③）：
|
||
1. **健康度打分**：每次成功 +1、失败 -1（下限 -5），分数高的先试 ——
|
||
即"上次能用的源优先"，与 App 的"成功源置顶"思路一致；
|
||
2. **失败冷却**：403（风控/额度）或连续失败 → 该源进入冷却（默认 10 分钟）并沉到队尾，
|
||
避免继续冲击它；冷却结束后自动回到候选；
|
||
3. 同分时按配置里的书写顺序，保证行为可预期；
|
||
4. 全部源都在冷却 / 全部失败 → 明确报错（下载任务转入重试，不会静默丢任务）；
|
||
5. **取链限速**：每次 i.php / do.php 请求前等 3~5 秒随机间隔（可配），全局串行、
|
||
不随并发放大 —— 避免给上游造成拥堵或被判风控。
|
||
统计与冷却状态会持久化到 SQLite meta，重启后不丢。
|
||
"""
|
||
|
||
BLOCK_COOLDOWN_SEC = 10 * 60
|
||
|
||
def __init__(self, cfg, store=None):
|
||
self.cfg = cfg
|
||
self.store = store
|
||
self.tokens: dict[str, str] = {}
|
||
self._lock = threading.Lock()
|
||
self.last_ok_host = ""
|
||
self.last_order: list[str] = []
|
||
self.health: dict[str, dict] = {}
|
||
self._load_health()
|
||
|
||
# ── 健康度持久化 ──
|
||
def _load_health(self) -> None:
|
||
if not self.store:
|
||
return
|
||
try:
|
||
raw = self.store.get_meta("upstream_health", "")
|
||
self.health = json.loads(raw) if raw else {}
|
||
except Exception:
|
||
self.health = {}
|
||
|
||
def _save_health(self) -> None:
|
||
if not self.store:
|
||
return
|
||
try:
|
||
self.store.set_meta("upstream_health", json.dumps(self.health, ensure_ascii=False))
|
||
except Exception:
|
||
pass
|
||
|
||
def _h(self, host: str) -> dict:
|
||
return self.health.setdefault(host, {"ok": 0, "fail": 0, "score": 0, "last_ok": 0,
|
||
"last_error": "", "blocked_until": 0})
|
||
|
||
def note_ok(self, host: str) -> None:
|
||
with self._lock:
|
||
h = self._h(host)
|
||
h["ok"] += 1
|
||
h["score"] = min(50, h["score"] + 1)
|
||
h["last_ok"] = int(time.time())
|
||
h["last_error"] = ""
|
||
h["blocked_until"] = 0
|
||
self._save_health()
|
||
|
||
def note_fail(self, host: str, reason: str) -> None:
|
||
with self._lock:
|
||
h = self._h(host)
|
||
h["fail"] += 1
|
||
h["score"] = max(-5, h["score"] - 1)
|
||
h["last_error"] = str(reason)[:200]
|
||
if "403" in str(reason):
|
||
h["blocked_until"] = int(time.time()) + self.BLOCK_COOLDOWN_SEC
|
||
elif h["fail"] >= 5 and h["ok"] == 0:
|
||
h["blocked_until"] = int(time.time()) + self.BLOCK_COOLDOWN_SEC
|
||
self._save_health()
|
||
|
||
def host_report(self) -> list[dict]:
|
||
"""给管理界面看的源列表（含健康度与冷却剩余时间）。"""
|
||
now = int(time.time())
|
||
out = []
|
||
for i, host in enumerate(self._configured_hosts()):
|
||
h = dict(self.health.get(host) or {})
|
||
out.append({"host": host, "order": i, "ok": h.get("ok", 0), "fail": h.get("fail", 0),
|
||
"score": h.get("score", 0), "last_ok": h.get("last_ok", 0),
|
||
"last_error": h.get("last_error", ""),
|
||
"cooldown_left": max(0, int(h.get("blocked_until", 0)) - now),
|
||
"cooling": int(h.get("blocked_until", 0)) > now})
|
||
return out
|
||
|
||
def clear_health(self) -> None:
|
||
with self._lock:
|
||
self.health = {}
|
||
self._save_health()
|
||
|
||
def _configured_hosts(self) -> list[str]:
|
||
hs = self.cfg.get("upstream_hosts") or []
|
||
if isinstance(hs, str):
|
||
hs = [h.strip() for h in hs.split(",") if h.strip()]
|
||
return [h if h.startswith("http") else "http://" + h for h in hs if h]
|
||
|
||
def mac_sn(self) -> tuple[str, str]:
|
||
"""本机上报给上游的设备身份 `(mac, sn)` —— **与 whoami 心跳同一份**。
|
||
|
||
`ktv/identity.py` 的 `apply_to_cfg()` 已把"自动生成的盒子身份"写进
|
||
`upstream_mac`/`upstream_sn`（用户显式配置的优先），因此这里读配置即可：
|
||
`i.php` 的 `mac=<mac>_<sn>`、`music/do.php` 的 `device=<mac>_<sn>` 与
|
||
whoami 的 `_mac`/`androidid`/`_cid` 全部来自同一个 mac/sn，不会再各读各处。
|
||
"""
|
||
return str(self.cfg.get("upstream_mac") or ""), str(self.cfg.get("upstream_sn") or "")
|
||
|
||
def clear_tokens(self) -> None:
|
||
"""清掉已缓存的 token（**设备身份变了以后必须清**：旧 token 绑在旧 device 上）。"""
|
||
with self._lock:
|
||
self.tokens.clear()
|
||
|
||
def hosts(self) -> list[str]:
|
||
"""**自动选择顺序**：按 (是否冷却, 健康分, 配置顺序) 排序。"""
|
||
now = int(time.time())
|
||
items = []
|
||
for i, host in enumerate(self._configured_hosts()):
|
||
h = self.health.get(host) or {}
|
||
cooling = int(h.get("blocked_until", 0)) > now
|
||
items.append((1 if cooling else 0, -int(h.get("score", 0)), i, host))
|
||
items.sort()
|
||
ordered = [x[3] for x in items]
|
||
if ordered:
|
||
with self._lock:
|
||
if self.last_order != ordered:
|
||
self.last_order = ordered
|
||
if self.store:
|
||
self.store.event("INFO", "上游源自动排序：" + " > ".join(ordered))
|
||
return ordered
|
||
|
||
def resolve(self, tid: int) -> tuple[str | None, str, str]:
|
||
"""返回 (url, host, detail)。"""
|
||
mac, sn = self.mac_sn() # 与 whoami 心跳同一份设备身份
|
||
resolution = str(self.cfg.get("upstream_resolution"))
|
||
ua = str(self.cfg.get("user_agent"))
|
||
gap = int(self.cfg.get("upstream_interval_min_ms", self.cfg.get("upstream_request_interval_ms", 3000)))
|
||
gap_max = int(self.cfg.get("upstream_interval_max_ms", gap))
|
||
timeout = int(self.cfg.get("timeout_sec"))
|
||
last = "no-source"
|
||
for host in self.hosts():
|
||
with self._lock:
|
||
token = self.tokens.get(host)
|
||
if not token:
|
||
token = get_token(host, mac, sn, timeout, ua, gap, gap_max)
|
||
if token:
|
||
with self._lock:
|
||
self.tokens[host] = token
|
||
if not token:
|
||
last = f"{host}: token-fail"
|
||
# i.php 拿不到 token（上游限流/风控）→ 记失败并进入冷却，别每 3~5 秒死磕
|
||
self.note_fail(host, "token-fail")
|
||
with self._lock:
|
||
h = self._h(host)
|
||
if h.get("fail", 0) >= 3:
|
||
h["blocked_until"] = int(time.time()) + int(
|
||
self.cfg.get("upstream_token_fail_cooldown_sec") or 900)
|
||
self._save_health()
|
||
continue
|
||
url, detail = get_song_url(host, tid, token, mac, sn, resolution,
|
||
False, timeout, ua, gap, gap_max)
|
||
if url:
|
||
self.last_ok_host = host
|
||
self.note_ok(host)
|
||
return url, host, "ok"
|
||
last = f"{host}: {detail}"
|
||
self.note_fail(host, detail)
|
||
with self._lock: # token 失效：下次重新登录
|
||
self.tokens.pop(host, None)
|
||
if self.store:
|
||
self.store.event("WARN", f"取址失败 编号={tid} 详情={last}")
|
||
return None, "", last
|
||
|
||
def note_failure(self, tid: int, detail: str) -> None:
|
||
if self.store:
|
||
self.store.event("WARN", f"取址失败 编号={tid} {detail}")
|