From 0df18f048ba969960f94bfdbf3a387686411f3c1 Mon Sep 17 00:00:00 2001 From: snowy-yang <102860110+snowy-yang@users.noreply.github.com> Date: Fri, 23 Oct 2026 23:40:55 +0800 Subject: [PATCH] =?UTF-8?q?fix(server):=20=E4=BF=AE=E5=A4=8D=E8=B7=AF?= =?UTF-8?q?=E7=94=B1=E9=81=AE=E8=94=BD=E3=80=81=E6=B5=81=E5=BC=8F=E8=83=8C?= =?UTF-8?q?=E5=8E=8B=E4=B8=8E=E9=89=B4=E6=9D=83=E9=99=90=E6=B5=81=E9=A1=BA?= =?UTF-8?q?=E5=BA=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 专辑搜索静态路由优先注册，避免被 /album/{source}/{id}/{page} 捕获 - mg 流改为异步背压队列，慢客户端不再 QueueFull 悬挂，错误中断响应 - 限流移到鉴权外层，401 爆破计入限流 - 酷我 token 重试移出锁，修复 asyncio.Lock 递归死锁 - 补 CORS、5sing 音质、parse _kind、wy 首次登录 cookie、stream 签名溢出 - mg 流直连、lifespan 关闭 aiohttp 会话、cookie 登录路由与 auth_store 健壮性 - 测试改用 aiohttp + 进程内 uvicorn，移除 httpx/TestClient - 更新 start.bat 与文档 --- README.md | 8 +- app/http/mg_stream.py | 10 +- app/http/stream_sign.py | 2 +- app/main.py | 33 ++-- app/routes/login.py | 41 ++++- app/routes/music.py | 167 ++++++++++-------- app/sdk/__init__.py | 13 -- app/sdk/apis/kw.py | 44 ++--- app/sdk/auth/wy.py | 11 +- app/sdk/auth_store.py | 70 +++++--- app/sdk/platforms/apple/__init__.py | 11 -- app/sdk/platforms/kg/__init__.py | 7 - docs/API.md | 8 +- docs/PORTING.md | 30 ++-- pyproject.toml | 1 - start.bat | 4 +- tests/conftest.py | 19 ++ tests/test_http.py | 257 ++++++++++++++++++++++++++++ uv.lock | 39 ----- 19 files changed, 534 insertions(+), 241 deletions(-) create mode 100644 tests/conftest.py create mode 100644 tests/test_http.py diff --git a/README.md b/README.md index 27773b5..88045cd 100644 --- a/README.md +++ b/README.md @@ -125,7 +125,9 @@ granian 也支持显式指定与默认不同的循环（如 Unix 上 `--loop uvl | `POST /login/me` | 绑定猫耳FM token（App 抓包或 m.missevan.com 复制，须含 `token=`，解锁付费声音取流） | | `POST /login/lr` | 懒人听书账号登录（account+password，懒人取流要求登录态，免费章节同样） | | `POST /login/bili/qr` → `POST /login/bili/qr/check` | 哔哩哔哩扫码登录（B站 App 扫一扫，登录后解锁大会员音质） | +| `POST /login/{mg,bd,joox,apple,5sing}` | 绑定对应平台 cookie（body `{"cookie": "..."}`；apple 需含 `media-user-token`、joox 通常需 `wmid`/`session_key`；服务端不做在线校验） | | `DELETE /login/{wy,kg,tx,xm,me,lr,bili}` | 清除对应平台凭据（退出登录） | +| `DELETE /login/{mg,bd,joox,apple,5sing}` | 清除对应平台 cookie 凭据 | | `GET /login/status` | 登录状态（登录后取流/请求自动携带凭据） | | `GET /sources` | 平台列表（音质与能力） | | `GET /health` | 健康检查 | @@ -186,7 +188,7 @@ app/ Python 应用（FastAPI，Python 3.12） config.py 环境变量、.env 加载、端口/鉴权/限流/数据目录 routes/ music.py 业务路由（搜索/取流/歌词/封面/榜单/歌单/专辑/歌手/解析/流代理） - login.py 登录路由（wy/kg/tx/bili 扫码与短信账密、xm/me cookie、lr 账密） + login.py 登录路由（wy/kg/tx/bili 扫码与短信账密、xm/me cookie、lr 账密、mg/bd/joox/apple/5sing cookie） refresh.py 登录态自动续期（启动后延时首轮 + 每 24h，kg/tx） system.py /health、/sources http/ @@ -214,7 +216,9 @@ docs/ PORTING.md TS → Python 平台移植契约（目录/子模块协议/签名差异/验收要求） pyproject.toml uv 项目定义（默认 granian/uvicorn 依赖、ruff、pyright、pytest 配置） tests/ - test_serve.py 启动器服务器/循环选择逻辑测试（TS 版 smoke 测试尚未移植） + conftest.py 测试隔离（临时 DATA_DIR，不触碰真实凭据） + test_serve.py 启动器服务器/循环选择逻辑测试 + test_http.py HTTP 回归（CORS/限流/路由/流式管道/登录存储；TS smoke 尚未全量移植） ``` > 分层依赖单向：`routes` → `sdk`，`sdk` 内部自包含。全源码 `.py` 由 uv 管理的 CPython 3.12 直接运行， diff --git a/app/http/mg_stream.py b/app/http/mg_stream.py index 02abecc..4f46500 100644 --- a/app/http/mg_stream.py +++ b/app/http/mg_stream.py @@ -15,7 +15,7 @@ from __future__ import annotations import hashlib import time import uuid -from collections.abc import Callable +from collections.abc import Awaitable, Callable from dataclasses import dataclass from urllib.parse import urljoin @@ -167,7 +167,7 @@ def migu_stream_content_type(url: str) -> str: async def stream_decrypted( url: str, feed: Callable[[bytes], bytes], - on_chunk: Callable[[bytes], None], + on_chunk: Callable[[bytes], Awaitable[None]], *, max_redirects: int = 3, ) -> None: @@ -175,12 +175,14 @@ async def stream_decrypted( 整段音频可达上百 MB，因此按块处理而不读入内存；上游 URL 常带 302， 这里手动跟随（默认不跟随重定向，与请求层语义一致）。 + on_chunk 为 async，调用方可用有界队列形成背压，避免慢客户端时丢块/死锁。 """ timeout = aiohttp.ClientTimeout(total=300, sock_read=60) headers = {"User-Agent": MIGU_STREAM_UA, "Referer": "https://music.migu.cn/"} current = url for _ in range(max_redirects + 1): - async with aiohttp.ClientSession(timeout=timeout) as session: + # trust_env=False：与 common/request 保持一致，不读系统代理（Windows 上 aiohttp 会走 IE 设置） + async with aiohttp.ClientSession(timeout=timeout, trust_env=False) as session: async with session.get(current, headers=headers, allow_redirects=False) as resp: if 300 <= resp.status < 400 and (location := resp.headers.get("location")): current = urljoin(current, location) @@ -190,6 +192,6 @@ async def stream_decrypted( async for chunk in resp.content.iter_any(): plain = feed(chunk) if plain: - on_chunk(plain) + await on_chunk(plain) return raise RuntimeError("重定向次数过多") diff --git a/app/http/stream_sign.py b/app/http/stream_sign.py index 00c3909..b3e772e 100644 --- a/app/http/stream_sign.py +++ b/app/http/stream_sign.py @@ -28,7 +28,7 @@ def verify_stream_sign(auth_key: str, path: str, exp: object, sig: object) -> bo """校验流地址签名（常量时间比较 + 时效窗口）。""" try: exp_num = int(float(str(exp))) - except (TypeError, ValueError): + except (TypeError, ValueError, OverflowError): return False sig_str = str(sig or "") if not sig_str: diff --git a/app/main.py b/app/main.py index 5b5f6c3..0f17f72 100644 --- a/app/main.py +++ b/app/main.py @@ -17,8 +17,9 @@ from collections.abc import AsyncIterator from contextlib import asynccontextmanager from typing import Any -from fastapi import APIRouter, FastAPI, Request +from fastapi import FastAPI, Request from fastapi.exceptions import RequestValidationError +from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse from starlette.exceptions import HTTPException as StarletteHTTPException @@ -28,6 +29,7 @@ from app.http.auth_gate import auth_gate from app.http.rate_limit import make_rate_limiter from app.openapi.components import COMPONENT_SCHEMAS from app.openapi.ui import redoc_html +from app.sdk.common.request import close_session API_TITLE = "MusicServer 聚合音乐 API" API_DESCRIPTION = ( @@ -38,21 +40,6 @@ API_DESCRIPTION = ( ) -class Envelope(JSONResponse): - """统一成功响应：{ code: 0, data }。文档里由路由的 responses 引用领域模型。""" - - def render(self, content: Any) -> bytes: - return json.dumps({"code": 0, "data": content}, ensure_ascii=False).encode("utf-8") - - -def ok(data: Any, status_code: int = 200) -> Envelope: - return Envelope(data, status_code=status_code) - - -def make_router(tag: str) -> APIRouter: - return APIRouter(tags=[tag]) - - @asynccontextmanager async def _lifespan(_app: FastAPI) -> AsyncIterator[None]: """启动期副作用 —— 登录态自动续期（启动后延时首刷 + 每 24h 一次，与 TS 版一致）。""" @@ -63,6 +50,7 @@ async def _lifespan(_app: FastAPI) -> AsyncIterator[None]: yield finally: await stop_refresh_scheduler() + await close_session() def create_app() -> FastAPI: @@ -89,9 +77,10 @@ def create_app() -> FastAPI: app.state.started_at = time.time() app.state.auth_key = config.auth_key() - # ---- 中间行为：限流 → 鉴权（限流先计数，401 爆破同样受限） ---- - app.middleware("http")(make_rate_limiter(config.rate_limit())) + # ---- 中间件顺序（后注册者在外层）：CORS → no-store → 限流 → 鉴权 ---- + # 限流必须在鉴权之外：错误秘钥的 401 也要计数，才能真正防爆破。 app.middleware("http")(auth_gate) + app.middleware("http")(make_rate_limiter(config.rate_limit())) @app.middleware("http") async def _no_store(request: Request, call_next): @@ -99,6 +88,14 @@ def create_app() -> FastAPI: response.headers["Cache-Control"] = "no-store" return response + app.add_middleware( + CORSMiddleware, + allow_origins=["*"], + allow_methods=["*"], + allow_headers=["*"], + max_age=600, + ) + # ---- 异常处理 ---- @app.exception_handler(ApiError) async def _api_error(_: Request, exc: ApiError): diff --git a/app/routes/login.py b/app/routes/login.py index 126aefd..10abd8f 100644 --- a/app/routes/login.py +++ b/app/routes/login.py @@ -9,12 +9,12 @@ from __future__ import annotations from typing import Any -from app.http.api_error import ApiError, upstream +from app.http.api_error import ApiError, not_found, upstream from app.http.envelope import ok from app.routes.refresh import refresh_all from app.sdk import auth as auth_mod from app.sdk.auth_store import clear_credentials, get_credentials, get_tx_auth, set_credentials -from fastapi import APIRouter, Body +from fastapi import APIRouter, Body, Path router = APIRouter() @@ -331,6 +331,43 @@ async def login_refresh() -> Any: return ok(await refresh_all()) +# 这些平台没有独立登录协议，只能从浏览器/App 复制 cookie（README 平台表已声明 cookie 登录）。 +# 白名单限定，避免动态路由把任意平台写进凭据文件。 +_COOKIE_LOGIN_SOURCES = {"mg", "bd", "joox", "apple", "5sing"} + + +@router.post( + "/login/{source}", + tags=[LOGIN_TAG], + summary="绑定平台 cookie", + description="mg/bd/joox/apple/5sing 用：从浏览器或 App 复制完整 cookie 字符串并持久化。" + "apple 需包含 media-user-token；joox 通常需 wmid/session_key。服务端不做在线校验。", +) +async def cookie_bind( + source: str = Path(..., description="平台 id（mg/bd/joox/apple/5sing）", examples=["bd"]), + cookie: str = Body(..., embed=True, description="完整 cookie 字符串"), +) -> Any: + if source not in _COOKIE_LOGIN_SOURCES: + raise not_found(f"不支持的 cookie 登录平台: {source}") + value = str(cookie or "").strip() + if not value: + raise ApiError(422, "cookie 不能为空") + set_credentials(source, {"cookie": value}) + return ok({"logged": True}) + + +@router.delete( + "/login/{source}", + tags=[LOGIN_TAG], + summary="退出平台登录（cookie 平台）", +) +async def cookie_logout(source: str = Path(..., description="平台 id（mg/bd/joox/apple/5sing）")) -> Any: + if source not in _COOKIE_LOGIN_SOURCES: + raise not_found(f"不支持的 cookie 登录平台: {source}") + clear_credentials(source) + return ok({"logged": False}) + + def _stored_status(package: str) -> dict[str, Any]: """按已存凭据判断登录态（xm/me/bd/joox/apple/5sing 用）。""" cred = get_credentials(package) diff --git a/app/routes/music.py b/app/routes/music.py index 59bab9f..3dd9871 100644 --- a/app/routes/music.py +++ b/app/routes/music.py @@ -6,9 +6,12 @@ from __future__ import annotations +import asyncio +import contextlib +from collections.abc import Callable from typing import Any -from app.http.api_error import not_found, upstream +from app.http.api_error import ApiError, not_found, upstream from app.http.envelope import ok from app.http.mg_stream import ( Z3DStreamDecoder, @@ -18,7 +21,7 @@ from app.http.mg_stream import ( take_migu_stream, ) from app.http.stream_sign import sign_stream -from app.sdk import SOURCE_MODULES, SUPPORTED_SOURCES, find_music +from app.sdk import SOURCE_MODULES, SUPPORTED_SOURCES, find_music, support_quality from app.sdk.types import cap as get_cap from fastapi import APIRouter, Body, Path, Query, Request from fastapi.responses import StreamingResponse @@ -27,15 +30,13 @@ router = APIRouter() QUALITIES = ["64k", "128k", "320k", "flac", "flac24bit"] + +def _allowed_qualities(source: str) -> list[str]: + """通用档位 + 平台自定义档位（如 5sing 的 sq/hq/lq）。""" + return list(dict.fromkeys([*QUALITIES, *support_quality.get(source, [])])) + + # ---- 常用 OpenAPI 响应片段 ---- -ENVELOPE_SCHEMA = { - "type": "object", - "properties": { - "code": {"type": "integer", "description": "0 表示成功", "example": 0}, - "data": {"description": "业务数据"}, - }, - "required": ["code", "data"], -} def envelope_ref(name: str) -> dict[str, Any]: @@ -75,7 +76,7 @@ def get_source(source: str, capability: str | None = None) -> Any: def platform_path(source: str, capability: str | None = None, description: str = "平台 id（见 /sources）") -> Any: - return Path(..., description=f"{description}{f'，需支持 {capability} 能力' if capability else ''}", example="kw") + return Path(..., description=f"{description}{f'，需支持 {capability} 能力' if capability else ''}", examples=["kw"]) # ---------- 搜索 ---------- @@ -125,8 +126,8 @@ async def search_platform(platform: str, word: str, page: int, limit: int) -> li ) async def search_first( source: str = platform_path("kw", "search"), - keyword: str = Path(..., description="搜索关键词（URL 编码的中文可直接使用）", example="稻香"), - singer: str | None = Query(None, description="歌手名（与关键词拼接搜索）", example="周杰伦"), + keyword: str = Path(..., description="搜索关键词（URL 编码的中文可直接使用）", examples=["稻香"]), + singer: str | None = Query(None, description="歌手名（与关键词拼接搜索）", examples=["周杰伦"]), limit: int = Query(20, ge=1, le=100, description="每页条数"), ) -> Any: return ok(await _search_handler(source, keyword, 1, singer, limit)) @@ -141,7 +142,7 @@ async def search_first( ) async def search_page( source: str = platform_path("kw", "search"), - keyword: str = Path(..., description="搜索关键词", example="稻香"), + keyword: str = Path(..., description="搜索关键词", examples=["稻香"]), page: int = Path(..., ge=1, description="页码"), singer: str | None = Query(None, description="歌手名"), limit: int = Query(20, ge=1, le=100, description="每页条数"), @@ -155,15 +156,15 @@ async def search_page( description="按歌手、歌名、专辑、时长综合排序的精确匹配。source 为平台 id（只在单平台内匹配）或 all" "（跨平台聚合，可用 excludeSource 跳过某平台；聚合自动排除 xm/me/qt/lr 有声平台）。", tags=["搜索"], - responses=responses_for("FindResult", not_found=False), + responses=responses_for("FindResult"), ) async def find( - source: str = Path(..., description="平台 id 或 all", example="all"), - keyword: str = Path(..., description="歌名", example="稻香"), - singer: str | None = Query(None, description="歌手名", example="周杰伦"), - albumName: str | None = Query(None, description="专辑名", example="魔杰座"), - interval: str | None = Query(None, description="时长（mm:ss），用于时长校验", example="03:43"), - excludeSource: str | None = Query(None, description="source=all 时跳过该平台", example="kw"), + source: str = Path(..., description="平台 id 或 all", examples=["all"]), + keyword: str = Path(..., description="歌名", examples=["稻香"]), + singer: str | None = Query(None, description="歌手名", examples=["周杰伦"]), + albumName: str | None = Query(None, description="专辑名", examples=["魔杰座"]), + interval: str | None = Query(None, description="时长（mm:ss），用于时长校验", examples=["03:43"]), + excludeSource: str | None = Query(None, description="source=all 时跳过该平台", examples=["kw"]), ) -> Any: kwargs: dict[str, Any] = { "name": keyword, @@ -172,8 +173,10 @@ async def find( "interval": interval, } if source != "all": + get_source(source, "music_search") kwargs["only"] = source elif excludeSource: + get_source(excludeSource, "music_search") kwargs["source"] = excludeSource try: return ok({"list": await find_music(**kwargs)}) @@ -198,17 +201,16 @@ async def find( async def music_url( request: Request, source: str = platform_path("kw", "url"), - id: str = Path(..., description="平台歌曲 id", example="28423286"), + id: str = Path(..., description="平台歌曲 id", examples=["28423286"]), quality: str = Query("128k", description="音质"), albumId: str | None = Query(None, description="专辑 id"), name: str | None = Query(None, description="歌名（酷我兜底匹配用）"), singer: str | None = Query(None, description="歌手名（酷我兜底匹配用）"), - interval: str | None = Query(None, description="时长 mm:ss（酷我兜底匹配用）", example="04:17"), + interval: str | None = Query(None, description="时长 mm:ss（酷我兜底匹配用）", examples=["04:17"]), ) -> Any: - if quality not in QUALITIES: - from app.http.api_error import ApiError - - raise ApiError(422, f"quality: 不支持的音质 {quality}（可选：{', '.join(QUALITIES)}）") + allowed = _allowed_qualities(source) + if quality not in allowed: + raise ApiError(422, f"quality: 不支持的音质 {quality}（{source} 可选：{', '.join(allowed)}）") module = get_source(source, "get_music_url") song_info = {"songmid": id, "hash": id, "copyrightId": id, "albumId": albumId or "", "_types": {}} @@ -226,7 +228,8 @@ async def music_url( if auth_key: path = raw_url.split("?")[0] suffix = f"{'&' if '?' in raw_url else '?'}{sign_stream(auth_key, path)}" - raw_url = f"{request.url.scheme}://{request.headers.get('host', '')}{raw_url}{suffix}" + base = str(request.base_url).rstrip("/") + raw_url = f"{base}{raw_url}{suffix}" payload: dict[str, Any] = {"type": result.get("type") or quality, "url": raw_url} if via != source: payload["via"] = via @@ -297,29 +300,36 @@ def _stream_response(url: str, decoder: Z3DStreamDecoder) -> StreamingResponse: """异步队列桥接：下载协程生产解密块，StreamingResponse 消费块。 解密按块增量进行（跨块保持全局偏移），整段音频可达上百 MB，不能读进内存。 + 队列有界并由生产者 ``await queue.put`` 形成背压；客户端断开时整个任务被取消。 + 上游/解密失败会作为异常项传给消费者并中断响应，避免把截断音频伪装成正常结束。 """ - import asyncio - - queue: asyncio.Queue[bytes | None] = asyncio.Queue(maxsize=32) + queue: asyncio.Queue[bytes | Exception | None] = asyncio.Queue(maxsize=32) async def produce() -> None: try: - await stream_decrypted(url, decoder.feed, queue.put_nowait) - except Exception as err: # noqa: BLE001 - 响应头已发出，只能中断连接`r`n # 日志留痕，只能中断连接；把错误留在服务端日志 + await stream_decrypted(url, decoder.feed, queue.put) + except asyncio.CancelledError: + raise + except Exception as err: # noqa: BLE001 - 响应头已发出，只能把错误交给消费者中断连接 print(f"[stream] 上游读取中断: {err!r}") - finally: - queue.put_nowait(None) + await queue.put(err) + return + await queue.put(None) async def body(): task = asyncio.create_task(produce()) try: while True: - chunk = await queue.get() - if chunk is None: + item = await queue.get() + if item is None: break - yield chunk + if isinstance(item, Exception): + raise item + yield item finally: task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await task return StreamingResponse( body(), @@ -405,7 +415,7 @@ async def _board_handler(source: str, board_id: str, page: int) -> Any: async def board_first( source: str = platform_path("kw", "boards"), boardId: str = Path( - ..., description="榜单 id，兼容榜单列表返回的带前缀 id（kw__93）与裸 bangid（93）", example="kw__93" + ..., description="榜单 id，兼容榜单列表返回的带前缀 id（kw__93）与裸 bangid（93）", examples=["kw__93"] ), ) -> Any: return await _board_handler(source, boardId, 1) @@ -602,12 +612,38 @@ async def _album_detail(source: str, album_id: str, page: int, limit: int) -> An ) async def album_first( source: str = platform_path("kg", "album"), - id: str = Path(..., description="平台专辑 id（搜索结果中的 albumId）", example="123456"), + id: str = Path(..., description="平台专辑 id（搜索结果中的 albumId）", examples=["123456"]), limit: int = Query(20, ge=1, le=100, description="每页条数"), ) -> Any: return await _album_detail(source, id, 1, limit) +# 注意：静态段路由必须先于 `/{id}/{page}` 注册。Starlette 按注册顺序匹配， +# 否则 /album/{source}/search/{keyword} 会被 /album/{source}/{id}/{page} 捕获， +# keyword 落到 page 参数上直接 422。 +@router.get( + "/album/{source}/search/{keyword}", + summary="搜索专辑", + description="支持 kg/kw/tx/wy/mg/bd/joox/apple（见 /sources 的 albumSearch 能力位）。", + tags=["歌手专辑"], + responses=responses_for("AlbumDetailResult", not_found=False), +) +async def album_search( + source: str = platform_path("kg", "album.search_album"), + keyword: str = Path(..., description="专辑名关键词", examples=["魔杰座"]), + page: int = Query(1, ge=1, description="页码"), + limit: int = Query(20, ge=1, le=100, description="每页条数"), +) -> Any: + module = get_source(source, "album.search_album") + try: + res = await module.album.search_album(keyword, page, limit) + except Exception as err: + raise upstream(str(err)) from err + if isinstance(res, list): + return ok({"source": source, "list": res, "page": page, "limit": limit, "total": len(res)}) + return ok({"source": source, **res}) + + @router.get( "/album/{source}/{id}/{page}", summary="获取专辑详情与曲目列表（指定页）", @@ -623,29 +659,6 @@ async def album_page( return await _album_detail(source, id, page, limit) -@router.get( - "/album/{source}/search/{keyword}", - summary="搜索专辑", - description="支持 kg/kw/tx/wy/mg/bd/joox/apple（见 /sources 的 albumSearch 能力位）。", - tags=["歌手专辑"], - responses=responses_for("AlbumDetailResult", not_found=False), -) -async def album_search( - source: str = platform_path("kg", "album.search_album"), - keyword: str = Path(..., description="专辑名关键词", example="魔杰座"), - page: int = Query(1, ge=1, description="页码"), - limit: int = Query(20, ge=1, le=100, description="每页条数"), -) -> Any: - module = get_source(source, "album.search_album") - try: - res = await module.album.search_album(keyword, page, limit) - except Exception as err: - raise upstream(str(err)) from err - if isinstance(res, list): - return ok({"source": source, "list": res, "page": page, "limit": limit, "total": len(res)}) - return ok({"source": source, **res}) - - # ---------- 歌单 ---------- @@ -659,7 +672,7 @@ async def album_search( ) async def playlist_search( source: str = platform_path("kw", "playlist.search_playlist"), - keyword: str = Path(..., description="歌单关键词", example="华语"), + keyword: str = Path(..., description="歌单关键词", examples=["华语"]), page: int = Query(1, ge=1, description="页码"), limit: int = Query(20, ge=1, le=100, description="每页条数"), ) -> Any: @@ -713,7 +726,7 @@ async def _category_playlists(source: str, category_id: str, page: int, limit: i ) async def playlist_category_items( source: str = platform_path("kw", "playlist.get_categories"), - categoryId: str = Path(..., description="分类 id（分类列表返回）", example="华语"), + categoryId: str = Path(..., description="分类 id（分类列表返回）", examples=["华语"]), page: int = Query(1, ge=1, description="页码"), limit: int = Query(20, ge=1, le=100, description="每页条数"), ) -> Any: @@ -755,7 +768,7 @@ async def _playlist_detail(source: str, playlist_id: str, page: int, limit: int) ) async def playlist_first( source: str = platform_path("kw", "playlist"), - id: str = Path(..., description="歌单 id（或平台歌单链接）", example="123456789"), + id: str = Path(..., description="歌单 id（或平台歌单链接）", examples=["123456789"]), limit: int = Query(20, ge=1, le=100, description="每页条数"), ) -> Any: return await _playlist_detail(source, id, 1, limit) @@ -818,19 +831,23 @@ async def _parse_handler(kind: str, source: str, link: str) -> Any: _PARSE_KINDS = {"song": "歌曲", "album": "专辑", "playlist": "歌单"} +def _make_parse_handler(kind: str) -> Callable[..., Any]: + """闭包固定解析类型 —— 不能让 kind 出现在 handler 签名里，否则 FastAPI 会把它当 query 参数。""" + + async def handler( + source: str = platform_path("kw", f"parse.parse_{kind}"), + link: str = Body(..., embed=True, description="平台链接或裸 id"), + ) -> Any: + return await _parse_handler(kind, source, str(link).strip()) + + return handler + + def _register_parse_routes() -> None: for kind, label in _PARSE_KINDS.items(): - - async def handler( - source: str = platform_path("kw", "parse.parse_song"), - link: str = Body(..., embed=True, description="平台链接或裸 id"), - _kind: str = kind, - ) -> Any: - return await _parse_handler(_kind, source, str(link).strip()) - router.add_api_route( f"/parse/{kind}/{{source}}", - handler, + _make_parse_handler(kind), methods=["POST"], summary=f"解析{label}链接", description="把平台链接（或裸 id）解析成结构化歌曲/曲目数据。支持平台见 /sources 的 parse 能力位；" diff --git a/app/sdk/__init__.py b/app/sdk/__init__.py index a3576a4..970a5f4 100644 --- a/app/sdk/__init__.py +++ b/app/sdk/__init__.py @@ -25,7 +25,6 @@ __all__ = [ "find_music", "format_play_count", "format_play_time", - "init", "search_music", "size_formate", "sources", @@ -113,18 +112,6 @@ def _quality_table() -> dict[str, list[str]]: support_quality: dict[str, list[str]] = _quality_table() -async def init() -> None: - """调用各平台的 init 钩子（无 init 的跳过；单个失败不阻塞启动，与原 init 一致）。""" - for item in SUPPORTED_SOURCES: - hook = getattr(item["module"], "init", None) - if hook is None: - continue - try: - await hook() - except Exception as err: # noqa: BLE001 - print(f"[sdk] {item['id']} init 失败（已跳过）: {err}") - - def _trim_str(value: Any) -> Any: return value.strip() if isinstance(value, str) else value diff --git a/app/sdk/apis/kw.py b/app/sdk/apis/kw.py index 81ffb6e..f6a0d28 100644 --- a/app/sdk/apis/kw.py +++ b/app/sdk/apis/kw.py @@ -153,28 +153,32 @@ async def get_token(retry_num: int = 0) -> KwToken: cached = _state["token"] if cached is not None: return cached - try: - cookies = await fetch_home_set_cookie() - except Exception as err: - if retry_num >= 2: - raise RuntimeError(f"获取酷我token失败: {err}") from err - return await get_token(retry_num + 1) + # 重试必须在锁内用循环完成：asyncio.Lock 不可重入，锁内递归会永久死锁。 + last_error: Exception | None = None + for _attempt in range(max(0, retry_num), 3): + try: + cookies = await fetch_home_set_cookie() + except Exception as err: # noqa: BLE001 - 网络抖动重试，最终统一抛出 + last_error = err + continue - found: list[tuple[str, str]] = [] - for raw in cookies: - name, _, rest = raw.partition("=") - name = name.strip() - value = rest.split(";")[0] if rest else "" - if _TOKEN_NAME_RE.match(name) and value: - found.append((name, value)) - if not found: - if retry_num >= 2: - raise RuntimeError("获取酷我token失败") - return await get_token(retry_num + 1) + found: list[tuple[str, str]] = [] + for raw in cookies: + name, _, rest = raw.partition("=") + name = name.strip() + value = rest.split(";")[0] if rest else "" + if _TOKEN_NAME_RE.match(name) and value: + found.append((name, value)) + if not found: + last_error = RuntimeError("响应中没有 Hm_Iuvt_* cookie") + continue - token = KwToken(found[0][0], found[0][1]) - _state["token"] = token - return token + token = KwToken(found[0][0], found[0][1]) + _state["token"] = token + return token + + detail = f": {last_error}" if last_error else "" + raise RuntimeError(f"获取酷我token失败{detail}") from last_error async def get_url_by_playurl(songmid: str, token: KwToken, retry_num: int = 0) -> dict[str, Any]: diff --git a/app/sdk/auth/wy.py b/app/sdk/auth/wy.py index 9de7d09..f7eb37f 100644 --- a/app/sdk/auth/wy.py +++ b/app/sdk/auth/wy.py @@ -62,9 +62,9 @@ def _extract_login_cookie(set_cookie: list[str]) -> tuple[str, str]: return joined, match.group(1) if match else "" -async def _fetch_profile() -> dict[str, Any]: - """登录成功后拉取账号资料。""" - body, _ = await _request_weapi("/api/w/nuser/account/get", {}) +async def _fetch_profile(cookie: str | None = None) -> dict[str, Any]: + """登录成功后拉取账号资料（cookie 显式传入新登录会话，避免首次登录读到旧 cookie）。""" + body, _ = await _request_weapi("/api/w/nuser/account/get", {}, cookie) body = as_dict(body) account = as_dict(body.get("account")) profile = as_dict(body.get("profile")) @@ -81,7 +81,10 @@ async def _apply_login(set_cookie: list[str]) -> dict[str, Any]: cookie, music_u = _extract_login_cookie(set_cookie) if not music_u: raise RuntimeError("登录响应中缺少 MUSIC_U") - profile = await _fetch_profile() + try: + profile = await _fetch_profile(cookie) + except Exception: # noqa: BLE001 - 资料拉取失败不影响已成功的登录，返回空资料 + profile = {"userid": None, "nickname": None, "avatar": None, "vipType": None} set_credentials("wy", {"cookie": cookie, "musicU": music_u, **profile}) return profile diff --git a/app/sdk/auth_store.py b/app/sdk/auth_store.py index 2000d10..20bedaa 100644 --- a/app/sdk/auth_store.py +++ b/app/sdk/auth_store.py @@ -14,6 +14,7 @@ import os import threading import time from datetime import UTC +from pathlib import Path from typing import Any from app.config import data_dir @@ -21,41 +22,58 @@ from app.config import data_dir _lock = threading.Lock() _cache: dict[str, Any] = {} _loaded = False +_cache_path: str | None = None +_cache_mtime: int | None = None -def _data_file(): +def _data_file() -> Path: return data_dir() / "auth.json" -def _load() -> dict[str, Any]: - global _cache, _loaded - if _loaded: - return _cache - path = _data_file() - if not path.exists(): - _cache = {} - _loaded = True - return _cache +def _mtime(path: Path) -> int | None: try: - _cache = json.loads(path.read_text(encoding="utf-8")) + return path.stat().st_mtime_ns + except OSError: + return None + + +def _backup_corrupt(path: Path, err: Exception) -> None: + """损坏文件先备份再重置，避免后续 set_credentials 用空表覆盖可抢救的数据。""" + backup = path.with_name(f"{path.name}.corrupt-{int(time.time() * 1000)}") + try: + path.rename(backup) + print(f"[auth/store] {path} 解析失败，已备份为 {backup}，本次以空凭据启动: {err}") + except OSError: + print(f"[auth/store] {path} 解析失败且备份未成功，本次以空凭据启动: {err}") + + +def _load() -> dict[str, Any]: + """读取凭据缓存；文件被外部修改（mtime 变化）时自动重载。""" + global _cache, _loaded, _cache_path, _cache_mtime + path = _data_file() + mtime = _mtime(path) + if _loaded and _cache_path == str(path) and _cache_mtime == mtime: + return _cache + + try: + if mtime is None: + _cache = {} + else: + data = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(data, dict): + raise ValueError("auth.json 顶层必须是 JSON 对象") + _cache = data except (ValueError, OSError) as err: - # 解析失败绝不能静默清空：先改名备份，让残留凭据仍可人工抢救； - # 否则下一次 set_credentials 就会用空表覆盖掉本可恢复的文件。 - backup = path.with_name(f"{path.name}.corrupt-{int(time.time() * 1000)}") - try: - path.rename(backup) - print(f"[auth/store] {path} 解析失败，已备份为 {backup}，本次以空凭据启动: {err}") - except OSError: - print(f"[auth/store] {path} 解析失败且备份未成功，本次以空凭据启动: {err}") + _backup_corrupt(path, err) _cache = {} _loaded = True + _cache_path = str(path) + _cache_mtime = _mtime(path) return _cache def _save(data: dict[str, Any]) -> None: - global _cache, _loaded - _cache = data - _loaded = True + global _cache, _loaded, _cache_path, _cache_mtime path = _data_file() path.parent.mkdir(parents=True, exist_ok=True) tmp = path.with_name(f"{path.name}.{os.getpid()}.tmp") @@ -70,6 +88,10 @@ def _save(data: dict[str, Any]) -> None: except OSError: pass raise + _cache = data + _loaded = True + _cache_path = str(path) + _cache_mtime = _mtime(path) def get_credentials(platform: str) -> dict[str, Any] | None: @@ -79,7 +101,7 @@ def get_credentials(platform: str) -> dict[str, Any] | None: def set_credentials(platform: str, credentials: dict[str, Any]) -> None: with _lock: - data = _load() + data = dict(_load()) merged = {**data.get(platform, {}), **credentials, "updatedAt": _iso_now()} data[platform] = merged _save(data) @@ -87,7 +109,7 @@ def set_credentials(platform: str, credentials: dict[str, Any]) -> None: def clear_credentials(platform: str) -> None: with _lock: - data = _load() + data = dict(_load()) data.pop(platform, None) _save(data) diff --git a/app/sdk/platforms/apple/__init__.py b/app/sdk/platforms/apple/__init__.py index 8bc562a..cd42bad 100644 --- a/app/sdk/platforms/apple/__init__.py +++ b/app/sdk/platforms/apple/__init__.py @@ -23,7 +23,6 @@ from app.sdk.platforms.apple.util import ( extract_apple_id, fetch_apple_song_resource, get_apple_config, - get_apple_token, ) @@ -31,15 +30,6 @@ def _song_id_of(song_info: dict[str, Any]) -> str: return extract_apple_id(str(song_info.get("songmid") or song_info.get("id") or ""), "song") -async def _init() -> bool: - """预热 developer token（失败不阻塞启动，路由调用时会重试）。""" - try: - await get_apple_token() - return True - except Exception: # noqa: BLE001 - token 预热失败不影响后续请求的重试 - return False - - async def _get_music_url(song_info: dict[str, Any], quality: str, retry_num: int = 0) -> dict[str, Any]: from app.sdk.apis.apple import get_music_url @@ -120,7 +110,6 @@ def status() -> dict[str, Any]: MODULE = Platform( id="apple", name="Apple Music", - init=_init, music_search=music_search, album=_AppleAlbum, playlist=playlist, diff --git a/app/sdk/platforms/kg/__init__.py b/app/sdk/platforms/kg/__init__.py index 8a521f3..ce9291a 100644 --- a/app/sdk/platforms/kg/__init__.py +++ b/app/sdk/platforms/kg/__init__.py @@ -25,12 +25,6 @@ from app.sdk.platforms.kg import ( singer, song_list, ) -from app.sdk.platforms.kg.util import get_device - - -async def _init() -> None: - """预先固化设备指纹（取流签名依赖 dfid/mid，首次调用时生成并持久化）。""" - get_device() def get_music_detail_page_url(song_info: dict[str, Any]) -> str: @@ -61,5 +55,4 @@ MODULE = Platform( get_pic=pic.get_pic, get_music_info=get_music_info, get_music_detail_page_url=get_music_detail_page_url, - init=_init, ) diff --git a/docs/API.md b/docs/API.md index 72c00c4..9f3ce23 100644 --- a/docs/API.md +++ b/docs/API.md @@ -65,7 +65,7 @@ | 方法 | 路径 | 说明 | | --- | --- | --- | -| GET | `/url/:source/:id` | 播放地址；`?quality=`（64k/128k/320k/flac/flac24bit，默认 128k）。取流链：外部音源（启用时）→ SDK 直连 → 酷我兜底（响应 `via` 标记实际命中通道） | +| GET | `/url/:source/:id` | 播放地址；`?quality=`（通用 64k/128k/320k/flac/flac24bit，5sing 另有 sq/hq/lq，默认 128k）。取流链：SDK 直连 → 酷我兜底（响应 `via` 标记实际命中通道） | | GET | `/stream/:source/:id` | 加密流解密代理（`mg` 咪咕 Z3D 加密流）。供 `/url` 返回的相对地址使用，不接受自行构造 | | GET | `/lyric/:source/:id` | 歌词；可带 `?name=` `?interval=`（部分平台搜索歌词时需要） | | GET | `/pic/:source/:id` | 封面；可带 `?albumId=`（kg 需要） | @@ -125,11 +125,13 @@ | DELETE | `/login/lr` | 退出懒人听书 | | POST | `/login/bili/qr` → `/login/bili/qr/check` | 哔哩哔哩扫码（解锁大会员音质） | | DELETE | `/login/bili` | 退出哔哩哔哩 | +| POST | `/login/{mg,bd,joox,apple,5sing}` | 绑定对应平台 cookie（body `{"cookie": "..."}`，服务端不做在线校验） | +| DELETE | `/login/{mg,bd,joox,apple,5sing}` | 退出对应平台 | | POST | `/login/refresh` | 手动触发登录态续期（kg/tx） | | GET | `/login/status` | 登录状态总览（各平台 `logged` 与账号资料） | -- 凭据持久化于 `data/auth.json`（不入库，文件权限 0600），登录后取流与平台请求自动携带。 -- 自动续期：启动 10s 后对 kg/tx 执行一次，此后每 24h 一次（音源模式下不参与取流，自动续期不启动）。 +- 凭据持久化于 `data/auth.json`（不入库，文件权限 0600），登录后取流与平台请求自动携带；文件被外部修改时会自动重载。 +- 自动续期：启动 10s 后对 kg/tx 执行一次，此后每 24h 一次。 ## 响应示例 diff --git a/docs/PORTING.md b/docs/PORTING.md index 7556100..e170c18 100644 --- a/docs/PORTING.md +++ b/docs/PORTING.md @@ -31,13 +31,13 @@ from app.sdk.platforms.kw import album, hot_search, leaderboard, lyric, music_se MODULE = Platform( id="kw", name="酷我音乐", - music_search=music_search, # 子模块对象（有 .search 方法） - leaderboard=leaderboard, # .get_boards() / .get_list(bangid, page) - song_list=song_list, # .get_list(sort_id, tag_id, page) / .sort_list - hot_search=hot_search, # .get_list() + music_search=music_search, # 子模块对象（有 .search 方法） + leaderboard=leaderboard, # .get_boards() / .get_list(bangid, page) + song_list=song_list, # .get_list(sort_id, tag_id, page) / .sort_list + hot_search=hot_search, # .get_list() album=album, playlist=playlist, - get_lyric=lyric.get_lyric, # 顶层方法直接挂函数 + get_lyric=lyric.get_lyric, # 顶层方法直接挂函数 get_pic=pic.get_pic, ) ``` @@ -118,14 +118,14 @@ async def handle_music_info(song_info: dict) -> dict # 可选：补全 name/ ```python from app.sdk.common.request import http_fetch -resp = await http_fetch(url, headers={...}) # 默认 GET -resp = await http_fetch(url, method="POST", form={...}) # urlencoded -resp = await http_fetch(url, method="POST", body={...}) # JSON -resp = await http_fetch(url, timeout=10_000) # 毫秒，与原文一致 +resp = await http_fetch(url, headers={...}) # 默认 GET +resp = await http_fetch(url, method="POST", form={...}) # urlencoded +resp = await http_fetch(url, method="POST", body={...}) # JSON +resp = await http_fetch(url, timeout=10_000) # 毫秒，与原文一致 -resp.status_code # int（Node 的 statusCode） -resp.body # 已解析：JSON → dict/list，否则 str -resp.raw # 原始 bytes +resp.status_code # int（Node 的 statusCode） +resp.body # 已解析：JSON → dict/list，否则 str +resp.raw # 原始 bytes resp.header("set-cookie") ``` @@ -175,9 +175,9 @@ from app.sdk.common.utils import size_formate ```python from app.sdk.auth_store import get_credentials, set_credentials, get_kg_auth, get_tx_auth, get_cookie -get_cookie("xm") # 空字符串表示未登录 -get_kg_auth() # {"token", "userid"} 或 None -get_tx_auth() # {"cookie", "uin"} 或 None +get_cookie("xm") # 空字符串表示未登录 +get_kg_auth() # {"token", "userid"} 或 None +get_tx_auth() # {"cookie", "uin"} 或 None ``` ## 8. 错误 diff --git a/pyproject.toml b/pyproject.toml index 890f74a..9f7aa0a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -22,7 +22,6 @@ dev = [ "pytest-asyncio>=0.24", "ruff>=0.6", "pyright>=1.1.380", - "httpx>=0.27", ] [tool.ruff] diff --git a/start.bat b/start.bat index 3521d75..5da1ca4 100644 --- a/start.bat +++ b/start.bat @@ -1,6 +1,6 @@ @echo off if "%1"=="dev" ( - pnpm run dev + uv run python scripts/serve.py --reload ) else ( - pnpm run start + uv run python scripts/serve.py %* ) diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..a3c22de --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,19 @@ +"""测试隔离：所有用例使用独立 DATA_DIR，绝不读写真实的 data/auth.json。 + +必须在任何 app 模块导入前设置；config.data_dir() 是运行时读取环境变量的。 +""" + +from __future__ import annotations + +import os +import sys +import tempfile +from pathlib import Path + +ROOT = Path(__file__).resolve().parent.parent +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +os.environ["DATA_DIR"] = tempfile.mkdtemp(prefix="music-server-tests-") +os.environ.pop("AUTH_KEY", None) +os.environ.pop("RATE_LIMIT", None) diff --git a/tests/test_http.py b/tests/test_http.py new file mode 100644 index 0000000..8899c2f --- /dev/null +++ b/tests/test_http.py @@ -0,0 +1,257 @@ +"""HTTP 层回归测试：CORS、限流顺序、路由遮蔽、流式管道与登录存储。 + +测试通过 tests/conftest.py 的临时 DATA_DIR 隔离，不会写真实 data/auth.json。 +HTTP 请求用 aiohttp 打到同一个进程内的 uvicorn，避免依赖 Starlette TestClient/httpx。 +""" + +from __future__ import annotations + +import asyncio +import importlib +import socket +import threading +import time +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager +from typing import Any +from urllib.parse import quote + +import aiohttp +import app.routes.music as music +import app.sdk.apis.kw as kw +import app.sdk.auth.wy as wy +import app.sdk.auth_store as auth_store +import pytest +import uvicorn +from app.main import create_app +from app.sdk import SOURCE_MODULES + + +def _free_port() -> int: + with socket.socket() as sock: + sock.bind(("127.0.0.1", 0)) + return int(sock.getsockname()[1]) + + +class _LiveServer: + """在后台线程运行真实 uvicorn，测试用 aiohttp 访问。""" + + def __init__(self, app: Any) -> None: + self.port = _free_port() + config = uvicorn.Config( + app, + host="127.0.0.1", + port=self.port, + log_level="error", + access_log=False, + lifespan="on", + ) + self.server = uvicorn.Server(config) + self.thread = threading.Thread(target=self.server.run, daemon=True) + + def start(self) -> None: + self.thread.start() + deadline = time.time() + 10 + while not self.server.started: + if not self.thread.is_alive() or time.time() > deadline: + raise RuntimeError("测试服务器启动失败") + time.sleep(0.01) + + def stop(self) -> None: + self.server.should_exit = True + self.thread.join(timeout=5) + + +@asynccontextmanager +async def live_client(app: Any) -> AsyncIterator[tuple[aiohttp.ClientSession, str]]: + server = _LiveServer(app) + await asyncio.to_thread(server.start) + try: + async with aiohttp.ClientSession() as session: + yield session, f"http://127.0.0.1:{server.port}" + finally: + await asyncio.to_thread(server.stop) + + +async def test_cors_headers_and_preflight() -> None: + async with live_client(create_app()) as (session, base): + async with session.get(f"{base}/health", headers={"Origin": "http://example.com"}) as response: + assert response.status == 200 + assert response.headers["access-control-allow-origin"] == "*" + + async with session.options( + f"{base}/health", + headers={"Origin": "http://example.com", "Access-Control-Request-Method": "GET"}, + ) as response: + assert response.status == 200 + assert response.headers["access-control-allow-origin"] == "*" + + +async def test_rate_limit_counts_unauthorized_requests(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("AUTH_KEY", "secret") + monkeypatch.setenv("RATE_LIMIT", "2") + statuses: list[int] = [] + async with live_client(create_app()) as (session, base): + for _ in range(3): + async with session.get(f"{base}/health", headers={"X-Server-Key": "wrong"}) as response: + statuses.append(response.status) + # 前两次进入鉴权返回 401，第三次在鉴权之前被限流拦截。 + assert statuses == [401, 401, 429] + + +async def test_album_search_is_not_shadowed(monkeypatch: pytest.MonkeyPatch) -> None: + kg = SOURCE_MODULES["kg"] + + async def fake_search(keyword: str, page: int = 1, limit: int = 20) -> dict: + return {"list": [{"name": keyword}], "page": page, "limit": limit, "total": 1} + + monkeypatch.setattr(kg.album, "search_album", fake_search) + async with live_client(create_app()) as (session, base): + async with session.get(f"{base}/album/kg/search/{quote('魔杰座')}") as response: + assert response.status == 200 + payload = await response.json() + assert payload["data"]["list"][0]["name"] == "魔杰座" + + +async def test_parse_kind_is_not_a_query_parameter(monkeypatch: pytest.MonkeyPatch) -> None: + seen: dict[str, str] = {} + + async def fake_parse(kind: str, source: str, link: str) -> dict: + seen.update(kind=kind, source=source, link=link) + return {"kind": kind} + + async with live_client(create_app()) as (session, base): + async with session.get(f"{base}/openapi.json") as response: + spec = await response.json() + params = spec["paths"]["/parse/song/{source}"]["post"].get("parameters", []) + assert "_kind" not in {item["name"] for item in params} + + monkeypatch.setattr(music, "_parse_handler", fake_parse) + async with session.post(f"{base}/parse/song/kw?_kind=playlist", json={"link": "x"}) as response: + assert response.status == 200 + + assert seen == {"kind": "song", "source": "kw", "link": "x"} + + +async def test_5sing_quality_from_sources_is_accepted(monkeypatch: pytest.MonkeyPatch) -> None: + fivesing = SOURCE_MODULES["5sing"] + + async def fake_url(song_info: dict, quality: str, retry_num: int = 0) -> dict: + return {"type": "sq", "url": "https://example.com/a.mp3"} + + monkeypatch.setattr(fivesing, "get_music_url", fake_url) + async with live_client(create_app()) as (session, base): + async with session.get(f"{base}/url/5sing/932557%7Cyc", params={"quality": "sq"}) as response: + assert response.status == 200 + payload = await response.json() + assert payload["data"] == {"type": "sq", "url": "https://example.com/a.mp3"} + + +async def test_find_rejects_unknown_source() -> None: + async with live_client(create_app()) as (session, base): + async with session.get(f"{base}/find/nope/{quote('稻香')}") as response: + assert response.status == 404 + + +async def test_stream_sign_rejects_infinite_exp(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("AUTH_KEY", "secret") + async with live_client(create_app()) as (session, base): + async with session.get(f"{base}/stream/mg/x", params={"exp": "inf", "sig": "x"}) as response: + assert response.status == 401 + + +async def test_stream_slow_consumer_does_not_hang(monkeypatch: pytest.MonkeyPatch) -> None: + class DummyDecoder: + @staticmethod + def feed(chunk: bytes) -> bytes: + return chunk + + async def fake_stream(url: str, feed, on_chunk) -> None: + for _ in range(100): + await on_chunk(b"x" * 1024) + + async def consume(response, delay: float) -> int: + count = 0 + async for _ in response.body_iterator: + count += 1 + await asyncio.sleep(delay) + return count + + monkeypatch.setattr(music, "stream_decrypted", fake_stream) + response = music._stream_response("http://example.com/a.mp3", DummyDecoder()) # type: ignore[arg-type] + assert await asyncio.wait_for(consume(response, 0.001), timeout=5) == 100 + + +async def test_stream_error_is_propagated(monkeypatch: pytest.MonkeyPatch) -> None: + class DummyDecoder: + @staticmethod + def feed(chunk: bytes) -> bytes: + return chunk + + async def failing_stream(url: str, feed, on_chunk) -> None: + await on_chunk(b"x") + raise RuntimeError("upstream boom") + + async def consume(response) -> None: + async for _ in response.body_iterator: + pass + + monkeypatch.setattr(music, "stream_decrypted", failing_stream) + response = music._stream_response("http://example.com/a.mp3", DummyDecoder()) # type: ignore[arg-type] + with pytest.raises(RuntimeError, match="upstream boom"): + await asyncio.wait_for(consume(response), timeout=5) + + +async def test_kw_token_retry_does_not_deadlock(monkeypatch: pytest.MonkeyPatch) -> None: + async def boom() -> list[str]: + raise RuntimeError("network down") + + monkeypatch.setattr(kw, "fetch_home_set_cookie", boom) + monkeypatch.setitem(kw._state, "token", None) + monkeypatch.setitem(kw._state, "lock", None) + with pytest.raises(RuntimeError, match="获取酷我token失败"): + await asyncio.wait_for(kw.get_token(), timeout=2) + + +def test_auth_store_rejects_non_dict_json(monkeypatch: pytest.MonkeyPatch, tmp_path) -> None: + monkeypatch.setenv("DATA_DIR", str(tmp_path)) + importlib.reload(auth_store) + (tmp_path / "auth.json").write_text("[]", encoding="utf-8") + + assert auth_store.get_credentials("wy") is None + assert any(path.name.startswith("auth.json.corrupt-") for path in tmp_path.iterdir()) + + +async def test_cookie_login_bind_and_logout() -> None: + async with live_client(create_app()) as (session, base): + async with session.post(f"{base}/login/bd", json={"cookie": "BDUSS=test"}) as response: + assert response.status == 200 + + async with session.get(f"{base}/login/status") as response: + payload = await response.json() + assert payload["data"]["bd"]["logged"] is True + + async with session.delete(f"{base}/login/bd") as response: + assert response.status == 200 + + async with session.get(f"{base}/login/status") as response: + payload = await response.json() + assert payload["data"]["bd"]["logged"] is False + + async with session.post(f"{base}/login/nope", json={"cookie": "x"}) as response: + assert response.status == 404 + + +async def test_wy_login_profile_uses_new_cookie(monkeypatch: pytest.MonkeyPatch) -> None: + seen: dict[str, str | None] = {} + + async def fake_fetch(cookie: str | None = None) -> dict: + seen["cookie"] = cookie + return {"userid": 1, "nickname": "n", "avatar": None, "vipType": None} + + monkeypatch.setattr(wy, "_fetch_profile", fake_fetch) + monkeypatch.setattr(wy, "set_credentials", lambda *args, **kwargs: None) + + profile = await wy._apply_login(["MUSIC_U=abc; Path=/"]) + assert seen["cookie"] is not None and seen["cookie"].startswith("MUSIC_U=abc") + assert profile["nickname"] == "n" diff --git a/uv.lock b/uv.lock index aea742d..5305293 100644 --- a/uv.lock +++ b/uv.lock @@ -127,15 +127,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/4d/3f/43ea0a58dcb50c820465a0d62ccbbf6464c2e2cc4ebf582e64225dd67b4b/brotli-1.2.1-pp312-pypy312_pp80-win_amd64.whl", hash = "sha256:06a22fcbcae059c0679098232779095fc50a735a3fea02db5c79cab10b84d877", size = 457800, upload-time = "2026-10-22T14:00:10.978Z" }, ] -[[package]] -name = "certifi" -version = "2026.10.23" -source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/fd/11/b524cbd9bbdaca58608e4cf4053fd463417549e1959fa8d3aa18319f9555/certifi-2026.10.23.tar.gz", hash = "sha256:a9895e6e2758b371e0545e321a3a8fd1fcfd9829ada8824a936dfb16aaaf36cd", size = 136225, upload-time = "2026-10-23T11:36:16.478Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/32/d1/a3652fe2c19134d0d15922177821d863cbd7e842d53a21bbe7e8e3a1b5ac/certifi-2026.10.23-py3-none-any.whl", hash = "sha256:e9b54de76e530bb6cf0ccd036787103acb58fd42ba930ea8bf948ddbaa2f61d1", size = 135082, upload-time = "2026-10-23T11:36:14.763Z" }, -] - [[package]] name = "cffi" version = "2.1.1" @@ -313,19 +304,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/04/4b/29cac41a4d98d144bf5f6d33995617b185d14b22401f75ca86f384e87ff1/h11-0.16.0-py3-none-any.whl", hash = "sha256:63cf8bbe7522de3bf65932fda1d9c2772064ffb3dae62d55932da54b31cb6c86", size = 37515, upload-time = "2025-04-24T03:35:24.344Z" }, ] -[[package]] -name = "httpcore" -version = "1.0.9" -source = { registry = "https://pypi.org/simple" } -dependencies = [ - { name = "certifi" }, - { name = "h11" }, -] -sdist = { url = "https://files.pythonhosted.org/packages/06/94/82699a10bca87a5556c9c59b5963f2d039dbd239f25bc2a63907a05a14cb/httpcore-1.0.9.tar.gz", hash = "sha256:6e34463af53fd2ab5d807f399a9b45ea31c3dfa2276f15a2c3f00afff6e176e8", size = 85484, upload-time = "2025-04-24T22:06:22.219Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/7e/f5/f66802a942d491edb555dd61e3a9961140fd64c90bce1eafd741609d334d/httpcore-1.0.9-py3-none-any.whl", hash = "sha256:2d400746a40668fc9dec9810239072b40b4484b640a8c38fd654a024c7a1bf55", size = 78784, upload-time = "2025-04-24T22:06:20.566Z" }, -] - [[package]] name = "httptools" version = "0.9.0" @@ -347,21 +325,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/ae/e0/1d29e328c4cafe843403341e1455e0aec18b0e6910fbb14f12b36b563f19/httptools-0.9.0-cp312-cp312-win_arm64.whl", hash = "sha256:813a32f94991b9627795528053c73a57d2ce3eb98ede89f0e1c7a31095938e81", size = 88676, upload-time = "2026-10-09T19:54:51.844Z" }, ] -[[package]] -name = "httpx" -version = "0.28.1" -source = { registry = "https://pypi.org/simple" } -dependencies = [ - { name = "anyio" }, - { name = "certifi" }, - { name = "httpcore" }, - { name = "idna" }, -] -sdist = { url = "https://files.pythonhosted.org/packages/b1/df/48c586a5fe32a0f01324ee087459e112ebb7224f646c0b5023f5e79e9956/httpx-0.28.1.tar.gz", hash = "sha256:75e98c5f16b0f35b567856f597f06ff2270a374470a5c2392242528e3e3e42fc", size = 141406, upload-time = "2024-12-06T15:37:23.222Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/2a/39/e50c7c3a983047577ee07d2a9e53faf5a69493943ec3f6a384bdc792deb2/httpx-0.28.1-py3-none-any.whl", hash = "sha256:d909fcccc110f8c7faf814ca82a9a4d816bc5a6dbfea25d6591d6985b8ba59ad", size = 73517, upload-time = "2024-12-06T15:37:21.509Z" }, -] - [[package]] name = "idna" version = "3.20" @@ -426,7 +389,6 @@ dependencies = [ [package.dev-dependencies] dev = [ - { name = "httpx" }, { name = "pyright" }, { name = "pytest" }, { name = "pytest-asyncio" }, @@ -447,7 +409,6 @@ requires-dist = [ [package.metadata.requires-dev] dev = [ - { name = "httpx", specifier = ">=0.27" }, { name = "pyright", specifier = ">=1.1.380" }, { name = "pytest", specifier = ">=8" }, { name = "pytest-asyncio", specifier = ">=0.24" },