From 9cbd341e2e65e4c0ec1a0fbde91ff26b641e50af Mon Sep 17 00:00:00 2001 From: snowy-yang <102860110+snowy-yang@users.noreply.github.com> Date: Sun, 27 Sep 2026 09:03:05 +0800 Subject: [PATCH] =?UTF-8?q?refactor:=20=E5=8D=95=E6=9B=B2=E4=B8=8B?= =?UTF-8?q?=E8=BD=BD=20worker=20=E6=8B=86=E8=A7=A3=EF=BC=88177=20=E8=A1=8C?= =?UTF-8?q?=20/=20=E5=9C=88=2036=20=E2=86=92=2043=20=E8=A1=8C=20/=20?= =?UTF-8?q?=E5=9C=88=208=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 为什么：_single_download_worker 是单曲下载的**唯一**后台执行路径（/download 的 file 模式由 start_task 启动它），177 行 / 圈复杂度 36 / 嵌套 5，且此前**零测试 覆盖**——改坏任意一处都不会有测试报警。它的三个 except 分支各写一遍「置终态 + 退款 + 删任务目录」（9 行 × 3），正文里又散着四处逐字相同的取消检查。 改法：一步一个具名函数，主函数只留线性流程。 SingleTaskParams 参数快照（frozen dataclass），免去每步重列长参数表 _begin_single_task 置 running + 取参数 _fetch_single_metadata 一次预取歌曲信息 + 算交付基础名 _single_progress_cb 构造字节级进度回调（限频 0.4s） _abort_if_canceled 取消检查点（原三处逐字重复 → 一处） _single_file_result 不打包的交付描述 _pack_single_zip 音频 + 歌词 + 封面 + info.txt 打成一包 _publish_single_task 置 done（已终态不覆盖）+ 结算 + 记日志 _fail_single_task 失败/取消统一收尾（原三处 × 9 行 → 一处） 复杂度（同口径 AST 实测，HEAD 与工作区由同一脚本测出）： _single_download_worker 177 行 / 圈 36 / 嵌套 5 → 43 行 / 圈 8 / 嵌套 2 文件最大圈复杂度 36（就是它） → 9（_single_progress_cb， 节流回调本身的形状） 文件行数 234 → 310（+76，全是新签名与 docstring）· 函数数 3 → 11 等价性证据（三道，互相独立）： 1. 19 例行为契约测试**在未改动的原始实现上一次性全绿**（先写测试、后动代码； 改完再跑仍全绿）——这批测试因此忠实编码了旧行为，构成等价性验证，而不是 「照着新实现写的断言」。断言只依赖任务终态 / 磁盘文件 / ZIP 内容 / 账本行 / 余额，不引用任何内部函数名，可跨后续重构复用。 2. 反向注入验证：把 HEAD 版注入回 services.downloads（**sys.modules 与父包属性 两处都要挂**——否则 monkeypatch.resolve 走 getattr 父包拿到真模块），同一套 19 例**全过**，与第 1 条相互独立地再次证明测试忠实。 3. AST 字面量多重集比对 + 逐条位置归因：字符串 69 种**一个不少**——没有一条 用户可见文案被动过。数值差异全部归因： 0 4→2：三处 ledger_settle(task_id, 0, ...) 收敛成一处（理由参数化） True 5→4：三处 rmtree(..., ignore_errors=True) 收敛成一处（另 +1 为新增 dataclass 的 frozen=True） '任务已取消' 5→3：三处 raise DownloadCanceled 收敛进 _abort_if_canceled 刻意保留的语义（是决定，不是疏漏）： - task_dir_path 仍在 try 外初始化、失败收尾仍用它判空——「下载未开始就失败」 时无目录可删，语义原样。 - 结算失败**不**把 done 改标 error（钱由重启对账兜底，宁可商家小损不坑用户）， 已用 test_settle_failure_does_not_flip_done_to_error 固化。 - 末段「已终态不得覆盖为 done」的状态检查与取消检查合流进 _publish_single_task， 仍是先到先得；用 test_terminal_state_not_overwritten_by_late_success 固化。 - 日志调用移到各自 except 分支开头（原在收尾之后），仅调整同一分支内的顺序， 不影响任何可观察行为；三处格式串原样保留。 - music_id 保持**原样透传**：params 里是数字字符串，而下游签名声明 int，两侧 一直不一致、运行时靠字符串参与上游查询拼装而正常工作。顺手 int() 会把异常面 从「上游查不到」变成 ValueError，属行为变更，故只把标注写成 Any 并在 docstring 记明。 测试：新增 tests/test_single_worker.py（19 例）——正常交付与交付副本位置、 ZIP 四件套与 SRT 歌词、空歌词跳过、取封面失败仍发布、文件名格式、操作记录与 remote_ip 兜底、下载失败退费、未知异常泛化文案（上游细节不外泄）、下载中/下载 后/打包中三种取消、终态不被覆盖、进度回调三态（有总量 / 无总量 / 自身抛错）、 结算失败不影响已交付、worker 崩在结算前留给对账兜底。 顺带修正一处自己的假设错误：最初把「取封面失败」写成 fetch_cover_bytes 抛异常， 实测该函数契约是**失败返回 None**（内部已 except Exception），抛异常是不存在的 形状——改为注入 None 后才在旧实现上跑绿。测试要贴合生产的真实形状。 验证：ruff 全过 · pyright 0/0 · pytest 294 passed（275→294）· 生产 data/ 未被 触碰（全部用例走 tmp_path 沙箱）。 --- services/downloads.py | 389 ++++++++++++++---------- tests/test_single_worker.py | 580 ++++++++++++++++++++++++++++++++++++ 2 files changed, 816 insertions(+), 153 deletions(-) create mode 100644 tests/test_single_worker.py diff --git a/services/downloads.py b/services/downloads.py index c8e2384..aef3b30 100644 --- a/services/downloads.py +++ b/services/downloads.py @@ -3,14 +3,22 @@ 账号探测（VIP/音质）与二维码登录客户端均已迁出（前者只剩 web.routers.account 的异步实现，后者只剩 web.routers.admin 自己的客户端），本模块只负责单曲任务。 + +``_single_download_worker`` 原是 177 行 / 圈复杂度 30 的单函数，三个 except 分支 +各写一遍「结算 + 删任务目录」，正文里又散着四处取消检查。此处按「一步一个具名 +函数」拆开：主函数只留线性流程，失败收尾收敛到 ``_fail_single_task`` 一处。 """ import os import shutil +import threading import time import traceback import zipfile +from collections.abc import Callable +from dataclasses import dataclass from pathlib import Path +from typing import Any import auth_system from core.state import ( @@ -21,6 +29,7 @@ from core.utils import format_lyric as _format_lyric from music_downloader import ( DownloadCanceled, DownloadException, + MusicInfo, fetch_cover_bytes, ) @@ -55,180 +64,254 @@ def _task_delivery_copy(task_id: str, src: Path, filename: str) -> str: # rmtree，本函数全文件零调用（审计确认）。 -def _single_download_worker(task_id: str) -> None: - """后台单曲下载：下载音频（可选 ZIP 打包），支持取消；完成后由 /download/result 交付，35 分钟保留期清理""" +@dataclass(frozen=True) +class SingleTaskParams: + """单曲任务参数快照（worker 只读）。 + + 原来是 10 个局部变量从 ``task['params']`` 逐个取出、再原样穿过整段流程； + 收成一个 frozen 对象后，各步骤按需取字段，调用点不必重列一遍长参数表。 + ``cancel_ev`` 不是 params 里的值，而是任务对象上的取消事件，一并放这里 + 以免每一步都多传一个参数。 + + ``music_id`` 刻意保持 **原样透传**（不转换）：它来自前端、经 ``_extract_id`` + 归一化成数字字符串，而下游 ``get_music_info`` / ``download_music_file`` 的 + 签名声明 ``int``——两侧一直不一致，运行时靠字符串参与上游查询拼装而正常工作 + （调用链上无算术）。类型标注为 ``Any`` 如实反映这个输入边界的实际形状； + 顺手 ``int()`` 会改变异常面（非数字时从「上游查不到」变成 ``ValueError``）， + 属于行为变更，不在重构范围内。 + """ + + music_id: Any + quality: str + filename_format: str + write_meta: bool + meta_fields: Any + lyric_format: str + need_zip: bool + cookies: dict[str, str] + remote_ip: str + cancel_ev: threading.Event + + +def _begin_single_task(task_id: str) -> SingleTaskParams: + """把任务置为 running 并取出参数快照。""" + with DOWNLOAD_TASKS_LOCK: + task = DOWNLOAD_TASKS[task_id] + task['status'] = 'running' + task['msg'] = '下载中...' + params = task['params'] + return SingleTaskParams( + music_id=params['music_id'], + quality=params['quality'], + filename_format=params['filename_format'], + write_meta=params['write_meta'], + meta_fields=params['meta_fields'], + lyric_format=params['lyric_format'], + need_zip=params['need_zip'], + cookies=params['cookies'], + remote_ip=params.get('remote_ip') or '-', + cancel_ev=task['cancel'], + ) + + +def _fetch_single_metadata(p: SingleTaskParams) -> tuple[MusicInfo, dict[str, Any], str, str]: + """一次预取歌曲完整信息（URL/详情/歌词三接口并行），并算出交付基础名。 + + 替代原先 name_v1 + url_v1 + 下载器内部 get_music_info 的重复串行调用。 + 返回 ``(MusicInfo, 下载器口径的 music_info, base, filename)``。 + """ + mi = api_service.downloader.get_music_info(p.music_id, p.quality, p.cookies) + music_info = { + 'id': p.music_id, + 'name': mi.name, + 'artist_string': mi.artists, + 'album': mi.album, + 'pic_url': mi.pic_url, + 'file_type': mi.file_type, + 'file_size': mi.file_size, + 'duration': mi.duration * 1000, + 'download_url': mi.download_url, + } + base = _build_download_base(p.filename_format, music_info) + return mi, music_info, base, f"{base}.{music_info['file_type']}" + + +def _single_progress_cb(task_id: str) -> Callable[[int, int], None]: + """构造字节级进度回调（限频 0.4s）：写进任务 live_percent。 + + /api/batch_progress 对运行中的单曲任务优先返回它——原实现 done 按首计数， + 单曲进度条全程 0%。回调内任何故障都被吞掉，绝不影响下载主流程。 + """ + last_ts = {'t': 0.0} + + def _cb(written: int, total_bytes: int) -> None: + try: + now = time.time() + if now - last_ts['t'] < 0.4 and written < (total_bytes or 0): + return + last_ts['t'] = now + with DOWNLOAD_TASKS_LOCK: + t = DOWNLOAD_TASKS.get(task_id) + if not t or t.get('status') != 'running': + return + if total_bytes: + pct = min(99.0, written / total_bytes * 100) + t['live_percent'] = pct + t['msg'] = (f"下载中 {int(pct)}% · " + f"{api_service._format_file_size(written)}/{api_service._format_file_size(total_bytes)}") + else: + t['msg'] = f"下载中 · 已接收 {api_service._format_file_size(written)}" + except Exception: + pass # 进度上报故障绝不影响下载 + + return _cb + + +def _abort_if_canceled(cancel_ev: threading.Event) -> None: + """取消检查点：下载完成后到发布前，每一步之间都要过这道闸。 + + 原先这三处检查（下载后 / 打包后 / 发布前）逐字重复，且末段那次还兼 + 「不得把已取消覆盖成 done」的职责——后者在 ``_publish_single_task`` 里另做。 + """ + if cancel_ev.is_set(): + raise DownloadCanceled("任务已取消") + + +def _single_file_result(file_path_str: str, filename: str, + music_info: dict[str, Any]) -> dict[str, Any]: + """不打包（zip=false）的交付描述。""" + return { + 'file_path': file_path_str, + 'filename': filename, + 'mimetype': f"audio/{music_info['file_type']}", + 'audio_files': [file_path_str], + } + + +def _pack_single_zip(file_path_str: str, base: str, filename: str, mi: MusicInfo, + music_info: dict[str, Any], p: SingleTaskParams) -> dict[str, Any]: + """把音频 + 歌词 + 封面 + info.txt 打成单曲 ZIP 交付包。 + + 歌词与封面都是**尽力而为**：歌词在预取时已一并拿到（无需再请求一次）， + 封面走共享缓存；两者拿不到就不写对应条目，不影响音频交付。 + """ import tempfile as _tempfile - file_path_str: str | None = None - zip_tmp_path: str | None = None + fd, zip_tmp_path = _tempfile.mkstemp(prefix='single_zip_', suffix='.zip', + dir=str(Path(file_path_str).parent)) + os.close(fd) + with zipfile.ZipFile(zip_tmp_path, 'w', zipfile.ZIP_DEFLATED) as zf: + zf.write(file_path_str, arcname=filename) + # 歌词已在预取时一并拿到，无需再请求一次 + if mi.lyric: + try: + zf.writestr(f"{base}.{p.lyric_format}", _format_lyric(mi.lyric, p.lyric_format)) + except Exception: + pass + # 封面走共享缓存（与元数据写入共用同一次网络请求） + cover = fetch_cover_bytes(mi.pic_url) + if cover: + zf.writestr(f"{base}.jpg", cover) + info_txt = ( + f"歌曲: {music_info['name']}\n" + f"歌手: {music_info['artist_string']}\n" + f"专辑: {music_info['album']}\n" + f"音质: {api_service._get_quality_display_name(p.quality)}\n" + ) + zf.writestr("info.txt", info_txt) + return { + 'file_path': file_path_str, + 'zip_path': zip_tmp_path, + 'filename': f"{base}.zip", + 'mimetype': 'application/zip', + 'audio_files': [file_path_str], + } + + +def _publish_single_task(task_id: str, result: dict[str, Any], remote_ip: str, + name: str, filename: str) -> None: + """最终发布：置 done（已终态不覆盖）→ 结算 → 记操作日志。 + + 交付已落定；结算失败**不再**把任务改标 error——钱由重启对账兜底， + 宁可商家小损不坑用户。末段取消检查与此处的状态检查合流：任务已被外部 + 置成 canceled/error 时，worker 的成功路径不得把它覆盖回 done。 + """ + with DOWNLOAD_TASKS_LOCK: + t = DOWNLOAD_TASKS.get(task_id) + if t is None or t.get('status') in ('canceled', 'error'): + raise DownloadCanceled("任务已取消") + t['status'] = 'done' + t['completed_at'] = time.time() + t['done'] = 1 + t['msg'] = '下载完成' + t['result'] = result + try: + auth_system.ledger_settle(task_id, 1, 'delivered') + except Exception as _e: + api_service.logger.error(f"单曲交付结算失败（重启对账兜底）: {task_id} {_e}") + try: + _record_log('下载', remote_ip, name) + except Exception: + pass + api_service.logger.info(f"单曲下载完成: {filename} ({task_id})") + + +def _fail_single_task(task_id: str, status: str, msg: str, ledger_status: str, + task_dir_path: str | None) -> None: + """失败/取消的统一收尾：置终态 → 退款 → 清掉任务目录。 + + 这三个 except 分支原先各写一遍（共 9 行 × 3 处）；收敛后新增一种失败 + 只需要挑一个 ``ledger_status``，不会再漏掉清理或退款。 + """ + _mark_task_terminal(task_id, status, msg) + try: + auth_system.ledger_settle(task_id, 0, ledger_status) + except Exception: + pass + if task_dir_path: + shutil.rmtree(task_dir_path, ignore_errors=True) + + +def _single_download_worker(task_id: str) -> None: + """后台单曲下载：下载音频（可选 ZIP 打包），支持取消；完成后由 /download/result 交付，35 分钟保留期清理""" task_dir_path: str | None = None try: - with DOWNLOAD_TASKS_LOCK: - task = DOWNLOAD_TASKS[task_id] - task['status'] = 'running' - task['msg'] = '下载中...' - params = task['params'] - cancel_ev = task['cancel'] - music_id = params['music_id'] - quality = params['quality'] - filename_format = params['filename_format'] - write_meta = params['write_meta'] - meta_fields = params['meta_fields'] - lyric_format = params['lyric_format'] - need_zip = params['need_zip'] - cookies = params['cookies'] - remote_ip = params.get('remote_ip') or '-' - - # 一次预取歌曲完整信息（URL/详情/歌词三接口并行）， - # 替代原先 name_v1 + url_v1 + 下载器内部 get_music_info 的重复串行调用 - mi = api_service.downloader.get_music_info(music_id, quality, cookies) - music_info = { - 'id': music_id, - 'name': mi.name, - 'artist_string': mi.artists, - 'album': mi.album, - 'pic_url': mi.pic_url, - 'file_type': mi.file_type, - 'file_size': mi.file_size, - 'duration': mi.duration * 1000, - 'download_url': mi.download_url, - } - base = _build_download_base(filename_format, music_info) - filename = f"{base}.{music_info['file_type']}" - - # 字节级进度回调（限频 0.4s）：写进任务 live_percent，/api/batch_progress - # 对运行中的单曲任务优先返回它——原实现 done 按首计数，单曲进度条全程 0% - _cb_ts = {'t': 0.0} - - def _progress_cb(written, total_bytes): - try: - now = time.time() - if now - _cb_ts['t'] < 0.4 and written < (total_bytes or 0): - return - _cb_ts['t'] = now - with DOWNLOAD_TASKS_LOCK: - t = DOWNLOAD_TASKS.get(task_id) - if not t or t.get('status') != 'running': - return - if total_bytes: - pct = min(99.0, written / total_bytes * 100) - t['live_percent'] = pct - t['msg'] = (f"下载中 {int(pct)}% · " - f"{api_service._format_file_size(written)}/{api_service._format_file_size(total_bytes)}") - else: - t['msg'] = f"下载中 · 已接收 {api_service._format_file_size(written)}" - except Exception: - pass # 进度上报故障绝不影响下载 + p = _begin_single_task(task_id) + mi, music_info, base, filename = _fetch_single_metadata(p) download_result = api_service.downloader.download_music_file( - music_id, quality, - write_meta=write_meta, - meta_fields=meta_fields, - cookies=cookies, - cancel_event=cancel_ev, + p.music_id, p.quality, + write_meta=p.write_meta, + meta_fields=p.meta_fields, + cookies=p.cookies, + cancel_event=p.cancel_ev, music_info=mi, - progress_cb=_progress_cb, + progress_cb=_single_progress_cb(task_id), ) if not download_result.success: raise DownloadException(download_result.error_message or "下载失败") - # 下载完成后的每一阶段都可能已被取消，发布前统一校验 - if cancel_ev.is_set(): - raise DownloadCanceled("任务已取消") + _abort_if_canceled(p.cancel_ev) assert download_result.file_path is not None - cache_path = Path(download_result.file_path) # 交付文件用任务专属副本，共享缓存不进清理清单 - file_path_str = _task_delivery_copy(task_id, cache_path, filename) + file_path_str = _task_delivery_copy(task_id, Path(download_result.file_path), filename) task_dir_path = str(Path(file_path_str).parent) - if need_zip: - fd, zip_tmp_path = _tempfile.mkstemp(prefix='single_zip_', suffix='.zip', - dir=str(Path(file_path_str).parent)) - os.close(fd) - with zipfile.ZipFile(zip_tmp_path, 'w', zipfile.ZIP_DEFLATED) as zf: - zf.write(file_path_str, arcname=filename) - # 歌词已在预取时一并拿到，无需再请求一次 - if mi.lyric: - try: - zf.writestr(f"{base}.{lyric_format}", _format_lyric(mi.lyric, lyric_format)) - except Exception: - pass - # 封面走共享缓存（与元数据写入共用同一次网络请求） - cover = fetch_cover_bytes(mi.pic_url) - if cover: - zf.writestr(f"{base}.jpg", cover) - info_txt = ( - f"歌曲: {music_info['name']}\n" - f"歌手: {music_info['artist_string']}\n" - f"专辑: {music_info['album']}\n" - f"音质: {api_service._get_quality_display_name(quality)}\n" - ) - zf.writestr("info.txt", info_txt) - # 打包阶段同样以取消检查收口 - if cancel_ev.is_set(): - raise DownloadCanceled("任务已取消") - result = { - 'file_path': file_path_str, - 'zip_path': zip_tmp_path, - 'filename': f"{base}.zip", - 'mimetype': 'application/zip', - 'audio_files': [file_path_str], - } + if p.need_zip: + result = _pack_single_zip(file_path_str, base, filename, mi, music_info, p) else: - result = { - 'file_path': file_path_str, - 'filename': filename, - 'mimetype': f"audio/{music_info['file_type']}", - 'audio_files': [file_path_str], - } + result = _single_file_result(file_path_str, filename, music_info) + # 打包阶段同样以取消检查收口 + _abort_if_canceled(p.cancel_ev) - # 最终发布前的取消检查（末段取消不得覆盖为 done） - if cancel_ev.is_set(): - raise DownloadCanceled("任务已取消") - with DOWNLOAD_TASKS_LOCK: - t = DOWNLOAD_TASKS.get(task_id) - if t is None or t.get('status') in ('canceled', 'error'): - raise DownloadCanceled("任务已取消") - t['status'] = 'done' - t['completed_at'] = time.time() - t['done'] = 1 - t['msg'] = '下载完成' - t['result'] = result - # 交付已落定；结算失败不再把任务改标 error（钱由重启对账兜底，宁可商家小损不坑用户） - try: - auth_system.ledger_settle(task_id, 1, 'delivered') - except Exception as _e: - api_service.logger.error(f"单曲交付结算失败（重启对账兜底）: {task_id} {_e}") - try: - _record_log('下载', remote_ip, music_info['name']) - except Exception: - pass - api_service.logger.info(f"单曲下载完成: {filename} ({task_id})") + _publish_single_task(task_id, result, p.remote_ip, music_info['name'], filename) except DownloadCanceled: - _mark_task_terminal(task_id, 'canceled', '任务已取消') - try: - auth_system.ledger_settle(task_id, 0, 'canceled') - except Exception: - pass - if task_dir_path: - shutil.rmtree(task_dir_path, ignore_errors=True) api_service.logger.info(f"单曲下载已取消: {task_id}") + _fail_single_task(task_id, 'canceled', '任务已取消', 'canceled', task_dir_path) except DownloadException as e: - _mark_task_terminal(task_id, 'error', str(e)) - try: - auth_system.ledger_settle(task_id, 0, 'error') - except Exception: - pass - if task_dir_path: - shutil.rmtree(task_dir_path, ignore_errors=True) api_service.logger.error(f"单曲下载失败: {task_id} {e}") + _fail_single_task(task_id, 'error', str(e), 'error', task_dir_path) except Exception as e: # 非预期异常的 repr 可能含上游主机/连接细节——用户可见文案 # 泛化为固定文案，原文 logger 先行留档（教训：先记录再泛化） api_service.logger.error(f"单曲下载异常: {task_id} {e}\n{traceback.format_exc()}") - _mark_task_terminal(task_id, 'error', '下载失败，请稍后重试') - try: - auth_system.ledger_settle(task_id, 0, 'error') - except Exception: - pass - if task_dir_path: - shutil.rmtree(task_dir_path, ignore_errors=True) + _fail_single_task(task_id, 'error', '下载失败，请稍后重试', 'error', task_dir_path) diff --git a/tests/test_single_worker.py b/tests/test_single_worker.py new file mode 100644 index 0000000..1a29915 --- /dev/null +++ b/tests/test_single_worker.py @@ -0,0 +1,580 @@ +# -*- coding: utf-8 -*- +"""单曲下载 worker 的行为契约测试（沙箱，不触碰生产 data/）。 + +为什么：``services/downloads.py::_single_download_worker`` 是单曲下载的**唯一**后台 +执行路径（``/download`` 的 file 模式走 ``start_task`` 启动它），177 行 / 圈复杂度 30， +此前**零测试覆盖**。它的三个 except 分支各写一遍「结算 + 删任务目录」，正文里又散着 +四处取消检查——改坏任意一处都不会有测试报警。 + +按惯例：**先在未改动的原始实现上跑绿**，这批测试因此忠实编码了旧行为，重构时本身 +即等价性验证器。断言只依赖「调用 worker 后世界里发生了什么」（任务终态、磁盘上的 +文件、ZIP 内容、账本行、余额），不引用任何内部函数名，故可跨后续重构复用。 + +账本口径：一律用 ``reserve_and_open`` 真预扣（``ledger_open`` 只记账不扣款， +会让「退款」凭空给余额加次数，守恒断言就失去意义）。 +""" +import threading +import time +import zipfile +from pathlib import Path +from typing import Any + +import pytest + +import auth_system as auth +import core.db as core_db +from core.state import DOWNLOAD_TASKS, DOWNLOAD_TASKS_LOCK +from music_downloader import DownloadCanceled, DownloadResult, MusicInfo +from services import downloads as downloads_svc +from services.api_service import api_service + +PW = 'Passw0rd12' +CREDITS = 50 + + +# ============ 测试替身 ============ + +class _NullLogger: + """吞掉 api_service.logger 输出，避免测试往项目根 music_api.log 追加。""" + + def debug(self, *a: Any, **k: Any) -> None: ... + def info(self, *a: Any, **k: Any) -> None: ... + def warning(self, *a: Any, **k: Any) -> None: ... + def error(self, *a: Any, **k: Any) -> None: ... + + +class _FakeDownloader: + """可编程下载器替身。 + + ``on_download`` 在 ``download_music_file`` 内部被调用，拿到调用时的 kwargs， + 用于在「下载中途」注入取消 / 调用进度回调 / 抛异常。 + ``info_error`` 让 ``get_music_info`` 抛异常（验证未知异常路径）。 + """ + + def __init__(self, cache_dir: Path, *, download_ok: bool = True, + info_error: Exception | None = None, + on_download: Any = None, lyric: str = '[00:01.00]第一行\n[00:02.00]第二行'): + self.cache_dir = cache_dir + self.download_ok = download_ok + self.info_error = info_error + self.on_download = on_download + self.lyric = lyric + self.info_calls = 0 + self.download_calls = 0 + + def get_music_info(self, mid: Any, quality: str, cookies: Any) -> MusicInfo: + self.info_calls += 1 + if self.info_error is not None: + raise self.info_error + return MusicInfo( + id=int(mid), name='歌名X', artists='歌手A', album='专辑A', + pic_url='https://pic.example/x.jpg', duration=200, track_number=1, + download_url='', file_type='mp3', file_size=1024, + quality='standard', lyric=self.lyric, + ) + + def download_music_file(self, mid: Any, quality: str, **kw: Any) -> DownloadResult: + self.download_calls += 1 + if self.on_download is not None: + self.on_download(kw) + cancel_ev = kw.get('cancel_event') + if cancel_ev is not None and cancel_ev.is_set(): + raise DownloadCanceled('任务已取消') + if not self.download_ok: + return DownloadResult(success=False, error_message='上游 404') + path = self.cache_dir / f'{mid}.mp3' + path.write_bytes(b'ID3\x03\x00\x00\x00' + b'\x00' * 64) + return DownloadResult(success=True, file_path=str(path), + file_size=path.stat().st_size, music_info=kw.get('music_info')) + + +# ============ 沙箱 ============ + +@pytest.fixture() +def sandbox(tmp_path, monkeypatch): + """临时站点库 + 临时下载目录 + 空任务表。""" + db_path = tmp_path / 'site.db' + monkeypatch.setattr(auth.db, 'DATA_DIR', tmp_path) + monkeypatch.setattr(auth.db, 'DB_PATH', db_path) + monkeypatch.setattr(auth.db, 'BACKUP_DIR', tmp_path / 'backups') + monkeypatch.setattr(auth.db, 'SECRET_KEY_PATH', tmp_path / 'secret.key') + monkeypatch.setattr(core_db, 'DATA_DIR', tmp_path) + monkeypatch.setattr(core_db, 'DB_PATH', db_path) + auth.init_db() + auth.create_user('payer', PW, email='p@example.com', credits=CREDITS) + + downloads = tmp_path / 'downloads' + downloads.mkdir() + cache = downloads / 'cache' + cache.mkdir() + monkeypatch.setattr(api_service, 'downloads_path', downloads) + monkeypatch.setattr(api_service, 'logger', _NullLogger()) + # 封面走网络——测试里默认返回固定字节，个例可覆盖 + monkeypatch.setattr(downloads_svc, 'fetch_cover_bytes', lambda url: b'\xff\xd8\xff cover') + # 操作记录写进程内共享列表——隔离掉，断言改由个例捕获 + monkeypatch.setattr(downloads_svc, '_record_log', lambda *a: None) + + with DOWNLOAD_TASKS_LOCK: + DOWNLOAD_TASKS.clear() + yield {'root': tmp_path, 'downloads': downloads, 'cache': cache} + with DOWNLOAD_TASKS_LOCK: + DOWNLOAD_TASKS.clear() + + +# ============ 工具 ============ + +def _uid(username: str = 'payer') -> int: + u = auth.get_user_by_name(username) + assert u is not None + return int(u['id']) + + +def _balance(username: str = 'payer') -> int: + u = auth.get_user_by_name(username) + assert u is not None + return int(u['duration_seconds'] or 0) + + +def _status(task_id: str) -> str: + with DOWNLOAD_TASKS_LOCK: + return str(DOWNLOAD_TASKS.get(task_id, {}).get('status') or '') + + +def _register_single(sandbox, *, music_id: int = 1, user_id: int, + reserved: int = 1, **params: Any) -> tuple[str, dict[str, Any]]: + tid = 's' + str(int(time.time() * 1000000) % 10 ** 10) + # reserve_and_open：真预扣 + 开账同事务（ledger_open 只记账不扣款，见模块 docstring） + left = auth.reserve_and_open(user_id, reserved, tid) + assert left is not None, '预扣失败（余额不足，检查 fixture 的 credits）' + task: dict[str, Any] = { + 'type': 'single', 'task_id': tid, + 'user_id': user_id, 'billing_user_id': user_id, 'billing_reserved': reserved, + 'total': 1, 'done': 0, 'failed': 0, + 'status': 'queued', 'start': time.time(), 'msg': '排队中...', + 'cancel': threading.Event(), 'pause': threading.Event(), + 'device_id': 'dev-test', 'result': None, + 'params': { + 'music_id': music_id, 'quality': 'standard', + 'filename_format': 'artist_song', 'write_meta': False, + 'meta_fields': None, 'lyric_format': 'lrc', + 'need_zip': False, 'cookies': {}, 'remote_ip': '1.2.3.4', + **params, + }, + } + with DOWNLOAD_TASKS_LOCK: + DOWNLOAD_TASKS[tid] = task + return tid, task + + +def _run(task_id: str) -> threading.Thread: + th = threading.Thread(target=downloads_svc._single_download_worker, + args=(task_id,), daemon=True) + th.start() + return th + + +def _zip_names(path: str) -> list[str]: + with zipfile.ZipFile(path) as zf: + return zf.namelist() + + +# ============ 正常路径 ============ + +def test_single_download_completes(sandbox, monkeypatch): + """常规单曲（不打包）：终态 done、交付文件落在任务目录、账本按交付结算。""" + uid = _uid() + before = _balance() + monkeypatch.setattr(api_service, 'downloader', _FakeDownloader(sandbox['cache'])) + tid, task = _register_single(sandbox, music_id=101, user_id=uid) + + th = _run(tid) + th.join(timeout=30) + assert not th.is_alive(), 'worker 未在 30s 内结束' + + assert _status(tid) == 'done', f"终态={_status(tid)} msg={task.get('msg')}" + assert task['msg'] == '下载完成' + assert task['done'] == 1 and task['completed_at'] > 0 + res = task['result'] + assert res is not None + assert res['filename'] == '歌手A - 歌名X.mp3' + assert res['mimetype'] == 'audio/mp3' + assert res['audio_files'] == [res['file_path']] + assert 'zip_path' not in res + + # 交付副本在 tasks// 下，与共享缓存分离 + delivered = Path(res['file_path']) + assert delivered.exists() + assert delivered.parent == sandbox['downloads'] / 'tasks' / tid + assert delivered.read_bytes() == (sandbox['cache'] / '101.mp3').read_bytes() + assert (sandbox['cache'] / '101.mp3').exists(), '共享缓存不应被搬走' + + row = auth.ledger_get(tid) + assert row is not None and row['status'] != 'open' + assert row['delivered'] == 1 + assert _balance() == before - 1, '交付成功：只扣预扣的 1 次，无退款' + + +def test_single_zip_contains_audio_lyric_cover_and_info(sandbox, monkeypatch): + """zip=true：包内含音频 + 歌词 + 封面 + info.txt（信息栏四行）。""" + uid = _uid() + monkeypatch.setattr(api_service, 'downloader', _FakeDownloader(sandbox['cache'])) + tid, task = _register_single(sandbox, music_id=102, user_id=uid, need_zip=True) + + th = _run(tid) + th.join(timeout=30) + + assert _status(tid) == 'done', f"终态={_status(tid)} msg={task.get('msg')}" + res = task['result'] + assert res is not None + assert res['filename'] == '歌手A - 歌名X.zip' + assert res['mimetype'] == 'application/zip' + names = sorted(_zip_names(res['zip_path'])) + assert names == ['info.txt', '歌手A - 歌名X.jpg', '歌手A - 歌名X.lrc', '歌手A - 歌名X.mp3'], names + + with zipfile.ZipFile(res['zip_path']) as zf: + info = zf.read('info.txt').decode('utf-8') + assert '歌曲: 歌名X' in info + assert '歌手: 歌手A' in info + assert '专辑: 专辑A' in info + assert info.startswith('歌曲: ') + + +def test_single_zip_lyric_format_srt(sandbox, monkeypatch): + """lyric_format=srt：歌词条目扩展名随之改变。""" + uid = _uid() + monkeypatch.setattr(api_service, 'downloader', _FakeDownloader(sandbox['cache'])) + tid, task = _register_single(sandbox, music_id=103, user_id=uid, + need_zip=True, lyric_format='srt') + th = _run(tid) + th.join(timeout=30) + + assert _status(tid) == 'done' + assert '歌手A - 歌名X.srt' in _zip_names(task['result']['zip_path']) + + +def test_empty_lyric_skips_lyric_entry(sandbox, monkeypatch): + """上游无歌词（mi.lyric 为空）：不写歌词条目，其余照常。""" + uid = _uid() + monkeypatch.setattr(api_service, 'downloader', + _FakeDownloader(sandbox['cache'], lyric='')) + tid, task = _register_single(sandbox, music_id=104, user_id=uid, need_zip=True) + th = _run(tid) + th.join(timeout=30) + + assert _status(tid) == 'done' + names = _zip_names(task['result']['zip_path']) + assert not [n for n in names if n.endswith('.lrc')], names + assert [n for n in names if n.endswith('.mp3')], names + + +def test_cover_failure_keeps_zip_without_cover(sandbox, monkeypatch): + """取封面失败：ZIP 仍发布，只是没有封面条目（封面非交付必要条件）。 + + 注意 ``fetch_cover_bytes`` 的契约是**失败返回 None**（内部已 except Exception）， + 所以这里注入的是返回 None，不是抛异常——后者是不存在的形状。 + """ + uid = _uid() + monkeypatch.setattr(api_service, 'downloader', _FakeDownloader(sandbox['cache'])) + monkeypatch.setattr(downloads_svc, 'fetch_cover_bytes', lambda url: None) + tid, task = _register_single(sandbox, music_id=105, user_id=uid, need_zip=True) + th = _run(tid) + th.join(timeout=30) + + assert _status(tid) == 'done', f"终态={_status(tid)} msg={task.get('msg')}" + names = _zip_names(task['result']['zip_path']) + assert not [n for n in names if n.endswith('.jpg')], names + + +def test_filename_format_song_only(sandbox, monkeypatch): + """filename_format=song：交付名只用歌名。""" + uid = _uid() + monkeypatch.setattr(api_service, 'downloader', _FakeDownloader(sandbox['cache'])) + tid, task = _register_single(sandbox, music_id=106, user_id=uid, + filename_format='song') + th = _run(tid) + th.join(timeout=30) + + assert _status(tid) == 'done' + assert task['result']['filename'] == '歌名X.mp3' + + +def test_record_log_gets_song_name(sandbox, monkeypatch): + """成功交付后按歌名记一条操作记录（IP 取 params.remote_ip）。""" + uid = _uid() + logs: list[tuple] = [] + monkeypatch.setattr(api_service, 'downloader', _FakeDownloader(sandbox['cache'])) + monkeypatch.setattr(downloads_svc, '_record_log', lambda *a: logs.append(a)) + tid, _ = _register_single(sandbox, music_id=107, user_id=uid) + th = _run(tid) + th.join(timeout=30) + + assert logs == [('下载', '1.2.3.4', '歌名X')], logs + + +def test_remote_ip_defaults_when_missing(sandbox, monkeypatch): + """params 无 remote_ip：按 '-' 记录，不因缺字段报错。""" + uid = _uid() + logs: list[tuple] = [] + monkeypatch.setattr(api_service, 'downloader', _FakeDownloader(sandbox['cache'])) + monkeypatch.setattr(downloads_svc, '_record_log', lambda *a: logs.append(a)) + tid, task = _register_single(sandbox, music_id=108, user_id=uid) + del task['params']['remote_ip'] + th = _run(tid) + th.join(timeout=30) + + assert _status(tid) == 'done' + assert logs == [('下载', '-', '歌名X')] + + +# ============ 失败与取消 ============ + +def test_download_failure_marks_error_and_refunds(sandbox, monkeypatch): + """下载器返回 success=False：转 error、全额退回预扣、任务目录不残留。""" + uid = _uid() + before = _balance() + monkeypatch.setattr(api_service, 'downloader', + _FakeDownloader(sandbox['cache'], download_ok=False)) + tid, task = _register_single(sandbox, music_id=201, user_id=uid) + + th = _run(tid) + th.join(timeout=30) + + assert _status(tid) == 'error', f"终态={_status(tid)} msg={task.get('msg')}" + assert task['msg'] == '上游 404' + assert task['result'] is None + assert _balance() == before, '失败应全额退回预扣' + row = auth.ledger_get(tid) + assert row is not None and row['delivered'] == 0 and row['status'] != 'open' + assert not (sandbox['downloads'] / 'tasks' / tid).exists(), '失败应清掉任务目录' + + +def test_unexpected_exception_uses_generic_message(sandbox, monkeypatch): + """非预期异常：用户可见文案泛化（不泄上游细节），原文进日志、余额退回。""" + uid = _uid() + before = _balance() + monkeypatch.setattr(api_service, 'downloader', + _FakeDownloader(sandbox['cache'], + info_error=RuntimeError('upstream host=10.0.0.9 refused'))) + tid, task = _register_single(sandbox, music_id=202, user_id=uid) + + th = _run(tid) + th.join(timeout=30) + + assert _status(tid) == 'error', f"终态={_status(tid)}" + assert task['msg'] == '下载失败，请稍后重试' + assert '10.0.0.9' not in task['msg'], '上游细节不得出现在用户可见文案里' + assert _balance() == before + assert not (sandbox['downloads'] / 'tasks' / tid).exists() + + +def test_cancel_mid_download(sandbox, monkeypatch): + """下载中被取消：worker 收敛 canceled、全额退费、任务目录清掉。""" + uid = _uid() + before = _balance() + holder: dict[str, Any] = {} + + def _cancel_inside(kw): + holder['task']['cancel'].set() + + monkeypatch.setattr(api_service, 'downloader', + _FakeDownloader(sandbox['cache'], on_download=_cancel_inside)) + tid, task = _register_single(sandbox, music_id=203, user_id=uid) + holder['task'] = task + + th = _run(tid) + th.join(timeout=30) + + assert _status(tid) == 'canceled', f"终态={_status(tid)} msg={task.get('msg')}" + assert task['msg'] == '任务已取消' + assert task['result'] is None + assert _balance() == before, '取消应全额退回预扣' + row = auth.ledger_get(tid) + assert row is not None and row['delivered'] == 0 + assert not (sandbox['downloads'] / 'tasks' / tid).exists() + + +def test_cancel_after_download_before_publish(sandbox, monkeypatch): + """下载已落盘但取消已置位：不发布交付（不发布 result），转 canceled 并退费。""" + uid = _uid() + before = _balance() + holder: dict[str, Any] = {} + + def _cancel_after(kw): + # 下载返回之后再置位——命中「下载完成后的取消检查」 + def _later(): + holder['task']['cancel'].set() + + threading.Timer(0.0, _later).start() + + monkeypatch.setattr(api_service, 'downloader', + _FakeDownloader(sandbox['cache'], on_download=_cancel_after)) + tid, task = _register_single(sandbox, music_id=204, user_id=uid) + holder['task'] = task + + th = _run(tid) + th.join(timeout=30) + + assert _status(tid) == 'canceled', f"终态={_status(tid)} msg={task.get('msg')}" + assert task['result'] is None, '取消后不得发布交付' + assert _balance() == before + assert not (sandbox['downloads'] / 'tasks' / tid).exists() + + +def test_cancel_during_zip_stage(sandbox, monkeypatch): + """打包阶段被取消：ZIP 不发布、转 canceled、退费。""" + uid = _uid() + before = _balance() + holder: dict[str, Any] = {} + + def _cancel_after(kw): + threading.Timer(0.0, lambda: holder['task']['cancel'].set()).start() + + monkeypatch.setattr(api_service, 'downloader', + _FakeDownloader(sandbox['cache'], on_download=_cancel_after)) + tid, task = _register_single(sandbox, music_id=205, user_id=uid, need_zip=True) + holder['task'] = task + + th = _run(tid) + th.join(timeout=30) + + assert _status(tid) == 'canceled', f"终态={_status(tid)} msg={task.get('msg')}" + assert task['result'] is None + assert _balance() == before + assert not (sandbox['downloads'] / 'tasks' / tid).exists() + + +def test_terminal_state_not_overwritten_by_late_success(sandbox, monkeypatch): + """任务已被外部置终态（canceled）时，worker 的成功路径不得覆盖为 done。""" + uid = _uid() + before = _balance() + holder: dict[str, Any] = {} + + def _mark_canceled(kw): + with DOWNLOAD_TASKS_LOCK: + holder['task']['status'] = 'canceled' + holder['task']['completed_at'] = time.time() + + monkeypatch.setattr(api_service, 'downloader', + _FakeDownloader(sandbox['cache'], on_download=_mark_canceled)) + tid, task = _register_single(sandbox, music_id=206, user_id=uid) + holder['task'] = task + + th = _run(tid) + th.join(timeout=30) + + assert _status(tid) == 'canceled', f"终态={_status(tid)}" + assert task['result'] is None, '已终态不得被覆盖为 done' + assert _balance() == before + + +# ============ 进度上报 ============ + +def test_progress_callback_writes_live_percent(sandbox, monkeypatch): + """进度回调写 live_percent 与「下载中 x% · 已收/总量」文案。""" + uid = _uid() + seen: dict[str, Any] = {} + + def _emit_progress(kw): + cb = kw.get('progress_cb') + if cb is not None: + cb(512, 1024) + with DOWNLOAD_TASKS_LOCK: + t = DOWNLOAD_TASKS.get(seen['tid']) or {} + seen['live'] = t.get('live_percent') + seen['msg'] = t.get('msg') + + monkeypatch.setattr(api_service, 'downloader', + _FakeDownloader(sandbox['cache'], on_download=_emit_progress)) + tid, _task = _register_single(sandbox, music_id=301, user_id=uid) + seen['tid'] = tid + + th = _run(tid) + th.join(timeout=30) + + assert seen.get('live') == 50.0, seen + assert '下载中 50%' in str(seen.get('msg')), seen + assert _status(tid) == 'done', '进度上报不应影响最终交付' + + +def test_progress_callback_without_total_uses_received_bytes(sandbox, monkeypatch): + """总量未知：文案退化为「已接收 <大小>」，不写 live_percent。""" + uid = _uid() + seen: dict[str, Any] = {} + + def _emit_progress(kw): + cb = kw.get('progress_cb') + if cb is not None: + cb(2048, 0) + with DOWNLOAD_TASKS_LOCK: + t = DOWNLOAD_TASKS.get(seen['tid']) or {} + seen['live'] = t.get('live_percent') + seen['msg'] = t.get('msg') + + monkeypatch.setattr(api_service, 'downloader', + _FakeDownloader(sandbox['cache'], on_download=_emit_progress)) + tid, _task = _register_single(sandbox, music_id=302, user_id=uid) + seen['tid'] = tid + + th = _run(tid) + th.join(timeout=30) + + assert '已接收' in str(seen.get('msg')), seen + assert seen.get('live') is None, seen + + +def test_progress_callback_errors_do_not_break_download(sandbox, monkeypatch): + """进度回调自身抛错：被吞掉，下载照常完成。""" + uid = _uid() + + def _bad_progress(kw): + cb = kw.get('progress_cb') + if cb is not None: + # 回调内部访问非法对象——必须被 except 吞掉 + cb('not-a-number', None) + + monkeypatch.setattr(api_service, 'downloader', + _FakeDownloader(sandbox['cache'], on_download=_bad_progress)) + tid, task = _register_single(sandbox, music_id=303, user_id=uid) + th = _run(tid) + th.join(timeout=30) + + assert _status(tid) == 'done', f"终态={_status(tid)} msg={task.get('msg')}" + + +# ============ 结算兜底 ============ + +def test_settle_failure_does_not_flip_done_to_error(sandbox, monkeypatch): + """交付已落定后结算抛错：任务仍为 done（钱由重启对账兜底，不坑已付费用户）。""" + uid = _uid() + + def _boom(*a, **k): + raise RuntimeError('库锁超时') + + monkeypatch.setattr(api_service, 'downloader', _FakeDownloader(sandbox['cache'])) + monkeypatch.setattr(downloads_svc.auth_system, 'ledger_settle', _boom) + tid, task = _register_single(sandbox, music_id=401, user_id=uid) + + th = _run(tid) + th.join(timeout=30) + + assert _status(tid) == 'done', f"终态={_status(tid)} msg={task.get('msg')}" + assert task['result'] is not None + + +def test_ledger_open_when_worker_crashes_before_settle(sandbox, monkeypatch): + """worker 在结算前崩掉：账本行仍为 open（留给重启对账兜底，不是静默丢失）。""" + uid = _uid() + monkeypatch.setattr(api_service, 'downloader', + _FakeDownloader(sandbox['cache'], + info_error=RuntimeError('boom'))) + tid, _ = _register_single(sandbox, music_id=402, user_id=uid) + + # 让结算也失败，模拟「连退款都没走成」的极端情况 + monkeypatch.setattr(downloads_svc.auth_system, 'ledger_settle', + lambda *a, **k: (_ for _ in ()).throw(RuntimeError('db busy'))) + th = _run(tid) + th.join(timeout=30) + + row = auth.ledger_get(tid) + assert row is not None + assert row['status'] == 'open', f"应留给对账兜底，实际 status={row['status']}"