diff --git a/services/podcast.py b/services/podcast.py index 7cf4a82..14f1c2e 100644 --- a/services/podcast.py +++ b/services/podcast.py @@ -56,68 +56,101 @@ def _podcast_safe_name(s: str, fallback: str = 'unknown') -> str: return s or fallback -def _podcast_stream_download(url: str, target: Path, total_size: int, - cancel_ev: threading.Event, - task_id: str = '', progress_label: str = '') -> int: - """流式下载播客音频到 target(先写 .part 再改名),返回实际字节数 +def _remove_quietly(path: Path) -> None: + """删除文件并忽略失败(Windows 上句柄未释放时 os.remove 会抛 OSError)。""" + try: + os.remove(str(path)) + except OSError: + pass - 自动加入全局下载信号量排队;每 0.5s 更新一次任务进度消息。 + +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: - 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 chunk: - f.write(chunk) - done += len(chunk) - now = time.time() - if now - last_update > 0.5: - last_update = now - pct = 0 - if total: - pct = min(99, int(done / total * 100)) - msg = f"{progress_label} {pct}% · {api_service._format_file_size(done)}/{api_service._format_file_size(total)}" - else: - msg = f"{progress_label} 已下载 {api_service._format_file_size(done)}" - if task_id: - 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)) + done, _total = _stream_to_part(url, part, total_size, cancel_ev, task_id, progress_label) finally: sem.release() if part.exists(): part.replace(target) - # 完整性校验:CDN 提前断流时 iter_content 会"正常"结束,文件静默截断 - # (API 报告的期望大小时精确相等——95% 容差会交付截断的坏文件还扣次数; - # 期望未知时退化为非空校验,Content-Length 兜底值可能是压缩口径不做精确比对) - if total_size and done != total_size: - try: - os.remove(str(target)) - except OSError: - pass - raise DownloadException(f"下载不完整({done}/{total_size} 字节),请重试") - if not total_size and done <= 0: - try: - os.remove(str(target)) - except OSError: - pass - raise DownloadException("下载内容为空,请重试") + _verify_downloaded(target, done, total_size) return done diff --git a/tests/test_podcast_stream.py b/tests/test_podcast_stream.py new file mode 100644 index 0000000..c8a232e --- /dev/null +++ b/tests/test_podcast_stream.py @@ -0,0 +1,220 @@ +# -*- coding: utf-8 -*- +"""`_podcast_stream_download` 的行为契约。 + +为什么:这是播客单集下载的**唯一落盘路径**(单集 worker 与整台打包的首集都走它), +却零测试覆盖。把 63 行 / 嵌套 10 层的实现拆成「拿名额 → 流式写 → 改名 → 校验」 +四步后,需要行为级测试钉住可观察契约:返回字节数、半成品文件名、失败时是否清掉 +已落地的坏文件、取消语义、进度消息写到哪。 + +断言只看「调用后世界里发生了什么」(文件系统 + 任务表),不引用私有函数名, +故可跨后续重构复用。 +""" +import threading + +import pytest + +from core.state import DOWNLOAD_TASKS, DOWNLOAD_TASKS_LOCK +from music_downloader import DownloadCanceled, DownloadException +from services import podcast as podcast_svc + + +class _FakeResp: + """`_stream_open` 的替身:可控状态码、Content-Length 与分块序列。 + + 注意不能用 list(chunks) 归一化——那会把传入的生成器**立即求值**,使「迭代中 + 才触发的副作用」(如中途置取消位)提前到构造时刻,测出来的就不是想测的路径。 + """ + + def __init__(self, chunks, status=200, content_length=None): + self._chunks = chunks + self.status = status + self.headers = {} if content_length is None else {'Content-Length': str(content_length)} + + def __enter__(self): + return self + + def __exit__(self, *exc): + return False + + def iter_content(self, chunk_size=0): + yield from self._chunks + + +class _CancelAfterFirstChunk: + """迭代器:吐第一块后置取消位,再吐第二块。 + + 这样取消发生在**迭代过程中**(下载已开始、.part 已建立),与「开始前取消」 + 是两条不同的路径。 + """ + + def __init__(self, cancel): + self._cancel = cancel + self._n = 0 + + def __iter__(self): + return self + + def __next__(self): + self._n += 1 + if self._n == 1: + return b'x' * 10 + if self._n == 2: + self._cancel.set() + return b'y' * 10 + raise StopIteration + + +@pytest.fixture() +def fake_stream(monkeypatch): + """替换 _stream_open;测试通过 holder['resp'] 指定本次响应。""" + holder: dict = {} + + def _open(url, timeout=60): + assert url == holder.get('url', url) + return holder['resp'] + + monkeypatch.setattr(podcast_svc, '_stream_open', _open) + return holder + + +@pytest.fixture() +def task_table(): + """在全局任务表里放一条任务,测试结束清理(该表是跨模块共享的进程内状态)。""" + tid = 'stream-test-task' + + def _put(**kw): + task = {'id': tid, 'status': 'running', 'type': 'single', 'msg': ''} + task.update(kw) + with DOWNLOAD_TASKS_LOCK: + DOWNLOAD_TASKS[tid] = task + return task + + yield _put + with DOWNLOAD_TASKS_LOCK: + DOWNLOAD_TASKS.pop(tid, None) + + +# ============ 正常路径 ============ + + +def test_writes_target_and_renames_part(tmp_path, fake_stream): + """正常下载:返回实际字节数,落地 target,不残留 .part。""" + fake_stream['resp'] = _FakeResp([b'a' * 600, b'b' * 400]) + target = tmp_path / 'ep.mp3' + done = podcast_svc._podcast_stream_download('http://u', target, 1000, threading.Event()) + assert done == 1000 + assert target.read_bytes() == b'a' * 600 + b'b' * 400 + assert not target.with_name('ep.mp3.part').exists() + + +def test_empty_chunks_are_skipped(tmp_path, fake_stream): + """空分块不计入字节数(上游偶发发空 chunk)。""" + fake_stream['resp'] = _FakeResp([b'', b'x' * 50, b'']) + target = tmp_path / 'ep.mp3' + assert podcast_svc._podcast_stream_download('http://u', target, 50, threading.Event()) == 50 + + +def test_content_length_used_when_total_unknown(tmp_path, fake_stream): + """未传期望大小时用 Content-Length 兜底,并据此校验完整性。""" + fake_stream['resp'] = _FakeResp([b'x' * 30], content_length=30) + target = tmp_path / 'ep.mp3' + assert podcast_svc._podcast_stream_download('http://u', target, 0, threading.Event()) == 30 + + +# ============ 完整性校验 ============ + + +def test_size_mismatch_removes_delivered_file(tmp_path, fake_stream): + """字节数不符:删掉已落地的文件并报错(不交付截断的坏文件)。""" + fake_stream['resp'] = _FakeResp([b'x' * 100]) + target = tmp_path / 'ep.mp3' + with pytest.raises(DownloadException) as ei: + podcast_svc._podcast_stream_download('http://u', target, 200, threading.Event()) + assert '下载不完整' in str(ei.value) + assert not target.exists(), '坏文件必须被清掉' + + +def test_empty_download_raises_and_removes(tmp_path, fake_stream): + """期望大小未知且一个字节都没拿到 → 报「下载内容为空」。""" + fake_stream['resp'] = _FakeResp([]) + target = tmp_path / 'ep.mp3' + with pytest.raises(DownloadException) as ei: + podcast_svc._podcast_stream_download('http://u', target, 0, threading.Event()) + assert '下载内容为空' in str(ei.value) + assert not target.exists() + + +def test_known_size_does_not_tolerate_partial(tmp_path, fake_stream): + """已知期望大小时要求精确相等——95% 容差会放行截断的坏文件。""" + fake_stream['resp'] = _FakeResp([b'x' * 95]) + target = tmp_path / 'ep.mp3' + with pytest.raises(DownloadException): + podcast_svc._podcast_stream_download('http://u', target, 100, threading.Event()) + + +# ============ 上游与取消 ============ + + +def test_upstream_error_status_raises(tmp_path, fake_stream): + fake_stream['resp'] = _FakeResp([], status=404) + target = tmp_path / 'ep.mp3' + with pytest.raises(DownloadException) as ei: + podcast_svc._podcast_stream_download('http://u', target, 100, threading.Event()) + assert '上游返回 404' in str(ei.value) + assert not target.exists() + + +def test_cancel_before_first_chunk(tmp_path, fake_stream): + """取消已置位:抛 DownloadCanceled,不发布 target。""" + cancel = threading.Event() + cancel.set() + fake_stream['resp'] = _FakeResp([b'x' * 10]) + target = tmp_path / 'ep.mp3' + with pytest.raises(DownloadCanceled): + podcast_svc._podcast_stream_download('http://u', target, 10, cancel) + assert not target.exists() + + +def test_cancel_midway_leaves_part_file(tmp_path, fake_stream): + """下载中途取消:.part 残留(不删)——改名不发生,下次同名下载直接覆盖。""" + cancel = threading.Event() + fake_stream['resp'] = _FakeResp(_CancelAfterFirstChunk(cancel)) + target = tmp_path / 'ep.mp3' + with pytest.raises(DownloadCanceled): + podcast_svc._podcast_stream_download('http://u', target, 20, cancel) + assert not target.exists() + assert target.with_name('ep.mp3.part').exists(), '既有行为:取消时 .part 留存' + + +# ============ 进度消息 ============ + + +def test_progress_written_for_running_single_task(tmp_path, fake_stream, task_table): + """running 的单集任务:写 msg,并按字节级口径写 live_percent。 + + 单块即完成 → 百分比被 min(99, ...) 封在 99(进度条不预支 100%,收尾靠终态)。 + """ + task = task_table() + fake_stream['resp'] = _FakeResp([b'x' * 100]) + target = tmp_path / 'ep.mp3' + podcast_svc._podcast_stream_download('http://u', target, 100, threading.Event(), + task_id=task['id'], progress_label='下载中') + assert '下载中' in task['msg'] and '99%' in task['msg'] + assert task['live_percent'] == 99.0 + + +def test_progress_not_written_for_finished_task(tmp_path, fake_stream, task_table): + """已终态的任务不再被写进度(避免覆盖终态消息),但下载照常完成。""" + task = task_table(status='done', msg='原消息') + fake_stream['resp'] = _FakeResp([b'x' * 50]) + target = tmp_path / 'ep.mp3' + assert podcast_svc._podcast_stream_download('http://u', target, 50, threading.Event(), + task_id=task['id']) == 50 + assert task['msg'] == '原消息' + + +def test_progress_skipped_without_task_id(tmp_path, fake_stream): + """task_id 为空时只下载不写任务表(不得因缺 id 而报错)。""" + fake_stream['resp'] = _FakeResp([b'x' * 50]) + target = tmp_path / 'ep.mp3' + assert podcast_svc._podcast_stream_download('http://u', target, 50, threading.Event()) == 50