diff --git a/core/http_client.py b/core/http_client.py index 2c8486c..a0c961f 100644 --- a/core/http_client.py +++ b/core/http_client.py @@ -254,6 +254,35 @@ class AsyncHTTPClient: except aiohttp.ClientError as e: raise UpstreamNetworkError(str(e)) from e + async def open( + 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, + ) -> aiohttp.ClientResponse: + """打开流式响应并返回**未关闭**的 ClientResponse（调用方负责 close）。 + + 与 stream() 的区别：不把响应交给 `async with`，便于「先读状态码/头决定是否 + 转发，再流式读取」的代理场景（如音频 Range 代理）。 + """ + session = await self._session_for_loop() + req_timeout = aiohttp.ClientTimeout(total=timeout) if timeout else None + try: + return await 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, + ) + except aiohttp.ClientError as e: + raise UpstreamNetworkError(str(e)) from e + # 进程级共享客户端（UA/Referer 由 music_api 在初始化后注入） client = AsyncHTTPClient() diff --git a/tests/test_stream.py b/tests/test_stream.py new file mode 100644 index 0000000..93a5140 --- /dev/null +++ b/tests/test_stream.py @@ -0,0 +1,88 @@ +# -*- coding: utf-8 -*- +"""音频流代理测试：签名校验、HEAD、Range 透传、无效 token。""" +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from core import security +from web.routers import stream as stream_router + +SECRET = 'stream-test-secret' +PAYLOAD = b'AUDIO-DATA-' * 1000 # 11000 字节 + + +@pytest.fixture() +def client(monkeypatch): + security.init_secret_key(SECRET) + app = FastAPI() + app.include_router(stream_router.router, prefix='/api/v1') + + class _Upstream: + def __init__(self): + self.status = 200 + self.headers = {'Content-Type': 'audio/mpeg', 'Content-Length': str(len(PAYLOAD)), + 'Accept-Ranges': 'bytes'} + self.closed = False + self.content = self + + async def iter_chunked(self, n): + for i in range(0, len(PAYLOAD), n): + yield PAYLOAD[i:i + n] + + def close(self): + self.closed = True + + state = {'last_range': None, 'upstream': None} + + async def fake_open(method, url, *, headers=None, timeout=None, **kw): + state['last_range'] = (headers or {}).get('Range') + up = _Upstream() + if state['last_range']: + up.status = 206 + up.headers = dict(up.headers) + up.headers['Content-Range'] = f'bytes 0-99/{len(PAYLOAD)}' + state['upstream'] = up + return up + + monkeypatch.setattr('web.routers.stream.http_client.open', fake_open) + return TestClient(app), state + + +def test_invalid_token(client): + c, _ = client + r = c.get('/api/v1/audio_stream?token=bogus') + assert r.status_code == 403 + + +def test_head_no_upstream(client): + c, state = client + r = c.head('/api/v1/audio_stream?token=x') + assert r.status_code == 200 + assert r.headers['accept-ranges'] == 'bytes' + assert state['upstream'] is None # HEAD 不应请求上游 + + +def test_stream_full(client): + c, _ = client + token = security.make_audio_token('http://m801.music.126.net/a.mp3') + r = c.get(f'/api/v1/audio_stream?token={token}') + assert r.status_code == 200 + assert r.content == PAYLOAD + assert r.headers['cache-control'] == 'no-store' + assert r.headers['content-length'] == str(len(PAYLOAD)) + + +def test_range_forwarded_and_206(client): + c, state = client + token = security.make_audio_token('http://m801.music.126.net/a.mp3') + r = c.get(f'/api/v1/audio_stream?token={token}', headers={'Range': 'bytes=0-99'}) + assert state['last_range'] == 'bytes=0-99' # Range 已透传上游 + assert r.status_code == 206 + assert 'content-range' in r.headers + + +def test_upstream_closed_on_complete(client): + c, state = client + token = security.make_audio_token('http://m801.music.126.net/a.mp3') + c.get(f'/api/v1/audio_stream?token={token}') + assert state['upstream'].closed is True diff --git a/web/app.py b/web/app.py index 202e8ee..7dcad78 100644 --- a/web/app.py +++ b/web/app.py @@ -22,9 +22,11 @@ from core.security import init_secret_key from web.errors import install_error_handlers from web.middleware import ResponseHygieneMiddleware from web.routers import account as account_router +from web.routers import delivery as delivery_router from web.routers import music as music_router from web.routers import podcast as podcast_router from web.routers import site as site_router +from web.routers import stream as stream_router from web.routers import system as system_router from web.routers import tasks as tasks_router @@ -75,6 +77,8 @@ def create_app(serve_flask_fallback: bool = False) -> FastAPI: app.include_router(account_router.router, prefix='/api/v1', tags=['account']) app.include_router(podcast_router.router, prefix='/api/v1', tags=['podcast']) app.include_router(tasks_router.router, prefix='/api/v1', tags=['tasks']) + app.include_router(delivery_router.router, prefix='/api/v1', tags=['delivery']) + app.include_router(stream_router.router, prefix='/api/v1', tags=['stream']) install_error_handlers(app) diff --git a/web/routers/delivery.py b/web/routers/delivery.py new file mode 100644 index 0000000..7528dbf --- /dev/null +++ b/web/routers/delivery.py @@ -0,0 +1,127 @@ +# -*- coding: utf-8 -*- +"""结果交付端点（FastAPI 栈）——与 Flask 侧同名端点响应契约逐字段对齐。 + +已迁移： +- POST/GET /api/v1/download/result ↔ Flask /download/result（单曲） +- POST/GET /api/v1/download/batch_zip/result ↔ Flask /download/batch_zip/result（批量 ZIP） + +语义对齐要点： +- 任务归属：仅任务本人的站点账号可领取（非本人 403） +- 状态语义：进行中 202 / 暂停 200+提示 / 取消 200+原因 / 失败 500+原因 / + 文件缺失 404（与 Flask 分支逐一对应） +- 交付物：FileResponse 流式发送；Cache-Control: no-store； + X-Download-Filename（URL 编码，供前端取回原始文件名） +- 文件清理仍由监控线程按 35 分钟保留期统一回收（交付请求不删文件） +""" +import asyncio +import os +from typing import Annotated, Any +from urllib.parse import quote + +from fastapi import APIRouter, Depends, Request +from fastapi.responses import FileResponse + +from core.state import DOWNLOAD_TASKS, DOWNLOAD_TASKS_LOCK +from web.deps import SiteUser, require_login +from web.response import error + +router = APIRouter() + +LoginUser = Annotated[SiteUser, Depends(require_login)] + + +async def _request_data(request: Request) -> dict[str, Any]: + data: dict[str, Any] = {} + if request.method == 'GET': + data.update(dict(request.query_params)) + return data + try: + body = await request.json() + if isinstance(body, dict): + data.update(body) + except Exception: + pass + if not data: + try: + form = await request.form() + data.update(dict(form.items())) + except Exception: + pass + return data + + +def _owned_by(task: dict[str, Any], user: SiteUser) -> bool: + return int(task.get('user_id') or 0) == int(user.id) + + +def _file_response(target: str, filename: str, mimetype: str) -> FileResponse: + resp = FileResponse(target, media_type=mimetype, filename=filename) + resp.headers['Cache-Control'] = 'no-store' + resp.headers['X-Download-Filename'] = quote(filename, safe='') + return resp + + +@router.api_route('/download/result', methods=['GET', 'POST']) +async def download_result(user: LoginUser, request: Request): + """单曲下载结果：任务完成后流式返回文件。""" + data = await _request_data(request) + task_id = str(data.get('task_id') or '').strip() + if not task_id: + return error('缺少 task_id 参数') + with DOWNLOAD_TASKS_LOCK: + task = DOWNLOAD_TASKS.get(task_id) + if not task or task.get('type') != 'single': + return error('任务不存在或已过期', 404) + if not _owned_by(task, user): + return error('无权下载该任务', 403) + + status = task.get('status') + if status in ('queued', 'running'): + return error('任务还在进行中，请稍后再试', 202) + if status == 'paused': + return error('任务已暂停，请重新打开页面后恢复', 200) + + result = task.get('result') or {} + target = result.get('zip_path') or result.get('file_path') + target_exists = bool(target) and await asyncio.to_thread(os.path.exists, target) + if status in ('canceled', 'error') and not target_exists: + return error(task.get('msg') or ('任务已取消' if status == 'canceled' else '任务失败'), + 200 if status == 'canceled' else 500) + if not target_exists: + return error('文件不存在或已被清理', 404) + filename = result.get('filename') or 'download' + assert target is not None + return _file_response(target, filename, result.get('mimetype') or 'application/octet-stream') + + +@router.api_route('/download/batch_zip/result', methods=['GET', 'POST']) +async def download_batch_zip_result(user: LoginUser, request: Request): + """批量打包结果：任务完成后流式返回 ZIP。""" + data = await _request_data(request) + task_id = str(data.get('task_id') or '').strip() + if not task_id: + return error('缺少 task_id 参数') + with DOWNLOAD_TASKS_LOCK: + task = DOWNLOAD_TASKS.get(task_id) + if not task or task.get('type') != 'batch': + return error('任务不存在或已过期', 404) + if not _owned_by(task, user): + return error('无权下载该任务', 403) + + status = task.get('status') + result = task.get('result') or {} + zip_path = result.get('zip_path') + has_deliverable = bool(zip_path) and await asyncio.to_thread(os.path.exists, zip_path) + if status in ('queued', 'running'): + return error('任务还在进行中，请稍后再试', 202) + if status == 'paused': + return error('任务已暂停，请重新打开页面后恢复', 200) + # 取消/失败但已有部分成果的任务，成果必须可领取 + if status in ('canceled', 'error') and not has_deliverable: + return error(task.get('msg') or ('任务已取消' if status == 'canceled' else '任务失败'), + 200 if status == 'canceled' else 500) + if not has_deliverable: + return error('打包文件不存在或已被清理', 404) + zip_name = result.get('filename') or '批量下载.zip' + assert zip_path is not None + return _file_response(zip_path, zip_name, 'application/zip') diff --git a/web/routers/stream.py b/web/routers/stream.py new file mode 100644 index 0000000..1b82c01 --- /dev/null +++ b/web/routers/stream.py @@ -0,0 +1,78 @@ +# -*- coding: utf-8 -*- +"""音频流代理（FastAPI 栈）——与 Flask /audio_stream 行为逐项对齐。 + +要点（对齐 Flask 实现）： +- 签名校验：token 由 core.security 校验（密钥同源，两栈 token 互认）；失效 403。 +- HEAD 无正文：直接 200 + Accept-Ranges，不请求上游。 +- Range 透传：客户端 Range 原样转发上游，206/Content-Range 原样回传（拖动进度可用）。 +- 编码透明：上游若带 Content-Encoding（aiohttp 会解压），不得沿用压缩前的 + Content-Length/Content-Range（否则头体不一致，同 Flask F92）。 +- 流式转发：16KB 分块，不整段入内存；下游中断时 finally 关闭上游响应。 +- Cache-Control: no-store。 +""" +from fastapi import APIRouter, Request +from fastapi.responses import Response, StreamingResponse + +from core import security +from core.http_client import client as http_client +from web.response import error + +router = APIRouter() + +_UPSTREAM_HEADERS = { + 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36', + 'Referer': 'https://music.163.com/', +} + +_CHUNK = 16 * 1024 + + +@router.api_route('/audio_stream', methods=['GET', 'HEAD']) +async def audio_stream(request: Request): + """签名音频代理流（支持 Range 拖动进度）。""" + if request.method == 'HEAD': + return Response(status_code=200, headers={'Accept-Ranges': 'bytes'}) + + token = request.query_params.get('token', '') + url = security.verify_audio_token(token) + if not url: + return error('音频链接已过期或无效，请重新解析', 403) + + upstream_headers = dict(_UPSTREAM_HEADERS) + range_header = request.headers.get('range') + if range_header: + upstream_headers['Range'] = range_header + + try: + upstream = await http_client.open('GET', url, headers=upstream_headers, timeout=20) + except Exception: + return error('音频获取失败', 502) + + if upstream.status >= 400: + upstream.close() + return error('音频获取失败', 502) + + upstream_encoded = bool(upstream.headers.get('Content-Encoding')) + pass_headers: dict[str, str] = {} + if not upstream_encoded: + for h in ('Content-Length', 'Content-Range', 'Accept-Ranges'): + if h in upstream.headers: + pass_headers[h] = upstream.headers[h] + elif 'Accept-Ranges' in upstream.headers: + pass_headers['Accept-Ranges'] = upstream.headers['Accept-Ranges'] + pass_headers['Cache-Control'] = 'no-store' + + async def _body(): + try: + async for chunk in upstream.content.iter_chunked(_CHUNK): + yield chunk + finally: + # 下游断开/提前关闭时也释放上游连接 + upstream.close() + + return StreamingResponse( + _body(), + status_code=upstream.status, + media_type=upstream.headers.get('Content-Type', 'audio/mpeg'), + headers=pass_headers, + )