refactor: 播客单集 worker 拆解（140 行 / 圈 35 → 40 行 / 圈 6 / 嵌套 1）

为什么：_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

@@ -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,61 +170,85 @@ def _podcast_resolve_audio(program: dict[str, Any], quality: str,
     return url_info
 
 
-def _podcast_download_worker(task_id: str) -> None:
+@dataclass(frozen=True)
-    """后台播客单集下载：解析单集 → 音质降级取直链 → 流式下载（可选 ZIP 打包）
+class _PodcastTaskSpec:
+    """worker 启动时从任务表取出的不可变执行参数。
 
-    任务 type 为 'single'，完成后经 /download/result 交付，与单曲下载共用交付通道。
+    ``program_id`` 标注 Any 与单曲 worker（SingleTaskParams.music_id）同因：
-    落盘文件进入任务专属目录且带节目 ID，跨任务永不重名；清理只动本任务目录。
+    任务 params 里历史存的是数字字符串，下游上游解析各自处理，worker 只透传
+    不做 int() 转换（那是行为变更）。
     """
-    import tempfile as _tempfile
 
-    file_path_str: str | None = None
+    program_id: Any
-    zip_tmp_path: str | None = None
+    quality: str
-    task_dir: Path | None = None
+    need_zip: bool
-    try:
+    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']
-        cancel_ev = task['cancel']
+    return _PodcastTaskSpec(
-        program_id = params['program_id']
+        program_id=params['program_id'],
-        quality = params['quality']
+        quality=params['quality'],
-        need_zip = params['need_zip']
+        need_zip=params['need_zip'],
-        cookies = params['cookies']
+        cookies=params['cookies'],
-        remote_ip = params.get('remote_ip') or '-'
+        remote_ip=params.get('remote_ip') or '-',
+        cancel_ev=task['cancel'],
+    )
 
-        program = podcast.get_program_detail(program_id, cookies)
-        url_info = _podcast_resolve_audio(program, quality, cookies)
 
+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('.')
-        # 磁盘名附节目 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}"
+    return _PodcastTarget(
-        target = task_dir / disk_name
+        task_dir=task_dir,
-        filename = f"{base}.{ext}"   # 交付名保持干净（不带 ID）
+        target=task_dir / f"{base} [{program_id}].{ext}",
+        filename=f"{base}.{ext}",
+        base=base,
+        name=name,
+        ext=ext,
+    )
 
-        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)
 
-        result: dict[str, Any]
+def _write_podcast_zip(zip_path: str, file_path_str: str, arcname: str, base: str,
-        if need_zip:
+                       name: str, program: dict[str, Any], url_info: dict[str, Any],
-            fd, zip_tmp_path = _tempfile.mkstemp(prefix='podcast_zip_', suffix='.zip',
+                       fallback_quality: str) -> None:
-                                                 dir=str(task_dir))
+    """单集入包：音频 + 封面（可选，失败不炸任务）+ info.txt。
-            os.close(fd)
+
-            with zipfile.ZipFile(zip_tmp_path, 'w', zipfile.ZIP_DEFLATED) as zf:
+    info.txt 不再写直链（真实 URL 不应随交付文件外泄）。
-                zf.write(file_path_str, arcname=filename)
+    """
+    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:
@@ -229,36 +257,52 @@ def _podcast_download_worker(task_id: str) -> None:
                     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"
+            f"音质: {api_service._get_quality_display_name(url_info.get('level', fallback_quality))}\n"
         )
         zf.writestr("info.txt", info_txt)
-            # 打包阶段后同样以取消检查收口
+
-            if cancel_ev.is_set():
+
-                raise DownloadCanceled("任务已取消")
+def _podcast_file_result(file_path_str: str, filename: str, ext: str) -> dict[str, Any]:
-            result = {
+    """不打包时的交付形状。"""
-                'file_path': file_path_str,
+    return {
-                'zip_path': zip_tmp_path,
-                'filename': f"{base}.zip",
-                'mimetype': 'application/zip',
-                'audio_files': [file_path_str],
-            }
-        else:
-            result = {
         'file_path': file_path_str,
         'filename': filename,
         'mimetype': f"audio/{ext}",
         'audio_files': [file_path_str],
     }
 
-        # 最终发布前的取消检查（末段取消不得覆盖为 done）
+
-        if cancel_ev.is_set():
+def _podcast_zip_result(prep: _PodcastTarget, file_path_str: str,
-            raise DownloadCanceled("任务已取消")
+                        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'):
@@ -268,7 +312,6 @@ def _podcast_download_worker(task_id: str) -> None:
         t['done'] = 1
         t['msg'] = '下载完成'
         t['result'] = result
-        # 交付已落定；结算失败不再把任务改标 error（同单曲）
     try:
         auth_system.ledger_settle(task_id, 1, 'delivered')
     except Exception as _e:
@@ -278,34 +321,60 @@ def _podcast_download_worker(task_id: str) -> None:
     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，跨任务永不重名；清理只动本任务目录。
+    """
+    task_dir: Path | None = None
+    try:
+        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'], prep.target, url_info.get('size', 0),
+                                 spec.cancel_ev, task_id, '下载中')
+        _abort_if_canceled(spec.cancel_ev)
+        file_path_str = str(prep.target)
+
+        if spec.need_zip:
+            result = _podcast_zip_result(prep, file_path_str, program, url_info, spec)
+        else:
+            result = _podcast_file_result(file_path_str, prep.filename, prep.ext)
+        # 打包阶段同样以取消检查收口（末段取消不得覆盖为 done）
+        _abort_if_canceled(spec.cancel_ev)
+
+        _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', '下载失败，请稍后重试')
+        _fail_podcast_task(task_id, 'error', '下载失败，请稍后重试', 'error', task_dir)
-        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)
 
 
 def _podcast_cleanup_files(paths: list[str | None]) -> None:

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']}"