feat(routes): 迁移任务状态端点(topen/queue/batch_progress/task/*)

- 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。
This commit is contained in:
snowy-yang committed 2026-09-26 01:18:51 +08:00
1 parent f45a9ccd3a
commit f588d313f0
2 files changed
+249

No files matched your search

+2
View File
@@ -26,6 +26,7 @@ from web.routers import music as music_router
from web.routers import podcast as podcast_router from web.routers import podcast as podcast_router
from web.routers import site as site_router from web.routers import site as site_router
from web.routers import system as system_router from web.routers import system as system_router
from web.routers import tasks as tasks_router
@asynccontextmanager @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(music_router.router, prefix='/api/v1', tags=['music'])
app.include_router(account_router.router, prefix='/api/v1', tags=['account']) 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(podcast_router.router, prefix='/api/v1', tags=['podcast'])
app.include_router(tasks_router.router, prefix='/api/v1', tags=['tasks'])
install_error_handlers(app) install_error_handlers(app)
+247
View File
@@ -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}, '获取任务列表成功')