# -*- coding: utf-8 -*- """下载提交端点。 端点： - POST /api/v1/download ↔ Flask /download（计费 1 次；file/json 两模式） - POST /api/v1/download/batch_zip ↔ Flask /download/batch_zip（按去重歌曲数计费） - 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 承担； worker 复用 services 层既有实现，账本与交付语义不变。 """ import asyncio import logging import threading import time from typing import Annotated, Any from fastapi import APIRouter, Depends, Request from core import media from core.config import VALID_QUALITY_LEVELS from music_api import APIException from web import ratelimit from web.billing import require_billing from web.deps import SiteUser, current_user, require_login from web.request_utils import cookies as _cookies from web.request_utils import extract_id as _extract_id from web.request_utils import request_data as _request_data from web.request_utils import safe_int as _safe_int from web.response import error, success from web.serializers import record_req_log as _record_log from web.tasks_registry import register_download_task, start_task router = APIRouter() LoginUser = Annotated[SiteUser, Depends(require_login)] logger = logging.getLogger('web.download') def _parse_meta_fields(raw: Any): from core.utils import parse_meta_fields return parse_meta_fields(raw) def _dedup_ids(raw_ids: Any) -> list[str]: if isinstance(raw_ids, str): raw_list = raw_ids.split(',') else: raw_list = [str(x) for x in (raw_ids or [])] out, seen = [], set() for x in raw_list: mid = _extract_id(x) s = str(mid).strip() if s and s not in seen: seen.add(s) out.append(s) return out async def _resolve_personal(data: dict[str, Any], user: SiteUser | None, request: Request): from web.routers.music import _resolve_user_cookies return await _resolve_user_cookies(data, user, request) # ============ 单曲下载 ============ @router.post('/download') async def download(user: Annotated[SiteUser | None, Depends(current_user)], request: Request): """下载单曲（file=后台任务 / json=同步返回）。""" await ratelimit.guard(request, bucket='dl', limit=20, window=60) data = await _request_data(request) music_id = data.get('id') quality = data.get('quality', 'lossless') return_format = data.get('format', 'file') ip = ratelimit.client_ip(request) 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) 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 await _fail('请先扫码登录个人账号后再使用个人解析', 401) else: cookies = _cookies() 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' need_zip = str(data.get('zip', 'false')).lower() in ('1', 'true', 'yes') import services.runtime as services downloader = services.api_service.downloader if return_format == 'json': 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) base = await asyncio.to_thread( services._build_download_base, filename_format, {'name': info.name, 'artist_string': info.artists}) filename = f'{base}.{info.file_type}' try: dl = await asyncio.to_thread( downloader.download_music_file, int(music_id), quality, write_meta=write_meta, meta_fields=meta_fields, 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) except Exception: pass response_data = { 'music_id': music_id, 'name': info.name, 'artist': info.artists.replace('/', ', '), 'album': info.album, 'quality': quality, 'quality_name': media.quality_display_name(quality), 'file_type': info.file_type, 'file_size': info.file_size, 'file_size_formatted': media.format_file_size(info.file_size), 'filename': filename, 'duration': info.duration * 1000, } await billing.settle(1, 'delivered') return success(response_data, '下载完成') # file 模式：注册后台任务 device_id = str(data.get('device_id') or '').strip()[:64] await asyncio.to_thread(services._touch_device, device_id) task = _single_task(data, music_id, quality, filename_format, write_meta, meta_fields, lyric_format, need_zip, cookies, 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._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) def _single_task(data, music_id, quality, filename_format, write_meta, meta_fields, lyric_format, need_zip, cookies, ip, device_id) -> dict[str, Any]: return { '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': { '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, }, } # ============ 批量 ZIP ============ @router.post('/download/batch_zip') async def download_batch_zip(user: Annotated[SiteUser | None, Depends(current_user)], request: Request): """批量打包 ZIP（按去重歌曲数计费）。""" await ratelimit.guard(request, bucket='dl', limit=20, window=60) data = await _request_data(request) ip = ratelimit.client_ip(request) def _cost() -> int: 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' 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] import services.runtime as services await asyncio.to_thread(services._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_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) # ============ 播客单集下载 ============ def _podcast_billing_count(data: dict[str, Any]) -> int: """播客打包计费。""" if data.get('radio_id'): return 1 limit = _safe_int(data.get('limit', 200) or 200, 200) limit = max(1, min(limit, 500)) ids = _dedup_ids(data.get('ids')) return max(1, min(len(ids), limit)) def _podcast_quality(data: dict[str, Any]) -> str: q = str(data.get('quality') or 'auto') return q if q in ('auto', 'standard', 'exhigh', 'lossless', 'hires') else 'auto' @router.post('/podcast/download') async def podcast_download(user: Annotated[SiteUser | None, Depends(current_user)], request: Request): """播客单集下载（计费 1 次）。""" await ratelimit.guard(request, bucket='dl', limit=20, window=60) data = await _request_data(request) raw = data.get('id') 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) import podcast_api quality = _podcast_quality(data) need_zip = str(data.get('zip', 'false')).lower() in ('1', 'true', 'yes') ref = podcast_api.extract_podcast_ref(raw) if ref['kind'] == 'radio': await billing.refund_open('rejected') 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() device_id = str(data.get('device_id') or '').strip()[:64] import services.runtime as services await asyncio.to_thread(services._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_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) @router.post('/podcast/download/batch_zip') async def podcast_batch_zip(user: Annotated[SiteUser | None, Depends(current_user)], request: Request): """整台播客打包（先扣 1（电台模式）或按去重单集数扣，worker 按实际集数补扣差额）。""" await ratelimit.guard(request, bucket='dl', limit=20, window=60) data = await _request_data(request) ip = ratelimit.client_ip(request) def _cost() -> int: 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] = [] import podcast_api if raw_radio: ref = podcast_api.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_api.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 '播客单集') billed = _podcast_billing_count(data) device_id = str(data.get('device_id') or '').strip()[:64] import services.runtime as services await asyncio.to_thread(services._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_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)