From f588d313f06f2eed9dd82707414d07e3e6cc53b2 Mon Sep 17 00:00:00 2001 From: snowy-yang <102860110+snowy-yang@users.noreply.github.com> Date: Sat, 26 Sep 2026 01:18:51 +0800 Subject: [PATCH] =?UTF-8?q?feat(routes):=20=E8=BF=81=E7=A7=BB=E4=BB=BB?= =?UTF-8?q?=E5=8A=A1=E7=8A=B6=E6=80=81=E7=AB=AF=E7=82=B9=EF=BC=88topen/que?= =?UTF-8?q?ue/batch=5Fprogress/task/*=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - web/routers/tasks.py：topen、queue、batch_progress、task/cancel、 task/heartbeat、task/list 迁至 /api/v1/* - 读写 core.state 共享任务表/统计（与 Flask 同一批对象）——两栈迁移期看到同一份 任务状态，这是单进程双框架成立的关键 - task/cancel 保持 POST-only（与 Flask 同口径，杜绝顶层导航 GET 触发取消） 契约奇偶校验：24 项全 OK（新增 queue/topen/task:list/heartbeat/progress404）； 验证：pytest 46/46；ruff All checks passed；pyright 0 errors；冒烟 10/10。 --- web/app.py | 2 + web/routers/tasks.py | 247 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 249 insertions(+) create mode 100644 web/routers/tasks.py diff --git a/web/app.py b/web/app.py index f249919..202e8ee 100644 --- a/web/app.py +++ b/web/app.py @@ -26,6 +26,7 @@ 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 system as system_router +from web.routers import tasks as tasks_router @asynccontextmanager @@ -73,6 +74,7 @@ def create_app(serve_flask_fallback: bool = False) -> FastAPI: app.include_router(music_router.router, prefix='/api/v1', tags=['music']) 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']) install_error_handlers(app) diff --git a/web/routers/tasks.py b/web/routers/tasks.py new file mode 100644 index 0000000..1f20255 --- /dev/null +++ b/web/routers/tasks.py @@ -0,0 +1,247 @@ +# -*- coding: utf-8 -*- +"""任务状态端点（FastAPI 栈）——与 Flask 侧同名端点响应契约逐字段对齐。 + +已迁移： +- POST/GET /api/v1/topen ↔ Flask /api/topen（请求统计，无鉴权） +- POST/GET /api/v1/queue ↔ Flask /api/queue（排队状态，无鉴权） +- POST/GET /api/v1/batch_progress ↔ Flask /api/batch_progress（需登录，仅本人任务） +- POST /api/v1/task/cancel ↔ Flask /api/task/cancel（需登录，仅本人任务） +- POST/GET /api/v1/task/heartbeat ↔ Flask /api/task/heartbeat（需登录） +- POST/GET /api/v1/task/list ↔ Flask /api/task/list（需登录） + +这些端点读写 core.state 的共享任务表（与 Flask 同一批对象），因此两栈在迁移期 +看到的是同一份任务状态——这是「单进程双框架」能成立的关键。 +""" +import time +from typing import Annotated, Any + +from fastapi import APIRouter, Depends, Request + +from core.state import ( + BATCH_DURATIONS, + BATCH_DURATIONS_LOCK, + DEVICE_LAST_SEEN, + DEVICE_LAST_SEEN_LOCK, + DOWNLOAD_TASKS, + DOWNLOAD_TASKS_LOCK, + REQUEST_STATS, + STATS_LOCK, +) +from web import ratelimit +from web.deps import SiteUser, require_login +from web.response import error, success + +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 _touch_device(device_id: str) -> None: + """记录设备心跳（与 Flask _touch_device 同口径，共享 DEVICE_LAST_SEEN）。""" + device_id = str(device_id or '')[:64] + if not device_id: + return + with DEVICE_LAST_SEEN_LOCK: + if len(DEVICE_LAST_SEEN) >= 5000: + for d in sorted(DEVICE_LAST_SEEN, key=lambda k: DEVICE_LAST_SEEN.get(k, 0.0))[:2500]: + DEVICE_LAST_SEEN.pop(d, None) + DEVICE_LAST_SEEN[device_id] = time.time() + + +def _owned_by(task: dict[str, Any], user: SiteUser) -> bool: + return int(task.get('user_id') or 0) == int(user.id) + + +def _semaphores(): + """批量打包/下载信号量（Flask 侧对象，懒取以共享同一计数）。""" + import main as legacy + return legacy.BATCH_ZIP_SEM, legacy.api_service.downloader._download_semaphore + + +@router.api_route('/topen', methods=['GET', 'POST']) +async def topen(request: Request): + """接口请求统计（实时）——锁内快照，避免 total 与明细不一致。""" + import main as legacy + with STATS_LOCK: + snapshot = {p: dict(v) for p, v in REQUEST_STATS.items()} + endpoints = [] + total = sum(v['count'] for v in snapshot.values()) + for path, info in snapshot.items(): + endpoints.append({ + 'path': path, + 'name': legacy.ENDPOINT_NAMES.get(path, path), + 'count': info['count'], + 'percent': round(info['count'] / total * 100, 1) if total else 0, + 'last': info['last'], + }) + endpoints.sort(key=lambda x: x['count'], reverse=True) + return success({ + 'total': total, + 'endpoint_count': len(endpoints), + 'max_count': endpoints[0]['count'] if endpoints else 0, + 'endpoints': endpoints, + }, '获取请求统计成功') + + +@router.api_route('/queue', methods=['GET', 'POST']) +async def queue(request: Request): + """实时排队状态（排队人数 + 预计等待时间）。""" + batch_sem, dl_sem = _semaphores() + avg_batch = None + with BATCH_DURATIONS_LOCK: + if BATCH_DURATIONS: + avg_batch = sum(BATCH_DURATIONS) / len(BATCH_DURATIONS) + return success({ + 'batch_capacity': batch_sem.capacity, + 'batch_active': batch_sem.active, + 'batch_waiting': batch_sem.waiting, + 'download_capacity': dl_sem.capacity, + 'download_active': dl_sem.active, + 'download_waiting': dl_sem.waiting, + 'avg_batch_seconds': round(avg_batch, 1) if avg_batch else None, + }, '获取排队状态成功') + + +@router.api_route('/batch_progress', methods=['GET', 'POST']) +async def batch_progress(user: LoginUser, request: Request): + """下载任务进度（已处理 X/Y + 百分比 + 预计剩余时间）。""" + data = await _request_data(request) + task_id = str(data.get('task_id') or '').strip() + if not task_id: + return error('缺少 task_id 参数') + now = time.time() + with DOWNLOAD_TASKS_LOCK: + task = DOWNLOAD_TASKS.get(task_id) + if not task: + return error('任务不存在或已过期', 404) + if not _owned_by(task, user): + return error('无权查询该任务', 403) + + total = task.get('total', 0) + done = task.get('done', 0) + live = task.get('live_percent') + if task.get('status') == 'running' and isinstance(live, (int, float)): + percent = round(min(99.0, max(0.0, float(live))), 1) + elif total: + percent = round(done / total * 100, 1) + else: + percent = 100.0 if task.get('status') in ('done',) else 0.0 + elapsed = now - task.get('start', now) + eta = None + if task.get('status') == 'running' and done > 0 and elapsed > 0 and total: + rate = done / elapsed + if rate > 0: + eta = round((total - done) / rate, 1) + result = task.get('result') or {} + has_result = bool(result.get('zip_path') or result.get('file_path')) + return success({ + 'total': total, + 'done': done, + 'failed': task.get('failed', 0), + 'percent': percent, + 'status': task.get('status', 'running'), + 'elapsed': round(elapsed, 1), + 'eta': eta, + 'msg': task.get('msg', ''), + 'refunded': task.get('refunded', 0), + 'has_result': has_result, + }, '获取任务进度成功') + + +@router.post('/task/cancel') +async def task_cancel(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: + return error('任务不存在或已过期', 404) + if not _owned_by(task, user): + return error('无权取消该任务', 403) + if task.get('status') in ('done', 'error', 'canceled'): + return error(f"任务已结束（{task.get('status')}）", 400) + task['status'] = 'canceled' + task['msg'] = '任务已取消' + task['completed_at'] = time.time() + task['cancel'].set() + return success({'task_id': task_id, 'status': 'canceled'}, '已请求取消') + + +@router.api_route('/task/heartbeat', methods=['GET', 'POST']) +async def task_heartbeat(user: LoginUser, request: Request): + """设备心跳：保持任务活跃；暂停中的任务自动恢复（重新进入排队）。""" + await ratelimit.guard(request, bucket='hb', limit=60, window=60) + data = await _request_data(request) + device_id = str(data.get('device_id') or '').strip()[:64] + if not device_id: + return error('缺少 device_id 参数') + _touch_device(device_id) + resumed = 0 + with DOWNLOAD_TASKS_LOCK: + for task in DOWNLOAD_TASKS.values(): + if (task.get('device_id') == device_id + and int(task.get('user_id') or 0) == int(user.id) + and task.get('status') == 'paused'): + task['status'] = 'queued' + task['msg'] = '恢复中...' + task['pause'].clear() + resumed += 1 + now = time.time() + with DEVICE_LAST_SEEN_LOCK: + stale = [d for d, ts in DEVICE_LAST_SEEN.items() if now - ts > 86400] + for d in stale: + DEVICE_LAST_SEEN.pop(d, None) + return success({'resumed': resumed, 'online': True}, '心跳成功') + + +@router.api_route('/task/list', methods=['GET', 'POST']) +async def task_list(user: LoginUser, request: Request): + """本设备未结束的任务列表（重新打开网站时恢复任务悬浮窗）。""" + data = await _request_data(request) + device_id = str(data.get('device_id') or '').strip()[:64] + if not device_id: + return error('缺少 device_id 参数') + _touch_device(device_id) + my_tasks = [] + with DOWNLOAD_TASKS_LOCK: + for tid, task in DOWNLOAD_TASKS.items(): + params = task.get('params') or {} + if (task.get('device_id') == device_id + and int(task.get('user_id') or 0) == int(user.id) + and task.get('status') in ('queued', 'running', 'paused')): + my_tasks.append({ + 'task_id': tid, + 'type': task.get('type', 'batch'), + 'kind': ('podcast' if ('program_id' in params + or 'radio_name' in params + or params.get('radio_id')) + else 'music'), + 'total': task.get('total', 0), + 'done': task.get('done', 0), + 'status': task.get('status', 'queued'), + 'msg': task.get('msg', ''), + }) + return success({'tasks': my_tasks}, '获取任务列表成功')