chore(async-infra): 异步基础层 + 工具链 + 清理

异步基础层(core/):
- db.py   aiosqlite + WAL:读路径无锁并发、写路径 asyncio.Lock 串行,
          busy_timeout 自旋——实现「读写同时进行」(旧实现用 RLock 全读串行)
- http_client.py  aiohttp 连接池 + 有界重试(5xx 重试 / 4xx 不重试)+
          响应体上限 + 流式接口;按事件循环隔离 session(支持迁移期双 loop)

工具链:
- pyproject.toml   ruff(E/F/W/I/B/UP/C4/ASYNC/S/RUF) + pyright(basic) + pytest(asyncio auto)
- requirements-fastapi.txt  FastAPI/aiohttp/aiosqlite/aiofiles 等目标栈
- requirements-dev.txt      开发依赖(pytest/ruff/pyright)

清理:删除无用脚本 _csscheck.py 与产物标记 _css_ok.txt/_pkg_ok.txt,同步 .gitignore

验证:pytest 10/10(含「写 200 条同时 4 读协程零异常」并发用例);
ruff All checks passed;pyright 0 errors。
This commit is contained in:
snowy-yang committed 2026-09-26 00:24:00 +08:00
1 parent 41356e89c4
commit 9f56f68b66
16 files changed
+706 -42

No files matched your search

+6 -3
View File
@@ -4,10 +4,10 @@
自 main.py 原样迁出,语义不变:仅常量、配置类与路径定义。
不导入 flask/fastapi,不持有任何运行时可变状态(运行时状态见 core.state)。
"""
import os
from dataclasses import dataclass
# 站点统一版本号(P3-12/P3-22:/health、/api/info、前端 CHANGELOG 全部以此为准,别处不得硬编码)
APP_VERSION = '2.4.0'
@@ -19,6 +19,7 @@ RESULT_RETENTION_SECONDS = 35 * 60
@dataclass
class Settings:
"""服务配置类(原 main.APIConfig;迁移期保留 APIConfig 别名以兼容既有引用)"""
host: str = '0.0.0.0'
port: int = 5000
debug: bool = False
@@ -26,7 +27,9 @@ class Settings:
# 第 98 条:max_file_size/request_timeout 两个架空字段已删除——请求体上限真实
# 生效点是 _body_size_limit(业务 2MB/备份 200MB),上游超时在 music_api 各调用点
log_level: str = 'INFO'
cors_origins: str = (os.environ.get('CORS_ORIGINS') or '*').strip() or '*' # 第 100 条:可经环境变量收敛同源(空值回落默认)
cors_origins: str = (
os.environ.get('CORS_ORIGINS') or '*'
).strip() or '*' # 第 100 条:可经环境变量收敛同源(空值回落默认)
# 向后兼容别名:既有代码/文档沿用 APIConfig 名称
@@ -57,7 +60,7 @@ BATCH_ZIP_CONCURRENCY = 3
# 音频直链多重防护(HMAC 签名 + 时效 + 代理流)
AUDIO_TOKEN_TTL = RESULT_RETENTION_SECONDS + 60 # 第 93 条(P2-4):= 36 分钟——原 30 分钟
AUDIO_TOKEN_TTL = RESULT_RETENTION_SECONDS + 60 # 第 93 条(P2-4):= 36 分钟——原 30 分钟
# 大于交付/解析结果 35 分钟保留期 60 秒:交付窗口末段试听不再 403(旧 30 分钟口径矛盾已修)
AUDIO_URL_KEYS = {'url', 'audioUrl'}
AUDIO_HOST_MARKS = ('126.net', 'music.126.net', '.netease')
+137
View File
@@ -0,0 +1,137 @@
# -*- coding: utf-8 -*-
"""异步 SQLite 访问层(aiosqlite + WAL)。
核心目标:**读写同时进行**。
- WAL 模式:读事务与写事务互不阻塞(读快照隔离),这是 SQLite 层并发的根本;
- 读路径:每次操作独立连接、无锁、可并发(`fetch_all`/`fetch_one`/`fetch_val`);
- 写路径:SQLite 全库同一时刻只允许一个写者,用 `asyncio.Lock` 串行化写事务,
但写者不阻塞读者(WAL);
- `busy_timeout`:多进程/多连接下遇到短暂写锁自旋等待而非立刻 SQLITE_BUSY。
与旧实现的差异:旧 `auth_system` 用 `threading.RLock` 把**全部**读写串行化,
读也被写挡住;本层只串行化写,读放行并发,即「读写同时进行」。
"""
import asyncio
import os
from collections.abc import AsyncIterator, Iterable, Sequence
from contextlib import asynccontextmanager
from pathlib import Path
from typing import Any
import aiosqlite
# 站点数据目录(可经环境变量覆盖,便于测试隔离)
DATA_DIR = Path(os.environ.get('SITE_DATA_DIR', 'data'))
DB_PATH = DATA_DIR / 'site.db'
# 写事务串行锁(惰性创建,避免跨事件循环绑定)
_WRITE_LOCK: asyncio.Lock | None = None
def _write_lock() -> asyncio.Lock:
global _WRITE_LOCK
if _WRITE_LOCK is None:
_WRITE_LOCK = asyncio.Lock()
return _WRITE_LOCK
async def _configure(conn: aiosqlite.Connection) -> None:
conn.row_factory = aiosqlite.Row
await conn.execute('PRAGMA journal_mode=WAL') # 读写并发的前提
await conn.execute('PRAGMA busy_timeout=15000') # 遇写锁自旋等待,而非立即报错
await conn.execute('PRAGMA synchronous=NORMAL') # WAL 下兼顾安全与吞吐
await conn.execute('PRAGMA foreign_keys=ON')
async def connect() -> aiosqlite.Connection:
"""新建一个已配置好的连接(调用方负责 close)。"""
DATA_DIR.mkdir(parents=True, exist_ok=True) # noqa: ASYNC240 - 幂等的微秒级一次目录创建,无需线程池
conn = await aiosqlite.connect(str(DB_PATH), timeout=15)
await _configure(conn)
return conn
# ============ 读路径(并发,无锁) ============
async def fetch_all(sql: str, params: Sequence = ()) -> list[aiosqlite.Row]:
conn = await connect()
try:
cur = await conn.execute(sql, params)
return list(await cur.fetchall())
finally:
await conn.close()
async def fetch_one(sql: str, params: Sequence = ()) -> aiosqlite.Row | None:
conn = await connect()
try:
cur = await conn.execute(sql, params)
return await cur.fetchone()
finally:
await conn.close()
async def fetch_val(sql: str, params: Sequence = (), default: Any = None) -> Any:
row = await fetch_one(sql, params)
return row[0] if row is not None else default
# ============ 写路径(串行,不阻塞读) ============
async def execute(sql: str, params: Sequence = ()) -> int:
"""执行单条写语句,返回 lastrowid(INSERT)或 rowcount(UPDATE/DELETE)。"""
async with _write_lock():
conn = await connect()
try:
cur = await conn.execute(sql, params)
await conn.commit()
return cur.lastrowid if cur.lastrowid else cur.rowcount
finally:
await conn.close()
async def execute_many(sql: str, seq: Iterable[Sequence]) -> int:
"""批量写,返回受影响行数。"""
async with _write_lock():
conn = await connect()
try:
cur = await conn.executemany(sql, seq)
await conn.commit()
return cur.rowcount
finally:
await conn.close()
@asynccontextmanager
async def write_transaction() -> AsyncIterator[aiosqlite.Connection]:
"""写事务上下文:多条写语句原子提交(串行锁内)。
用法:
async with write_transaction() as conn:
await conn.execute(...)
await conn.execute(...)
# 退出即 commit;异常则 rollback
"""
async with _write_lock():
conn = await connect()
try:
yield conn
await conn.commit()
except Exception:
await conn.rollback()
raise
finally:
await conn.close()
@asynccontextmanager
async def read_connection() -> AsyncIterator[aiosqlite.Connection]:
"""只读连接上下文(无锁、可并发);需要在一个连接上做多次读时使用。"""
conn = await connect()
try:
yield conn
finally:
await conn.close()
+259
View File
@@ -0,0 +1,259 @@
# -*- coding: utf-8 -*-
"""异步 HTTP 客户端(aiohttp):连接池 + 有界重试 + 响应体上限。
替代旧的长连接 `requests.Session`/裸 `requests.get/post`:
- 单一 `ClientSession` 复用 TCP 连接(keep-alive),连接池限制防上游打爆;
- 重试口径与旧实现一致:网络异常/5xx 退避重试一次,4xx(429/460 风控)**不**重试;
- 响应体超过 `max_bytes` 直接拒绝(防止被劫持/异常上游的超大 body 进内存)。
异常层次(调用方按需映射为业务异常,例如 music_api.APIException):
UpstreamError
├── UpstreamNetworkError 网络/超时(可重试)
├── UpstreamHTTPError 非 2xx(5xx 可重试,4xx 不可)
└── UpstreamTooLarge 响应体超限
"""
import asyncio
import json
from collections.abc import AsyncIterator, Mapping
from contextlib import asynccontextmanager
from dataclasses import dataclass, field
from typing import Any
import aiohttp
DEFAULT_TIMEOUT = 30
DEFAULT_MAX_BYTES = 8 * 1024 * 1024
# 第 95 条(P2-6)/第 98 条口径:UA 与 Referer 由调用方传入(music_api.APIConstants)
_EMPTY: Mapping[str, str] = {}
class UpstreamError(Exception):
"""上游请求错误基类。"""
class UpstreamNetworkError(UpstreamError):
"""网络异常/超时(可重试)。"""
class UpstreamHTTPError(UpstreamError):
"""上游返回非 2xx 状态码。"""
def __init__(self, status: int, message: str = ''):
super().__init__(f"HTTP {status} {message}".strip())
self.status = status
class UpstreamTooLarge(UpstreamError):
"""响应体超过大小上限。"""
@dataclass
class HTTPResult:
"""一次请求的完整结果(供需要状态码/头/cookie 的调用点使用)。"""
status: int
url: str
text: str
headers: Mapping[str, str] = field(default_factory=dict)
cookies: Mapping[str, str] = field(default_factory=dict)
def json(self) -> Any:
return json.loads(self.text)
class AsyncHTTPClient:
"""带连接池的异步 HTTP 客户端。"""
def __init__(
self,
*,
total_timeout: float = DEFAULT_TIMEOUT,
max_bytes: int = DEFAULT_MAX_BYTES,
pool_limit: int = 100,
pool_per_host: int = 32,
user_agent: str | None = None,
referer: str | None = None,
):
self._total_timeout = total_timeout
self._max_bytes = max_bytes
self._pool_limit = pool_limit
self._pool_per_host = pool_per_host
self._default_headers: dict[str, str] = {}
if user_agent:
self._default_headers['User-Agent'] = user_agent
if referer:
self._default_headers['Referer'] = referer
self._session: aiohttp.ClientSession | None = None
self._loop_sessions: dict[int, aiohttp.ClientSession] = {}
self._lock = asyncio.Lock()
async def _session_get(self) -> aiohttp.ClientSession:
if self._session is not None and not self._session.closed:
return self._session
async with self._lock:
if self._session is None or self._session.closed:
connector = aiohttp.TCPConnector(
limit=self._pool_limit,
limit_per_host=self._pool_per_host,
ttl_dns_cache=300,
)
self._session = aiohttp.ClientSession(
connector=connector,
timeout=aiohttp.ClientTimeout(total=self._total_timeout),
headers=dict(self._default_headers),
)
return self._session
async def _session_for_loop(self) -> aiohttp.ClientSession:
"""按事件循环隔离 session。
迁移期同一进程内可能有两个 loop(FastAPI 主 loop + 遗留 Flask 的同步桥 loop)。
aiohttp 的 ClientSession 不能跨 loop 复用,故按 loop 分别持有。
"""
key = id(asyncio.get_running_loop())
sess = self._loop_sessions.get(key)
if sess is not None and not sess.closed:
return sess
async with self._lock:
sess = self._loop_sessions.get(key)
if sess is None or sess.closed:
connector = aiohttp.TCPConnector(
limit=self._pool_limit,
limit_per_host=self._pool_per_host,
ttl_dns_cache=300,
)
sess = aiohttp.ClientSession(
connector=connector,
timeout=aiohttp.ClientTimeout(total=self._total_timeout),
headers=dict(self._default_headers),
)
self._loop_sessions[key] = sess
return sess
async def close(self) -> None:
if self._session is not None and not self._session.closed:
await self._session.close()
self._session = None
for sess in self._loop_sessions.values():
if not sess.closed:
await sess.close()
self._loop_sessions.clear()
# ---------- 底层:单次请求(不重试) ----------
async def _once(
self,
method: str,
url: str,
*,
headers: Mapping[str, str] | None = None,
cookies: Mapping[str, str] | None = None,
params: Mapping[str, Any] | None = None,
data: Any = None,
json_body: Any = None,
timeout: float | None = None,
allow_redirects: bool = True,
) -> HTTPResult:
session = await self._session_for_loop()
req_timeout = aiohttp.ClientTimeout(total=timeout) if timeout else None
try:
async with session.request(
method,
url,
headers=dict(headers) if headers else None,
cookies=dict(cookies) if cookies else None,
params=params,
data=data,
json=json_body,
timeout=req_timeout,
allow_redirects=allow_redirects,
) as resp:
body = await resp.read()
if len(body) > self._max_bytes:
raise UpstreamTooLarge(f"上游响应体超过上限 {self._max_bytes} 字节")
if resp.status >= 400:
raise UpstreamHTTPError(resp.status, resp.reason or '')
return HTTPResult(
status=resp.status,
url=str(resp.url),
text=body.decode(resp.charset or 'utf-8', errors='replace'),
headers=dict(resp.headers),
cookies={k: v.value for k, v in resp.cookies.items()},
)
except aiohttp.ClientError as e:
raise UpstreamNetworkError(str(e)) from e
except TimeoutError as e:
raise UpstreamNetworkError('请求超时') from e
async def _with_retry(self, fn, *, attempts: int = 2, backoff: float = 1.0) -> HTTPResult:
"""网络异常/5xx 退避重试;4xx 与超限立即抛出不重试。"""
last: UpstreamError | None = None
for i in range(attempts):
try:
return await fn()
except UpstreamHTTPError as e:
if e.status >= 500 and i + 1 < attempts:
last = e
await asyncio.sleep(backoff)
continue
raise
except UpstreamNetworkError as e:
last = e
if i + 1 < attempts:
await asyncio.sleep(backoff)
continue
raise
# 理论上不可达
raise last or UpstreamError('上游请求失败')
# ---------- 高层 API ----------
async def request(self, method: str, url: str, **kw) -> HTTPResult:
return await self._with_retry(lambda: self._once(method, url, **kw))
async def get_text(self, url: str, **kw) -> str:
return (await self.request('GET', url, **kw)).text
async def post_text(self, url: str, **kw) -> str:
return (await self.request('POST', url, **kw)).text
async def get_json(self, url: str, **kw) -> Any:
return (await self.request('GET', url, **kw)).json()
async def post_json(self, url: str, **kw) -> Any:
return (await self.request('POST', url, **kw)).json()
@asynccontextmanager
async def stream(
self,
method: str,
url: str,
*,
headers: Mapping[str, str] | None = None,
cookies: Mapping[str, str] | None = None,
timeout: float | None = None,
allow_redirects: bool = True,
) -> AsyncIterator[aiohttp.ClientResponse]:
"""流式请求(下载/音视频代理用):yield 原始响应,边读边写不整段入内存。"""
session = await self._session_for_loop()
req_timeout = aiohttp.ClientTimeout(total=timeout) if timeout else None
try:
async with session.request(
method,
url,
headers=dict(headers) if headers else None,
cookies=dict(cookies) if cookies else None,
timeout=req_timeout,
allow_redirects=allow_redirects,
) as resp:
if resp.status >= 400:
raise UpstreamHTTPError(resp.status, resp.reason or '')
yield resp
except aiohttp.ClientError as e:
raise UpstreamNetworkError(str(e)) from e
# 进程级共享客户端(UA/Referer 由 music_api 在初始化后注入)
client = AsyncHTTPClient()
+4 -4
View File
@@ -5,16 +5,16 @@
改为**启动时注入**(`init_secret_key`),从而解除对本模块对 Flask app 的依赖;
签名/校验算法与口径保持逐字一致,迁移期新旧应用共用同一密钥即可互认 token。
"""
import base64
import hashlib
import hmac
import secrets
import time
from typing import Any, Optional
from typing import Any
from core.config import AUDIO_HOST_MARKS, AUDIO_TOKEN_TTL, AUDIO_URL_KEYS
# 由应用启动时注入(来源:auth_system.get_secret_key() → app.secret_key)
_SECRET_KEY: bytes = b''
@@ -36,11 +36,11 @@ def sign_payload(payload: str) -> str:
def make_audio_token(url: str) -> str:
exp = int(time.time()) + AUDIO_TOKEN_TTL
nonce = secrets.token_hex(4)
payload = base64.urlsafe_b64encode(f'{exp}|{nonce}|{url}'.encode('utf-8')).decode().rstrip('=')
payload = base64.urlsafe_b64encode(f'{exp}|{nonce}|{url}'.encode()).decode().rstrip('=')
return payload + '.' + sign_payload(payload)
def verify_audio_token(token: str) -> Optional[str]:
def verify_audio_token(token: str) -> str | None:
try:
payload, sig = token.rsplit('.', 1)
if not hmac.compare_digest(sig, sign_payload(payload)):
+10 -10
View File
@@ -12,14 +12,14 @@ Flask/FastAPI 迁移期共用,并为后续封装成带锁的单例(State 对
依赖仅限标准库 threading/collections——任何下载器/HTTP 相关对象
(如批量打包信号量)不放在这里,随对应服务模块搬迁。
"""
import collections
import threading
from typing import Any, Dict, List
from typing import Any
# ============ 短链展开结果缓存 ============
# 用户反复解析同一分享链接不再重复请求 163cn.tv(有界防膨胀)
SHORTLINK_CACHE: Dict[str, str] = {}
SHORTLINK_CACHE: dict[str, str] = {}
SHORTLINK_CACHE_LOCK = threading.Lock()
@@ -37,23 +37,23 @@ BATCH_DURATIONS_LOCK = threading.Lock()
# ============ 下载任务表 ============
# (批量打包 + 单曲下载统一):task_id -> {type, total, done, failed, status, start, msg, cancel, pause, device_id, result, params}
# 前端通过 /api/batch_progress 轮询进度,支持取消;完成后文件按 35 分钟保留期清理
DOWNLOAD_TASKS: Dict[str, Dict[str, Any]] = {}
DOWNLOAD_TASKS: dict[str, dict[str, Any]] = {}
DOWNLOAD_TASKS_LOCK = threading.Lock()
# ============ 设备在线心跳 ============
# device_id -> 最近心跳时间。心跳超时(PAUSE_AFTER_SECONDS)未上线的设备,其批量任务
# 自动暂停让出排队名额;超时超过 CANCEL_AFTER_SECONDS(30 分钟)则自动取消任务并删文件
DEVICE_LAST_SEEN: Dict[str, float] = {}
DEVICE_LAST_SEEN: dict[str, float] = {}
DEVICE_LAST_SEEN_LOCK = threading.Lock()
# ============ 接口请求统计 ============
REQUEST_STATS: Dict[str, Dict[str, Any]] = {}
REQUEST_STATS: dict[str, dict[str, Any]] = {}
STATS_LOCK = threading.Lock()
# 用户解析/下载操作记录(IP);仅保留最近 2000 条
REQ_LOG: List[Dict[str, Any]] = []
REQ_LOG: list[dict[str, Any]] = []
# ============ 上游探测缓存 ============
@@ -62,10 +62,10 @@ GLOBAL_MUSIC_U_CACHE = {'v': None, 'ts': 0.0}
# 服务器网易云账号昵称缓存(10 分钟):非管理员可见响应中遮蔽真实昵称
SERVER_NICK_CACHE = {'nick': None, 'ts': 0.0}
SERVER_NICK_LOCK = threading.Lock() # 第 91 条(P3-9):刷新单飞锁,防并发重复打上游
VIP_PROBE_LOCK = threading.Lock() # 第 100 条:音质探测单飞锁(有界等待)
SERVER_NICK_LOCK = threading.Lock() # 第 91 条(P3-9):刷新单飞锁,防并发重复打上游
VIP_PROBE_LOCK = threading.Lock() # 第 100 条:音质探测单飞锁(有界等待)
# 全局账号 VIP 探测缓存(5 分钟):probe 一次要打 8 次网易云接口,
# 公网下账号栏高频刷新会连累服务器账号触发风控。
VIP_CACHE = {'data': None, 'ts': 0.0}
VIP_USER_CACHE = {} # user 路径按 cookie 哈希的 5 分钟缓存
VIP_USER_CACHE = {} # user 路径按 cookie 哈希的 5 分钟缓存
+6 -5
View File
@@ -4,10 +4,10 @@
自 main.py 原样迁出:安全整数解析、控制台输出兜底、SRT 时间轴、
LRC→SRT 转换、字段参数解析、163cn.tv 短链展开(含私有缓存)。
"""
import re
import sys
import threading
from typing import Any, Dict, List, Optional
def safe_int(value, default: int) -> int:
@@ -34,7 +34,7 @@ def srt_time(t: float) -> str:
h = int(t // 3600)
m = int((t % 3600) // 60)
s = int(t % 60)
ms = int(round((t - int(t)) * 1000))
ms = round((t - int(t)) * 1000)
if ms >= 1000:
ms = 999
return f"{h:02d}:{m:02d}:{s:02d},{ms:03d}"
@@ -73,14 +73,14 @@ def format_lyric(lrc: str, lyric_format: str) -> str:
for i, (t, text) in enumerate(timed):
end = timed[i + 1][0] if i + 1 < len(timed) else t + 5
if end <= t:
end = t + 1 # 同时间多行:给 1 秒最小区间,不再生成倒挂区间
end = t + 1 # 同时间多行:给 1 秒最小区间,不再生成倒挂区间
out.append(f"{i + 1}\n{srt_time(t)} --> {srt_time(end)}\n{text}\n")
if untimed:
out.append('\n'.join(untimed))
return '\n'.join(out)
def parse_meta_fields(raw) -> Optional[List[str]]:
def parse_meta_fields(raw) -> list[str] | None:
"""解析写入字段参数,空返回 None(表示全部写入)"""
if raw is None or raw == '':
return None
@@ -93,7 +93,7 @@ def parse_meta_fields(raw) -> Optional[List[str]]:
# 短链展开结果缓存:用户反复解析同一分享链接不再重复请求 163cn.tv(有界防膨胀)
_SHORTLINK_CACHE: Dict[str, str] = {}
_SHORTLINK_CACHE: dict[str, str] = {}
_SHORTLINK_CACHE_LOCK = threading.Lock()
@@ -106,6 +106,7 @@ def expand_shortlink(url: str) -> str:
expanded = url
try:
import requests
response = requests.get(url, allow_redirects=False, timeout=10)
expanded = response.headers.get('Location', url)
except Exception: