# -*- 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 import auth_system from core.state import ( DOWNLOAD_TASKS, DOWNLOAD_TASKS_LOCK, ) from music_api import ( APIException, podcast, ) from music_downloader import ( DownloadCanceled, DownloadException, fetch_cover_bytes, ) from ._shared import ( _record_log, ) from ._taskcore import ( _mark_task_terminal, _settle_batch_billing, ) from ._zipcore import ( ZipContext, ZipJob, run_zip_job, ) from .api_service import ( api_service, ) from .streaming import ( _stream_open, ) def _podcast_safe_name(s: str, fallback: str = 'unknown') -> str: """文件名安全化（去除非法字符）""" s = ''.join(c for c in str(s or '') if c not in r'<>:"/\|?*' and ord(c) >= 32).strip() return s or fallback def _remove_quietly(path: Path) -> None: """删除文件并忽略失败（Windows 上句柄未释放时 os.remove 会抛 OSError）。""" try: os.remove(str(path)) except OSError: pass def _report_progress(task_id: str, progress_label: str, done: int, total: int) -> None: """把一次进度写进任务表；task_id 为空或任务不在 running 时只计算不写。 msg 的计算与「是否真有可取的任务」无关（历史实现如此），故 task_id 为空时 仍会格式化字符串——保持这一无害的既有行为，不做「优化」。 """ pct = 0 if total: pct = min(99, int(done / total * 100)) msg = (f"{progress_label} {pct}% · " f"{api_service._format_file_size(done)}/{api_service._format_file_size(total)}") else: msg = f"{progress_label} 已下载 {api_service._format_file_size(done)}" if not task_id: return with DOWNLOAD_TASKS_LOCK: t = DOWNLOAD_TASKS.get(task_id) if t and t['status'] == 'running': t['msg'] = msg # 单集任务的字节级进度（批量任务保持按集数口径） if t.get('type') == 'single' and total: t['live_percent'] = min(99.0, float(pct)) def _stream_to_part(url: str, part: Path, total_size: int, cancel_ev: threading.Event, task_id: str, progress_label: str) -> tuple[int, int]: """把响应体流式写入 part，返回 (已写字节数, 期望总大小)。 期望总大小取入参，缺失时回落到 Content-Length（仍可能为 0，表示上游没给）。 """ done = 0 last_update = 0.0 with _stream_open(url, timeout=60) as r: if r.status >= 400: raise DownloadException(f"上游返回 {r.status}") total = total_size or int(r.headers.get('Content-Length') or 0) with open(part, 'wb') as f: for chunk in r.iter_content(chunk_size=256 * 1024): if cancel_ev.is_set(): raise DownloadCanceled("任务已取消") if not chunk: continue f.write(chunk) done += len(chunk) now = time.time() if now - last_update > 0.5: last_update = now _report_progress(task_id, progress_label, done, total) return done, total def _verify_downloaded(target: Path, done: int, total_size: int) -> None: """完整性校验：不达标则删掉已落地的文件并抛 DownloadException。 为什么非要校验：CDN 提前断流时 iter_content 会「正常」结束，文件静默截断。 已知期望大小时要求**精确相等**（API 报的大小与实际一致；放宽到 95% 容差会 交付截断的坏文件还照扣次数）；期望未知时退化为非空校验——Content-Length 兜底值可能是压缩口径，不做精确比对。 """ if total_size and done != total_size: _remove_quietly(target) raise DownloadException(f"下载不完整（{done}/{total_size} 字节），请重试") if not total_size and done <= 0: _remove_quietly(target) raise DownloadException("下载内容为空，请重试") def _podcast_stream_download(url: str, target: Path, total_size: int, cancel_ev: threading.Event, task_id: str = '', progress_label: str = '') -> int: """流式下载播客音频到 target（先写 .part 再改名），返回实际字节数。 自动加入全局下载信号量排队；每 0.5s 更新一次任务进度消息。 取消时 .part 会留存（不删）——下次下载同名文件直接覆盖，是既有行为。 """ part = target.with_name(target.name + '.part') # 等待全局下载名额期间可被取消（与音乐批量 BATCH_ZIP_SEM 同语义） sem = api_service.downloader._download_semaphore if not sem.acquire_cancellable(cancel_ev): raise DownloadCanceled("任务已取消") try: done, _total = _stream_to_part(url, part, total_size, cancel_ev, task_id, progress_label) finally: sem.release() if part.exists(): part.replace(target) _verify_downloaded(target, done, total_size) return done def _podcast_resolve_audio(program: dict[str, Any], quality: str, cookies: dict[str, str]) -> dict[str, Any]: """解析单集音频直链，失败抛 DownloadException""" song_id = program.get('main_song_id') if not song_id: raise DownloadException("该单集缺少音频信息（mainSong）") url_info = podcast.resolve_episode_url(song_id, cookies, quality) if not url_info or not url_info.get('url'): raise DownloadException("未获取到音频直链（单集可能已下架或受版权限制）") return url_info def _podcast_download_worker(task_id: str) -> None: """后台播客单集下载：解析单集 → 音质降级取直链 → 流式下载（可选 ZIP 打包） 任务 type 为 'single'，完成后经 /download/result 交付，与单曲下载共用交付通道。 落盘文件进入任务专属目录且带节目 ID，跨任务永不重名；清理只动本任务目录。 """ import tempfile as _tempfile file_path_str: str | None = None zip_tmp_path: str | None = None task_dir: Path | None = None try: with DOWNLOAD_TASKS_LOCK: task = DOWNLOAD_TASKS[task_id] task['status'] = 'running' task['msg'] = '解析单集信息...' params = task['params'] cancel_ev = task['cancel'] program_id = params['program_id'] quality = params['quality'] need_zip = params['need_zip'] cookies = params['cookies'] remote_ip = params.get('remote_ip') or '-' program = podcast.get_program_detail(program_id, cookies) url_info = _podcast_resolve_audio(program, quality, cookies) serial = int(program.get('serial_num') or 0) name = program.get('name') or program.get('main_song_name') or '播客单集' prefix = f"{serial:03d} - " if serial else '' base = _podcast_safe_name(f"{prefix}{name}") ext = (url_info.get('type') or 'mp3').lower().lstrip('.') # 磁盘名附节目 ID + 任务目录隔离（两个任务同下载一集互不覆盖、互不删除） task_dir = api_service.downloads_path / 'tasks' / task_id task_dir.mkdir(parents=True, exist_ok=True) disk_name = f"{base} [{program_id}].{ext}" target = task_dir / disk_name filename = f"{base}.{ext}" # 交付名保持干净（不带 ID） 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] if need_zip: fd, zip_tmp_path = _tempfile.mkstemp(prefix='podcast_zip_', suffix='.zip', dir=str(task_dir)) os.close(fd) with zipfile.ZipFile(zip_tmp_path, 'w', zipfile.ZIP_DEFLATED) as zf: zf.write(file_path_str, arcname=filename) cover_url = program.get('cover_url') or '' if cover_url: try: cover = fetch_cover_bytes(cover_url) if cover: zf.writestr(f"{base}.jpg", cover) except Exception: pass # info.txt 不再写直链（真实 URL 不应随交付文件外泄） info_txt = ( f"单集: {name}\n" f"播客: {program.get('radio_name') or ''}\n" f"主播: {program.get('dj_name') or ''}\n" f"时长: {program.get('duration_text') or ''}\n" f"音质: {api_service._get_quality_display_name(url_info.get('level', quality))}\n" ) zf.writestr("info.txt", info_txt) # 打包阶段后同样以取消检查收口 if cancel_ev.is_set(): raise DownloadCanceled("任务已取消") result = { 'file_path': file_path_str, 'zip_path': zip_tmp_path, 'filename': f"{base}.zip", 'mimetype': 'application/zip', 'audio_files': [file_path_str], } else: result = { 'file_path': file_path_str, 'filename': filename, 'mimetype': f"audio/{ext}", 'audio_files': [file_path_str], } # 最终发布前的取消检查（末段取消不得覆盖为 done） if cancel_ev.is_set(): raise DownloadCanceled("任务已取消") with DOWNLOAD_TASKS_LOCK: t = DOWNLOAD_TASKS.get(task_id) if t is None or t.get('status') in ('canceled', 'error'): raise DownloadCanceled("任务已取消") t['status'] = 'done' t['completed_at'] = time.time() t['done'] = 1 t['msg'] = '下载完成' t['result'] = result # 交付已落定；结算失败不再把任务改标 error（同单曲） try: auth_system.ledger_settle(task_id, 1, 'delivered') except Exception as _e: api_service.logger.error(f"播客单集交付结算失败（重启对账兜底）: {task_id} {_e}") try: _record_log('下载', remote_ip, f"[播客] {name}") except Exception: pass api_service.logger.info(f"播客单集下载完成: {filename} ({task_id})") 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}") 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}") except Exception as e: # 非预期异常原文 logger 先行，用户可见文案泛化（教训） api_service.logger.error(f"播客单集下载异常: {task_id} {e}\n{traceback.format_exc()}") _mark_task_terminal(task_id, 'error', '下载失败，请稍后重试') try: auth_system.ledger_settle(task_id, 0, 'error') except Exception: pass if task_dir is not None: shutil.rmtree(str(task_dir), ignore_errors=True) def _podcast_cleanup_files(paths: list[str | None]) -> None: """清理播客任务产生的临时/残留文件""" for p in paths: if not p: continue try: os.remove(p) except OSError: pass # 连带清理可能残留的 .part 文件 try: part = Path(str(p) + '.part') if part.exists(): part.unlink() except OSError: pass @dataclass(frozen=True) class _EpisodePayload: """一集下载完成后的入包素材（下载线程产出，落包线程消费）。""" 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'] 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) if not episodes: raise DownloadException("该电台暂无可下载的单集") else: episodes = [] for pid in (params.get('ids') or []): try: prog = podcast.get_program_detail(pid, cookies) episodes.append({ 'id': prog['id'], 'name': prog['name'], 'main_song_id': prog['main_song_id'], 'serial_num': prog['serial_num'], 'cover_url': prog['cover_url'], 'detail': prog, }) except APIException as ae: failed.append(f"{pid}({ae})") episodes = episodes[:limit] if not episodes: raise DownloadException("没有可下载的单集") self._radio_name = _podcast_safe_name(params.get('radio_name') or '', '播客') return episodes def supplement(self, task_id: str, params: dict[str, Any], count: int) -> None: """按次计费补扣：整台电台提交时只预扣 1 次，此处按实际单集数补足差额。 补扣失败（余额不足/数据库异常）任务必须失败并按账本退费——不允许无账本继续交付。 """ try: _buid = params.get('billing_uid') _extra = count - int(params.get('billed') or 1) if _buid and _extra > 0: if not auth_system.ledger_supplement(task_id, _extra): raise DownloadException( f"补扣下载次数失败（需补扣 {_extra} 次），任务已取消，预扣部分将退回") if _buid: with DOWNLOAD_TASKS_LOCK: DOWNLOAD_TASKS[task_id]['billing_user_id'] = int(_buid) DOWNLOAD_TASKS[task_id]['billing_cost'] = count except DownloadCanceled: raise except DownloadException: raise except Exception as e: raise DownloadException(f"补扣下载次数异常: {e}") from e 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: _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 def _podcast_zip_worker(task_id: str) -> None: """整台播客打包：逐集 下载→入包即删，支持取消与断线暂停（复用批量任务机制）""" run_zip_job(_PodcastZipJob(), task_id, settle=_settle_batch_billing)