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 例）
314 lines
17 KiB
Python
314 lines
17 KiB
Python
"""App 点歌推送：把 App 取到的**歌曲下载地址**同步进本服务器的下载队列（服务端口接口）。
|
||
|
||
为什么要有它：App 点歌时已经向上游取到了一条可用的媒体地址（`…/<编号>.ts?token=…`）。
|
||
把这条地址顺带推给自建服务器，服务器就能**省掉一次上游取址**（取址有 5~10 秒的全局串行节奏、
|
||
还会被上游风控），直接按同一个地址下载入库。
|
||
|
||
接口（都在**服务端口**，即 App 面向的命名空间）：
|
||
* `POST /push_download` —— 接收一条推送（JSON 或 `application/x-www-form-urlencoded`）
|
||
* `GET /push_download/status` —— 运维核对计数：accepted / duplicate / rejected / rate_limited / last_at
|
||
|
||
处理顺序（**绝不盲信 App**）：
|
||
① 鉴权：**两条通道任一即可** —— 合法的 App 令牌（请求头 `X-App-Token`，由
|
||
`ktv/apptoken.py` 的守卫在路由之前验过并置 `req.app_token_ok`）**或** `push_token`
|
||
（请求头 `X-Push-Token` 或 body 的 `token`，默认 "local"）；两者都不满足 → 401；
|
||
⚠️ `/push_download` 的 body `token` 永远是 `push_token` 口径（守卫不会把它当 App 令牌，
|
||
见 `apptoken.TokenService._is_legacy_param`），自有 App 请用 `X-App-Token` 头。
|
||
② 限流：同一**真实客户端 IP** 每分钟最多 `push_rate_limit_per_min` 条（默认 60），超出 429（0 = 关闭）。
|
||
"真实客户端 IP" = 直连对端；反代部署（IIS/ARR）下由 `trusted_proxies` + `trust_proxy_headers`
|
||
认定（见 ktv/proxy.py）—— 否则所有设备会共用代理地址的一份额度；
|
||
③ 校验：`tid` 必须是正整数；`url` 必须是 http(s) 且必须通过服务端既有闸门
|
||
`resolver.is_trusted_url(url, tid)`（域名而非裸 IP、剥离 query 后路径以 `/<tid>.ts` 结尾、
|
||
不在 wb66/demo/ac16.vip/ad_files/my_ad_video/pub-*.r2.dev/ss 黑名单）—— 不通过一律 400 且**绝不入库**；
|
||
④ 去重（核心，按 `tid` 判，命中任一种即视为重复、**不重复下载**）：
|
||
a) `keeps.state='done'` 且媒体文件确实在（`already_done`）；
|
||
b) `keeps.state in ('want','downloading')` 或 `downloads.state in ('pending','downloading')`
|
||
（`already_queued`）；
|
||
c) 同一 tid 在过去 `push_dedupe_sec` 秒内已被推送过（`recently_pushed`）；
|
||
⑤ 接受：登记歌曲表（`state='want'`、`user='app'`、`note='App 点歌推送'`；**曲库里没有该编号也允许**，
|
||
`matched=false`）+ 用推送来的 URL 入队下载（`downloads.source='app-push'`、`push_url=<URL>`）。
|
||
|
||
事件日志口径（**这个项目很在意日志量**）：
|
||
* **接受**：每 tid 一条 INFO（`App 点歌推送：接受 编号=…`）—— 真实点歌事件，量小、可追溯；
|
||
* **重复 / 拒绝 / 限流**：按**自然分钟**聚合，分钟翻页（下一次推送或 status 查询）时汇总成一条
|
||
（`App 点歌推送：重复跳过 N 首（already_done=a/already_queued=b/recently_pushed=c）`、
|
||
`App 点歌推送：拒绝 N 条（其中触发限流 M 条），示例：…`）。这三类才可能被刷屏，所以刻意合并成一条；
|
||
拒绝那条保留**一条示例明细**（编号 + 原因），便于定位坏地址从哪来。尚未汇总的当前分钟计数在
|
||
`/push_download/status` 的 `pending_log` 里也能看到，不会丢。
|
||
|
||
⚠️ 本模块只是"入口 + 判定"：真正的队列写入在 `Store.push_enqueue()`，
|
||
固定地址的下载与失败回退在 `Downloader._download()`（见 ktv/downloader.py）。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import hmac
|
||
import os
|
||
import re
|
||
import threading
|
||
import time
|
||
|
||
from . import web
|
||
from .resolver import is_trusted_url
|
||
|
||
# 三种"重复"原因（响应里的 reason 取值）
|
||
DUP_DONE = "already_done"
|
||
DUP_QUEUED = "already_queued"
|
||
DUP_RECENT = "recently_pushed"
|
||
|
||
_KINDS = ("accepted", "duplicate", "rejected", "rate_limited")
|
||
# 重复原因 → 聚合日志里的细分键
|
||
_DUP_KEYS = {DUP_DONE: "done", DUP_QUEUED: "queued", DUP_RECENT: "recent"}
|
||
|
||
|
||
class PushDownloadService:
|
||
"""`/push_download` 的判定与计数（一个实例挂在服务端口路由上）。
|
||
|
||
全部状态都在**内存**里（计数 / 按 IP 的限流窗口 / 按 tid 的近期推送）：
|
||
* 计数是运维核对用的，重启归零，**不落盘**（避免每来一条推送就写一次库）；
|
||
* 限流窗口（60 秒）与近期推送（`push_dedupe_sec`）天然是短时窗，不需要持久化。
|
||
"""
|
||
|
||
MAX_IPS = 512 # 限流窗口最多保留多少个来源 IP（防内存被撑爆）
|
||
|
||
def __init__(self, cfg, store, downloader=None):
|
||
self.cfg = cfg
|
||
self.store = store
|
||
self.downloader = downloader
|
||
self._lock = threading.Lock()
|
||
self.counters = {k: 0 for k in _KINDS}
|
||
self.last_at = 0
|
||
self._hits: dict[str, list[float]] = {} # 来源 IP → 最近一分钟内的请求时刻
|
||
self._recent: dict[int, float] = {} # tid → 上次**接受**推送的时刻
|
||
self._buckets: dict[int, dict] = {} # 自然分钟 → 重复/拒绝/限流计数（聚合日志）
|
||
|
||
# ── 配置（每次现读，改设置后立刻生效）──
|
||
def token(self) -> str:
|
||
"""鉴权串：空值回落到 "local"（与 `service_token` 同口径，绝不等于"谁都能推"）。"""
|
||
return str(self.cfg.get("push_token") or "local")
|
||
|
||
def rate_per_min(self) -> int:
|
||
"""同一 IP 每分钟的推送上限；`0` = 关闭限流。填错按默认 60。"""
|
||
try:
|
||
raw = self.cfg.get("push_rate_limit_per_min")
|
||
return max(0, int(60 if raw is None else raw))
|
||
except Exception:
|
||
return 60
|
||
|
||
def dedupe_sec(self) -> int:
|
||
"""同一编号的推送去重窗口（秒）；`0` = 关闭（永远判 recently_pushed 为否）。"""
|
||
try:
|
||
raw = self.cfg.get("push_dedupe_sec")
|
||
return max(0, int(300 if raw is None else raw))
|
||
except Exception:
|
||
return 300
|
||
|
||
# ── 计数与聚合日志 ──
|
||
def _bump(self, kind: str, sub: str = "", sample: str = "") -> None:
|
||
"""计数 +1 并刷新 `last_at`。
|
||
|
||
* `accepted` **不进桶**（由调用方每 tid 写一条 INFO，真实点歌事件、可追溯）；
|
||
* `duplicate/rejected/rate_limited` 进"当前分钟桶"，分钟翻页时汇总成**一条**事件
|
||
（这三类才可能被刷屏）。`sample` 只记本分钟第一条的明细，便于定位坏地址从哪来。
|
||
|
||
⚠️ 事件落库放在**释放锁之后**（`_flush` 只生成文本），避免持锁写库。
|
||
"""
|
||
now = time.time()
|
||
with self._lock:
|
||
self.counters[kind] = int(self.counters.get(kind, 0)) + 1
|
||
self.last_at = int(now)
|
||
if kind != "accepted":
|
||
bucket = self._buckets.setdefault(int(now // 60), {})
|
||
bucket[kind] = int(bucket.get(kind, 0)) + 1
|
||
if sub:
|
||
bucket[sub] = int(bucket.get(sub, 0)) + 1
|
||
if sample and not bucket.get("sample"):
|
||
bucket["sample"] = str(sample)[:200]
|
||
# 每次推送都看一眼"上一分钟有没有攒着没写的汇总"（接受也要看：否则下一分钟的
|
||
# 第一条是接受时，上一分钟的重复汇总要等到再有重复/拒绝才写出来）
|
||
lines = self._flush(now)
|
||
self._emit(lines)
|
||
|
||
def _flush(self, now: float) -> list[tuple[str, str]]:
|
||
"""挑出**已经过去的**分钟桶，生成汇总事件文本（**不写库、不改计数**）。
|
||
|
||
只汇总"已经翻页"的分钟（`minute < 当前分钟`）：写当前分钟会碎成一行一条。
|
||
返回 `[(level, text), …]`，由调用方在释放锁后交给 `store.event()`。
|
||
"""
|
||
out: list[tuple[str, str]] = []
|
||
cur = int(now // 60)
|
||
for minute in sorted([m for m in self._buckets if m < cur]):
|
||
b = self._buckets.pop(minute)
|
||
if sum(int(b.get(k, 0)) for k in _KINDS) <= 0:
|
||
continue
|
||
if b.get("duplicate"):
|
||
out.append(("INFO", f"App 点歌推送：重复跳过 {b['duplicate']} 首"
|
||
f"（already_done={b.get('done', 0)}"
|
||
f"/already_queued={b.get('queued', 0)}"
|
||
f"/recently_pushed={b.get('recent', 0)}）"))
|
||
if b.get("rejected") or b.get("rate_limited"):
|
||
tail = f"，示例：{b['sample']}" if b.get("sample") else ""
|
||
out.append(("WARN", f"App 点歌推送：拒绝 {b.get('rejected', 0)} 条"
|
||
f"（其中触发限流 {b.get('rate_limited', 0)} 条）{tail}"))
|
||
return out
|
||
|
||
def _emit(self, lines: list[tuple[str, str]]) -> None:
|
||
for level, text in lines:
|
||
self.store.event(level, text)
|
||
|
||
# ── 限流（同 IP 滑动 60 秒窗口）──
|
||
def _rate_exceeded(self, ip: str) -> int:
|
||
"""返回还需等待的秒数（0 = 放行，并记一次请求时刻）。"""
|
||
limit = self.rate_per_min()
|
||
if limit <= 0:
|
||
return 0
|
||
now = time.time()
|
||
with self._lock:
|
||
hits = [t for t in self._hits.get(ip, []) if now - t < 60.0]
|
||
if len(hits) >= limit:
|
||
self._hits[ip] = hits
|
||
return max(1, int(60.0 - (now - hits[0])) + 1)
|
||
hits.append(now)
|
||
self._hits[ip] = hits
|
||
if len(self._hits) > self.MAX_IPS: # 淘汰最久没活动的来源
|
||
overflow = len(self._hits) - self.MAX_IPS
|
||
for k in sorted(self._hits, key=lambda x: self._hits[x][-1])[:overflow]:
|
||
self._hits.pop(k, None)
|
||
return 0
|
||
|
||
# ── 去重 ──
|
||
def _media_exists(self, tid: int) -> bool:
|
||
"""媒体文件是否真的在（`<media_dir>/<tid>.ts`）。只 stat 一个确定路径，不列举目录。"""
|
||
try:
|
||
return os.path.isfile(os.path.join(str(self.cfg.media_dir), f"{tid}.ts"))
|
||
except Exception:
|
||
return False
|
||
|
||
def duplicate_reason(self, tid: int) -> str:
|
||
"""命中返回 `already_done|already_queued|recently_pushed`，都不命中返回空串。"""
|
||
keeps = self.store.q1("SELECT state FROM keeps WHERE tid=?", (int(tid),))
|
||
state = str(keeps["state"]) if keeps else ""
|
||
if state == "done" and self._media_exists(tid):
|
||
return DUP_DONE
|
||
if state in ("want", "downloading"):
|
||
return DUP_QUEUED
|
||
row = self.store.q1("SELECT state FROM downloads WHERE tid=?", (int(tid),))
|
||
if row is not None and str(row["state"]) in ("pending", "downloading"):
|
||
return DUP_QUEUED
|
||
with self._lock:
|
||
last = self._recent.get(int(tid))
|
||
if last and (time.time() - last) < self.dedupe_sec():
|
||
return DUP_RECENT
|
||
return ""
|
||
|
||
# ── 接口 ──
|
||
def handle_push(self, req: web.Request) -> web.Response:
|
||
form = req.form()
|
||
# ⚠️ 来源 IP 必须是**真实客户端 IP**（`req.client_ip` 由 web.py 按 trusted_proxies 认定）：
|
||
# 反向代理（IIS）后面若还用 TCP 对端，所有设备会共用代理地址的一份限流额度。
|
||
ip = str(req.client_ip or "") or "unknown"
|
||
# ① 鉴权（用 bytes 比较：令牌可能含中文，`compare_digest` 对非 ASCII str 会抛异常）
|
||
# **两条合法通道任一即可**：① 合法的 App 令牌（`X-App-Token`，由 ktv/apptoken.py
|
||
# 的守卫在路由之前验过并置 `req.app_token_ok`）；② 既有的 `push_token`。
|
||
# 两者都没有/都不对 → 401「鉴权失败」（口径不变，老 App 不受影响）。
|
||
got = str(req.header("x-push-token") or form.get("token") or "")
|
||
if not bool(getattr(req, "app_token_ok", False)) \
|
||
and not hmac.compare_digest(got.encode("utf-8"), self.token().encode("utf-8")):
|
||
self._bump("rejected", sample=f"来源IP={ip}（鉴权失败）")
|
||
return web.json_response({"ok": False, "error": "鉴权失败"}, 401)
|
||
|
||
# ② 限流（同 IP 每分钟）
|
||
wait = self._rate_exceeded(ip)
|
||
if wait > 0:
|
||
self._bump("rate_limited", sample=f"来源IP={ip}（触发限流，{wait} 秒后可再试）")
|
||
resp = web.json_response(
|
||
{"ok": False, "retry_after": wait, "client_ip": ip,
|
||
"error": f"推送过于频繁：来源 {ip} 每分钟最多 {self.rate_per_min()} 条，请 {wait} 秒后再试"},
|
||
429)
|
||
resp.headers["Retry-After"] = str(wait)
|
||
return resp
|
||
|
||
# ③ 校验（绝不盲信 App）
|
||
tid, err = self._parse_tid(form.get("tid"))
|
||
if err:
|
||
self._bump("rejected", sample=f"来源IP={ip} 编号={form.get('tid')!r}（{err}）")
|
||
return web.json_response({"ok": False, "error": err}, 400)
|
||
url = str(form.get("url") or "").strip()
|
||
# ⚠️ `url` 必须是**已编码的表单字段值**（App 侧 PushDownload.buildForm() 用 URLEncoder 整体编码）：
|
||
# 真源地址自带 `?token=…&t=…`，不编码会被 parse_qs 拆成额外字段、只存下被截断的地址。
|
||
# 这里无法可靠识别"被截断的地址"—— 那种行会下载失败并自动回退取址（不会下错内容）。
|
||
if not re.match(r"^https?://", url, re.I):
|
||
self._bump("rejected", sample=f"来源IP={ip} 编号={tid}（url 不是 http(s)）")
|
||
return web.json_response(
|
||
{"ok": False, "error": "url 必须是以 http:// 或 https:// 开头的下载地址"}, 400)
|
||
ok, why = is_trusted_url(url, tid)
|
||
if not ok:
|
||
self._bump("rejected", sample=f"来源IP={ip} 编号={tid}（地址未通过闸门：{why}）")
|
||
return web.json_response({"ok": False, "error": f"地址未通过闸门：{why}"}, 400)
|
||
|
||
# ④ 去重三态
|
||
reason = self.duplicate_reason(tid)
|
||
if reason:
|
||
self._bump("duplicate", _DUP_KEYS.get(reason, ""))
|
||
return web.json_response({"ok": True, "tid": tid, "duplicate": True,
|
||
"reason": reason, "queued": False})
|
||
|
||
# ⑤ 接受：登记歌曲表 + 用推送来的 URL 入队（source='app-push'）
|
||
# `name` 只是 App 捎带的歌名（可选）：**不入库**（歌名由曲库 songs 提供），仅写进事件便于核对。
|
||
name = str(form.get("name") or "")[:120]
|
||
self.store.push_enqueue(tid, url)
|
||
with self._lock:
|
||
self._recent[int(tid)] = time.time()
|
||
self._bump("accepted")
|
||
self.store.event("INFO", f"App 点歌推送：接受 编号={tid} 来源IP={ip}"
|
||
f"{(' 歌名=' + name) if name else ''}（已入队，使用推送地址）")
|
||
return web.json_response({"ok": True, "tid": tid, "duplicate": False, "queued": True})
|
||
|
||
def handle_status(self, req: web.Request) -> web.Response:
|
||
"""运维核对：累计计数 + 最近一次推送时间 + 未汇总的当前分钟计数 + 生效的限流/去重口径。"""
|
||
with self._lock:
|
||
lines = self._flush(time.time())
|
||
bucket = self._buckets.get(int(time.time() // 60)) or {}
|
||
pending = {k: v for k, v in bucket.items() if k in _KINDS}
|
||
if bucket.get("sample"):
|
||
# 当前分钟的第一条明细也放出来（含**真实客户端 IP**），便于立刻定位是哪台设备
|
||
pending["sample"] = str(bucket["sample"])
|
||
data = dict(self.counters)
|
||
data.update({"ok": True, "last_at": int(self.last_at),
|
||
"rate_limit_per_min": self.rate_per_min(),
|
||
"dedupe_sec": self.dedupe_sec(),
|
||
"pending_log": pending})
|
||
self._emit(lines) # 顺手把已翻页的分钟汇总写掉（不额外制造调用）
|
||
return web.json_response(data)
|
||
|
||
@staticmethod
|
||
def _parse_tid(raw) -> tuple[int, str]:
|
||
"""`tid` 必须是**正整数**（字符串数字也接受：表单/JSON 两种 body 都可能）。"""
|
||
if isinstance(raw, bool):
|
||
return 0, "tid 必须是正整数"
|
||
text = str(raw if raw is not None else "").strip()
|
||
if not text:
|
||
return 0, "缺少 tid（必须是正整数）"
|
||
try:
|
||
tid = int(text)
|
||
except Exception:
|
||
return 0, "tid 必须是正整数"
|
||
if tid <= 0:
|
||
return 0, "tid 必须是正整数"
|
||
return tid, ""
|
||
|
||
|
||
def build_router(cfg, store, downloader=None, service: PushDownloadService | None = None) -> web.Router:
|
||
"""把两个推送接口挂到**服务端口**的路由（由 `service.build_router` 调用）。"""
|
||
r = web.Router()
|
||
svc = service or PushDownloadService(cfg, store, downloader)
|
||
|
||
@r.post("/push_download")
|
||
def push_download(req: web.Request):
|
||
return svc.handle_push(req)
|
||
|
||
@r.get("/push_download/status")
|
||
def push_download_status(req: web.Request):
|
||
return svc.handle_status(req)
|
||
|
||
return r
|