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 例）
368 lines
16 KiB
Python
368 lines
16 KiB
Python
"""极简 HTTP 框架（纯标准库）：路由 + JSON/表单解析 + 静态文件（支持 Range/206）。
|
||
|
||
为什么不用框架：Debian 上零依赖即可部署（`apt install python3` 就够），
|
||
省掉 pip/venv 与版本漂移问题。性能对 KTV 场景（1~3 台设备）足够。
|
||
|
||
**反向代理友好**（IIS/ARR，见 `deploy/iis/`）：
|
||
* 每个响应都带精确 `Content-Length`（HTTP/1.1 keep-alive 的前提，缺了代理会挂住）；
|
||
* 媒体流式发送（256KB 分块），**绝不整块读进内存**，发完连接照常复用；
|
||
* `HEAD`/`Range` 与直连时行为一致（206 + `Content-Length` + `Accept-Ranges`）；
|
||
* 请求体超过 `max_body_bytes` → **413 + 中文说明**（并主动关闭连接，不让管道错位）；
|
||
* 监听队列长度可配（`http_backlog`，默认 128）：探活/突发请求不会被拒。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import re
|
||
import threading
|
||
import urllib.parse
|
||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||
from typing import Callable
|
||
|
||
from . import proxy
|
||
|
||
# 请求体上限的**兜底值**（实际以配置 `max_body_bytes` 为准，见 make_handler）
|
||
MAX_BODY = 8 << 20
|
||
# 监听队列长度兜底（实际以配置 `http_backlog` 为准，见 start_server）
|
||
DEFAULT_BACKLOG = 128
|
||
# 媒体流式发送的分块大小（75MB 的文件整块读进内存会拖垮 2GB 的机器 —— 部署实测）
|
||
CHUNK = 256 * 1024
|
||
|
||
|
||
class Request:
|
||
def __init__(self, method: str, path: str, query: dict, headers: dict, body: bytes,
|
||
client_ip: str = "", peer_ip: str = "", via_proxy: bool = False,
|
||
proxy_source: str = "", forwarded_proto: str = "", forwarded_host: str = "",
|
||
forwarded_for: str = ""):
|
||
self.method = method
|
||
self.path = path
|
||
self.query = query
|
||
# ⚠️ 统一小写：`dict(HTTPMessage)` 之后失去大小写不敏感特性，
|
||
# 若保留原样，`Range` 永远匹配不到 `range` → 媒体分发退回全量 200（部署实测踩过）。
|
||
self.headers = {(k or "").lower(): v for k, v in dict(headers).items()}
|
||
self.body = body
|
||
self.params: dict = {}
|
||
# ── 来源 IP 的三层口径（见 ktv/proxy.py）──
|
||
# `client_ip` = **真实客户端 IP**（限流与日志都用它；经可信代理时来自 XFF/X-Real-IP）；
|
||
# `peer_ip` = TCP 直连对端（反向代理部署时就是代理地址）；
|
||
# 手工构造 Request（测试）时通常只给 `client_ip`，其余为空 —— 与直连语义一致。
|
||
self.client_ip = client_ip
|
||
self.peer_ip = peer_ip or client_ip
|
||
self.via_proxy = bool(via_proxy)
|
||
self.proxy_source = proxy_source
|
||
self.forwarded_proto = forwarded_proto
|
||
self.forwarded_host = forwarded_host
|
||
self.forwarded_for = forwarded_for
|
||
|
||
|
||
def text(self) -> str:
|
||
return self.body.decode("utf-8", "replace")
|
||
|
||
def json(self) -> dict:
|
||
try:
|
||
data = json.loads(self.text() or "{}")
|
||
return data if isinstance(data, dict) else {}
|
||
except Exception:
|
||
return {}
|
||
|
||
def form(self) -> dict:
|
||
ctype = (self.headers.get("content-type") or "").lower()
|
||
if "application/json" in ctype:
|
||
return self.json()
|
||
parsed = urllib.parse.parse_qs(self.text(), keep_blank_values=True)
|
||
return {k: v[0] for k, v in parsed.items()}
|
||
|
||
def q(self, key: str, default: str = "") -> str:
|
||
return self.query.get(key, default)
|
||
|
||
def qi(self, key: str, default: int = 0) -> int:
|
||
try:
|
||
return int(self.query.get(key, default))
|
||
except Exception:
|
||
return default
|
||
|
||
def header(self, key: str, default: str = "") -> str:
|
||
return self.headers.get(key, default)
|
||
|
||
|
||
class Response:
|
||
def __init__(self, body: bytes = b"", status: int = 200, content_type: str = "application/json; charset=utf-8",
|
||
headers: dict | None = None, file_path: str | None = None,
|
||
file_start: int = 0, file_len: int = 0):
|
||
self.body = body
|
||
self.status = status
|
||
self.content_type = content_type
|
||
self.headers = headers or {}
|
||
# 媒体走流式发送：75MB 的文件整块读进内存会拖垮 2GB 的机器（部署实测）
|
||
self.file_path = file_path
|
||
self.file_start = file_start
|
||
self.file_len = file_len
|
||
|
||
|
||
def json_response(data, status: int = 200) -> Response:
|
||
return Response(json.dumps(data, ensure_ascii=False).encode(), status)
|
||
|
||
|
||
def text_response(text: str, status: int = 200) -> Response:
|
||
return Response(str(text).encode(), status, "text/plain; charset=utf-8")
|
||
|
||
|
||
def html_response(html: str, status: int = 200) -> Response:
|
||
return Response(str(html).encode(), status, "text/html; charset=utf-8")
|
||
|
||
|
||
class Router:
|
||
def __init__(self):
|
||
self.routes: list[tuple[str, re.Pattern, Callable, bool]] = []
|
||
# **前置守卫**（可选）：`guard(req)` 返回 Response 就短路（不再走路由），返回 None 放行。
|
||
# 用途：App 令牌校验（ktv/apptoken.py）与双站点宿主隔离 —— 都必须发生在**路由匹配之前**
|
||
# （否则 `/media/x.ts` 会先被媒体路由读到文件，隔离就形同虚设）。
|
||
self.guard: Callable[[Request], Response | None] | None = None
|
||
|
||
def add(self, method: str, pattern: str, handler: Callable, auth: bool = False) -> None:
|
||
self.routes.append((method.upper(), re.compile("^" + pattern + "$"), handler, auth))
|
||
|
||
def get(self, pattern: str, auth: bool = False):
|
||
def deco(fn):
|
||
self.add("GET", pattern, fn, auth)
|
||
return fn
|
||
return deco
|
||
|
||
def post(self, pattern: str, auth: bool = False):
|
||
def deco(fn):
|
||
self.add("POST", pattern, fn, auth)
|
||
return fn
|
||
return deco
|
||
|
||
def dispatch(self, req: Request) -> Response:
|
||
# 前置守卫（App 令牌 / 宿主隔离）：返回响应即短路；**必须在路由之前**。
|
||
# 守卫自身出错也不能打死请求 —— 按"放行"处理并在响应里体现（见 apptoken 的防御式写法）。
|
||
if self.guard is not None:
|
||
early = self.guard(req)
|
||
if early is not None:
|
||
return early
|
||
# ⚠️ HEAD 按 GET 路由（curl -I / OkDownload 的探测都会用到），
|
||
# 否则 HEAD 一律 404（部署实测）；响应体在 _send 里对 HEAD 主动跳过。
|
||
want = "GET" if req.method == "HEAD" else req.method
|
||
for method, rx, handler, _auth in self.routes:
|
||
if method != want:
|
||
continue
|
||
m = rx.match(req.path)
|
||
if m:
|
||
req.params = {k: v for k, v in m.groupdict().items()}
|
||
return handler(req)
|
||
return json_response({"error": "not found", "path": req.path}, 404)
|
||
|
||
|
||
def serve_file(path: str, req: Request, content_type: str = "video/mp2t") -> Response:
|
||
"""支持 Range/206 的文件响应（App 的 OkDownload 断点续传依赖它）。"""
|
||
if not os.path.isfile(path):
|
||
return text_response("not found", 404)
|
||
size = os.path.getsize(path)
|
||
rng = req.header("range")
|
||
start, end = 0, size - 1
|
||
status = 200
|
||
headers = {"Accept-Ranges": "bytes", "Cache-Control": "public, max-age=86400"}
|
||
if rng:
|
||
m = re.match(r"bytes=(\d*)-(\d*)", rng.strip())
|
||
if m:
|
||
if m.group(1):
|
||
start = int(m.group(1))
|
||
end = int(m.group(2)) if m.group(2) else size - 1
|
||
else: # bytes=-N 后缀范围
|
||
start = max(0, size - int(m.group(2) or 0))
|
||
if start >= size or start > end:
|
||
return Response(b"", 416, content_type, {"Content-Range": f"bytes */{size}"})
|
||
status = 206
|
||
headers["Content-Range"] = f"bytes {start}-{end}/{size}"
|
||
length = end - start + 1
|
||
# Content-Length 由 _send 统一发送（这里再写一次会出现重复头）
|
||
return Response(b"", status, content_type, headers, file_path=path, file_start=start, file_len=length)
|
||
|
||
|
||
def make_handler(router: Router, auth_token: str = "", log_fn: Callable[[str], None] | None = None,
|
||
cfg=None, max_body_bytes: int | None = None):
|
||
"""构造 handler；`cfg` 非空时按配置认定真实客户端 IP（反向代理）与请求体上限。"""
|
||
limit = max_body_bytes
|
||
if limit is None and cfg is not None:
|
||
try:
|
||
limit = int(cfg.get("max_body_bytes") or 0)
|
||
except Exception:
|
||
limit = 0
|
||
limit = int(limit if limit else MAX_BODY)
|
||
if limit <= 0:
|
||
limit = MAX_BODY
|
||
|
||
class Handler(BaseHTTPRequestHandler):
|
||
server_version = "KtvSource/1.0"
|
||
protocol_version = "HTTP/1.1"
|
||
|
||
def log_message(self, fmt, *args):
|
||
# 交给上层日志，不直接写 stderr；**用真实客户端 IP**（经可信代理时是设备 IP，
|
||
# 否则反代后面所有日志都是代理地址，排查设备问题等于没日志）。
|
||
if log_fn:
|
||
who = str(getattr(self, "_log_client_ip", "") or self.address_string())
|
||
log_fn(f"{who} {fmt % args}")
|
||
|
||
# ── 入口 ──
|
||
def _handle(self):
|
||
try:
|
||
parsed = urllib.parse.urlparse(self.path)
|
||
query = {k: v[0] for k, v in urllib.parse.parse_qs(parsed.query, keep_blank_values=True).items()}
|
||
# 真实客户端 IP：直连 = TCP 对端；经**可信**代理 = XFF 最左非代理 / X-Real-IP
|
||
# （未注入 cfg 时退回"直连"口径，行为与改造前一致）
|
||
peer = str(getattr(self, "client_address", ("", 0))[0] or "")
|
||
if cfg is not None:
|
||
info = proxy.resolve(peer, self.headers, cfg)
|
||
else:
|
||
info = proxy.ClientInfo(ip=proxy.norm_ip(peer), peer_ip=proxy.norm_ip(peer))
|
||
self._log_client_ip = info.ip or info.peer_ip
|
||
if not (self.headers.get("Content-Length") or "").strip() \
|
||
and str(self.headers.get("Transfer-Encoding") or "").strip():
|
||
# 分块请求体（chunked）本框架不支持：**必须明确拒绝并关连接**，否则那串 chunk
|
||
# 会被当成下一个请求解析（keep-alive 管道错位，经代理时表现为"偶发乱响应"）。
|
||
self.close_connection = True
|
||
resp = json_response({
|
||
"ok": False,
|
||
"error": "不支持分块请求体（Transfer-Encoding: chunked），请带 Content-Length"}, 411)
|
||
resp.headers["Connection"] = "close"
|
||
self._send(resp)
|
||
return
|
||
length = self._content_length()
|
||
if length is None:
|
||
self.close_connection = True
|
||
self._send(json_response({"ok": False, "error": "Content-Length 非法"}, 400))
|
||
return
|
||
limit_now = self._body_limit()
|
||
if length > limit_now:
|
||
# 超过上限**不读进内存**：直接 413 + 中文说明，并关掉连接（否则剩余字节
|
||
# 会被当成下一个请求解析 —— keep-alive 管道错位）。
|
||
self.close_connection = True
|
||
resp = json_response({
|
||
"ok": False,
|
||
"error": f"请求体过大：{length} 字节，超过上限 {limit_now} 字节"
|
||
f"（配置项 max_body_bytes）"}, 413)
|
||
resp.headers["Connection"] = "close"
|
||
self._send(resp)
|
||
return
|
||
body = self.rfile.read(length) if length > 0 else b""
|
||
req = Request(self.command, urllib.parse.unquote(parsed.path), query,
|
||
dict(self.headers), body, client_ip=info.ip,
|
||
peer_ip=info.peer_ip, via_proxy=info.via_proxy,
|
||
proxy_source=info.source, forwarded_proto=info.proto,
|
||
forwarded_host=info.host, forwarded_for=info.xff)
|
||
if auth_token and self._needs_auth(parsed.path):
|
||
if not self._authorized(req):
|
||
self._send(json_response({"error": "unauthorized"}, 401))
|
||
return
|
||
resp = router.dispatch(req)
|
||
self._send(resp)
|
||
except BrokenPipeError:
|
||
pass
|
||
except Exception as exc: # pragma: no cover
|
||
if log_fn:
|
||
log_fn(f"handler error: {type(exc).__name__}: {exc}")
|
||
try:
|
||
self._send(json_response({"error": f"{type(exc).__name__}: {exc}"}, 500))
|
||
except Exception:
|
||
pass
|
||
|
||
def _content_length(self) -> int | None:
|
||
raw = self.headers.get("Content-Length")
|
||
if raw is None or str(raw).strip() == "":
|
||
return 0
|
||
try:
|
||
n = int(str(raw).strip())
|
||
except Exception:
|
||
return None
|
||
return n if n >= 0 else None
|
||
|
||
def _body_limit(self) -> int:
|
||
"""请求体上限：**每次现读配置**（改完立刻生效）；读不到就用构造时的兜底值。"""
|
||
if cfg is not None:
|
||
try:
|
||
v = int(cfg.get("max_body_bytes") or 0)
|
||
if v > 0:
|
||
return v
|
||
except Exception:
|
||
pass
|
||
return limit
|
||
|
||
def _authorized(self, req: Request) -> bool:
|
||
if req.header("x-auth-token") == auth_token:
|
||
return True
|
||
if req.query.get("token") == auth_token:
|
||
return True
|
||
auth = req.header("authorization")
|
||
return auth == f"Bearer {auth_token}"
|
||
|
||
def _needs_auth(self, path: str) -> bool:
|
||
return path.startswith("/api/") and not path.startswith("/api/public")
|
||
|
||
def _send(self, resp: Response):
|
||
self.send_response(resp.status)
|
||
self.send_header("Content-Type", resp.content_type)
|
||
if resp.file_path:
|
||
# 精确长度 = keep-alive 的前提；HEAD 也要带（App/代理的探测靠它）
|
||
self.send_header("Content-Length", str(resp.file_len))
|
||
else:
|
||
self.send_header("Content-Length", str(len(resp.body)))
|
||
for k, v in resp.headers.items():
|
||
self.send_header(k, v)
|
||
self.end_headers()
|
||
if self.command == "HEAD":
|
||
return
|
||
if resp.file_path:
|
||
# 流式分块发送：绝不整块读进内存；发完刚好等于 Content-Length，连接可复用
|
||
remaining = resp.file_len
|
||
with open(resp.file_path, "rb") as fh:
|
||
fh.seek(resp.file_start)
|
||
while remaining > 0:
|
||
chunk = fh.read(min(CHUNK, remaining))
|
||
if not chunk:
|
||
break
|
||
self.wfile.write(chunk)
|
||
remaining -= len(chunk)
|
||
return
|
||
if resp.body:
|
||
self.wfile.write(resp.body)
|
||
|
||
def do_GET(self):
|
||
self._handle()
|
||
|
||
def do_HEAD(self):
|
||
self._handle()
|
||
|
||
def do_POST(self):
|
||
self._handle()
|
||
|
||
def do_PUT(self):
|
||
self._handle()
|
||
|
||
def do_DELETE(self):
|
||
self._handle()
|
||
|
||
return Handler
|
||
|
||
|
||
def start_server(host: str, port: int, router: Router, auth_token: str = "",
|
||
log_fn: Callable[[str], None] | None = None, cfg=None,
|
||
backlog: int | None = None) -> ThreadingHTTPServer:
|
||
"""起一个监听线程并返回 server（`shutdown()` 关闭）。
|
||
|
||
`backlog` = `request_queue_size`（监听队列长度，配置项 `http_backlog`，默认 128）：
|
||
IIS/负载均衡的探活 + 设备突发请求并发时，队列太短会直接 Connection refused。
|
||
"""
|
||
size = int(backlog if backlog is not None else DEFAULT_BACKLOG)
|
||
size = max(5, min(4096, size))
|
||
|
||
class _Server(ThreadingHTTPServer):
|
||
request_queue_size = size # 必须在 __init__（= server_activate/listen）之前生效
|
||
|
||
srv = _Server((host, port), make_handler(router, auth_token, log_fn, cfg=cfg))
|
||
srv.daemon_threads = True
|
||
t = threading.Thread(target=srv.serve_forever, name=f"http-{port}", daemon=True)
|
||
t.start()
|
||
return srv
|