From 6fb37534d35a9f43ca38e960b82244b79294fd36 Mon Sep 17 00:00:00 2001 From: snowy-yang <102860110+snowy-yang@users.noreply.github.com> Date: Sun, 27 Sep 2026 12:13:00 +0800 Subject: [PATCH] =?UTF-8?q?refactor:=20=E8=AE=A1=E8=B4=B9=E9=80=80?= =?UTF-8?q?=E6=AC=BE=E4=B8=8D=E5=8F=98=E9=87=8F=E6=94=B6=E6=95=9B=E4=B8=BA?= =?UTF-8?q?=20Billing.scope()=EF=BC=88download.py=2025=20=E5=A4=84?= =?UTF-8?q?=E6=89=8B=E5=B7=A5=E9=80=80=E6=AC=BE=20=E2=86=92=200=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 为什么：「任何失败路径必须退款」此前靠路由里每条早退/except 手工调 refund_open 维持（download.py 单文件 25 处），上一轮拆 register/login 时已经体会过这个模式的代价——改任何一条返回路径都要核对还原时序， 漏一处不会有测试报警，只剩重启对账兜底在跑。 改动： - Billing.scope()（asynccontextmanager）：作用域退出时自动退还仍 open 的账本行。正确性完全依赖 refund_open 的既有幂等口径（只退仍 open 的行）——settle()/rename_to() 消费过的行天然豁免，成功路径零动作， 无需手动「取消自动退」；未捕获异常先退款再原样穿透。 - download.py 四个端点全部改用 async with billing.scope()，删除全部 手工退款与 _fail 收尾闭包；_sync_download_json 的内部退款一并移除 （docstring 标明必须在 scope 内运行）。 - 语义逐路径等价：错误早退→自动退（同原手工）；register 成功后 rename→自动退 no-op（同原手工退的 no-op）；start 失败行挂 task_id 留给对账（同原）；CancelledError 语义不变（退款随取消中断，对账兜底）。 先写测试后动代码：tests/test_billing_scope.py 5 例（TDD 红→绿）， 对真实沙箱账本断言——open 行自动退、settle 后 no-op 且余额恰 -1、 rename 后请求级行消失且任务级行保持 open、异常穿透、free 路径零动作。 既有 30 例计费/下载端到端测试作迁移回归网。 门禁：ruff 全过 · pyright 0/0 · pytest 387 passed（382→387）。 --- tests/test_billing_scope.py | 123 ++++++++++ web/billing.py | 22 +- web/routers/download.py | 463 +++++++++++++++++------------------- 3 files changed, 364 insertions(+), 244 deletions(-) create mode 100644 tests/test_billing_scope.py diff --git a/tests/test_billing_scope.py b/tests/test_billing_scope.py new file mode 100644 index 0000000..c3c2ad3 --- /dev/null +++ b/tests/test_billing_scope.py @@ -0,0 +1,123 @@ +# -*- coding: utf-8 -*- +"""Billing.scope() 契约：作用域退出自动退还仍 open 的账本行。 + +设计前提（已由既有实现保证）：refund_open 只退「仍 open」的行—— +settle/rename_to 消费过的行天然不在退款范围，因此自动退与成功路径 +互不干扰，调用方无需手动取消。测试直接对真实账本（沙箱库）断言。 +""" +import pytest + +import auth_system as auth +from web.billing import Billing +from web.deps import SiteUser + + +@pytest.fixture() +def ledger_env(tmp_path, monkeypatch): + monkeypatch.setattr(auth.db, 'DATA_DIR', tmp_path) + monkeypatch.setattr(auth.db, 'DB_PATH', tmp_path / 'site.db') + monkeypatch.setattr(auth.db, 'BACKUP_DIR', tmp_path / 'backups') + monkeypatch.setattr(auth.db, 'SECRET_KEY_PATH', tmp_path / 'secret.key') + auth.init_db() + row = auth.create_user('payer', 'Passw0rd12', email='p@example.com', credits=5) + assert row is not None, '建号应成功' + yield {'uid': int(row['id'])} + # 无需清理：tmp_path 由 pytest 回收 + + +def _row(key: str): + """取账本行，不存在则测试失败（供「行必须存在」的断言用）。""" + row = auth.ledger_get(key) + assert row is not None, f'账本行 {key} 应存在' + return row + + +def _user_credits() -> int: + u = auth.get_user_by_name('payer') + assert u is not None, '用户 payer 应存在' + return int(u['duration_seconds'] or 0) + + +def _make_billing(ledger_env, key: str | None) -> Billing: + user = SiteUser(id=ledger_env['uid'], username='payer', role='user', credits=5) + return Billing(user=user, ledger_key=key, need=1, free=False) + + +@pytest.mark.anyio +async def test_scope_auto_refunds_open_row(ledger_env): + """正常退出作用域：仍 open 的行被退（rejected），余额恢复。""" + key = 'req-aaa' + reserved = auth.reserve_and_open(ledger_env['uid'], 1, key) + assert reserved is not None + assert _row(key)['status'] == 'open' + + billing = _make_billing(ledger_env, key) + async with billing.scope(): + pass # 业务体正常走完但未结算（等价于路由里的错误早退） + + assert _row(key)['status'] == 'rejected' + assert _user_credits() == 5 + + +@pytest.mark.anyio +async def test_scope_noop_after_settle(ledger_env): + """作用域内已结算：退出自动退为 no-op，交付行不受影响。""" + key = 'req-bbb' + auth.reserve_and_open(ledger_env['uid'], 1, key) + billing = _make_billing(ledger_env, key) + + async with billing.scope(): + await billing.settle(1, 'delivered') + + assert _row(key)['status'] == 'delivered' + assert _user_credits() == 4 # 恰好扣 1，未被二次退 + + +@pytest.mark.anyio +async def test_scope_noop_after_rename(ledger_env): + """作用域内账本已改挂 task_id：请求级行不存在，自动退为 no-op； + 任务级行保持 open 交给 worker 结算——与迁移前手工语义逐路径一致。""" + key = 'req-ccc' + auth.reserve_and_open(ledger_env['uid'], 1, key) + billing = _make_billing(ledger_env, key) + + async with billing.scope(): + ok = await billing.rename_to('task-xyz') + assert ok is True + + assert auth.ledger_get(key) is None + moved = _row('task-xyz') + assert moved['status'] == 'open' + + +@pytest.mark.anyio +async def test_scope_refunds_on_exception_and_reraises(ledger_env): + """业务体抛异常：先退款，异常原样穿透（路由自行转错误响应）。""" + key = 'req-ddd' + auth.reserve_and_open(ledger_env['uid'], 1, key) + billing = _make_billing(ledger_env, key) + + with pytest.raises(ValueError, match='boom'): + async with billing.scope(): + raise ValueError('boom') + + assert _row(key)['status'] == 'rejected' + + +@pytest.mark.anyio +async def test_scope_free_billing_noop(ledger_env): + """free/unlimited（ledger_key=None）：作用域进出零动作。""" + billing = _make_billing(ledger_env, None) + async with billing.scope(): + pass + # 无账本行可动，不抛异常即为通过 + assert _open_rows() == [] + + +def _open_rows(): + conn = auth.get_conn() + try: + rows = conn.execute("SELECT task_id FROM billing_ledger WHERE status='open'").fetchall() + return [dict(r) for r in rows] + finally: + conn.close() diff --git a/web/billing.py b/web/billing.py index c69b501..73a37c6 100644 --- a/web/billing.py +++ b/web/billing.py @@ -12,7 +12,8 @@ import asyncio import logging import secrets -from collections.abc import Callable +from collections.abc import AsyncGenerator, Callable +from contextlib import asynccontextmanager from dataclasses import dataclass import auth_system @@ -63,6 +64,25 @@ class Billing: return True return await asyncio.to_thread(auth_system.ledger_rename, self.ledger_key, task_id) + @asynccontextmanager + async def scope(self) -> AsyncGenerator['Billing', None]: + """计费作用域：退出时自动退还仍 open 的账本行。 + + 把「任何失败路径必须退款」这条不变量从路由的每条早退/except 里 + 收敛到这里。正确性依赖 refund_open 的既有幂等口径（只退仍 open + 的行），因此： + - 错误早退 / 异常转错误响应 → 正常退出作用域 → 自动退； + - settle()/rename_to() 已消费的行不在退款范围，成功路径自然豁免， + 无需手动「取消自动退」； + - 未捕获异常先退款再原样穿透，由路由既有 except 转错误响应； + - 客户端断连（CancelledError）与迁移前行为一致：退款可能被一并 + 取消，open 行留给重启对账兜底，不额外加 shield 复杂化。 + """ + try: + yield self + finally: + await self.refund_open('rejected') + async def require_billing( user: SiteUser | None, diff --git a/web/routers/download.py b/web/routers/download.py index 508acfc..a01d338 100644 --- a/web/routers/download.py +++ b/web/routers/download.py @@ -7,8 +7,9 @@ - POST /api/v1/podcast/download ↔ Flask /podcast/download（计费 1 次） - POST /api/v1/podcast/download/batch_zip ↔ Flask /podcast/download/batch_zip -计费：经 web.billing.require_billing（预扣+开账同事务 / 失败只退 open 行）， -任务登记与启动由 web.tasks_registry 承担； +计费：经 web.billing.require_billing（预扣+开账同事务）+ billing.scope() +（「任何失败路径必须退款」不变量收敛在作用域退出：自动退仍 open 的行， +settle/rename 消费过的行天然豁免）；任务登记与启动由 web.tasks_registry 承担； worker 复用 services 层既有实现，账本与交付语义不变。 """ import asyncio @@ -67,14 +68,14 @@ async def _sync_download_json(downloader, music_id: str, quality: str, cookies: billing, ip: str): """format=json 的同步下载分支：取信息 → 下载 → 组响应 → 结算。 - 语义：任一步失败退回预扣（账本守恒），只有成功才 settle(1, 'delivered')。 + 必须运行在 billing.scope() 内：失败退款由作用域退出统一完成，本函数 + 只负责在成功路径 settle(1, 'delivered')（结算后自动退自然豁免）。 """ try: info = await asyncio.to_thread( downloader.get_music_info, int(music_id), quality, cookies) except Exception as e: logger.error('获取音乐信息失败: %s', e) - await billing.refund_open('rejected') return error('获取失败：歌曲不存在或暂无版权', 404) from services._shared import _build_download_base base = await asyncio.to_thread( @@ -88,11 +89,9 @@ async def _sync_download_json(downloader, music_id: str, quality: str, cookies: cookies=cookies, music_info=info) if not dl.success: logger.error('下载失败: %s', dl.error_message) - await billing.refund_open('rejected') return error('下载失败，请稍后重试或更换音质', 500) except Exception as e: logger.error('下载异常: %s', e) - await billing.refund_open('rejected') return error('服务暂时不可用，请稍后再试', 500) try: _record_log('下载', ip, info.name) @@ -143,8 +142,8 @@ async def _resolve_personal(data: dict[str, Any], user: SiteUser | None, request async def download(user: Annotated[SiteUser | None, Depends(current_user)], request: Request): """下载单曲（file=后台任务 / json=同步返回）。 - 顺序：限流 → 预扣计费 → 四道参数闸（失败即退款）→ 账号 Cookie → - 可选参数 → json 同步分支 / file 后台任务分支。 + 顺序：限流 → 预扣计费 → billing.scope()（失败自动退款）→ 四道参数闸 → + 账号 Cookie → 可选参数 → json 同步分支 / file 后台任务分支。 """ await ratelimit.guard(request, bucket='dl', limit=20, window=60) data = await _request_data(request) @@ -155,69 +154,62 @@ async def download(user: Annotated[SiteUser | None, Depends(current_user)], requ billing = await require_billing(user, action='音乐下载', ip=ip) - async def _fail(msg: str, code: int = 400): - """参数/校验失败：退回本次预扣后返回错误（账本守恒）。""" - await billing.refund_open('rejected') - return error(msg, code) + async with billing.scope(): + try: + if not music_id: + return error("参数 'music_id' 不能为空", 400) + music_id = str(music_id).strip() + if not music_id.isdigit(): + return error('无效的歌曲 ID') + if quality not in VALID_QUALITY_LEVELS: + return error(f"无效的音质参数，支持: {', '.join(VALID_QUALITY_LEVELS)}") + if return_format not in ('file', 'json'): + return error("返回格式只支持 'file' 或 'json'") + music_id = _extract_id(music_id) - try: - if not music_id: - return await _fail("参数 'music_id' 不能为空", 400) - music_id = str(music_id).strip() - if not music_id.isdigit(): - return await _fail('无效的歌曲 ID') - if quality not in VALID_QUALITY_LEVELS: - return await _fail(f"无效的音质参数，支持: {', '.join(VALID_QUALITY_LEVELS)}") - if return_format not in ('file', 'json'): - return await _fail("返回格式只支持 'file' 或 'json'") - music_id = _extract_id(music_id) + account = data.get('account', 'admin') + if account == 'user': + cookies = await _resolve_personal(data, user, request) + if not cookies: + return error('请先扫码登录个人账号后再使用个人解析', 401) + else: + cookies = _cookies() - account = data.get('account', 'admin') - if account == 'user': - cookies = await _resolve_personal(data, user, request) - if not cookies: - return await _fail('请先扫码登录个人账号后再使用个人解析', 401) - else: - cookies = _cookies() + (filename_format, write_meta, meta_fields, + lyric_format, need_zip) = _parse_download_options(data) - (filename_format, write_meta, meta_fields, - lyric_format, need_zip) = _parse_download_options(data) + from services.api_service import api_service + downloader = api_service.downloader - from services.api_service import api_service - downloader = api_service.downloader + if return_format == 'json': + return await _sync_download_json( + downloader, music_id, quality, cookies, + filename_format, write_meta, meta_fields, billing, ip) - if return_format == 'json': - return await _sync_download_json( - downloader, music_id, quality, cookies, - filename_format, write_meta, meta_fields, billing, ip) - - # file 模式：注册后台任务 - from services import downloads as services_downloads - from services import tasks as services_tasks - device_id = str(data.get('device_id') or '').strip()[:64] - await asyncio.to_thread(services_tasks._touch_device, device_id) - task = _single_task({ - 'music_id': music_id, 'quality': quality, - 'filename_format': filename_format, 'write_meta': write_meta, - 'meta_fields': meta_fields, 'lyric_format': lyric_format, - 'need_zip': need_zip, 'cookies': cookies, 'remote_ip': ip, - }, device_id) - reg_err = await register_download_task(billing, task, str(data.get('task_id') or '').strip()) - if reg_err: - await billing.refund_open('rejected') - return error(reg_err, 429) - task_id = task['task_id'] - start_err = await start_task(task_id, services_downloads._single_download_worker, (task_id,)) - if start_err: - return error(start_err, 500) - return success({'task_id': task_id, 'status': 'queued'}, '下载任务已提交') - except APIException: - await billing.refund_open('rejected') - return error('服务暂时不可用，请稍后再试', 500) - except Exception as e: - logger.error('下载音乐异常: %s', e) - await billing.refund_open('rejected') - return error('服务暂时不可用，请稍后再试', 500) + # file 模式：注册后台任务 + from services import downloads as services_downloads + from services import tasks as services_tasks + device_id = str(data.get('device_id') or '').strip()[:64] + await asyncio.to_thread(services_tasks._touch_device, device_id) + task = _single_task({ + 'music_id': music_id, 'quality': quality, + 'filename_format': filename_format, 'write_meta': write_meta, + 'meta_fields': meta_fields, 'lyric_format': lyric_format, + 'need_zip': need_zip, 'cookies': cookies, 'remote_ip': ip, + }, device_id) + reg_err = await register_download_task(billing, task, str(data.get('task_id') or '').strip()) + if reg_err: + return error(reg_err, 429) + task_id = task['task_id'] + start_err = await start_task(task_id, services_downloads._single_download_worker, (task_id,)) + if start_err: + return error(start_err, 500) + return success({'task_id': task_id, 'status': 'queued'}, '下载任务已提交') + except APIException: + return error('服务暂时不可用，请稍后再试', 500) + except Exception as e: + logger.error('下载音乐异常: %s', e) + return error('服务暂时不可用，请稍后再试', 500) def _single_task(params: dict[str, Any], device_id: str) -> dict[str, Any]: @@ -250,69 +242,64 @@ async def download_batch_zip(user: Annotated[SiteUser | None, Depends(current_us return max(1, min(len(_dedup_ids(data.get('ids') or data.get('id') or [])), 500)) billing = await require_billing(user, action='批量ZIP下载', cost=_cost, ip=ip) - try: - raw_ids = data.get('ids') or data.get('id') - ids = _dedup_ids(raw_ids) - if not ids: - await billing.refund_open('rejected') - return error('缺少 ids 参数') - non_digit = [x for x in ids if not x.isdigit()] - if non_digit: - await billing.refund_open('rejected') - return error('ids 中含无效歌曲 ID') - if len(ids) > 500: - await billing.refund_open('rejected') - return error('单个批量任务最多 500 首，请分批提交') - - quality = data.get('quality', 'lossless') - if quality not in VALID_QUALITY_LEVELS: - await billing.refund_open('rejected') - return error(f"无效的音质参数，支持: {', '.join(VALID_QUALITY_LEVELS)}") - filename_format = data.get('filename_format', 'artist_song') - write_meta = str(data.get('write_meta', 'true')).lower() in ('1', 'true', 'yes') - meta_fields = _parse_meta_fields(data.get('meta_fields')) - lyric_format = data.get('lyric_format', 'lrc') - if lyric_format not in ('lrc', 'srt'): - lyric_format = 'lrc' + async with billing.scope(): try: - concurrency = int(data.get('concurrency', 8) or 8) - except (TypeError, ValueError): - concurrency = 8 - concurrency = max(1, min(10, concurrency)) + raw_ids = data.get('ids') or data.get('id') + ids = _dedup_ids(raw_ids) + if not ids: + return error('缺少 ids 参数') + non_digit = [x for x in ids if not x.isdigit()] + if non_digit: + return error('ids 中含无效歌曲 ID') + if len(ids) > 500: + return error('单个批量任务最多 500 首，请分批提交') - cookies = _cookies() - device_id = str(data.get('device_id') or '').strip()[:64] - from services import batch as services_batch - from services import tasks as services_tasks - await asyncio.to_thread(services_tasks._touch_device, device_id) - task = { - 'type': 'batch', - 'total': len(ids), 'done': 0, 'failed': 0, - 'status': 'queued', 'start': time.time(), 'msg': '排队中...', - 'cancel': threading.Event(), 'pause': threading.Event(), - 'device_id': device_id, 'result': None, - 'params': { - 'ids': ids, 'quality': quality, 'filename_format': filename_format, - 'write_meta': write_meta, 'meta_fields': meta_fields, - 'lyric_format': lyric_format, 'concurrency': concurrency, - 'cookies': cookies, 'remote_ip': ip, - 'want_zip': True, - }, - } - reg_err = await register_download_task(billing, task, str(data.get('task_id') or '').strip()) - if reg_err: - await billing.refund_open('rejected') - return error(reg_err, 429) - task_id = task['task_id'] - start_err = await start_task(task_id, services_batch._batch_zip_worker, (task_id,)) - if start_err: - return error(start_err, 500) - return success({'task_id': task_id, 'status': 'queued', 'total': len(ids)}, - '批量打包任务已提交') - except Exception as e: - logger.error('批量打包提交异常: %s', e) - await billing.refund_open('rejected') - return error('服务暂时不可用，请稍后再试', 500) + quality = data.get('quality', 'lossless') + if quality not in VALID_QUALITY_LEVELS: + return error(f"无效的音质参数，支持: {', '.join(VALID_QUALITY_LEVELS)}") + filename_format = data.get('filename_format', 'artist_song') + write_meta = str(data.get('write_meta', 'true')).lower() in ('1', 'true', 'yes') + meta_fields = _parse_meta_fields(data.get('meta_fields')) + lyric_format = data.get('lyric_format', 'lrc') + if lyric_format not in ('lrc', 'srt'): + lyric_format = 'lrc' + try: + concurrency = int(data.get('concurrency', 8) or 8) + except (TypeError, ValueError): + concurrency = 8 + concurrency = max(1, min(10, concurrency)) + + cookies = _cookies() + device_id = str(data.get('device_id') or '').strip()[:64] + from services import batch as services_batch + from services import tasks as services_tasks + await asyncio.to_thread(services_tasks._touch_device, device_id) + task = { + 'type': 'batch', + 'total': len(ids), 'done': 0, 'failed': 0, + 'status': 'queued', 'start': time.time(), 'msg': '排队中...', + 'cancel': threading.Event(), 'pause': threading.Event(), + 'device_id': device_id, 'result': None, + 'params': { + 'ids': ids, 'quality': quality, 'filename_format': filename_format, + 'write_meta': write_meta, 'meta_fields': meta_fields, + 'lyric_format': lyric_format, 'concurrency': concurrency, + 'cookies': cookies, 'remote_ip': ip, + 'want_zip': True, + }, + } + reg_err = await register_download_task(billing, task, str(data.get('task_id') or '').strip()) + if reg_err: + return error(reg_err, 429) + task_id = task['task_id'] + start_err = await start_task(task_id, services_batch._batch_zip_worker, (task_id,)) + if start_err: + return error(start_err, 500) + return success({'task_id': task_id, 'status': 'queued', 'total': len(ids)}, + '批量打包任务已提交') + except Exception as e: + logger.error('批量打包提交异常: %s', e) + return error('服务暂时不可用，请稍后再试', 500) # ============ 播客单集下载 ============ @@ -343,56 +330,51 @@ async def podcast_download(user: Annotated[SiteUser | None, Depends(current_user ip = ratelimit.client_ip(request) billing = await require_billing(user, action='播客单集下载', ip=ip) - try: - if not raw: - await billing.refund_open('rejected') - return error("参数 'id' 不能为空", 400) - from music_api import podcast - quality = _podcast_quality(data) - need_zip = str(data.get('zip', 'false')).lower() in ('1', 'true', 'yes') - ref = podcast.extract_podcast_ref(raw) - if ref['kind'] == 'radio': - await billing.refund_open('rejected') - return error('这是电台链接，请先解析电台后选择单集下载，或使用「打包整个播客」') + async with billing.scope(): + try: + if not raw: + return error("参数 'id' 不能为空", 400) + from music_api import podcast + quality = _podcast_quality(data) + need_zip = str(data.get('zip', 'false')).lower() in ('1', 'true', 'yes') + ref = podcast.extract_podcast_ref(raw) + if ref['kind'] == 'radio': + return error('这是电台链接，请先解析电台后选择单集下载，或使用「打包整个播客」') - account = data.get('account', 'admin') - if account == 'user': - cookies = await _resolve_personal(data, user, request) - if not cookies: - await billing.refund_open('rejected') - return error('请先扫码登录个人账号后再使用个人解析', 401) - else: - cookies = _cookies() + account = data.get('account', 'admin') + if account == 'user': + cookies = await _resolve_personal(data, user, request) + if not cookies: + return error('请先扫码登录个人账号后再使用个人解析', 401) + else: + cookies = _cookies() - device_id = str(data.get('device_id') or '').strip()[:64] - from services import podcast as services_podcast - from services import tasks as services_tasks - await asyncio.to_thread(services_tasks._touch_device, device_id) - task = { - 'type': 'single', - 'total': 1, 'done': 0, 'failed': 0, - 'status': 'queued', 'start': time.time(), 'msg': '排队中...', - 'cancel': threading.Event(), 'pause': threading.Event(), - 'device_id': device_id, 'result': None, - 'params': {'program_id': ref['id'], 'quality': quality, - 'need_zip': need_zip, 'cookies': cookies, 'remote_ip': ip}, - } - reg_err = await register_download_task(billing, task, str(data.get('task_id') or '').strip()) - if reg_err: - await billing.refund_open('rejected') - return error(reg_err, 429) - task_id = task['task_id'] - start_err = await start_task(task_id, services_podcast._podcast_download_worker, (task_id,)) - if start_err: - return error(start_err, 500) - return success({'task_id': task_id, 'status': 'queued'}, '播客单集下载任务已提交') - except APIException as e: - await billing.refund_open('rejected') - return error(str(e), 404) - except Exception as e: - logger.error('播客单集下载提交异常: %s', e) - await billing.refund_open('rejected') - return error('服务暂时不可用，请稍后再试', 500) + device_id = str(data.get('device_id') or '').strip()[:64] + from services import podcast as services_podcast + from services import tasks as services_tasks + await asyncio.to_thread(services_tasks._touch_device, device_id) + task = { + 'type': 'single', + 'total': 1, 'done': 0, 'failed': 0, + 'status': 'queued', 'start': time.time(), 'msg': '排队中...', + 'cancel': threading.Event(), 'pause': threading.Event(), + 'device_id': device_id, 'result': None, + 'params': {'program_id': ref['id'], 'quality': quality, + 'need_zip': need_zip, 'cookies': cookies, 'remote_ip': ip}, + } + reg_err = await register_download_task(billing, task, str(data.get('task_id') or '').strip()) + if reg_err: + return error(reg_err, 429) + task_id = task['task_id'] + start_err = await start_task(task_id, services_podcast._podcast_download_worker, (task_id,)) + if start_err: + return error(start_err, 500) + return success({'task_id': task_id, 'status': 'queued'}, '播客单集下载任务已提交') + except APIException as e: + return error(str(e), 404) + except Exception as e: + logger.error('播客单集下载提交异常: %s', e) + return error('服务暂时不可用，请稍后再试', 500) @router.post('/podcast/download/batch_zip') @@ -406,73 +388,68 @@ async def podcast_batch_zip(user: Annotated[SiteUser | None, Depends(current_use return _podcast_billing_count(data) billing = await require_billing(user, action='播客整台打包', cost=_cost, ip=ip) - try: - raw_radio = data.get('radio_id') - raw_ids = data.get('ids') - if not raw_radio and not raw_ids: - await billing.refund_open('rejected') - return error('缺少 radio_id 或 ids 参数') - limit = max(1, min(_safe_int(data.get('limit', 200) or 200, 200), 500)) - quality = _podcast_quality(data) - cookies = _cookies() - # id_list 只在「按单集 ID 列表」分支赋值，但下面组装任务参数时两个分支都会 - # 引用它（radio_id 为假则取 id_list）——先绑定，避免异常/类型检查下的未绑定名 - id_list: list[str] = [] + async with billing.scope(): + try: + raw_radio = data.get('radio_id') + raw_ids = data.get('ids') + if not raw_radio and not raw_ids: + return error('缺少 radio_id 或 ids 参数') + limit = max(1, min(_safe_int(data.get('limit', 200) or 200, 200), 500)) + quality = _podcast_quality(data) + cookies = _cookies() + # id_list 只在「按单集 ID 列表」分支赋值，但下面组装任务参数时两个分支都会 + # 引用它（radio_id 为假则取 id_list）——先绑定，避免异常/类型检查下的未绑定名 + id_list: list[str] = [] - from music_api import podcast - if raw_radio: - ref = podcast.extract_podcast_ref(raw_radio) - if ref['kind'] == 'program': - await billing.refund_open('rejected') - return error('这是单集链接，请使用「下载单集」或直接输入电台链接/ID') - radio_id = ref['id'] - radio_name = '' - try: - radio_name = podcast.get_radio_detail(radio_id, cookies).get('name', '') - except APIException: - pass - else: - id_list = _dedup_ids(raw_ids)[:limit] - if not id_list: - await billing.refund_open('rejected') - return error('缺少 ids 参数') - radio_id = None - radio_name = str(data.get('name') or '播客单集') + from music_api import podcast + if raw_radio: + ref = podcast.extract_podcast_ref(raw_radio) + if ref['kind'] == 'program': + return error('这是单集链接，请使用「下载单集」或直接输入电台链接/ID') + radio_id = ref['id'] + radio_name = '' + try: + radio_name = podcast.get_radio_detail(radio_id, cookies).get('name', '') + except APIException: + pass + else: + id_list = _dedup_ids(raw_ids)[:limit] + if not id_list: + return error('缺少 ids 参数') + radio_id = None + radio_name = str(data.get('name') or '播客单集') - billed = _podcast_billing_count(data) - device_id = str(data.get('device_id') or '').strip()[:64] - from services import podcast as services_podcast - from services import tasks as services_tasks - await asyncio.to_thread(services_tasks._touch_device, device_id) - task = { - 'type': 'batch', - 'total': 0, 'done': 0, 'failed': 0, - 'status': 'queued', 'start': time.time(), 'msg': '排队中...', - 'cancel': threading.Event(), 'pause': threading.Event(), - 'device_id': device_id, 'result': None, - 'params': { - 'radio_id': radio_id, - 'ids': None if radio_id else id_list, - 'limit': limit, 'quality': quality, 'radio_name': radio_name, - 'cookies': cookies, 'remote_ip': ip, - # 整台模式：worker 解析完实际集数后按差额补扣（一首 = 1 次） - 'billing_uid': user.id if user else None, - 'billed': billed, - }, - } - reg_err = await register_download_task(billing, task, str(data.get('task_id') or '').strip()) - if reg_err: - await billing.refund_open('rejected') - return error(reg_err, 429) - task_id = task['task_id'] - start_err = await start_task(task_id, services_podcast._podcast_zip_worker, (task_id,)) - if start_err: - return error(start_err, 500) - return success({'task_id': task_id, 'status': 'queued'}, '播客打包任务已提交') - except APIException as e: - await billing.refund_open('rejected') - return error(str(e), 404) - except Exception as e: - logger.error('播客打包提交异常: %s', e) - await billing.refund_open('rejected') - return error('服务暂时不可用，请稍后再试', 500) + billed = _podcast_billing_count(data) + device_id = str(data.get('device_id') or '').strip()[:64] + from services import podcast as services_podcast + from services import tasks as services_tasks + await asyncio.to_thread(services_tasks._touch_device, device_id) + task = { + 'type': 'batch', + 'total': 0, 'done': 0, 'failed': 0, + 'status': 'queued', 'start': time.time(), 'msg': '排队中...', + 'cancel': threading.Event(), 'pause': threading.Event(), + 'device_id': device_id, 'result': None, + 'params': { + 'radio_id': radio_id, + 'ids': None if radio_id else id_list, + 'limit': limit, 'quality': quality, 'radio_name': radio_name, + 'cookies': cookies, 'remote_ip': ip, + # 整台模式：worker 解析完实际集数后按差额补扣（一首 = 1 次） + 'billing_uid': user.id if user else None, + 'billed': billed, + }, + } + reg_err = await register_download_task(billing, task, str(data.get('task_id') or '').strip()) + if reg_err: + return error(reg_err, 429) + task_id = task['task_id'] + start_err = await start_task(task_id, services_podcast._podcast_zip_worker, (task_id,)) + if start_err: + return error(start_err, 500) + return success({'task_id': task_id, 'status': 'queued'}, '播客打包任务已提交') + except APIException as e: + return error(str(e), 404) + except Exception as e: + logger.error('播客打包提交异常: %s', e) + return error('服务暂时不可用，请稍后再试', 500)