This repository has been archived on 2026-10-02. You can view files and clone it. You cannot open issues or pull requests or push a commit.
Files
NeteaseUrl/services/podcast.py
T
snowy-yang a05af4a880 refactor: 播客流式下载拆解(嵌套 10 层 → 1,三段职责各就其位)
为什么:_podcast_stream_download 是播客单集下载的**唯一落盘路径**(单集 worker
与整台打包的首集都走它),却零测试覆盖;而且它把三段互不相关的职责——全局名额
排队、流式写盘与进度上报、完整性校验与坏文件清理——叠在 try/with/with/for 加四层
if 里,最大嵌套 10 层,是全仓最深的一个。

改法:拆成四个具名函数,主函数只留编排。
  _stream_to_part    :拿响应 → 写 .part → 按 0.5s 节流上报进度,返回 (已写, 期望)
  _report_progress   :把一次进度写进任务表(含单集任务的 live_percent)
  _verify_downloaded :完整性判定 + 清掉坏文件(精确相等 / 非空两条口径)
  _remove_quietly    :删文件并吞掉 OSError(Windows 句柄未释放时 remove 会抛)

复杂度(同口径 AST 实测):
  _podcast_stream_download  63 行 / 圈 19 / 嵌套 10  →  21 行 / 圈 3 / 嵌套 1
  新增 _stream_to_part 25 行/圈 8/嵌套 4 · _report_progress 22/6/3 · _verify_downloaded 14/3/1
  文件内嵌套 ≥5 层 2 → 1(余下那个是本次未动的 _podcast_download_worker)

刻意保留的语义(是决定,不是疏漏):
- 取消时 **.part 留存不删**(改名不发生,下次同名下载直接覆盖)——已用测试固化。
  不「顺手」改成删除,那是行为变更。
- 进度 msg 的计算与「是否真有可取的任务」无关:task_id 为空时仍会格式化字符串再
  return。这是既有的无害浪费,一并保留。
- 完整性口径原样:已知期望大小要求**精确相等**(放宽到 95% 容差会交付截断的坏文件
  还照扣次数),期望未知则退化为非空校验。

等价性证据(两道):
1. 反向注入验证:把 HEAD 的旧实现换回去,用同一套 tests/test_podcast_stream.py 跑
   → 12 例**全过**,证明这批测试忠实编码旧行为、对新实现构成等价性验证。
2. 新增 12 例行为契约(该函数此前零覆盖):正常落盘与改名、空分块跳过、
   Content-Length 兜底、字节数不符清坏文件、空下载报错、已知大小不容忍部分交付、
   上游 4xx、开始前取消、中途取消留 .part、running 任务写 msg/live_percent、
   终态任务不被覆盖、无 task_id 不报错。

顺带记录一个测试自身的坑:替身响应最初用 list(chunks) 归一化,导致传入生成器的
副作用(中途置取消位)在**构造时刻**就执行,测出来变成「开始前取消」——两条本不
相同的路径被混成一条。已改为保留原 iterable 并写明原因。

验证:ruff 全过 · pyright 0/0 · pytest 245 passed(233→245)· 生产 data/ 未被触碰。
2026-09-27 07:45:49 +08:00

487 lines
20 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# -*- 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)