From 1c6b17f718b88427cebe3ac44135eb5dfe241d0a Mon Sep 17 00:00:00 2001 From: snowy-yang <102860110+snowy-yang@users.noreply.github.com> Date: Sun, 27 Sep 2026 09:30:52 +0800 Subject: [PATCH] =?UTF-8?q?refactor:=20=E6=92=AD=E5=AE=A2=E5=8D=95?= =?UTF-8?q?=E9=9B=86=20worker=20=E6=8B=86=E8=A7=A3=EF=BC=88140=20=E8=A1=8C?= =?UTF-8?q?=20/=20=E5=9C=88=2035=20=E2=86=92=2040=20=E8=A1=8C=20/=20?= =?UTF-8?q?=E5=9C=88=206=20/=20=E5=B5=8C=E5=A5=97=201=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 为什么：_podcast_download_worker 是播客单集下载的唯一后台执行路径， 此前零直接测试覆盖；三条 except 分支各写一遍「结算+删任务目录」， 正文散着三处取消检查，改坏任意一处不会有测试报警。 拆出（与单曲 worker 9cbd341 同风格）： - _PodcastTaskSpec（frozen dataclass，program_id 标注 Any——params 历史 存数字字符串，worker 只透传，int() 是行为变更） - _PodcastTarget（磁盘名/交付名/任务目录的命名组装） - _begin_podcast_task / _prepare_podcast_target / _write_podcast_zip / _podcast_file_result / _podcast_zip_result / _publish_podcast_task / _fail_podcast_task（三处 ×9 行收尾收敛） - 取消检查复用 downloads._abort_if_canceled（无循环依赖）；tempfile 局部 import 上移模块顶 先写测试后动代码：新增 tests/test_podcast_worker.py 18 例，在**未改动** 实现上一次跑绿 = 忠实编码旧行为，重构后同一套测试即等价性验证。 唯一语义顺序变化：except 分支的日志从「收尾后」挪到「收尾前」（与 单曲 worker 对齐，日志内容不变，无行为影响）。 门禁：ruff 全过 · pyright 0/0 · pytest 312 passed（294→312）。 --- services/podcast.py | 305 +++++++++++-------- tests/test_podcast_worker.py | 550 +++++++++++++++++++++++++++++++++++ 2 files changed, 737 insertions(+), 118 deletions(-) create mode 100644 tests/test_podcast_worker.py diff --git a/services/podcast.py b/services/podcast.py index 14f1c2e..0d2dfe2 100644 --- a/services/podcast.py +++ b/services/podcast.py @@ -7,6 +7,7 @@ """ import os import shutil +import tempfile import threading import time import traceback @@ -45,6 +46,9 @@ from ._zipcore import ( from .api_service import ( api_service, ) +from .downloads import ( + _abort_if_canceled, +) from .streaming import ( _stream_open, ) @@ -166,146 +170,211 @@ def _podcast_resolve_audio(program: dict[str, Any], quality: str, return url_info +@dataclass(frozen=True) +class _PodcastTaskSpec: + """worker 启动时从任务表取出的不可变执行参数。 + + ``program_id`` 标注 Any 与单曲 worker（SingleTaskParams.music_id）同因： + 任务 params 里历史存的是数字字符串，下游上游解析各自处理，worker 只透传 + 不做 int() 转换（那是行为变更）。 + """ + + program_id: Any + quality: str + need_zip: bool + cookies: dict[str, Any] + remote_ip: str + cancel_ev: threading.Event + + +@dataclass(frozen=True) +class _PodcastTarget: + """单集落盘目标与交付命名。 + + 磁盘名附节目 ID + 任务目录隔离（两个任务同下载一集互不覆盖、互不删除）； + 交付名保持干净（不带 ID）。 + """ + + task_dir: Path + target: Path + filename: str # 交付名（含扩展名） + base: str # 交付名去扩展名（ZIP 条目名与封面名共用） + name: str # 单集原始名（info.txt 与操作记录用） + ext: str + + +def _begin_podcast_task(task_id: str) -> _PodcastTaskSpec: + """置 running 并取出执行参数（任务不存在时 KeyError → 通用 except 收敛 error）。""" + with DOWNLOAD_TASKS_LOCK: + task = DOWNLOAD_TASKS[task_id] + task['status'] = 'running' + task['msg'] = '解析单集信息...' + params = task['params'] + return _PodcastTaskSpec( + program_id=params['program_id'], + quality=params['quality'], + need_zip=params['need_zip'], + cookies=params['cookies'], + remote_ip=params.get('remote_ip') or '-', + cancel_ev=task['cancel'], + ) + + +def _prepare_podcast_target(task_id: str, program_id: Any, program: dict[str, Any], + url_info: dict[str, Any]) -> _PodcastTarget: + """组装文件名与任务专属目录（顺带 mkdir）。""" + serial = int(program.get('serial_num') or 0) + name = program.get('name') or program.get('main_song_name') or '播客单集' + prefix = f"{serial:03d} - " if serial else '' + base = _podcast_safe_name(f"{prefix}{name}") + ext = (url_info.get('type') or 'mp3').lower().lstrip('.') + task_dir = api_service.downloads_path / 'tasks' / task_id + task_dir.mkdir(parents=True, exist_ok=True) + return _PodcastTarget( + task_dir=task_dir, + target=task_dir / f"{base} [{program_id}].{ext}", + filename=f"{base}.{ext}", + base=base, + name=name, + ext=ext, + ) + + +def _write_podcast_zip(zip_path: str, file_path_str: str, arcname: str, base: str, + name: str, program: dict[str, Any], url_info: dict[str, Any], + fallback_quality: str) -> None: + """单集入包：音频 + 封面（可选，失败不炸任务）+ info.txt。 + + info.txt 不再写直链（真实 URL 不应随交付文件外泄）。 + """ + with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zf: + zf.write(file_path_str, arcname=arcname) + cover_url = program.get('cover_url') or '' + if cover_url: + try: + cover = fetch_cover_bytes(cover_url) + if cover: + zf.writestr(f"{base}.jpg", cover) + except Exception: + pass + info_txt = ( + f"单集: {name}\n" + f"播客: {program.get('radio_name') or ''}\n" + f"主播: {program.get('dj_name') or ''}\n" + f"时长: {program.get('duration_text') or ''}\n" + f"音质: {api_service._get_quality_display_name(url_info.get('level', fallback_quality))}\n" + ) + zf.writestr("info.txt", info_txt) + + +def _podcast_file_result(file_path_str: str, filename: str, ext: str) -> dict[str, Any]: + """不打包时的交付形状。""" + return { + 'file_path': file_path_str, + 'filename': filename, + 'mimetype': f"audio/{ext}", + 'audio_files': [file_path_str], + } + + +def _podcast_zip_result(prep: _PodcastTarget, file_path_str: str, + program: dict[str, Any], url_info: dict[str, Any], + spec: _PodcastTaskSpec) -> dict[str, Any]: + """打包交付形状：ZIP 中间文件建在任务目录下（失败收尾 rmtree 一并覆盖）。""" + fd, zip_tmp_path = tempfile.mkstemp(prefix='podcast_zip_', suffix='.zip', + dir=str(prep.task_dir)) + os.close(fd) + _write_podcast_zip(zip_tmp_path, file_path_str, arcname=prep.filename, + base=prep.base, name=prep.name, program=program, + url_info=url_info, fallback_quality=spec.quality) + return { + 'file_path': file_path_str, + 'zip_path': zip_tmp_path, + 'filename': f"{prep.base}.zip", + 'mimetype': 'application/zip', + 'audio_files': [file_path_str], + } + + +def _publish_podcast_task(task_id: str, result: dict[str, Any], remote_ip: str, + name: str, filename: str) -> None: + """最终发布：置 done（已终态不覆盖）→ 结算 → 记操作日志。 + + 交付已落定；结算失败**不再**把任务改标 error——钱由重启对账兜底， + 宁可商家小损不坑用户（与单曲 _publish_single_task 同口径）。 + """ + 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, f"[播客] {name}") + except Exception: + pass + api_service.logger.info(f"播客单集下载完成: {filename} ({task_id})") + + +def _fail_podcast_task(task_id: str, status: str, msg: str, ledger_status: str, + task_dir: Path | None) -> None: + """失败/取消的统一收尾：置终态 → 退款 → 清掉任务目录（含 ZIP 中间件）。""" + _mark_task_terminal(task_id, status, msg) + try: + auth_system.ledger_settle(task_id, 0, ledger_status) + except Exception: + pass + if task_dir is not None: + shutil.rmtree(str(task_dir), ignore_errors=True) + + def _podcast_download_worker(task_id: str) -> None: """后台播客单集下载：解析单集 → 音质降级取直链 → 流式下载（可选 ZIP 打包） 任务 type 为 'single'，完成后经 /download/result 交付，与单曲下载共用交付通道。 落盘文件进入任务专属目录且带节目 ID，跨任务永不重名；清理只动本任务目录。 """ - import tempfile as _tempfile - - file_path_str: str | None = None - zip_tmp_path: str | None = None task_dir: Path | None = None try: - with DOWNLOAD_TASKS_LOCK: - task = DOWNLOAD_TASKS[task_id] - task['status'] = 'running' - task['msg'] = '解析单集信息...' - params = task['params'] - cancel_ev = task['cancel'] - program_id = params['program_id'] - quality = params['quality'] - need_zip = params['need_zip'] - cookies = params['cookies'] - remote_ip = params.get('remote_ip') or '-' - - program = podcast.get_program_detail(program_id, cookies) - url_info = _podcast_resolve_audio(program, quality, cookies) - - serial = int(program.get('serial_num') or 0) - name = program.get('name') or program.get('main_song_name') or '播客单集' - prefix = f"{serial:03d} - " if serial else '' - base = _podcast_safe_name(f"{prefix}{name}") - ext = (url_info.get('type') or 'mp3').lower().lstrip('.') - # 磁盘名附节目 ID + 任务目录隔离（两个任务同下载一集互不覆盖、互不删除） - task_dir = api_service.downloads_path / 'tasks' / task_id - task_dir.mkdir(parents=True, exist_ok=True) - disk_name = f"{base} [{program_id}].{ext}" - target = task_dir / disk_name - filename = f"{base}.{ext}" # 交付名保持干净（不带 ID） + spec = _begin_podcast_task(task_id) + program = podcast.get_program_detail(spec.program_id, spec.cookies) + url_info = _podcast_resolve_audio(program, spec.quality, spec.cookies) + prep = _prepare_podcast_target(task_id, spec.program_id, program, url_info) + task_dir = prep.task_dir with DOWNLOAD_TASKS_LOCK: t = DOWNLOAD_TASKS[task_id] t['msg'] = '下载中...' - _podcast_stream_download(url_info['url'], target, url_info.get('size', 0), - cancel_ev, task_id, '下载中') - if cancel_ev.is_set(): - raise DownloadCanceled("任务已取消") - file_path_str = str(target) + _podcast_stream_download(url_info['url'], prep.target, url_info.get('size', 0), + spec.cancel_ev, task_id, '下载中') + _abort_if_canceled(spec.cancel_ev) + file_path_str = str(prep.target) - result: dict[str, Any] - if need_zip: - fd, zip_tmp_path = _tempfile.mkstemp(prefix='podcast_zip_', suffix='.zip', - dir=str(task_dir)) - os.close(fd) - with zipfile.ZipFile(zip_tmp_path, 'w', zipfile.ZIP_DEFLATED) as zf: - zf.write(file_path_str, arcname=filename) - cover_url = program.get('cover_url') or '' - if cover_url: - try: - cover = fetch_cover_bytes(cover_url) - if cover: - zf.writestr(f"{base}.jpg", cover) - except Exception: - pass - # info.txt 不再写直链（真实 URL 不应随交付文件外泄） - info_txt = ( - f"单集: {name}\n" - f"播客: {program.get('radio_name') or ''}\n" - f"主播: {program.get('dj_name') or ''}\n" - f"时长: {program.get('duration_text') or ''}\n" - f"音质: {api_service._get_quality_display_name(url_info.get('level', 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 spec.need_zip: + result = _podcast_zip_result(prep, file_path_str, program, url_info, spec) else: - result = { - 'file_path': file_path_str, - 'filename': filename, - 'mimetype': f"audio/{ext}", - 'audio_files': [file_path_str], - } + result = _podcast_file_result(file_path_str, prep.filename, prep.ext) + # 打包阶段同样以取消检查收口（末段取消不得覆盖为 done） + _abort_if_canceled(spec.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, f"[播客] {name}") - except Exception: - pass - api_service.logger.info(f"播客单集下载完成: {filename} ({task_id})") + _publish_podcast_task(task_id, result, spec.remote_ip, prep.name, prep.filename) except DownloadCanceled: - _mark_task_terminal(task_id, 'canceled', '任务已取消') - try: - auth_system.ledger_settle(task_id, 0, 'canceled') - except Exception: - pass - if task_dir is not None: - shutil.rmtree(str(task_dir), ignore_errors=True) api_service.logger.info(f"播客单集下载已取消: {task_id}") + _fail_podcast_task(task_id, 'canceled', '任务已取消', 'canceled', task_dir) 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 is not None: - shutil.rmtree(str(task_dir), ignore_errors=True) api_service.logger.error(f"播客单集下载失败: {task_id} {e}") + _fail_podcast_task(task_id, 'error', str(e), 'error', task_dir) except Exception as e: # 非预期异常原文 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 is not None: - shutil.rmtree(str(task_dir), ignore_errors=True) + _fail_podcast_task(task_id, 'error', '下载失败，请稍后重试', 'error', task_dir) def _podcast_cleanup_files(paths: list[str | None]) -> None: diff --git a/tests/test_podcast_worker.py b/tests/test_podcast_worker.py new file mode 100644 index 0000000..870216c --- /dev/null +++ b/tests/test_podcast_worker.py @@ -0,0 +1,550 @@ +# -*- coding: utf-8 -*- +"""播客单集下载 worker 的行为契约测试（沙箱，不触碰生产 data/）。 + +为什么：``services/podcast.py::_podcast_download_worker`` 是播客单集下载的**唯一** +后台执行路径（解析 → 音质降级直链 → 流式下载 → 可选 ZIP → /download/result 交付）， +此前**零直接测试覆盖**（test_podcast_stream.py 只覆盖到流式下载这一层）。它有三条 +except 分支各写一遍「结算 + 删任务目录」，正文里散着三处取消检查——改坏任意一处 +都不会有测试报警。 + +按惯例：**先在未改动的原始实现上跑绿**，这批测试因此忠实编码了旧行为，重构时本身 +即等价性验证器。断言只依赖「调用 worker 后世界里发生了什么」（任务终态、磁盘上的 +文件、ZIP 内容、账本行、余额），不引用 worker 内部函数名。 + +替身边界（与真实契约一一对应）： +- ``podcast.get_program_detail`` / ``podcast.resolve_episode_url``：可编程节目信息与直链； +- ``_podcast_stream_download``：契约是「成功写 target 并返回字节数；取消抛 + DownloadCanceled；失败抛 DownloadException」，替身照此实现，不模拟字节流； +- ``fetch_cover_bytes``：失败返回 None（真实契约内部吞异常），不抛。 + +账本口径：一律用 ``reserve_and_open`` 真预扣（与 test_single_worker.py 同理由）。 +""" +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, DownloadException +from services import podcast as podcast_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: ... + + +def _program(**over: Any) -> dict[str, Any]: + """标准节目详情替身（字段与 music_api.podcast.get_program_detail 对齐）。""" + d: dict[str, Any] = { + 'main_song_id': 999, + 'name': '第一集 测试', + 'serial_num': 3, + 'cover_url': 'https://cover.example/x.jpg', + 'radio_name': '测试电台', + 'dj_name': '主播甲', + 'duration_text': '10:00', + } + d.update(over) + return d + + +def _url_info(**over: Any) -> dict[str, Any]: + d: dict[str, Any] = {'url': 'https://cdn.example/a.mp3', 'type': 'mp3', + 'size': 11, 'level': 'standard'} + d.update(over) + return d + + +class _FakeStream: + """``_podcast_stream_download`` 替身。 + + ``on_call`` 在写入 target 之前被调用（注入取消置位 / 外部终态 / 抛错）； + ``fail`` 模拟流式下载失败（如完整性不达标）。 + """ + + def __init__(self, *, fail: Exception | None = None, on_call: Any = None): + self.fail = fail + self.on_call = on_call + self.calls: list[dict[str, Any]] = [] + + def __call__(self, url: str, target: Path, total_size: int, + cancel_ev: threading.Event, task_id: str = '', + progress_label: str = '') -> int: + self.calls.append({'url': url, 'target': Path(target), + 'total_size': total_size}) + if self.on_call is not None: + self.on_call() + if cancel_ev.is_set(): + raise DownloadCanceled('任务已取消') + if self.fail is not None: + raise self.fail + Path(target).write_bytes(b'audio-bytes') + return 11 + + +# ============ 沙箱 ============ + +@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() + monkeypatch.setattr(api_service, 'downloads_path', downloads) + monkeypatch.setattr(api_service, 'logger', _NullLogger()) + # 操作记录写进程内共享列表——隔离掉，断言改由个例捕获 + monkeypatch.setattr(podcast_svc, '_record_log', lambda *a: None) + + with DOWNLOAD_TASKS_LOCK: + DOWNLOAD_TASKS.clear() + yield {'root': tmp_path, 'downloads': downloads} + 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_podcast(sandbox, *, program_id: int = 777, user_id: int, + reserved: int = 1, **params: Any) -> tuple[str, dict[str, Any]]: + tid = 'p' + str(int(time.time() * 1000000) % 10 ** 10) + 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': { + 'program_id': program_id, 'quality': 'standard', + 'need_zip': False, 'cookies': {}, 'remote_ip': '1.2.3.4', + **params, + }, + } + with DOWNLOAD_TASKS_LOCK: + DOWNLOAD_TASKS[tid] = task + return tid, task + + +def _install_upstream(monkeypatch, *, program: dict[str, Any] | None = None, + url_info: dict[str, Any] | None = None, + stream: Any = None, + cover: Any = lambda url: b'\xff\xd8\xff cover') -> None: + """装上播客上游替身（节目详情 / 直链 / 流式下载 / 封面）。""" + monkeypatch.setattr(podcast_svc.podcast, 'get_program_detail', + lambda pid, cookies: program if program is not None else _program()) + monkeypatch.setattr(podcast_svc.podcast, 'resolve_episode_url', + lambda sid, cookies, quality: + url_info if url_info is not None else _url_info()) + if stream is None: + stream = _FakeStream() + monkeypatch.setattr(podcast_svc, '_podcast_stream_download', stream) + monkeypatch.setattr(podcast_svc, 'fetch_cover_bytes', cover) + + +def _run(task_id: str) -> threading.Thread: + th = threading.Thread(target=podcast_svc._podcast_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_podcast_download_completes(sandbox, monkeypatch): + """常规单集（不打包）：终态 done、音频落在任务目录、账本按交付结算、余额 -1。""" + uid = _uid() + before = _balance() + _install_upstream(monkeypatch) + tid, task = _register_podcast(sandbox, program_id=777, 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 + # 交付名干净（不带节目 ID），磁盘名带 [节目ID] + assert res['filename'] == '003 - 第一集 测试.mp3' + assert res['mimetype'] == 'audio/mp3' + assert res['audio_files'] == [res['file_path']] + assert 'zip_path' not in res + + delivered = Path(res['file_path']) + assert delivered.exists() + assert delivered.name == '003 - 第一集 测试 [777].mp3' + assert delivered.parent == sandbox['downloads'] / 'tasks' / tid + assert delivered.read_bytes() == b'audio-bytes' + + 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_no_serial_prefix_when_serial_zero(sandbox, monkeypatch): + """serial_num 缺失/为 0：文件名不带序号前缀，直接用单集名。""" + uid = _uid() + _install_upstream(monkeypatch, program=_program(serial_num=0)) + tid, task = _register_podcast(sandbox, user_id=uid) + th = _run(tid) + th.join(timeout=30) + + assert _status(tid) == 'done' + assert task['result']['filename'] == '第一集 测试.mp3' + + +def test_ext_follows_url_info_type(sandbox, monkeypatch): + """直链 type=m4a：交付名/磁盘名扩展名随之变化。""" + uid = _uid() + _install_upstream(monkeypatch, url_info=_url_info(type='m4a')) + tid, task = _register_podcast(sandbox, user_id=uid) + th = _run(tid) + th.join(timeout=30) + + assert _status(tid) == 'done' + res = task['result'] + assert res['filename'] == '003 - 第一集 测试.m4a' + assert res['mimetype'] == 'audio/m4a' + + +def test_record_log_gets_episode_name(sandbox, monkeypatch): + """成功交付后按单集名记一条操作记录（IP 取 params.remote_ip）。""" + uid = _uid() + logs: list[tuple] = [] + _install_upstream(monkeypatch) + monkeypatch.setattr(podcast_svc, '_record_log', lambda *a: logs.append(a)) + tid, _ = _register_podcast(sandbox, user_id=uid) + th = _run(tid) + th.join(timeout=30) + + assert logs == [('下载', '1.2.3.4', '[播客] 第一集 测试')], logs + + +def test_remote_ip_defaults_when_missing(sandbox, monkeypatch): + """params 无 remote_ip：按 '-' 记录，不因缺字段报错。""" + uid = _uid() + logs: list[tuple] = [] + _install_upstream(monkeypatch) + monkeypatch.setattr(podcast_svc, '_record_log', lambda *a: logs.append(a)) + tid, task = _register_podcast(sandbox, user_id=uid) + del task['params']['remote_ip'] + th = _run(tid) + th.join(timeout=30) + + assert _status(tid) == 'done' + assert logs == [('下载', '-', '[播客] 第一集 测试')], logs + + +def test_zip_contains_audio_cover_and_info(sandbox, monkeypatch): + """need_zip=true：包内含音频 + 封面 + info.txt，result 走 zip 交付形状。""" + uid = _uid() + _install_upstream(monkeypatch) + tid, task = _register_podcast(sandbox, 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'] == '003 - 第一集 测试.zip' + assert res['mimetype'] == 'application/zip' + assert sorted(_zip_names(res['zip_path'])) == [ + '003 - 第一集 测试.jpg', '003 - 第一集 测试.mp3', 'info.txt'] + + with zipfile.ZipFile(res['zip_path']) as zf: + info = zf.read('info.txt').decode('utf-8') + assert '单集: 第一集 测试' in info + assert '播客: 测试电台' in info + assert '主播: 主播甲' in info + assert '时长: 10:00' in info + assert '音质: ' in info + # 音频正文同时留在任务目录（result.file_path 指向它） + assert Path(res['file_path']).exists() + + +def test_zip_without_cover_when_cover_url_empty(sandbox, monkeypatch): + """节目无 cover_url：ZIP 不含封面条目，其余照常（封面非交付必要条件）。""" + uid = _uid() + _install_upstream(monkeypatch, program=_program(cover_url='')) + tid, task = _register_podcast(sandbox, 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('.jpg')], names + assert [n for n in names if n.endswith('.mp3')], names + + +def test_cover_fetch_failure_still_delivers_zip(sandbox, monkeypatch): + """取封面抛异常：被吞掉，ZIP 照常交付（历史行为：封面失败不炸任务）。""" + uid = _uid() + + def _boom(url: str) -> bytes: + raise RuntimeError('cover cdn down') + + _install_upstream(monkeypatch, cover=_boom) + tid, task = _register_podcast(sandbox, 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_missing_song_id_marks_error_and_refunds(sandbox, monkeypatch): + """单集缺 main_song_id：转 error（文案透出）、退费、任务目录清理。""" + uid = _uid() + before = _balance() + _install_upstream(monkeypatch, program=_program(main_song_id=None)) + tid, task = _register_podcast(sandbox, user_id=uid) + + th = _run(tid) + th.join(timeout=30) + + assert _status(tid) == 'error', f"终态={_status(tid)} msg={task.get('msg')}" + assert task['msg'] == '该单集缺少音频信息（mainSong）' + 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_no_url_marks_error_and_refunds(sandbox, monkeypatch): + """直链解析为空（下架/版权）：转 error、退费、目录清理。""" + uid = _uid() + before = _balance() + _install_upstream(monkeypatch, url_info=_url_info(url='')) + tid, task = _register_podcast(sandbox, user_id=uid) + th = _run(tid) + th.join(timeout=30) + + assert _status(tid) == 'error' + assert task['msg'] == '未获取到音频直链（单集可能已下架或受版权限制）' + assert _balance() == before + assert not (sandbox['downloads'] / 'tasks' / tid).exists() + + +def test_stream_failure_marks_error_and_refunds(sandbox, monkeypatch): + """流式下载失败（如完整性不达标）：转 error、退费、目录清理。""" + uid = _uid() + before = _balance() + _install_upstream(monkeypatch, + stream=_FakeStream(fail=DownloadException('下载不完整（5/11 字节）'))) + tid, task = _register_podcast(sandbox, user_id=uid) + th = _run(tid) + th.join(timeout=30) + + assert _status(tid) == 'error' + assert task['msg'] == '下载不完整（5/11 字节）' + assert _balance() == before + assert not (sandbox['downloads'] / 'tasks' / tid).exists() + + +def test_unexpected_exception_uses_generic_message(sandbox, monkeypatch): + """非预期异常：用户可见文案泛化（不泄上游细节），原文进日志、余额退回。""" + uid = _uid() + before = _balance() + + def _boom(pid: Any, cookies: Any) -> dict[str, Any]: + raise RuntimeError('upstream host=10.0.0.9 refused') + + _install_upstream(monkeypatch) + monkeypatch.setattr(podcast_svc.podcast, 'get_program_detail', _boom) + tid, task = _register_podcast(sandbox, 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_before_stream_start(sandbox, monkeypatch): + """流式下载启动前取消已置位：canceled、退费、目录清理。""" + uid = _uid() + before = _balance() + holder: dict[str, Any] = {} + + def _cancel_first(): + holder['task']['cancel'].set() + + _install_upstream(monkeypatch, stream=_FakeStream(on_call=_cancel_first)) + tid, task = _register_podcast(sandbox, 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_after_download_before_publish(sandbox, monkeypatch): + """下载已落盘但取消随即置位：不发布交付，转 canceled 并退费。""" + uid = _uid() + before = _balance() + holder: dict[str, Any] = {} + + def _cancel_after(): + threading.Timer(0.0, lambda: holder['task']['cancel'].set()).start() + + _install_upstream(monkeypatch, stream=_FakeStream(on_call=_cancel_after)) + tid, task = _register_podcast(sandbox, 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_in_cover(url: str) -> bytes: + holder['task']['cancel'].set() + return b'\xff\xd8\xff cover' + + _install_upstream(monkeypatch, cover=_cancel_in_cover) + tid, task = _register_podcast(sandbox, 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(): + with DOWNLOAD_TASKS_LOCK: + holder['task']['status'] = 'canceled' + holder['task']['completed_at'] = time.time() + + _install_upstream(monkeypatch, stream=_FakeStream(on_call=_mark_canceled)) + tid, task = _register_podcast(sandbox, 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_settle_failure_does_not_flip_done_to_error(sandbox, monkeypatch): + """交付已落定后结算抛错：任务仍为 done（钱由重启对账兜底，不坑已付费用户）。""" + uid = _uid() + + def _boom(*a: Any, **k: Any) -> None: + raise RuntimeError('库锁超时') + + _install_upstream(monkeypatch) + monkeypatch.setattr(podcast_svc.auth_system, 'ledger_settle', _boom) + tid, task = _register_podcast(sandbox, 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（留给重启对账兜底，不是静默丢失）。 + + 注意：光让 ledger_settle 抛错不够——成功路径对结算失败有 try/except 吞掉 + （上一条测试验证的行为）。必须让下载本身先抛非预期异常走进通用 except， + 连「退款结算」也没走成，账本才停留在 open。 + """ + uid = _uid() + + def _boom(*a: Any, **k: Any) -> None: + raise RuntimeError('db busy') + + _install_upstream(monkeypatch, + stream=_FakeStream(fail=RuntimeError('boom'))) + monkeypatch.setattr(podcast_svc.auth_system, 'ledger_settle', _boom) + tid, _task = _register_podcast(sandbox, user_id=uid) + th = _run(tid) + th.join(timeout=30) + + assert _status(tid) == 'error' + row = auth.ledger_get(tid) + assert row is not None + assert row['status'] == 'open', f"应留给对账兜底，实际 status={row['status']}"