diff --git a/services/_zipcore.py b/services/_zipcore.py new file mode 100644 index 0000000..3fa5d37 --- /dev/null +++ b/services/_zipcore.py @@ -0,0 +1,439 @@ +# -*- coding: utf-8 -*- +"""批量 ZIP 打包引擎：两个打包 worker 的共用骨架。 + +音乐歌单（``services.batch``）与播客整台（``services.podcast``）的打包 worker +此前是两份约 300 行的平行实现——任务状态机、断线暂停、全局批量名额、 +「逐条失败不计整批」、取消与异常的四路终态收敛、账本结算全部各写一遍， +连注释都逐字重复（上一轮修掉的 ``zip_valid`` 未绑定缺陷正是同型出现在这两处）。 +真正不同的只有五件事：清单来源、单条怎么下、包内条目怎么拼、并发模型、 +交付包名与文案。 + +因此本模块保留共用骨架，把上述差异收敛到 :class:`ZipJob`；引擎内任何一处 +修订（账本口径、取消检查点、结算幂等）从此只需改一次。 + +术语：下文「条目」指一个下载单元——音乐歌单里是一首歌，播客整台里是一集。 + +线程模型（本轮最容易踩坏的地方，改动前请先读）： +- 下载 I/O 在 ``ThreadPoolExecutor`` 里跑，窗口大小 = ``job.concurrency``； +- **ZIP 写入与包内名去重只发生在调用线程**（``as_completed`` 消费结果时）， + :class:`zipfile.ZipFile` 不是线程安全的，且去重读改写 ``used_arcnames`` + 需要串行——故下载与落包被拆成 ``download_one`` / ``write_entry`` 两个钩子， + 只有前者进线程池。 +""" +import os +import tempfile +import threading +import time +import traceback +import zipfile +from collections.abc import Callable +from concurrent.futures import ThreadPoolExecutor, as_completed +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +from core.state import ( + BATCH_DURATIONS, + BATCH_DURATIONS_LOCK, + DOWNLOAD_TASKS, + DOWNLOAD_TASKS_LOCK, +) +from music_downloader import ( + DownloadCanceled, + DownloadException, +) + +from ._shared import ( + BATCH_ZIP_SEM, +) +from ._taskcore import ( + _mark_task_terminal, + _settle_batch_billing, + _task_work_dir, + _wait_if_paused, +) +from .api_service import ( + api_service, +) + + +class ItemFailed(Exception): + """单个条目失败（不影响整批）。 + + ``label`` 已是最终展示文案（引擎不再拼后缀）。与 + :class:`~music_downloader.DownloadCanceled` 必须区分——取消要中止整批， + 本异常只记一次失败。 + """ + + def __init__(self, label: str) -> None: + super().__init__(label) + self.label = label + + +@dataclass +class ZipContext: + """单次打包运行中，引擎交给 ZipJob 的上下文（引擎持有，任务侧只读）。""" + + task_id: str + params: dict[str, Any] + cancel_ev: threading.Event + work_dir: Path + zf: zipfile.ZipFile + progress: Callable[..., None] + total: int + used_arcnames: set[str] = field(default_factory=set) + + +@dataclass +class _Tally: + """成功/失败计数。 + + 两个 worker 此前把 ``ok_count`` 当外层局部变量，异常处理器读到的是「异常 + 发生那一刻」的值（``zip_valid`` 为假时该值不影响退款，但日志里的「成功N」 + 会失真）。收成一个对象后，异常路径读到的恒是真实值。 + """ + + ok: int = 0 + failed: list[str] = field(default_factory=list) + + +class ZipJob: + """一次批量打包任务的「因任务而异」部分。 + + 子类至少实现 :meth:`prepare` / :meth:`download_one` / :meth:`write_entry` / + :meth:`zip_filename`，其余钩子按需覆写。**实例每次运行新建**——下载器信息 + 预取池等运行期状态就挂在实例上。 + """ + + tag = '批量打包' # 日志前缀 + noun = '项' # 计数单位（文案里的「N 首/N 集」） + item_noun = '条目' # 全称（文案里的「部分xx已打包完成」） + tmp_prefix = 'batch_zip_' # 临时 ZIP 文件名前缀 + start_msg = '打包中...' # 进入 running 时的进度文案 + concurrency = 1 # 每窗口条目数（1 = 顺序） + record_duration = False # 是否把耗时记入 BATCH_DURATIONS（供前端估算等待） + + # ---- 必须实现 ---- + + def prepare(self, params: dict[str, Any], task_id: str, failed: list[str]) -> list[Any]: + """组装条目清单。可抛 ``DownloadException`` 让任务按业务失败收敛。 + + ``failed`` 可被直接追加——清单阶段就有条目已知失败（如按 ID 列表时个别 + 详情拉取失败），这些失败要与下载阶段的一起计入。 + """ + raise NotImplementedError + + def download_one(self, item: Any, idx: int, ctx: ZipContext) -> Any: + """下载单个条目（**在下载线程池内执行**）。 + + 返回的 ``payload`` 会原样交给 :meth:`write_entry`（放临时文件名、 + 已取到的元信息等）。失败请抛异常：``ItemFailed`` 用其自带文案， + 其他异常由引擎按 ``fail_label(item)(异常)`` 记录。 + """ + raise NotImplementedError + + def write_entry(self, item: Any, idx: int, payload: Any, ctx: ZipContext) -> None: + """把下载结果写入 ZIP（**在引擎调用线程内执行，天然串行**）。 + + 包内名去重在这里做——需要读写 ``ctx.used_arcnames``，必须串行。 + """ + raise NotImplementedError + + def zip_filename(self, ok_count: int) -> str: + """交付 ZIP 的文件名。""" + raise NotImplementedError + + # ---- 可选钩子 ---- + + def prefetch(self, window: list[Any], ctx: ZipContext) -> None: + """窗口级预热（每窗口一次、提交下载前调用）。默认无操作。""" + + def fail_label(self, item: Any) -> str: + """单条失败文案前缀（引擎拼成 ``f'{label}({异常})'``）。""" + return str(item) + + def done_msg(self, item: Any, idx: int, ctx: ZipContext) -> str: + """条目成功后的进度文案（空串 = 沿用当前文案）。默认不追加文案。""" + return '' + + def log_label(self) -> str: + """日志里区隔本次任务的名字（如电台名），默认空串。""" + return '' + + def all_failed_msg(self, failed: list[str], refunded: int) -> str: + """全部条目失败时的用户可见文案（含退费说明）。""" + return f"全部失败（{len(failed)} {self.noun}）" + ( + f"，已退回 {refunded} 次" if refunded else '') + + def partial_msg(self, refunded: int) -> str: + """已打包部分交付时的用户可见文案（仅异常路径会用到）。""" + return f"部分{self.item_noun}已打包完成，可领取" + ( + f"，退回 {refunded} 次" if refunded else '') + + def supplement(self, task_id: str, params: dict[str, Any], count: int) -> None: + """清单确定后的计费补扣（默认无操作）。""" + + def on_delivered(self, items: list[Any], task_id: str, params: dict[str, Any]) -> None: + """交付落定后的操作记录（默认无操作）。""" + + def cleanup(self) -> None: + """释放运行期资源（如信息预取池）。默认无操作。""" + + +def _discard(path: str | None) -> None: + """删除未发布的临时 ZIP（坏包/空包/取消包）——失败静默（清理非关键路径）。""" + if not path: + return + try: + os.remove(path) + except OSError: + pass + + +def run_zip_job(job: ZipJob, task_id: str, *, + settle: Callable[[str, int, str], int] = _settle_batch_billing) -> None: + """执行一次批量打包任务（在 worker 线程内调用，自身不向外抛异常）。 + + ``settle`` 由调用方注入，默认走 :func:`services._taskcore._settle_batch_billing`。 + 做成参数是为了让测试能在**模块级**替换结算函数（包装层在调用时读取模块全局， + 替换即生效）。 + """ + zip_tmp_path: str | None = None + # zip_valid 必须在函数头初始化：ZIP 构建前的业务错误（参数缺失/无条目/ + # 补扣失败）会 early raise，异常处理器若引用未绑定名会 NameError 二次崩线程 + zip_valid = False + tally = _Tally() + items: list[Any] = [] + try: + with DOWNLOAD_TASKS_LOCK: + task = DOWNLOAD_TASKS[task_id] + task['status'] = 'running' + task['msg'] = job.start_msg + params = task['params'] + cancel_ev = task['cancel'] + pause_ev = task['pause'] + # 计耗时只算「下载+打包」这一段（不含排队），故不用 task['start'] + t0 = time.time() + + items = job.prepare(params, task_id, tally.failed) + with DOWNLOAD_TASKS_LOCK: + DOWNLOAD_TASKS[task_id]['total'] = len(items) + job.supplement(task_id, params, len(items)) + + def _progress(msg: str = '') -> None: + with DOWNLOAD_TASKS_LOCK: + t = DOWNLOAD_TASKS[task_id] + t['done'] = tally.ok + len(tally.failed) + t['failed'] = len(tally.failed) + if msg: + t['msg'] = msg + + # ZIP 构建放 downloads 卷（原实现落系统 %TEMP%——500 首无损可达 GB 级， + # 小系统盘一次大单即塞爆），且建在任务专属目录（见 _task_work_dir）： + # downloads 根目录会被缓存清扫按 mtime 回收 + work_dir = _task_work_dir(task_id) + with tempfile.NamedTemporaryFile(prefix=job.tmp_prefix, suffix='.zip', + dir=str(work_dir), delete=False) as zip_tmp: + zip_tmp_path = zip_tmp.name + # 音频本身已压缩，条目按 ZIP_STORED（免压缩）大幅提速；文本条目仍压缩 + with zipfile.ZipFile(zip_tmp, 'w', zipfile.ZIP_DEFLATED) as zf: + ctx = ZipContext(task_id=task_id, params=params, cancel_ev=cancel_ev, + work_dir=work_dir, zf=zf, progress=_progress, + total=len(items)) + _fill_zip(job, items, ctx, tally, task, pause_ev) + # zip_valid 只在 ZIP with 块正常退出（中央目录完整写出）后才置 True。 + # 此前两个 worker 在这之后还各留了一段 `if not zip_valid:` 失败处理—— + # 紧跟赋值、中间无任何可改zip_valid的语句，是**不可达**的复审残留，已删。 + zip_valid = True + + if job.record_duration and tally.ok > 0: + with BATCH_DURATIONS_LOCK: + BATCH_DURATIONS.append(time.time() - t0) + + # 全部失败（含清单非空但一条都没成）：不发布空 ZIP、不按交付结算， + # 转 error 并全额退费。原音乐批量用 `ok==0 and failed`、播客用早抛 + # DownloadException——这里统一为显式收敛，两边的账本终态一致 + if tally.ok == 0: + _discard(zip_tmp_path) + zip_tmp_path = None + refunded = settle(task_id, 0, 'error') + _mark_task_terminal(task_id, 'error', job.all_failed_msg(tally.failed, refunded)) + with DOWNLOAD_TASKS_LOCK: + t = DOWNLOAD_TASKS.get(task_id) + if t is not None: + t['failed'] = len(tally.failed) + t['refunded'] = refunded + api_service.logger.info( + f"{job.tag}全部失败: {task_id} 失败{len(tally.failed)} 退回{refunded}次") + return + + # 正常交付：按入包成功数结算（'settled' 是 _settle_batch_billing 的默认值， + # 此处显式写出，让「成功/取消/失败」三种账本终态在本文件里一目了然） + refunded = settle(task_id, tally.ok, 'settled') + with DOWNLOAD_TASKS_LOCK: + t = DOWNLOAD_TASKS.get(task_id) + # 发布前锁内复查取消/失效状态——取消请求落在「检查后、拿锁前」的窗口 + # 也不得覆盖终态 + if t is None or t.get('status') in ('canceled', 'error'): + raise DownloadCanceled("任务已取消") + if cancel_ev.is_set(): + raise DownloadCanceled("任务已取消") + t['status'] = 'done' + t['completed_at'] = time.time() + t['done'] = tally.ok + len(tally.failed) + t['failed'] = len(tally.failed) + t['msg'] = f"成功{tally.ok} 失败{len(tally.failed)}" + ( + f" 退回{refunded}次" if refunded else '') + t['refunded'] = refunded + t['result'] = { + 'zip_path': zip_tmp_path, + 'filename': job.zip_filename(tally.ok), + 'audio_files': [], # 缓存文件不归任务所有，不进入清理清单 + } + job.on_delivered(items, task_id, params) + api_service.logger.info( + f"{job.tag}完成: {job.log_label()}成功{tally.ok} 失败{len(tally.failed)} " + f"退回{refunded}次 ({task_id})") + except DownloadCanceled: + # 部分成功后取消——已入包部分照常交付（成果守恒），未完成部分按账本退费。 + # ZIP 未正常完成（写坏）时不按 ok_count 交付——按全额退费收敛 + delivered = tally.ok if zip_valid else 0 + refunded = settle(task_id, delivered, 'canceled') + if delivered > 0 and zip_tmp_path: + _mark_task_terminal(task_id, 'canceled', + f"已取消（已打包 {tally.ok} {job.noun}可领取）" + + (f"，退回 {refunded} 次" if refunded else '')) + with DOWNLOAD_TASKS_LOCK: + t = DOWNLOAD_TASKS.get(task_id) + if t is not None: + t['refunded'] = refunded + t['result'] = { + 'zip_path': zip_tmp_path, + 'filename': job.zip_filename(tally.ok), + 'audio_files': [], + } + else: + _discard(zip_tmp_path) + _mark_task_terminal(task_id, 'canceled', + '任务已取消' + ( + f"（已退回 {refunded} 次下载次数）" if refunded else '')) + with DOWNLOAD_TASKS_LOCK: + t = DOWNLOAD_TASKS.get(task_id) + if t is not None: + t['refunded'] = refunded + api_service.logger.info( + f"{job.tag}已取消: {task_id}（成功{tally.ok} 退回{refunded}次）") + except DownloadException as e: + # 业务失败（清单阶段：无可下载条目 / 补扣下载次数失败 / 上游不可用）。 + # 文案是我们自己拼的用户可读文本，保留原文（同播客单集下载） + delivered = tally.ok if zip_valid else 0 + refunded = settle(task_id, delivered, 'error') + _apply_error(task_id, str(e), refunded) + _discard(zip_tmp_path) + api_service.logger.error( + f"{job.tag}失败: {task_id} {e}（成功{tally.ok} 退回{refunded}次）") + except Exception as e: + # 非预期异常原文 logger 先行（结算本身可能再抛，先落下 traceback）， + # 用户可见文案泛化；带上退款结果再补一条，避免排障时只看到「异常」 + api_service.logger.error(f"{job.tag}异常: {task_id} {e}\n{traceback.format_exc()}") + delivered = tally.ok if zip_valid else 0 + refunded = settle(task_id, delivered, 'error') + if delivered > 0 and zip_tmp_path: + _apply_error(task_id, job.partial_msg(refunded), refunded) + with DOWNLOAD_TASKS_LOCK: + t = DOWNLOAD_TASKS.get(task_id) + if t is not None: + t['result'] = { + 'zip_path': zip_tmp_path, + 'filename': job.zip_filename(tally.ok), + 'audio_files': [], + } + else: + _apply_error(task_id, '打包失败，请稍后重试' + ( + f"（已退回 {refunded} 次下载次数）" if refunded else ''), refunded) + _discard(zip_tmp_path) + api_service.logger.error( + f"{job.tag}异常: {task_id}（成功{tally.ok} 退回{refunded}次）") + finally: + # 回收信息预取线程池，避免线程泄漏（未启动的预取任务一并取消） + job.cleanup() + + +def _apply_error(task_id: str, msg: str, refunded: int) -> None: + """异常/业务失败的终态收敛：置 error（终态先到先得）、记账退款额。 + + 走 :func:`_mark_task_terminal` 而非就地写任务字典——后者是播客侧的旧写法， + 除了重复一遍「已终态不覆盖」判断外，还漏掉了 worker 线程已死时按账本补结算 + 的兜底（``/F-E``）。 + """ + _mark_task_terminal(task_id, 'error', msg) + with DOWNLOAD_TASKS_LOCK: + t = DOWNLOAD_TASKS.get(task_id) + if t is not None: + t['refunded'] = refunded + + +def _fill_zip(job: ZipJob, items: list[Any], ctx: ZipContext, tally: _Tally, + task: dict[str, Any], pause_ev: threading.Event) -> None: + """逐窗口下载并写入 ZIP（返回时中央目录尚未写出，由调用方的 with 退出时落盘）。 + + 窗口 = ``job.concurrency`` 个条目：下载并行、落包串行。窗口大小取 1 时 + 退化为顺序下载（播客整台即如此）。 + """ + concurrency = max(1, int(job.concurrency)) + pool: ThreadPoolExecutor | None = None + try: + # 不用 with 管理下载池——取消时 with 退出会 wait=True 等在途 future + # （其内部阻塞在预取 HTTP 上最长 30s），改为不等待 + 取消排队任务 + pool = ThreadPoolExecutor(max_workers=concurrency) + offset = 0 + while offset < len(items): + if ctx.cancel_ev.is_set(): + raise DownloadCanceled("任务已取消") + # 断线暂停：释放名额等待恢复/取消，恢复后重新进入排队 + _wait_if_paused(task, pause_ev, ctx.cancel_ev, tally.ok + len(tally.failed)) + # 每个窗口持有 BATCH_ZIP_SEM（同一时刻全局最多 N 个批量任务在下载） + # 等待名额期间可被取消，不再无限阻塞线程与退款 + if not BATCH_ZIP_SEM.acquire_cancellable(ctx.cancel_ev): + raise DownloadCanceled("任务已取消") + try: + # 暂停/取消落在「等待恢复返回」与「拿到名额」之间：立刻放名额回到 + # 循环顶部重新处理暂停（原播客侧缺这一步，会多下完这一集才停） + if (ctx.cancel_ev.is_set() or pause_ev.is_set() + or task['status'] == 'paused'): + continue + window = items[offset:offset + concurrency] + start_idx = offset + 1 # 条目序号从 1 起（进度文案用） + offset += len(window) + job.prefetch(window, ctx) # 只预取本窗口 + futures = { + pool.submit(job.download_one, item, start_idx + i, ctx): (item, start_idx + i) + for i, item in enumerate(window) + } + for future in as_completed(futures): + if ctx.cancel_ev.is_set(): + raise DownloadCanceled("任务已取消") + item, idx = futures[future] + try: + payload = future.result() + # 落包在主线程：ZipFile 非线程安全 + 包内名去重要串行 + job.write_entry(item, idx, payload, ctx) + except DownloadCanceled: + raise + except ItemFailed as f: + tally.failed.append(f.label) + except Exception as e: + tally.failed.append(f"{job.fail_label(item)}({e})") + else: + tally.ok += 1 + ctx.progress(job.done_msg(item, idx, ctx)) + finally: + BATCH_ZIP_SEM.release() + finally: + if pool is not None: + try: + # 取消/异常时不等待在途下载线程（其可能阻塞在预取 HTTP 上）， + # 排队任务一并取消；在途线程自行收尾（.part 残片由续传/清扫兜底） + pool.shutdown(wait=False, cancel_futures=True) + except TypeError: + pool.shutdown(wait=False) diff --git a/services/batch.py b/services/batch.py index 4c248ad..e214b71 100644 --- a/services/batch.py +++ b/services/batch.py @@ -1,40 +1,174 @@ # -*- coding: utf-8 -*- """歌单/专辑批量打包 worker。 + +打包骨架（状态机 / 暂停 / 名额 / 终态收敛 / 结算）在 :mod:`services._zipcore`， +本模块只保留「音乐歌单」这一侧的策略：窗口化预取与并行下载、每首三条目 +（音频 + 歌词 + 封面）入包、同名去重、包名与文案。 """ - -import os -import time -import traceback import zipfile -from pathlib import Path - -from core.state import ( - BATCH_DURATIONS, - BATCH_DURATIONS_LOCK, - DOWNLOAD_TASKS, - DOWNLOAD_TASKS_LOCK, +from concurrent.futures import ( + Future, + ThreadPoolExecutor, ) +from dataclasses import dataclass +from pathlib import Path +from typing import Any + from core.utils import format_lyric as _format_lyric from music_downloader import ( - DownloadCanceled, fetch_cover_bytes, ) from ._shared import ( - BATCH_ZIP_SEM, _build_download_base, ) from ._taskcore import ( - _mark_task_terminal, _settle_batch_billing, - _task_work_dir, - _wait_if_paused, +) +from ._zipcore import ( + ItemFailed, + ZipContext, + ZipJob, + run_zip_job, ) from .api_service import ( api_service, ) +@dataclass(frozen=True) +class _SongPayload: + """一首歌下载完成后的入包素材（下载线程产出，落包线程消费）。""" + + path: str + name: str + artists: str + file_type: str + lyric: str + cover: bytes + + +class _SonglistZipJob(ZipJob): + """音乐歌单打包：窗口化并行下载，每首写 音频 + 歌词 + 封面 三条目。""" + + tag = '批量ZIP' + noun = '首' + item_noun = '歌曲' + tmp_prefix = 'batch_zip_' + start_msg = '下载打包中...' + record_duration = True + + def __init__(self) -> None: + self._info_pool: ThreadPoolExecutor | None = None + self._info_futures: dict[Any, Future[Any]] = {} + + def prepare(self, params: dict[str, Any], task_id: str, failed: list[str]) -> list[Any]: + self.concurrency = int(params['concurrency']) + return list(params['ids']) + + def prefetch(self, window: list[Any], ctx: ZipContext) -> None: + """按窗口预取歌曲信息。 + + 原实现一次性提交全部（最多 500 首 ×3 接口）预取任务，大歌单瞬间以 10 并发 + 连发上游约 1500 次请求打风控；改为每个下载窗口只预取本窗口的歌（与下载节奏 + 对齐），``download_one`` 语义不变（等预取结果再下载）。 + """ + if self._info_pool is None: + self._info_pool = ThreadPoolExecutor(max_workers=min(4, max(1, self.concurrency))) + quality = ctx.params['quality'] + cookies = ctx.params['cookies'] + for mid in window: + if mid not in self._info_futures: + self._info_futures[mid] = self._info_pool.submit( + api_service.downloader.get_music_info, mid, quality, cookies + ) + + def download_one(self, item: Any, idx: int, ctx: ZipContext) -> _SongPayload: + """并行下载单元：等待预取信息 → 下载 → 复用预取歌词与缓存封面（零重复请求）。""" + mid = item + info = self._info_futures[mid].result() + params = ctx.params + result = api_service.downloader.download_music_file( + mid, params['quality'], + write_meta=params['write_meta'], + meta_fields=params['meta_fields'], + cookies=params['cookies'], + cancel_event=ctx.cancel_ev, + music_info=info, + ) + lyric = '' + cover = b'' + if result.success: + # 歌词已随 get_music_info 预取返回，无需再调一次歌词接口 + lyric = getattr(info, 'lyric', '') or '' + # 封面走共享缓存（同一专辑只拉一次网络）；取消后不再发起新请求 + if not ctx.cancel_ev.is_set() and info.pic_url: + cover = fetch_cover_bytes(info.pic_url) or b'' + if not result.success: + raise ItemFailed(str(mid)) + assert result.file_path is not None + if not Path(result.file_path).exists(): + # 文件可能已被同名并发任务入包删除，按失败处理而非炸掉整个任务 + raise ItemFailed(str(mid)) + mi = result.music_info + if mi is None: + # 与上面 exists() 同口径：成功但缺元信息（上游详情缺失）按单曲失败计， + # 不让 mi.name 抛 AttributeError 把整批转 error + raise ItemFailed(str(mid)) + return _SongPayload(path=result.file_path, name=mi.name, artists=mi.artists, + file_type=mi.file_type, lyric=lyric, cover=cover) + + def write_entry(self, item: Any, idx: int, payload: _SongPayload, ctx: ZipContext) -> None: + mid = item + base = _build_download_base(ctx.params['filename_format'], + {'name': payload.name, 'artist_string': payload.artists}) + lyric_format = ctx.params['lyric_format'] + # 同名去重：歌单里同一首歌出现两次/同名不同版本时，追加序号避免 ZIP 内重名 + # 互相覆盖（用户被扣 2 次只拿到 1 首）。F-C：音频/歌词/封面三个 arcname 都要 + # 查重——同名同歌手但音源扩展不同（flac vs mp3）时歌词/封面也会互相覆盖 + final_name = f"{base}.{payload.file_type}" + lrc_name = f"{base}.{lyric_format}" + jpg_name = f"{base}.jpg" + k = 2 + while (final_name in ctx.used_arcnames + or lrc_name in ctx.used_arcnames + or jpg_name in ctx.used_arcnames): + final_name = f"{base} ({k}).{payload.file_type}" + lrc_name = f"{base} ({k}).{lyric_format}" + jpg_name = f"{base} ({k}).jpg" + k += 1 + ctx.used_arcnames.update((final_name, lrc_name, jpg_name)) + # F-D：exists 与 zf.write 之间的缓存竞态（另一任务缓存 miss 删文件） + # 按单曲失败计，不再让整批转 error + try: + ctx.zf.write(payload.path, arcname=final_name, compress_type=zipfile.ZIP_STORED) + except OSError as e: + raise ItemFailed(f"{mid}(缓存文件读取失败: {e})") from e + # 缓存文件是跨任务共享资源，任务不得删除；磁盘回收交给监控线程的下载缓存 + # 清扫（按文件年龄）。歌词与封面已在并行下载线程中拉取，这里直接写包（无网络等待） + if payload.lyric: + try: + ctx.zf.writestr(lrc_name, _format_lyric(payload.lyric, lyric_format), + compress_type=zipfile.ZIP_DEFLATED) + except Exception: + pass + if payload.cover: + try: + ctx.zf.writestr(jpg_name, payload.cover, compress_type=zipfile.ZIP_STORED) + except Exception: + pass + + def zip_filename(self, ok_count: int) -> str: + return f"批量下载_{ok_count}首.zip" + + def cleanup(self) -> None: + if self._info_pool is not None: + try: + self._info_pool.shutdown(wait=False, cancel_futures=True) + except TypeError: + self._info_pool.shutdown(wait=False) + + def _batch_zip_worker(task_id: str) -> None: """后台批量打包：窗口化下载 → 写 ZIP（音频免压缩），支持取消与断线暂停 @@ -42,323 +176,4 @@ def _batch_zip_worker(task_id: str) -> None: 保存进度进入暂停等待；重新打开网站（心跳恢复）后自动回到队列继续下载 - 音频入包即删，缓解磁盘占用 """ - import tempfile as _tempfile - from concurrent.futures import ThreadPoolExecutor, as_completed - - zip_tmp_path: str | None = None - # zip_valid 必须在函数头初始化：ZIP 构建前的业务错误（参数缺失/无歌曲等） - # 会 early raise，异常处理器若引用未绑定名会 NameError 二次崩溃 worker 线程 - zip_valid = False - ok_count = 0 - failed: list[str] = [] - used_arcnames = set() # ZIP 内已占用文件名（同名歌曲去重，防止包内互相覆盖） - info_pool = None - info_futures = {} - try: - with DOWNLOAD_TASKS_LOCK: - task = DOWNLOAD_TASKS[task_id] - task['status'] = 'running' - task['msg'] = '下载打包中...' - params = task['params'] - cancel_ev = task['cancel'] - pause_ev = task['pause'] - ids = list(params['ids']) - quality = params['quality'] - filename_format = params['filename_format'] - write_meta = params['write_meta'] - meta_fields = params['meta_fields'] - lyric_format = params['lyric_format'] - concurrency = params['concurrency'] - cookies = params['cookies'] - batch_start = time.time() - - # 预取按窗口进行——原实现一次性提交全部（最多 500 首 ×3 接口）预取任务， - # 大歌单瞬间以 10 并发连发上游约 1500 次请求打风控；改为每个下载窗口只预取本窗口 - # 的歌（与下载节奏对齐），_download_one 语义不变（等预取结果再下载） - from concurrent.futures import ThreadPoolExecutor as _InfoPool - info_pool = _InfoPool(max_workers=min(4, max(1, concurrency))) - - def _prefetch(window_ids): - for mid in window_ids: - if mid not in info_futures: - info_futures[mid] = info_pool.submit( - api_service.downloader.get_music_info, mid, quality, cookies - ) - - def _download_one(mid): - """并行下载单元：等待预取信息 → 下载 → 复用预取歌词与缓存封面（零重复请求）""" - info = info_futures[mid].result() - result = api_service.downloader.download_music_file( - mid, quality, - write_meta=write_meta, - meta_fields=meta_fields, - cookies=cookies, - cancel_event=cancel_ev, - music_info=info, - ) - extra = {'lyric': '', 'cover': b''} - if result.success: - # 歌词已随 get_music_info 预取返回，无需再调一次歌词接口 - extra['lyric'] = getattr(info, 'lyric', '') or '' - # 封面走共享缓存（同一专辑只拉一次网络）；取消后不再发起新请求 - if not cancel_ev.is_set() and info.pic_url: - cover = fetch_cover_bytes(info.pic_url) - if cover: - extra['cover'] = cover - return result, extra - - def _update_progress() -> None: - with DOWNLOAD_TASKS_LOCK: - DOWNLOAD_TASKS[task_id]['done'] = ok_count + len(failed) - DOWNLOAD_TASKS[task_id]['failed'] = len(failed) - - # ZIP 构建放 downloads 卷（原实现落系统 %TEMP%——500 首无损可达 GB 级， - # 小系统盘一次大单即塞爆；单曲/播客早已指定 dir，唯批量漏了） - # 且建在任务专属目录（见 _task_work_dir）：根目录会被缓存清扫按 mtime 回收 - _task_dir_for_zip = _task_work_dir(task_id) - # 中央目录完整写出后（下方）才置 True，否则一律按失败全额退费 - with _tempfile.NamedTemporaryFile(prefix='batch_zip_', suffix='.zip', - dir=str(_task_dir_for_zip), - delete=False) as zip_tmp: - zip_tmp_path = zip_tmp.name - # 音频本身已压缩，用 ZIP_STORED（免压缩）大幅提速；歌词等文本仍压缩 - with zipfile.ZipFile(zip_tmp, 'w', zipfile.ZIP_DEFLATED) as zf: - # 不用 with 管理下载池——取消时 with 退出会 wait=True 等在途 - # future（其内部阻塞在预取 HTTP 上最长 30s），改为不等待 + 取消排队任务 - pool = ThreadPoolExecutor(max_workers=concurrency) - try: - while ids: - if cancel_ev.is_set(): - raise DownloadCanceled("任务已取消") - # 断线暂停：释放名额等待恢复/取消，恢复后重新进入排队 - _wait_if_paused(task, pause_ev, cancel_ev, ok_count + len(failed)) - # 每个窗口持有 BATCH_ZIP_SEM（同一时刻全局最多 3 个批量任务在下载） - # 等待名额期间可被取消，不再无限阻塞线程与退款 - if not BATCH_ZIP_SEM.acquire_cancellable(cancel_ev): - raise DownloadCanceled("任务已取消") - try: - if cancel_ev.is_set() or pause_ev.is_set() or task['status'] == 'paused': - continue # 立即释放名额，回到循环顶部处理暂停 - window = ids[:concurrency] - ids = ids[concurrency:] - _prefetch(window) # 只预取本窗口 - future_map = { - pool.submit(_download_one, mid): mid - for mid in window - } - for future in as_completed(future_map): - if cancel_ev.is_set(): - raise DownloadCanceled("任务已取消") - mid = future_map[future] - try: - result, extra = future.result() - except DownloadCanceled: - raise - except Exception as e: - failed.append(f"{mid}({e})") - _update_progress() - continue - if not result.success: - failed.append(str(mid)) - _update_progress() - continue - assert result.file_path is not None - fp = Path(result.file_path) - if not fp.exists(): - # 文件可能已被同名并发任务入包删除，按失败处理而非炸掉整个任务 - failed.append(str(mid)) - _update_progress() - continue - mi = result.music_info - if mi is None: - # 与上面 fp.exists() 同口径：成功但缺元信息（上游详情缺失） - # 按单曲失败计，不让 mi.name 抛 AttributeError 把整批转 error - failed.append(str(mid)) - _update_progress() - continue - base = _build_download_base(filename_format, {'name': mi.name, 'artist_string': mi.artists}) - # 同名去重：歌单里同一首歌出现两次/同名不同版本时， - # 追加序号避免 ZIP 内重名互相覆盖（用户被扣 2 次只拿到 1 首） - # F-C：音频/歌词/封面三个 arcname 都要查重——同名同歌手但 - # 音源扩展不同（flac vs mp3）时歌词/封面条目也会互相覆盖 - final_name = f"{base}.{mi.file_type}" - lrc_name = f"{base}.{lyric_format}" - jpg_name = f"{base}.jpg" - k = 2 - while (final_name in used_arcnames - or lrc_name in used_arcnames - or jpg_name in used_arcnames): - final_name = f"{base} ({k}).{mi.file_type}" - lrc_name = f"{base} ({k}).{lyric_format}" - jpg_name = f"{base} ({k}).jpg" - k += 1 - used_arcnames.update((final_name, lrc_name, jpg_name)) - # F-D：exists 与 zf.write 之间的缓存竞态（另一任务缓存 miss - # 删文件）按单曲失败计，不再让整批转 error - try: - zf.write(str(fp), arcname=final_name, compress_type=zipfile.ZIP_STORED) - except OSError as e: - failed.append(f"{mid}(缓存文件读取失败: {e})") - _update_progress() - continue - # 缓存文件是跨任务共享资源，任务不得删除； - # 磁盘回收交给监控线程的下载缓存清扫（按文件年龄） - # 歌词与封面已在并行下载线程中拉取，直接写入 ZIP（无网络等待） - lrc = extra.get('lyric', '') - if lrc: - try: - zf.writestr(lrc_name, _format_lyric(lrc, lyric_format), compress_type=zipfile.ZIP_DEFLATED) - except Exception: - pass - if extra.get('cover'): - try: - zf.writestr(jpg_name, extra['cover'], compress_type=zipfile.ZIP_STORED) - except Exception: - pass - ok_count += 1 - _update_progress() - finally: - BATCH_ZIP_SEM.release() - finally: - # 取消/异常时不等待在途下载线程（其可能阻塞在预取 HTTP 上）， - # 排队任务一并取消；在途线程自行收尾（.part 残片由续传/清扫兜底） - try: - pool.shutdown(wait=False, cancel_futures=True) - except TypeError: - pool.shutdown(wait=False) - # 复审 修正：zip_valid = True 必须在内层 with zipfile 退出之后 - # （中央目录在 close 时才写出）——原缩进 16 在 with 内部，守卫不可达 - zip_valid = True - - if ok_count > 0: - with BATCH_DURATIONS_LOCK: - BATCH_DURATIONS.append(time.time() - batch_start) - - # F-B：ZIP 未正常完成（磁盘满写坏中央目录等）——不发布坏包，按全额退费收敛 - if not zip_valid: - if zip_tmp_path: - try: - os.remove(zip_tmp_path) - except OSError: - pass - zip_tmp_path = None - refunded = _settle_batch_billing(task_id, 0, 'error') - _mark_task_terminal(task_id, 'error', - f"ZIP 打包失败（可能磁盘空间不足），已退回 {refunded} 次" if refunded - else "ZIP 打包失败（可能磁盘空间不足），费用未扣或已退回") - with DOWNLOAD_TASKS_LOCK: - t = DOWNLOAD_TASKS.get(task_id) - if t is not None: - t['failed'] = t.get('failed', 0) + ok_count - t['refunded'] = refunded - api_service.logger.error(f"批量ZIP构建失败未发布: {task_id} ok_count={ok_count} 退回{refunded}次") - return - - # 全部失败不发布空 ZIP、不标 done —— 按失败收敛并全额退费 - if ok_count == 0 and failed: - if zip_tmp_path: - try: - os.remove(zip_tmp_path) - except OSError: - pass - zip_tmp_path = None - refunded = _settle_batch_billing(task_id, 0, 'error') - _mark_task_terminal(task_id, 'error', - f"全部失败（{len(failed)} 首）" + (f"，已退回 {refunded} 次" if refunded else '')) - with DOWNLOAD_TASKS_LOCK: - DOWNLOAD_TASKS[task_id]['failed'] = len(failed) - DOWNLOAD_TASKS[task_id]['refunded'] = refunded - api_service.logger.info(f"批量ZIP全部失败: {task_id} 失败{len(failed)} 退回{refunded}次") - return - - refunded = _settle_batch_billing(task_id, ok_count) - with DOWNLOAD_TASKS_LOCK: - t = DOWNLOAD_TASKS.get(task_id) - # 发布前锁内复查取消/失效状态——取消请求落在"检查后、拿锁前"的窗口也不得覆盖终态 - if t is None or t.get('status') in ('canceled', 'error'): - raise DownloadCanceled("任务已取消") - if cancel_ev.is_set(): - raise DownloadCanceled("任务已取消") - t['status'] = 'done' - t['completed_at'] = time.time() - t['done'] = ok_count + len(failed) - t['failed'] = len(failed) - t['msg'] = f"成功{ok_count} 失败{len(failed)}" + (f" 退回{refunded}次" if refunded else '') - t['refunded'] = refunded - t['result'] = { - 'zip_path': zip_tmp_path, - 'filename': f"批量下载_{ok_count}首.zip", - 'audio_files': [], # 缓存文件不归任务所有，不进入清理清单 - } - api_service.logger.info(f"批量ZIP完成: 成功{ok_count} 失败{len(failed)} 退回{refunded}次") - except DownloadCanceled: - # 部分成功后取消——已入包部分照常交付（成果守恒），未完成部分按账本退费 - # F-B：ZIP 未正常完成（写坏）时不按 ok_count 交付——按全额退费收敛 - delivered = ok_count if zip_valid else 0 - refunded = _settle_batch_billing(task_id, delivered, 'canceled') - if delivered > 0 and zip_tmp_path: - _mark_task_terminal(task_id, 'canceled', - f"已取消（已打包 {ok_count} 首可领取）" + - (f"，退回 {refunded} 次" if refunded else '')) - with DOWNLOAD_TASKS_LOCK: - t = DOWNLOAD_TASKS.get(task_id) - if t is not None: - t['refunded'] = refunded - t['result'] = { - 'zip_path': zip_tmp_path, - 'filename': f"批量下载_{ok_count}首.zip", - 'audio_files': [], - } - else: - if zip_tmp_path: - try: - os.remove(zip_tmp_path) - except OSError: - pass - _mark_task_terminal(task_id, 'canceled', - '任务已取消' + (f"（已退回 {refunded} 次下载次数）" if refunded else '')) - with DOWNLOAD_TASKS_LOCK: - t = DOWNLOAD_TASKS.get(task_id) - if t is not None: - t['refunded'] = refunded - api_service.logger.info(f"批量ZIP已取消: {task_id}（成功{ok_count} 退回{refunded}次）") - except Exception as e: - # 通用异常与取消同一套交付/结算规则：有成果则交付，按账本退未交付部分 - # F-B：ZIP 未正常完成（写坏）时按全额退费，不发布坏包 - # 非预期异常原文 logger 先行，用户可见文案泛化 - api_service.logger.error(f"批量打包异常: {task_id} {e}\n{traceback.format_exc()}") - delivered = ok_count if zip_valid else 0 - refunded = _settle_batch_billing(task_id, delivered, 'error') - if delivered > 0 and zip_tmp_path: - _mark_task_terminal(task_id, 'error', - "部分歌曲已打包完成，可领取" + - (f"，退回 {refunded} 次" if refunded else '')) - with DOWNLOAD_TASKS_LOCK: - t = DOWNLOAD_TASKS.get(task_id) - if t is not None: - t['refunded'] = refunded - t['result'] = { - 'zip_path': zip_tmp_path, - 'filename': f"批量下载_{ok_count}首.zip", - 'audio_files': [], - } - else: - _mark_task_terminal(task_id, 'error', - '打包失败，请稍后重试' + (f"（已退回 {refunded} 次下载次数）" if refunded else '')) - with DOWNLOAD_TASKS_LOCK: - t = DOWNLOAD_TASKS.get(task_id) - if t is not None: - t['refunded'] = refunded - if zip_tmp_path: - try: - os.remove(zip_tmp_path) - except OSError: - pass - api_service.logger.error(f"批量ZIP任务异常: {task_id} {e}（成功{ok_count} 退回{refunded}次）\n{traceback.format_exc()}") - finally: - # 回收信息预取线程池，避免线程泄漏（未启动的预取任务一并取消） - if info_pool is not None: - try: - info_pool.shutdown(wait=False, cancel_futures=True) - except TypeError: - info_pool.shutdown(wait=False) + run_zip_job(_SonglistZipJob(), task_id, settle=_settle_batch_billing) diff --git a/services/podcast.py b/services/podcast.py index bcbabf3..7cf4a82 100644 --- a/services/podcast.py +++ b/services/podcast.py @@ -1,13 +1,17 @@ # -*- coding: utf-8 -*- """播客单集与整台打包 worker。 -""" +整台打包的骨架（状态机 / 暂停 / 名额 / 终态收敛 / 结算）在 +:mod:`services._zipcore`，本模块只保留播客侧策略：清单组装（整台或指定单集）、 +逐集顺序下载、按实际集数补扣、单条目入包（首集附封面）。 +""" import os import shutil import threading import time import traceback import zipfile +from dataclasses import dataclass from pathlib import Path from typing import Any @@ -27,14 +31,16 @@ from music_downloader import ( ) from ._shared import ( - BATCH_ZIP_SEM, _record_log, ) from ._taskcore import ( _mark_task_terminal, _settle_batch_billing, - _task_work_dir, - _wait_if_paused, +) +from ._zipcore import ( + ZipContext, + ZipJob, + run_zip_job, ) from .api_service import ( api_service, @@ -287,32 +293,37 @@ def _podcast_cleanup_files(paths: list[str | None]) -> None: pass -def _podcast_zip_worker(task_id: str) -> None: - """整台播客打包：逐集 下载→入包即删，支持取消与断线暂停（复用批量任务机制）""" - import tempfile as _tempfile +@dataclass(frozen=True) +class _EpisodePayload: + """一集下载完成后的入包素材（下载线程产出，落包线程消费）。""" - zip_tmp_path: str | None = None - ok_count = 0 - failed: list[str] = [] - # zip_valid / radio_name 必须在函数头初始化——空电台/无单集/补扣失败等 - # 正常业务错误在 try 内早期 raise，异常处理器引用未绑定名会二次崩溃 worker 线程 - zip_valid = False - radio_name = '' - try: - with DOWNLOAD_TASKS_LOCK: - task = DOWNLOAD_TASKS[task_id] - task['status'] = 'running' - task['msg'] = '解析单集列表...' - params = task['params'] - cancel_ev = task['cancel'] - pause_ev = task['pause'] - radio_id = params['radio_id'] - preset_ids = params.get('ids') - limit = params['limit'] - quality = params['quality'] + path: str + base: str + ext: str + + +class _PodcastZipJob(ZipJob): + """整台播客打包：逐集顺序下载（窗口=1）→ 单条目入包，首集附封面。 + + 并发刻意等于音乐批量：同样占用全局批量名额（BATCH_ZIP_SEM，全局 3）， + 逐集 acquire/release——暂停等待/取消时不长期占名额；音频本身仍受全局 + 下载信号量约束。原实现不受批量并发控制，多用户同时打包会无限并发。 + """ + + tag = '播客打包' + noun = '集' + item_noun = '单集' + tmp_prefix = 'podcast_batch_' + start_msg = '解析单集列表...' + concurrency = 1 + + def __init__(self) -> None: + self._radio_name = '' + + def prepare(self, params: dict[str, Any], task_id: str, failed: list[str]) -> list[Any]: cookies = params['cookies'] - remote_ip = params.get('remote_ip') or '-' - + radio_id = params['radio_id'] + limit = params['limit'] # 组装单集清单：整台电台（正序全量）或指定单集ID列表 if radio_id: episodes = podcast.list_radio_all_program_ids(radio_id, cookies, max_count=limit) @@ -320,7 +331,7 @@ def _podcast_zip_worker(task_id: str) -> None: raise DownloadException("该电台暂无可下载的单集") else: episodes = [] - for pid in (preset_ids or []): + for pid in (params.get('ids') or []): try: prog = podcast.get_program_detail(pid, cookies) episodes.append({ @@ -337,15 +348,17 @@ def _podcast_zip_worker(task_id: str) -> None: if not episodes: raise DownloadException("没有可下载的单集") - radio_name = _podcast_safe_name(params.get('radio_name') or '', '播客') - with DOWNLOAD_TASKS_LOCK: - DOWNLOAD_TASKS[task_id]['total'] = len(episodes) + self._radio_name = _podcast_safe_name(params.get('radio_name') or '', '播客') + return episodes - # 按次计费补扣：整台电台提交时只预扣 1 次，此处按实际单集数补足差额。 - # 补扣失败（余额不足/数据库异常）任务必须失败并按账本退费——不允许无账本继续交付 + def supplement(self, task_id: str, params: dict[str, Any], count: int) -> None: + """按次计费补扣：整台电台提交时只预扣 1 次，此处按实际单集数补足差额。 + + 补扣失败（余额不足/数据库异常）任务必须失败并按账本退费——不允许无账本继续交付。 + """ try: _buid = params.get('billing_uid') - _extra = len(episodes) - int(params.get('billed') or 1) + _extra = count - int(params.get('billed') or 1) if _buid and _extra > 0: if not auth_system.ledger_supplement(task_id, _extra): raise DownloadException( @@ -353,7 +366,7 @@ def _podcast_zip_worker(task_id: str) -> None: if _buid: with DOWNLOAD_TASKS_LOCK: DOWNLOAD_TASKS[task_id]['billing_user_id'] = int(_buid) - DOWNLOAD_TASKS[task_id]['billing_cost'] = len(episodes) + DOWNLOAD_TASKS[task_id]['billing_cost'] = count except DownloadCanceled: raise except DownloadException: @@ -361,206 +374,80 @@ def _podcast_zip_worker(task_id: str) -> None: except Exception as e: raise DownloadException(f"补扣下载次数异常: {e}") from e - def _update_progress(msg: str = '') -> None: - with DOWNLOAD_TASKS_LOCK: - t = DOWNLOAD_TASKS[task_id] - t['done'] = ok_count + len(failed) - t['failed'] = len(failed) - if msg: - t['msg'] = msg - - # 对齐音乐批量 F-B 口径——zip_valid 仅在 ZIP with 块正常 - # 退出（中央目录完整写出）后才置 True（初始 False 已上移函数头）； - # 未 valid 一律不发布、不按 ok_count 结算 - # 同音乐批量——ZIP 构建放 downloads 卷，不落系统 %TEMP% - # 与音乐批量打包同口径：建在任务专属目录 tasks//，根目录缓存清扫不再触及交付物 - _podcast_task_dir = _task_work_dir(task_id) - with _tempfile.NamedTemporaryFile(prefix='podcast_batch_', suffix='.zip', - dir=str(_podcast_task_dir), - delete=False) as zip_tmp: - zip_tmp_path = zip_tmp.name - used_arcnames = set() # 包内条目名去重（同名单集/同号不同集不得互相覆盖） - with zipfile.ZipFile(zip_tmp, 'w', zipfile.ZIP_DEFLATED) as zf: - for idx, ep in enumerate(episodes, 1): - if cancel_ev.is_set(): - raise DownloadCanceled("任务已取消") - _wait_if_paused(task, pause_ev, cancel_ev, ok_count + len(failed)) - # 并发对齐音乐批量：播客打包同样占用全局批量名额（BATCH_ZIP_SEM，全局 3）， - # 逐集 acquire/release——暂停等待/取消时不长期占名额；音频本身仍受 - # 全局下载信号量约束。原实现不受批量并发控制，多用户同时打包会无限并发 - if not BATCH_ZIP_SEM.acquire_cancellable(cancel_ev): - raise DownloadCanceled("任务已取消") - try: - label = f"[{idx}/{len(episodes)}] {ep['name'][:24]}" - _update_progress(f"{label} 解析中...") - try: - prog = ep.get('detail') or podcast.get_program_detail(ep['id'], cookies) - url_info = _podcast_resolve_audio(prog, quality, cookies) - _update_progress(f"{label} 下载中...") - ext = (url_info.get('type') or 'mp3').lower().lstrip('.') - serial = int(ep.get('serial_num') or 0) - prefix = f"{serial:03d} - " if serial else '' - base = _podcast_safe_name(f"{prefix}{ep['name']}") - # 中间文件移入任务目录——落 downloads 根目录会被 - # 缓存清扫按 mtime 打断慢速单集（Linux 上 unlink 打开中的文件） - tmp_audio = _podcast_task_dir / f"podcast_zip_tmp_{task_id}_{idx}.{ext}" - try: - _podcast_stream_download(url_info['url'], tmp_audio, - url_info.get('size', 0), cancel_ev) - # 重名条目自动追加序号 - arcname = f"{base}.{ext}" - k = 2 - while arcname in used_arcnames: - arcname = f"{base} ({k}).{ext}" - k += 1 - used_arcnames.add(arcname) - zf.write(str(tmp_audio), arcname=arcname, compress_type=zipfile.ZIP_STORED) - finally: - # finally 连带清理 .part 残片（原实现只删正文， - # 取消/失败时残片要等 1 小时缓存清扫兜底） - _podcast_cleanup_files([str(tmp_audio)]) - # 首集附带封面，便于识别 - if idx == 1 and ep.get('cover_url'): - try: - cover = fetch_cover_bytes(ep['cover_url']) - if cover: - zf.writestr("cover.jpg", cover, compress_type=zipfile.ZIP_STORED) - except Exception: - pass - ok_count += 1 - _update_progress(f"{label} 完成") - except DownloadCanceled: - raise - except Exception as e: - failed.append(f"{ep['name']}({e})") - _update_progress() - finally: - BATCH_ZIP_SEM.release() - - # 走到这里说明 ZIP with 块正常退出、中央目录完整，才允许发布 - zip_valid = True - - if ok_count == 0: - raise DownloadException(f"全部单集下载失败: {'; '.join(failed[:3])}") - - # ZIP 未正常完成（磁盘满写坏中央目录等）——不发布坏包，按全额退费收敛 - if not zip_valid: - if zip_tmp_path: - try: - os.remove(zip_tmp_path) - except OSError: - pass - zip_tmp_path = None - refunded = _settle_batch_billing(task_id, 0, 'error') - _mark_task_terminal(task_id, 'error', - f"ZIP 打包失败（可能磁盘空间不足），已退回 {refunded} 次" if refunded - else "ZIP 打包失败（可能磁盘空间不足），费用未扣或已退回") - with DOWNLOAD_TASKS_LOCK: - t = DOWNLOAD_TASKS.get(task_id) - if t is not None: - t['failed'] = t.get('failed', 0) + ok_count - t['refunded'] = refunded - api_service.logger.error(f"播客打包构建失败未发布: {task_id} ok_count={ok_count} 退回{refunded}次") - return - - # 最终发布前的取消检查（末段取消不得覆盖为 done） - if cancel_ev.is_set(): - raise DownloadCanceled("任务已取消") - refunded = _settle_batch_billing(task_id, ok_count) - with DOWNLOAD_TASKS_LOCK: - t = DOWNLOAD_TASKS.get(task_id) - # 发布前锁内复查取消/失效状态（同音乐批量） - if t is None or t.get('status') in ('canceled', 'error'): - raise DownloadCanceled("任务已取消") - if cancel_ev.is_set(): - raise DownloadCanceled("任务已取消") - t['status'] = 'done' - t['completed_at'] = time.time() - t['done'] = ok_count + len(failed) - t['failed'] = len(failed) - t['msg'] = f"成功{ok_count} 失败{len(failed)}" + (f" 退回{refunded}次" if refunded else '') - t['refunded'] = refunded - t['result'] = { - 'zip_path': zip_tmp_path, - 'filename': f"播客_{radio_name}_{ok_count}集.zip", - 'audio_files': [], - } + def download_one(self, item: Any, idx: int, ctx: ZipContext) -> _EpisodePayload: + ep = item + label = f"[{idx}/{ctx.total}] {ep['name'][:24]}" + ctx.progress(f"{label} 解析中...") + cookies = ctx.params['cookies'] + prog = ep.get('detail') or podcast.get_program_detail(ep['id'], cookies) + url_info = _podcast_resolve_audio(prog, ctx.params['quality'], cookies) + ctx.progress(f"{label} 下载中...") + ext = (url_info.get('type') or 'mp3').lower().lstrip('.') + serial = int(ep.get('serial_num') or 0) + prefix = f"{serial:03d} - " if serial else '' + base = _podcast_safe_name(f"{prefix}{ep['name']}") + # 中间文件移入任务目录——落 downloads 根目录会被缓存清扫按 mtime 打断慢速 + # 单集（Linux 上 unlink 打开中的文件） + tmp_audio = ctx.work_dir / f"podcast_zip_tmp_{ctx.task_id}_{idx}.{ext}" try: - first_name = episodes[0].get('name') if episodes else '' - _record_log('下载', remote_ip, f"[播客打包] {radio_name} · {first_name}") + _podcast_stream_download(url_info['url'], tmp_audio, + url_info.get('size', 0), ctx.cancel_ev) + except Exception: + # 下载失败即清（.part 残片同清）；成功时由 write_entry 的 finally 清 + _podcast_cleanup_files([str(tmp_audio)]) + raise + return _EpisodePayload(path=str(tmp_audio), base=base, ext=ext) + + def write_entry(self, item: Any, idx: int, payload: _EpisodePayload, ctx: ZipContext) -> None: + ep = item + try: + # 重名条目自动追加序号（同名单集/同号不同集不得互相覆盖） + arcname = f"{payload.base}.{payload.ext}" + k = 2 + while arcname in ctx.used_arcnames: + arcname = f"{payload.base} ({k}).{payload.ext}" + k += 1 + ctx.used_arcnames.add(arcname) + ctx.zf.write(payload.path, arcname=arcname, compress_type=zipfile.ZIP_STORED) + # 首集附带封面，便于识别 + if idx == 1 and ep.get('cover_url'): + try: + cover = fetch_cover_bytes(ep['cover_url']) + if cover: + ctx.zf.writestr("cover.jpg", cover, compress_type=zipfile.ZIP_STORED) + except Exception: + pass + finally: + # finally 连带清理 .part 残片（原实现只删正文，取消/失败时残片要等 + # 1 小时缓存清扫兜底） + _podcast_cleanup_files([payload.path]) + + def fail_label(self, item: Any) -> str: + """失败文案用单集名而非整个条目字典（引擎拼成「单集名(异常)」）。""" + return str(item.get('name') or '') + + def done_msg(self, item: Any, idx: int, ctx: ZipContext) -> str: + return f"[{idx}/{ctx.total}] {item['name'][:24]} 完成" + + def zip_filename(self, ok_count: int) -> str: + return f"播客_{self._radio_name or '打包'}_{ok_count}集.zip" + + def log_label(self) -> str: + return f"{self._radio_name} " + + def all_failed_msg(self, failed: list[str], refunded: int) -> str: + return f"全部单集下载失败: {'; '.join(failed[:3])}" + ( + f"（已退回 {refunded} 次下载次数）" if refunded else '') + + def on_delivered(self, items: list[Any], task_id: str, params: dict[str, Any]) -> None: + try: + first_name = items[0].get('name') if items else '' + _record_log('下载', params.get('remote_ip') or '-', + f"[播客打包] {self._radio_name} · {first_name}") except Exception: pass - api_service.logger.info(f"播客打包完成: {radio_name} 成功{ok_count} 失败{len(failed)} ({task_id})") - except DownloadCanceled: - # 部分成功后取消——已入包部分照常交付，未完成部分按账本退费 - # ZIP 未正常完成（写坏）时不按 ok_count 交付——按全额退费收敛 - delivered = ok_count if zip_valid else 0 - refunded = _settle_batch_billing(task_id, delivered, 'canceled') - if delivered > 0 and zip_tmp_path: - _mark_task_terminal(task_id, 'canceled', - f"已取消（已打包 {ok_count} 集可领取）" + - (f"，退回 {refunded} 次" if refunded else '')) - with DOWNLOAD_TASKS_LOCK: - t = DOWNLOAD_TASKS.get(task_id) - if t is not None: - t['refunded'] = refunded - t['result'] = { - 'zip_path': zip_tmp_path, - 'filename': f"播客_{radio_name or '打包'}_{ok_count}集.zip", - 'audio_files': [], - } - else: - _mark_task_terminal(task_id, 'canceled', - '任务已取消' + (f"（已退回 {refunded} 次下载次数）" if refunded else '')) - with DOWNLOAD_TASKS_LOCK: - t = DOWNLOAD_TASKS.get(task_id) - if t is not None: - t['refunded'] = refunded - if zip_tmp_path: - try: - os.remove(zip_tmp_path) - except OSError: - pass - api_service.logger.info(f"播客打包已取消: {task_id}（成功{ok_count} 退回{refunded}次）") - except DownloadException as e: - delivered = ok_count if zip_valid else 0 - refunded = _settle_batch_billing(task_id, delivered) - with DOWNLOAD_TASKS_LOCK: - _t = DOWNLOAD_TASKS.get(task_id) - # 终态先到先得——已终态（如用户取消/超时兜底） - # 不再被覆盖为 error（堵 canceled→error 非法迁移边） - if _t and _t.get('status') not in ('done', 'error', 'canceled'): - _t['status'] = 'error' - _t['completed_at'] = time.time() # 保留期从完成时刻起算 - _t['msg'] = str(e) + (f"（已退回 {refunded} 次下载次数）" if refunded else '') - _t['refunded'] = refunded - elif _t: - _t['refunded'] = refunded - if zip_tmp_path: - try: - os.remove(zip_tmp_path) - except OSError: - pass - api_service.logger.error(f"播客打包失败: {task_id} {e}（退回{refunded}次）") - except Exception as e: - # 非预期异常原文 logger 先行，用户可见文案泛化 - # （DownloadException 的文案是我们自己拼的用户可读文本，保留 str(e)） - api_service.logger.error(f"播客打包异常: {task_id} {e}\n{traceback.format_exc()}") - delivered = ok_count if zip_valid else 0 - refunded = _settle_batch_billing(task_id, delivered) - with DOWNLOAD_TASKS_LOCK: - _t = DOWNLOAD_TASKS.get(task_id) - # 同上——已终态不再覆盖为 error - if _t and _t.get('status') not in ('done', 'error', 'canceled'): - _t['status'] = 'error' - _t['completed_at'] = time.time() # 保留期从完成时刻起算 - _t['msg'] = '打包失败，请稍后重试' + (f"（已退回 {refunded} 次下载次数）" if refunded else '') - _t['refunded'] = refunded - elif _t: - _t['refunded'] = refunded - if zip_tmp_path: - try: - os.remove(zip_tmp_path) - except OSError: - pass - api_service.logger.error(f"播客打包异常: {task_id} {e}（退回{refunded}次）\n{traceback.format_exc()}") + + +def _podcast_zip_worker(task_id: str) -> None: + """整台播客打包：逐集 下载→入包即删，支持取消与断线暂停（复用批量任务机制）""" + run_zip_job(_PodcastZipJob(), task_id, settle=_settle_batch_billing) diff --git a/tests/test_zip_workers.py b/tests/test_zip_workers.py index a17ae7a..b655a4f 100644 --- a/tests/test_zip_workers.py +++ b/tests/test_zip_workers.py @@ -1,13 +1,17 @@ # -*- coding: utf-8 -*- """两个 ZIP 打包 worker 的行为契约测试（沙箱，不触碰生产 data/）。 -为什么：``_batch_zip_worker`` / ``_podcast_zip_worker`` 合计约 642 行此前**零测试覆盖**， -而两者高度重复——连 ``_wait_if_paused`` 那 16 行都逐字相同（上一次修的 -``zip_valid`` 可能未绑定缺陷就是同型出现在这两处，重复代码在持续制造同类 bug）。 +为什么：``_batch_zip_worker`` / ``_podcast_zip_worker`` 合并前合计约 642 行、 +**零测试覆盖**，且两者高度重复——连 ``_wait_if_paused`` 那 16 行都逐字相同 +（``zip_valid`` 未绑定那类缺陷就是同型出现在这两处）。抽取公共实现 +（``services/_zipcore.py``）之前，先用**行为级**测试钉住可观察契约：任务终态、 +ZIP 内容、账本守恒、暂停/恢复/取消语义。断言只依赖「调用 worker 后世界里发生了 +什么」，不引用任何内部函数名——所以合并实现时本文件一行未改，它同时就是等价性 +验证器。 -抽公共实现之前，先用**行为级**测试钉住可观察契约：任务终态、ZIP 内容、 -账本守恒、暂停/恢复/取消语义。断言只依赖「调用 worker 后世界里发生了什么」， -不引用任何内部函数名——这样重构时测试不必跟着改，才具备等价性验证能力。 +末尾「合并后的契约固化」一节是**新增**的钉子：合并时逐条对齐了几处分叉 +（详见各用例 docstring），这些用例在合并前必然失败——留着它们，那次对齐才不会 +被后来者当成「本来就如此」而回退。 账本口径：一律用 ``reserve_and_open`` 真预扣（``ledger_open`` 只记账不扣款， 会让「退款」凭空给余额加次数，守恒断言就失去意义）。 @@ -15,6 +19,7 @@ import threading import time import zipfile +from dataclasses import replace from pathlib import Path from typing import Any @@ -23,7 +28,7 @@ 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 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 @@ -276,6 +281,7 @@ def test_batch_zip_pause_then_cancel(sandbox, monkeypatch): 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 = auth.ledger_get(tid) assert row is not None and row['delivered'] == 0 and row['status'] != 'open' @@ -454,3 +460,125 @@ def test_podcast_zip_all_fail_refunds(podcast_stub, monkeypatch): assert _balance() == before, '全失败应全额退回' row = auth.ledger_get(tid) assert row is not None and row['delivered'] == 0 + + +# ============ 合并后的契约固化（分叉点对齐 / 拆分风险点） ============ +# +# 以下用例在「两个 worker 各写一份」的旧实现上**必然失败**——它们钉的是合并时 +# 逐条对齐的结果，以及「下载」与「落包」拆成两个钩子后新增的边界。 + +class _SameNameDownloader(_FakeDownloader): + """所有歌曲同名同歌手、且都带封面——用于验证包内三条目共享序号后缀。""" + + def get_music_info(self, mid: Any, quality: str, cookies: Any) -> MusicInfo: + return replace(super().get_music_info(mid, quality, cookies), + 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'])) + monkeypatch.setattr(batch_svc, 'fetch_cover_bytes', lambda _u: b'\xff\xd8\xff\xe0') + 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 = 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() + + 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 = 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() + + 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}'