From 9f56f68b6697ce6c80cf9a953a053a561a539fe7 Mon Sep 17 00:00:00 2001 From: snowy-yang <102860110+snowy-yang@users.noreply.github.com> Date: Sat, 26 Sep 2026 00:24:00 +0800 Subject: [PATCH] =?UTF-8?q?chore(async-infra):=20=E5=BC=82=E6=AD=A5?= =?UTF-8?q?=E5=9F=BA=E7=A1=80=E5=B1=82=20+=20=E5=B7=A5=E5=85=B7=E9=93=BE?= =?UTF-8?q?=20+=20=E6=B8=85=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 异步基础层（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。 --- _csscheck.py | 10 -- core/config.py | 9 +- core/db.py | 137 ++++++++++++++++++++ core/http_client.py | 259 ++++++++++++++++++++++++++++++++++++++ core/security.py | 8 +- core/state.py | 20 +-- core/utils.py | 11 +- pyproject.toml | 63 ++++++++++ requirements-dev.txt | 6 + requirements-fastapi.txt | 17 ++- tests/test_db.py | 79 ++++++++++++ tests/test_http_client.py | 113 +++++++++++++++++ web/app.py | 3 + web/main.py | 1 + web/response.py | 3 +- web/routers/system.py | 9 +- 16 files changed, 706 insertions(+), 42 deletions(-) delete mode 100644 _csscheck.py create mode 100644 core/db.py create mode 100644 core/http_client.py create mode 100644 pyproject.toml create mode 100644 requirements-dev.txt create mode 100644 tests/test_db.py create mode 100644 tests/test_http_client.py diff --git a/_csscheck.py b/_csscheck.py deleted file mode 100644 index 1a99e8e..0000000 --- a/_csscheck.py +++ /dev/null @@ -1,10 +0,0 @@ -# -*- coding: utf-8 -*- -"""CSS 完整性检查：花括号平衡 + 关键美化标记存活""" -import sys -src = open('templates/index.html', encoding='utf-8').read() -style = src.split('')[0] -ok = style.count('{') == style.count('}') -marks = sum(1 for m in ['font-family: inherit', 'kw.on { text-shadow', 'uc-card:hover', 'theme-fading', 'textarea.input', 'prefers-reduced-motion', '美化层 v4', '美化层 v5', 'safe-area-inset'] if m in style) -with open('_css_ok.txt', 'w', encoding='utf-8') as f: - f.write(f'braces={style.count("{")}=={style.count("}")} {ok} | marks {marks}/9') -print('ok') diff --git a/core/config.py b/core/config.py index f3cb186..cfb81e0 100644 --- a/core/config.py +++ b/core/config.py @@ -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') diff --git a/core/db.py b/core/db.py new file mode 100644 index 0000000..be3b081 --- /dev/null +++ b/core/db.py @@ -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() diff --git a/core/http_client.py b/core/http_client.py new file mode 100644 index 0000000..2c8486c --- /dev/null +++ b/core/http_client.py @@ -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() diff --git a/core/security.py b/core/security.py index 2143a5a..5553a62 100644 --- a/core/security.py +++ b/core/security.py @@ -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)): diff --git a/core/state.py b/core/state.py index abc623e..a9bd80b 100644 --- a/core/state.py +++ b/core/state.py @@ -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 分钟缓存 diff --git a/core/utils.py b/core/utils.py index edfe353..e45923b 100644 --- a/core/utils.py +++ b/core/utils.py @@ -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: diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..42b934c --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,63 @@ +# -*- coding: utf-8 -*- +[project] +name = "yunmusic-assistant" +version = "3.0.0" +description = "云村小助手 · 网易云音乐解析/下载服务（FastAPI + asyncio + aiohttp）" +requires-python = ">=3.12" + +# ============ Ruff：静态检查 + 导入排序 + 格式化 ============ +[tool.ruff] +target-version = "py312" +line-length = 120 +extend-exclude = [".venv", ".zcode", "templates", "static", "data"] + +[tool.ruff.lint] +# E/F 基础错误、I 导入排序、B bugbear、UP 升级、ASYNC 异步陷阱、S 安全、RUF +select = ["E", "F", "W", "I", "B", "UP", "C4", "ASYNC", "S", "RUF"] +ignore = [ + "E501", # 行宽交给 formatter，不强制 + "E402", # main.py 存在「先定义后 import」的历史结构，迁移期暂放行 + "S101", # 生产代码外的 assert（测试/断言脚本） + "S104", # 绑定 0.0.0.0（服务端有意为之） + "S110", # try/except/pass（本项目大量「失败静默降级」策略，见交接文档） + "S112", # try/except/continue + "S324", # hashlib.md5 用于非安全场景（上游签名口径） + "S603", "S607", # subprocess（打包/运维脚本） + "UP009", # 保留 `# -*- coding: utf-8 -*-` 文件头（全项目既有约定） + "RUF001", "RUF002", "RUF003", # 全角字符（中文注释/文案，误报） +] + +[tool.ruff.lint.per-file-ignores] +"tests/**" = ["S", "ASYNC"] +".zcode/**" = ["S", "ASYNC", "F"] +"_mkpkg.py" = ["S"] +# ASYNC109：timeout 参数是有意 API——直接转发给 aiohttp 的 ClientTimeout， +# 不是「用 sleep 当超时」的反模式。 +"core/http_client.py" = ["ASYNC109"] + +[tool.ruff.lint.isort] +known-first-party = ["core", "web", "music_api", "auth_system"] + +[tool.ruff.format] +quote-style = "preserve" + +# ============ Pyright：类型检查 ============ +[tool.pyright] +include = ["core", "web"] +exclude = [".venv", ".zcode", "data", "downloads", "templates", "static"] +venvPath = "." +venv = ".venv" +pythonVersion = "3.12" +typeCheckingMode = "basic" +# 迁移期口径：存量 Flask 巨石（main.py/auth_*.py）暂不纳入严格检查， +# 新写的 core/ 与 web/ 要求类型干净。随迁移推进逐步把模块移入 include。 +reportMissingImports = true +reportOptionalMemberAccess = "warning" +reportAttributeAccessIssue = "warning" +reportGeneralTypeIssues = "warning" + +# ============ Pytest ============ +[tool.pytest.ini_options] +asyncio_mode = "auto" +testpaths = ["tests"] +addopts = "-q" diff --git a/requirements-dev.txt b/requirements-dev.txt new file mode 100644 index 0000000..32e5cfc --- /dev/null +++ b/requirements-dev.txt @@ -0,0 +1,6 @@ +# 开发依赖（开发/CI 专用，不进生产镜像） +pytest>=8.0 +pytest-asyncio>=0.24 +ruff>=0.6 +pyright>=1.1.390 +httpx>=0.27 diff --git a/requirements-fastapi.txt b/requirements-fastapi.txt index bfdb4ae..ab4622c 100644 --- a/requirements-fastapi.txt +++ b/requirements-fastapi.txt @@ -1,5 +1,14 @@ -# FastAPI 迁移目标依赖（当前生产仍以 Flask + waitress 运行，本文件不影响既有部署） -# 待阶段 C 切换生产服务器后，合并进 requirements.txt 并移除 Flask 相关项。 -# 版本上界口径同 requirements.txt（第 95 条 P2-6：防上游大版本静默破坏）。 +# 生产运行时依赖（FastAPI 版） +# 口径同旧 requirements.txt：加版本上界，防上游大版本静默破坏（第 95 条 P2-6）。 +# +# 迁移说明：本文件是 FASTAPI 目标栈。当前仓库同时保留 requirements.txt（Flask 旧栈） +# 直到路由全部迁移完成、生产服务器切换为止，两者可并存。 fastapi>=0.115,<1.0 -uvicorn>=0.30,<1.0 +uvicorn[standard]>=0.30,<1.0 +aiohttp>=3.10,<4.0 +aiosqlite>=0.20,<1.0 +aiofiles>=24.1 +itsdangerous>=2.2,<3.0 +cryptography>=50.0,<51.0 +qrcode>=7.0,<8.3 +python-multipart>=0.0.9 diff --git a/tests/test_db.py b/tests/test_db.py new file mode 100644 index 0000000..30af5d0 --- /dev/null +++ b/tests/test_db.py @@ -0,0 +1,79 @@ +# -*- coding: utf-8 -*- +"""异步 SQLite 层测试：读写并发、WAL、事务原子性。""" + +import asyncio + +import pytest + +import core.db as db + + +@pytest.fixture() +def temp_db(tmp_path, monkeypatch): + """把 DB 指向临时目录，独立于生产 data/site.db。""" + monkeypatch.setattr(db, 'DATA_DIR', tmp_path) + monkeypatch.setattr(db, 'DB_PATH', tmp_path / 'site.db') + return tmp_path / 'site.db' + + +async def test_write_then_read(temp_db): + await db.execute('CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT)') + rid = await db.execute('INSERT INTO t (v) VALUES (?)', ('hello',)) + assert rid == 1 + row = await db.fetch_one('SELECT v FROM t WHERE id=?', (1,)) + assert row['v'] == 'hello' + assert await db.fetch_val('SELECT COUNT(*) FROM t') == 1 + + +async def test_wal_mode_enabled(temp_db): + await db.execute('CREATE TABLE t (id INTEGER PRIMARY KEY)') + mode = await db.fetch_val('PRAGMA journal_mode') + assert str(mode).lower() == 'wal' + + +async def test_concurrent_read_write(temp_db): + """核心诉求：写的同时读不被阻塞、且读到一致快照（WAL 快照隔离）。""" + await db.execute('CREATE TABLE nums (id INTEGER PRIMARY KEY, v INTEGER)') + + stop = asyncio.Event() + read_errors = [] + write_count = 0 + + async def writer(): + nonlocal write_count + for i in range(200): + await db.execute('INSERT INTO nums (v) VALUES (?)', (i,)) + write_count += 1 + stop.set() + + async def reader(): + # 写入过程中持续读，不应抛 SQLITE_BUSY / 不应被阻塞致死 + while not stop.is_set(): + try: + n = await db.fetch_val('SELECT COUNT(*) FROM nums') + assert isinstance(n, int) + except Exception as e: + read_errors.append(e) + await asyncio.sleep(0) + + await asyncio.gather(writer(), *[reader() for _ in range(4)]) + + assert not read_errors, f'读期间出现异常: {read_errors[:3]}' + assert write_count == 200 + assert await db.fetch_val('SELECT COUNT(*) FROM nums') == 200 + + +async def test_write_transaction_commit_and_rollback(temp_db): + await db.execute('CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT)') + + async with db.write_transaction() as conn: + await conn.execute('INSERT INTO t (v) VALUES (?)', ('a',)) + await conn.execute('INSERT INTO t (v) VALUES (?)', ('b',)) + assert await db.fetch_val('SELECT COUNT(*) FROM t') == 2 + + with pytest.raises(RuntimeError): + async with db.write_transaction() as conn: + await conn.execute('INSERT INTO t (v) VALUES (?)', ('c',)) + raise RuntimeError('boom') + # 回滚：c 不应落库 + assert await db.fetch_val('SELECT COUNT(*) FROM t') == 2 diff --git a/tests/test_http_client.py b/tests/test_http_client.py new file mode 100644 index 0000000..191b5ac --- /dev/null +++ b/tests/test_http_client.py @@ -0,0 +1,113 @@ +# -*- coding: utf-8 -*- +"""异步 HTTP 客户端测试：连接复用、有界重试、4xx 不重试、响应体上限。""" + +import pytest +from aiohttp import web + +from core.http_client import AsyncHTTPClient, UpstreamHTTPError, UpstreamTooLarge + + +@pytest.fixture() +async def server(): + """本地 aiohttp 测试服务器。""" + hits = {'flaky': 0} + + async def ok(request): + return web.json_response({'ok': True}) + + async def flaky(request): + # 首次 500，之后 200 —— 验证 5xx 退避重试 + hits['flaky'] += 1 + if hits['flaky'] == 1: + return web.Response(status=500, text='boom') + return web.json_response({'retried': True}) + + async def notfound(request): + return web.Response(status=404, text='nope') + + async def big(request): + return web.Response(body=b'x' * (2 * 1024 * 1024)) + + app = web.Application() + app.router.add_get('/ok', ok) + app.router.add_get('/flaky', flaky) + app.router.add_get('/notfound', notfound) + app.router.add_get('/big', big) + + runner = web.AppRunner(app) + await runner.setup() + site = web.TCPSite(runner, '127.0.0.1', 0) + await site.start() + port = site._server.sockets[0].getsockname()[1] + base = f'http://127.0.0.1:{port}' + try: + yield base, hits + finally: + await runner.cleanup() + + +async def test_get_json(server): + base, _ = server + c = AsyncHTTPClient() + try: + assert await c.get_json(base + '/ok') == {'ok': True} + finally: + await c.close() + + +async def test_retry_on_5xx(server): + base, hits = server + c = AsyncHTTPClient() + try: + data = await c.get_json(base + '/flaky') + assert data == {'retried': True} + assert hits['flaky'] == 2 # 第一次 500 + 重试成功 + finally: + await c.close() + + +async def test_no_retry_on_4xx(server): + base, _ = server + c = AsyncHTTPClient() + try: + with pytest.raises(UpstreamHTTPError) as ei: + await c.get_text(base + '/notfound') + assert ei.value.status == 404 + finally: + await c.close() + + +async def test_response_size_limit(server): + base, _ = server + c = AsyncHTTPClient(max_bytes=1024 * 1024) # 1MB 上限，服务端返回 2MB + try: + with pytest.raises(UpstreamTooLarge): + await c.get_text(base + '/big') + finally: + await c.close() + + +async def test_session_reuse(server): + """同一客户端多次请求复用同一（按 loop 隔离的）session —— 连接池生效。""" + base, _ = server + c = AsyncHTTPClient() + try: + await c.get_json(base + '/ok') + s1 = await c._session_for_loop() + await c.get_json(base + '/ok') + s2 = await c._session_for_loop() + assert s1 is s2 and not s1.closed + finally: + await c.close() + + +async def test_stream(server): + base, _ = server + c = AsyncHTTPClient() + try: + async with c.stream('GET', base + '/ok') as resp: + assert resp.status == 200 + body = await resp.read() + assert b'ok' in body + finally: + await c.close() diff --git a/web/app.py b/web/app.py index 3c78612..5e62338 100644 --- a/web/app.py +++ b/web/app.py @@ -12,6 +12,7 @@ 两套独立进程语义。生产切换前需按计划把状态与任务模型统一（见迁移计划阶段 C）， 当前仅用于开发期联调。 """ + from fastapi import FastAPI from core.config import APP_VERSION @@ -40,5 +41,7 @@ def create_app(serve_flask_fallback: bool = False) -> FastAPI: def _mount_flask_fallback(app: FastAPI) -> None: """把遗留 Flask WSGI 应用挂到根路径，兜底未迁移的路由。""" from starlette.middleware.wsgi import WSGIMiddleware + import main as legacy_flask + app.mount('/', WSGIMiddleware(legacy_flask.app)) diff --git a/web/main.py b/web/main.py index f7ad6d0..7089bdc 100644 --- a/web/main.py +++ b/web/main.py @@ -4,6 +4,7 @@ 开发期启动： .venv/Scripts/python.exe web/main.py 生产切换前需先将状态/任务模型与既有 Flask 进程统一（见迁移计划阶段 C）。 """ + import os import sys diff --git a/web/response.py b/web/response.py index 7712ce9..ef75b08 100644 --- a/web/response.py +++ b/web/response.py @@ -7,6 +7,7 @@ FastAPI 侧改为直接返回 JSONResponse，但**包体结构完全一致**： error: {"status": sc, "success": false, "message": msg, ["error_code": ...]} 这样前端无需任何改动即可同时对接两个框架。 """ + from typing import Any from fastapi.responses import JSONResponse @@ -24,7 +25,7 @@ def success(data: Any = None, message: str = 'success', status_code: int = 200) return JSONResponse(content=body, status_code=status_code) -def error(message: str, status_code: int = 400, error_code: str = None) -> JSONResponse: +def error(message: str, status_code: int = 400, error_code: str | None = None) -> JSONResponse: """错误响应（对应 Flask APIResponse.error）""" body = { 'status': status_code, diff --git a/web/routers/system.py b/web/routers/system.py index af907eb..a3dde99 100644 --- a/web/routers/system.py +++ b/web/routers/system.py @@ -5,6 +5,7 @@ 此处实现与 main.py 的 health_check / api_info **响应契约逐字段对齐**， 用于验证双框架一致性（见 .zcode/refactor/parity.py）。 """ + import sys import time @@ -41,6 +42,7 @@ def health_check(): """健康检查API（对应 Flask health_check）""" try: from music_api import load_cookies + cookies = load_cookies() cookie_status = 'valid' if cookies else 'invalid' @@ -49,7 +51,7 @@ def health_check(): 'timestamp': int(time.time()) if 'time' in sys.modules else None, 'cookie_status': cookie_status, 'cookie_count': len(cookies), - 'downloads_dir': 'configured', # R5-P3：不向匿名访客回显服务器绝对路径 + 'downloads_dir': 'configured', # R5-P3：不向匿名访客回显服务器绝对路径 'version': APP_VERSION, } return success(health_info, "API服务运行正常") @@ -66,10 +68,7 @@ def api_info(): 'version': APP_VERSION, 'description': '云村小助手 · 网易云音乐解析/下载服务', 'endpoints': dict(_ENDPOINTS), - 'supported_qualities': [ - 'standard', 'exhigh', 'lossless', - 'hires', 'sky', 'jyeffect', 'jymaster', 'dolby' - ], + 'supported_qualities': ['standard', 'exhigh', 'lossless', 'hires', 'sky', 'jyeffect', 'jymaster', 'dolby'], 'config': { 'downloads_dir': 'configured', 'max_body_bytes': '2MB（备份上传 200MB）',