# -*- coding: utf-8 -*-
"""两个 ZIP 打包 worker 的行为契约测试（沙箱，不触碰生产 data/）。

为什么：``_batch_zip_worker`` / ``_podcast_zip_worker`` 合并前合计约 642 行、
**零测试覆盖**，且两者高度重复——连 ``_wait_if_paused`` 那 16 行都逐字相同
（``zip_valid`` 未绑定那类缺陷就是同型出现在这两处）。抽取公共实现
（``services/_zipcore.py``）之前，先用**行为级**测试钉住可观察契约：任务终态、
ZIP 内容、账本守恒、暂停/恢复/取消语义。断言只依赖「调用 worker 后世界里发生了
什么」，不引用任何内部函数名——所以合并实现时本文件一行未改，它同时就是等价性
验证器。

末尾「合并后的契约固化」一节是**新增**的钉子：合并时逐条对齐了几处分叉
（详见各用例 docstring），这些用例在合并前必然失败——留着它们，那次对齐才不会
被后来者当成「本来就如此」而回退。

账本口径：一律用 ``reserve_and_open`` 真预扣（``ledger_open`` 只记账不扣款，
会让「退款」凭空给余额加次数，守恒断言就失去意义）。
"""
import asyncio
import threading
import time
import zipfile
from dataclasses import replace
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, DownloadResult, MusicInfo
from services import batch as batch_svc
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: ...


class _FakeDownloader:
    """可编程下载器替身：记录调用序，支持预设失败集与按序触发钩子。"""

    def __init__(self, cache_dir: Path, *, fail_ids: tuple[int, ...] = (),
                 hooks: dict[int, Any] | None = None):
        self.cache_dir = cache_dir
        self.fail_ids = set(fail_ids)
        self.hooks = hooks or {}
        self.calls: list[int] = []
        self._lock = threading.Lock()

    async def get_music_info(self, mid: Any, quality: str, cookies: Any) -> MusicInfo:
        return MusicInfo(
            id=int(mid), name=f'歌{mid}', artists='歌手A', album='专辑A',
            pic_url='', duration=200, track_number=1, download_url='',
            file_type='mp3', file_size=1024, quality='standard',
            lyric='[00:01.00]第一行\n[00:02.00]第二行',
        )

    async def download_music_file(self, mid: Any, quality: str, **kw: Any) -> DownloadResult:
        with self._lock:
            self.calls.append(int(mid))
            seq = len(self.calls)
        hook = self.hooks.get(seq)
        if hook is not None:
            hook()
        cancel_ev = kw.get('cancel_event')
        if cancel_ev is not None and cancel_ev.is_set():
            raise DownloadCanceled('任务已取消')
        if int(mid) in self.fail_ids:
            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)
    asyncio.run(auth.init_db())
    asyncio.run(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())

    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 = asyncio.run(auth.get_user_by_name(username))
    assert u is not None
    return int(u['id'])


def _balance(username: str = 'payer') -> int:
    u = asyncio.run(auth.get_user_by_name(username))
    assert u is not None
    return int(u['duration_seconds'] or 0)


def _wait_for(pred, timeout: float = 30.0, interval: float = 0.05) -> bool:
    deadline = time.time() + timeout
    while time.time() < deadline:
        if pred():
            return True
        time.sleep(interval)
    return False


def _status(task_id: str) -> str:
    with DOWNLOAD_TASKS_LOCK:
        return str(DOWNLOAD_TASKS.get(task_id, {}).get('status') or '')


def _resume_like_heartbeat(task: dict[str, Any]) -> None:
    """完全对齐生产心跳（web/routers/tasks.py:task_heartbeat）的恢复动作：
    置 status='queued' 且 clear pause 事件——两者缺一 worker 都不会醒来。"""
    with DOWNLOAD_TASKS_LOCK:
        task['status'] = 'queued'
        task['pause'].clear()


def _register_batch(sandbox, ids: list[int], *, reserved: int, user_id: int,
                    concurrency: int = 1) -> tuple[str, dict[str, Any]]:
    tid = 't' + str(int(time.time() * 1000000) % 10 ** 10)
    # 必须走 reserve_and_open（真预扣+开账同事务）——ledger_open 只记账不扣款，
    # 用它会让「退款」凭空给余额加次数，账本守恒断言失去意义
    left = asyncio.run(auth.reserve_and_open(user_id, reserved, tid))
    assert left is not None, '预扣失败（余额不足，检查 fixture 的 credits）'
    task: dict[str, Any] = {
        'type': 'batch', 'task_id': tid,
        'user_id': user_id, 'billing_user_id': user_id, 'billing_reserved': reserved,
        'total': len(ids), 'done': 0, 'failed': 0,
        'status': 'queued', 'start': time.time(), 'msg': '排队中...',
        'cancel': threading.Event(), 'pause': threading.Event(),
        'device_id': 'dev-test', 'result': None,
        'params': {
            'ids': list(ids), 'quality': 'standard', 'filename_format': 'artist_song',
            'write_meta': False, 'meta_fields': None, 'lyric_format': 'lrc',
            'concurrency': concurrency, 'cookies': {}, 'remote_ip': '-',
            'want_zip': True,
        },
    }
    with DOWNLOAD_TASKS_LOCK:
        DOWNLOAD_TASKS[tid] = task
    return tid, task


def _register_podcast(sandbox, pids: list[int], *, reserved: int, user_id: int) -> tuple[str, dict[str, Any]]:
    tid = 'p' + str(int(time.time() * 1000000) % 10 ** 10)
    left = asyncio.run(auth.reserve_and_open(user_id, reserved, tid))
    assert left is not None, '预扣失败（余额不足）'
    task: dict[str, Any] = {
        'type': 'podcast_batch', 'task_id': tid,
        'user_id': user_id, 'billing_user_id': user_id, 'billing_reserved': reserved,
        'total': len(pids), 'done': 0, 'failed': 0,
        'status': 'queued', 'start': time.time(), 'msg': '排队中...',
        'cancel': threading.Event(), 'pause': threading.Event(),
        'device_id': 'dev-test', 'result': None,
        'params': {
            'radio_id': None, 'ids': list(pids), 'limit': max(1, len(pids)),
            'quality': 'standard', 'cookies': {}, 'remote_ip': '-',
            'radio_name': '测试电台', 'billing_uid': user_id, 'billed': reserved,
        },
    }
    with DOWNLOAD_TASKS_LOCK:
        DOWNLOAD_TASKS[tid] = task
    return tid, task


def _run(worker, task_id: str) -> threading.Thread:
    """把 worker 协程跑在独立线程的私有事件循环里（等价于生产的任务循环）。"""
    def _target() -> None:
        asyncio.run(worker(task_id))

    th = threading.Thread(target=_target, daemon=True)
    th.start()
    return th


def _zip_names(path: str) -> list[str]:
    with zipfile.ZipFile(path) as zf:
        return zf.namelist()


# ============ batch worker ============

def test_batch_zip_completes(sandbox, monkeypatch):
    """3 首全部成功：终态 done、ZIP 含音频与歌词、账本按交付结算。"""
    uid = _uid()
    before = _balance()
    monkeypatch.setattr(api_service, 'downloader', _FakeDownloader(sandbox['cache']))
    tid, task = _register_batch(sandbox, [101, 102, 103], reserved=3, user_id=uid)

    th = _run(batch_svc._batch_zip_worker, 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['done'] == 3 and task['failed'] == 0
    assert task['result'] is not None
    names = _zip_names(task['result']['zip_path'])
    assert sum(1 for n in names if n.endswith('.mp3')) == 3, names
    assert sum(1 for n in names if n.endswith('.lrc')) == 3, names

    row = asyncio.run(auth.ledger_get(tid))
    assert row is not None and row['status'] != 'open'
    assert row['delivered'] == 3
    assert _balance() == before - 3, '全部成功：只扣预扣的 3 次，无退款'


def test_batch_zip_pause_then_resume(sandbox, monkeypatch):
    """断线暂停 → 心跳恢复：ZIP 最终完整，任务走完 done。"""
    uid = _uid()
    before = _balance()
    holder: dict[str, Any] = {}

    monkeypatch.setattr(api_service, 'downloader',
                        _FakeDownloader(sandbox['cache'],
                                        hooks={1: lambda: holder['task']['pause'].set()}))
    tid, task = _register_batch(sandbox, [201, 202, 203], reserved=3, user_id=uid)
    holder['task'] = task

    th = _run(batch_svc._batch_zip_worker, tid)

    assert _wait_for(lambda: _status(tid) == 'paused'), f"未进入 paused（当前 {_status(tid)}）"
    assert '已暂停' in task['msg']
    assert task['done'] >= 1, '暂停前应至少完成 1 首'

    _resume_like_heartbeat(task)
    th.join(timeout=30)

    assert _status(tid) == 'done', f"恢复后终态={_status(tid)} msg={task.get('msg')}"
    names = _zip_names(task['result']['zip_path'])
    assert sum(1 for n in names if n.endswith('.mp3')) == 3, names
    assert _balance() == before - 3


def test_batch_zip_pause_then_cancel(sandbox, monkeypatch):
    """暂停期间取消：任务收敛 canceled，未交付则不发布 ZIP、账本全额退费。"""
    uid = _uid()
    before = _balance()
    holder: dict[str, Any] = {}
    monkeypatch.setattr(api_service, 'downloader',
                        _FakeDownloader(sandbox['cache'],
                                        hooks={1: lambda: holder['task']['pause'].set()}))
    tid, task = _register_batch(sandbox, [301, 302, 303], reserved=3, user_id=uid)
    holder['task'] = task

    th = _run(batch_svc._batch_zip_worker, tid)
    assert _wait_for(lambda: _status(tid) == 'paused'), f"未进入 paused（{_status(tid)}）"

    task['cancel'].set()
    th.join(timeout=30)

    assert _status(tid) == 'canceled', f"终态={_status(tid)} msg={task.get('msg')}"
    assert task['result'] is None, '未交付不应发布 ZIP'
    assert task['msg'] == '任务已取消（已退回 3 次下载次数）', task['msg']
    assert _balance() == before, '取消后应全额退回预扣'
    row = asyncio.run(auth.ledger_get(tid))
    assert row is not None and row['delivered'] == 0 and row['status'] != 'open'


def test_batch_zip_all_fail_refunds(sandbox, monkeypatch):
    """全部失败：不发布空 ZIP、转 error、预扣全额退回。"""
    uid = _uid()
    before = _balance()
    monkeypatch.setattr(api_service, 'downloader',
                        _FakeDownloader(sandbox['cache'], fail_ids=(401, 402)))
    tid, task = _register_batch(sandbox, [401, 402], reserved=2, user_id=uid)

    th = _run(batch_svc._batch_zip_worker, tid)
    th.join(timeout=30)

    assert _status(tid) == 'error', f"终态={_status(tid)} msg={task.get('msg')}"
    assert task['failed'] == 2
    assert task['result'] is None, '全失败不应发布 ZIP'
    assert _balance() == before, '全失败应全额退回预扣'
    row = asyncio.run(auth.ledger_get(tid))
    assert row is not None and row['delivered'] == 0 and row['status'] != 'open'


def test_batch_zip_midway_cancel_refunds_all(sandbox, monkeypatch):
    """循环中途取消：按「不发布 + 全额退费」收敛（记录现状）。

    注意这与 ``test_batch_zip_cancel_at_publish_checkpoint_delivers`` 的差别：
    worker 的 ``zip_valid = True`` 位于 ``with zipfile`` 块**之后**，而 ``while ids``
    循环内的任何取消都以异常**穿过**该块（``zipfile.__exit__`` 仍会写出中央目录，
    但那行赋值不会执行）→ ``zip_valid`` 保持 False → ``delivered = ok_count if
    zip_valid else 0`` 恒为 0。因此源码注释所称的「已入包部分照常交付（成果守恒）」
    在循环内取消时并不可达，只在发布前最后一个检查点可达。
    """
    uid = _uid()
    before = _balance()
    holder: dict[str, Any] = {}
    monkeypatch.setattr(api_service, 'downloader',
                        _FakeDownloader(sandbox['cache'],
                                        hooks={2: lambda: holder['task']['cancel'].set()}))
    tid, task = _register_batch(sandbox, [501, 502, 503, 504], reserved=4, user_id=uid)
    holder['task'] = task

    th = _run(batch_svc._batch_zip_worker, tid)
    th.join(timeout=30)

    assert _status(tid) == 'canceled', f"终态={_status(tid)} msg={task.get('msg')}"
    assert task['result'] is None, '当前实现：循环内取消不发布'
    assert _balance() == before, '未交付部分全额退回'
    row = asyncio.run(auth.ledger_get(tid))
    assert row is not None and row['delivered'] == 0


def test_batch_zip_cancel_at_publish_checkpoint_delivers(sandbox, monkeypatch):
    """发布前最后一刻取消：已打包部分照常交付、按交付数结算（成果守恒）。

    这是「部分成功后取消仍交付」唯一的可达路径，用替身包一层结算函数
    在其返回后立即置取消位，从而精确命中该检查点。
    """
    uid = _uid()
    before = _balance()
    holder: dict[str, Any] = {}
    real_settle = batch_svc._settle_batch_billing

    async def settle_then_cancel(task_id: str, ok_count: int,
                                 final_status: str = 'settled') -> int:
        refunded = await real_settle(task_id, ok_count, final_status)
        holder['task']['cancel'].set()   # 结算已完成、发布前一刻取消
        return refunded

    monkeypatch.setattr(batch_svc, '_settle_batch_billing', settle_then_cancel)
    monkeypatch.setattr(api_service, 'downloader', _FakeDownloader(sandbox['cache']))
    tid, task = _register_batch(sandbox, [601, 602, 603], reserved=3, user_id=uid)
    holder['task'] = task

    th = _run(batch_svc._batch_zip_worker, tid)
    th.join(timeout=30)

    assert _status(tid) == 'canceled', f"终态={_status(tid)} msg={task.get('msg')}"
    assert task['result'] is not None, '已打包部分应交付'
    names = _zip_names(task['result']['zip_path'])
    assert sum(1 for n in names if n.endswith('.mp3')) == 3, names
    assert _balance() == before - 3, '交付 3 首即不退款'


# ============ podcast worker ============

@pytest.fixture()
def podcast_stub(monkeypatch, sandbox):
    """把播客上游与流式下载替换为本地确定性实现。"""
    async def fake_detail(pid: Any, cookies: Any) -> dict[str, Any]:
        n = int(pid)
        return {'id': n, 'name': f'单集{n}', 'main_song_id': 1,
                'serial_num': n, 'cover_url': '', 'songs': []}

    async def fake_resolve(prog: dict[str, Any], quality: str, cookies: Any) -> dict[str, Any]:
        return {'url': f"http://example.invalid/{prog['id']}.mp3", 'size': 2048, 'type': 'mp3'}

    async def fake_stream(url: str, target: Path, total_size: int, cancel_ev: Any,
                          task_id: str = '', progress_label: str = '') -> int:
        if cancel_ev is not None and cancel_ev.is_set():
            raise DownloadCanceled('任务已取消')
        target.write_bytes(b'\xff\xfb\x90\x00' + b'\x00' * 128)
        return target.stat().st_size

    async def _no_cover(_u: str) -> None:
        return None

    monkeypatch.setattr(podcast_svc, '_podcast_resolve_audio', fake_resolve)
    monkeypatch.setattr(podcast_svc, '_podcast_stream_download', fake_stream)
    monkeypatch.setattr(podcast_svc, 'fetch_cover_bytes', _no_cover)
    monkeypatch.setattr(podcast_svc.podcast, 'get_program_detail', fake_detail)
    return sandbox


def test_podcast_zip_completes(podcast_stub):
    """整台打包正常路径：终态 done、ZIP 含全部单集、账本按交付结算。"""
    uid = _uid()
    before = _balance()
    tid, task = _register_podcast(podcast_stub, [11, 12, 13], reserved=3, user_id=uid)

    th = _run(podcast_svc._podcast_zip_worker, tid)
    th.join(timeout=30)

    assert _status(tid) == 'done', f"终态={_status(tid)} msg={task.get('msg')}"
    assert task['done'] == 3 and task['failed'] == 0
    names = _zip_names(task['result']['zip_path'])
    assert sum(1 for n in names if n.endswith('.mp3')) == 3, names
    row = asyncio.run(auth.ledger_get(tid))
    assert row is not None and row['delivered'] == 3
    assert _balance() == before - 3


def test_podcast_zip_pause_then_resume(podcast_stub, monkeypatch):
    """播客打包同样支持断线暂停 → 心跳恢复，最终完整交付。"""
    uid = _uid()
    before = _balance()
    holder: dict[str, Any] = {}
    fired = {'n': 0}

    async def _pause_once(url, target, total_size, cancel_ev, task_id='', progress_label=''):
        target.write_bytes(b'\xff\xfb\x90\x00' + b'\x00' * 128)
        fired['n'] += 1
        if fired['n'] == 1:      # 只在第一集触发——否则恢复后立刻又被暂停
            holder['task']['pause'].set()
        return target.stat().st_size

    monkeypatch.setattr(podcast_svc, '_podcast_stream_download', _pause_once)
    tid, task = _register_podcast(podcast_stub, [21, 22, 23], reserved=3, user_id=uid)
    holder['task'] = task

    th = _run(podcast_svc._podcast_zip_worker, tid)
    assert _wait_for(lambda: _status(tid) == 'paused'), f"未进入 paused（{_status(tid)}）"
    assert '已暂停' in task['msg']

    _resume_like_heartbeat(task)
    th.join(timeout=30)

    assert _status(tid) == 'done', f"恢复后终态={_status(tid)} msg={task.get('msg')}"
    names = _zip_names(task['result']['zip_path'])
    assert sum(1 for n in names if n.endswith('.mp3')) == 3, names
    assert _balance() == before - 3


def test_podcast_zip_all_fail_refunds(podcast_stub, monkeypatch):
    """全部单集失败：转 error 并全额退费，不发布空包。"""
    uid = _uid()
    before = _balance()

    async def _boom(prog, quality, cookies):
        raise RuntimeError('上游不可用')

    monkeypatch.setattr(podcast_svc, '_podcast_resolve_audio', _boom)
    tid, task = _register_podcast(podcast_stub, [31, 32], reserved=2, user_id=uid)

    th = _run(podcast_svc._podcast_zip_worker, tid)
    th.join(timeout=30)

    assert _status(tid) == 'error', f"终态={_status(tid)} msg={task.get('msg')}"
    assert _balance() == before, '全失败应全额退回'
    row = asyncio.run(auth.ledger_get(tid))
    assert row is not None and row['delivered'] == 0


# ============ 合并后的契约固化（分叉点对齐 / 拆分风险点） ============
#
# 以下用例在「两个 worker 各写一份」的旧实现上**必然失败**——它们钉的是合并时
# 逐条对齐的结果，以及「下载」与「落包」拆成两个钩子后新增的边界。

class _SameNameDownloader(_FakeDownloader):
    """所有歌曲同名同歌手、且都带封面——用于验证包内三条目共享序号后缀。"""

    async def get_music_info(self, mid: Any, quality: str, cookies: Any) -> MusicInfo:
        base = await super().get_music_info(mid, quality, cookies)
        return replace(base, name='同名歌', artists='同歌手',
                       pic_url='http://example.invalid/cover.jpg')


def test_batch_zip_dedups_three_arcnames_with_shared_suffix(sandbox, monkeypatch):
    """同名同歌手两首：音频/歌词/封面三条目**共享同一序号后缀**。

    F-C 去重口径：同名同歌手但音源扩展不同（flac vs mp3）时，歌词/封面条目也会
    互相覆盖。合并把「落包」拆成独立钩子后，去重读取的 ``used_arcnames`` 仍是
    串行读改写，且三条目必须同步跳号（只给音频加序号就等于没修 F-C）。
    """
    uid = _uid()
    monkeypatch.setattr(api_service, 'downloader', _SameNameDownloader(sandbox['cache']))
    async def _cover(_u: str) -> bytes:
        return b'\xff\xd8\xff\xe0'

    monkeypatch.setattr(batch_svc, 'fetch_cover_bytes', _cover)
    tid, task = _register_batch(sandbox, [701, 702], reserved=2, user_id=uid)

    th = _run(batch_svc._batch_zip_worker, tid)
    th.join(timeout=30)

    assert _status(tid) == 'done', f"终态={_status(tid)} msg={task.get('msg')}"
    base = '同歌手 - 同名歌'
    assert sorted(_zip_names(task['result']['zip_path'])) == sorted([
        f'{base}.mp3', f'{base}.lrc', f'{base}.jpg',
        f'{base} (2).mp3', f'{base} (2).lrc', f'{base} (2).jpg',
    ])


def test_batch_all_fail_message_and_ledger_status(sandbox, monkeypatch):
    """全失败：用户可见文案含条目数与退费说明；账本终态 error。

    纯固化钉——合并前后的文案与账本终态在此路径上本就一致（对齐时以音乐侧为准）。
    """
    uid = _uid()
    monkeypatch.setattr(api_service, 'downloader',
                        _FakeDownloader(sandbox['cache'], fail_ids=(801, 802)))
    tid, task = _register_batch(sandbox, [801, 802], reserved=2, user_id=uid)

    th = _run(batch_svc._batch_zip_worker, tid)
    th.join(timeout=30)

    assert task['msg'] == '全部失败（2 首），已退回 2 次', task['msg']
    row = asyncio.run(auth.ledger_get(tid))
    assert row is not None and row['status'] == 'error'


def test_podcast_all_fail_message_and_ledger_status(podcast_stub, monkeypatch):
    """全失败：文案含前 3 条失败明细与退费说明；**账本终态统一为 error**。

    合并前这条路径走播客自己的 ``except DownloadException`` 处理器，账本落
    ``'settled'``，与音乐侧同一语义事件落的 ``'error'`` 不一致。整个代码库里
    只有 ``'open'`` 被读取（对账、清理、计数），其余取值仅作排障留痕，故合并时
    统一为语义更准确的 ``'error'``——本用例就是那次对齐的印记。
    """
    uid = _uid()
    before = _balance()

    async def _boom(prog, quality, cookies):
        raise RuntimeError('上游不可用')

    monkeypatch.setattr(podcast_svc, '_podcast_resolve_audio', _boom)
    tid, task = _register_podcast(podcast_stub, [31, 32], reserved=2, user_id=uid)

    th = _run(podcast_svc._podcast_zip_worker, tid)
    th.join(timeout=30)

    # 失败文案一律「单集名(异常)」——引擎的 fail_label 钩子，漏覆盖会退化成整个字典
    assert task['msg'] == (
        '全部单集下载失败: 单集31(上游不可用); 单集32(上游不可用)（已退回 2 次下载次数）'
    ), task['msg']
    assert task['result'] is None, '不发布空包'
    assert _balance() == before, '全失败应全额退回'
    row = asyncio.run(auth.ledger_get(tid))
    assert row is not None and row['status'] == 'error' and row['delivered'] == 0


def test_podcast_cleans_temp_files_on_episode_failure(podcast_stub, monkeypatch):
    """单集下载中途失败：任务目录不留中间音频与 .part 残片。

    合并把「下载」与「落包」拆成两个钩子后，临时文件生命周期跨越两者：失败路径
    由 download_one 自己清，成功路径由 write_entry 的 finally 清（原实现在同一段
    try/finally 里，拆开后若只保留一处就会漏清）。
    """
    uid = _uid()
    before = _balance()

    async def _flaky(url: str, target: Path, total_size: int, cancel_ev: Any,
                     task_id: str = '', progress_label: str = '') -> int:
        if '/102.' in url:
            target.write_bytes(b'\x00' * 64)                                    # 正文残留
            target.with_name(target.name + '.part').write_bytes(b'\x00' * 32)   # 残片
            raise DownloadException('上游断流')
        target.write_bytes(b'\xff\xfb\x90\x00' + b'\x00' * 128)
        return target.stat().st_size

    monkeypatch.setattr(podcast_svc, '_podcast_stream_download', _flaky)
    tid, task = _register_podcast(podcast_stub, [101, 102], reserved=2, user_id=uid)

    th = _run(podcast_svc._podcast_zip_worker, tid)
    th.join(timeout=30)

    assert _status(tid) == 'done', f"终态={_status(tid)} msg={task.get('msg')}"
    # done 是「已完成条目数」（含失败），不是成功数——与旧实现同口径
    assert (task['done'], task['failed']) == (2, 1)
    assert task['msg'] == '成功1 失败1 退回1次', task['msg']
    assert sum(1 for n in _zip_names(task['result']['zip_path']) if n.endswith('.mp3')) == 1
    assert _balance() == before - 1, '只对已入包的 1 集扣费'

    work_dir = podcast_stub['downloads'] / 'tasks' / tid
    leftover = sorted(p.name for p in work_dir.glob('podcast_zip_tmp_*'))
    assert leftover == [], f'中间文件未清理: {leftover}'