Archived
snowy-yang
1d86ec87ec
为什么先测后改：_batch_zip_worker(347 行) 与 _podcast_zip_worker(295 行) 近乎 复制——连 _wait_if_paused 那 16 行都逐字相同（注释只差一行）。重复的代价是实打实 的：上一次修的 zip_valid「变量可能未绑定」缺陷就是**同型出现在这两处**。 但暂停/恢复是一条并发状态机，此前**零测试覆盖**，直接改不符合先测量后改动。 所以本 commit 分两步落地。 第一步（新增 tests/test_zip_workers.py，9 个用例）——行为级契约 断言只用「调用 worker 后世界里发生了什么」：任务终态、ZIP 内条目、账本守恒 （delivered / refunded）、暂停/恢复/取消后的余额。不引用任何内部函数名， 这样第二步重构时测试不必跟着改，才具备等价性验证能力。 - batch：正常完成 / 暂停→恢复 / 暂停→取消 / 全失败退费 / 循环中途取消 / 发布前一刻取消 - podcast：正常完成 / 暂停→恢复 / 全失败退费 - 脚手架一律用 reserve_and_open 真预扣（ledger_open 只记账不扣款， 会让「退款」凭空给余额加次数，守恒断言就失去意义） 写测试过程中暴露两处代码与注释不符，如实钉住而非粉饰： 1. **循环内取消并不会「交付已打包部分」**。zip_valid = True 位于 with zipfile 块**之后**，而 while 循环内的任何取消都以异常**穿过**该块 （zipfile.__exit__ 仍写出中央目录，但那行赋值不执行）→ zip_valid 恒为 False → delivered = ok_count if zip_valid else 0 恒为 0。源码注释所称的 「部分成功后取消照常交付（成果守恒）」只在发布前最后一个检查点可达， 已分别用两个用例把两条路径都钉住。 2. podcast 的 _wait_if_paused 与 batch 版本除 docstring 外完全一致。 第二步——抽取（只抽逐字相同、零语义差异的部分） - _taskcore._wait_if_paused(task, pause_ev, cancel_ev, done)：done 写成参数 而非捕获闭包，使函数脱离两个 worker 各自的局部状态 - _taskcore._task_work_dir(task_id)：任务专属目录约定（ZIP 与中间文件都落 tasks/<task_id>/，对根目录的缓存清扫免疫） - 两个 worker 各自的 16 行副本与 2 行 mkdir 删除，调用点改为显式传参 复杂度实测量（同口径，git HEAD 对比）： 两个 worker 合计 行数 642 → 604（-38） 决策点 152 → 140（-12） 新增共享实现 _wait_if_paused 27 行/7 决策点 · _task_work_dir 11 行/1 **净行数基本持平（总量 -38 +38）**——重复代码抽取的收益从来不是行数变少， 而是**唯一真源**：以后改暂停语义只需改一处，不会再出现「同型缺陷出现在两地」。 验证：pytest 184 passed（175 → 184，+9）· ruff 0 · pyright 0 errors/0 warnings
365 lines
19 KiB
Python
365 lines
19 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""歌单/专辑批量打包 worker。
|
||
"""
|
||
|
||
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 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 .api_service import (
|
||
api_service,
|
||
)
|
||
|
||
|
||
def _batch_zip_worker(task_id: str) -> None:
|
||
"""后台批量打包：窗口化下载 → 写 ZIP（音频免压缩），支持取消与断线暂停
|
||
|
||
- 断线暂停：设备心跳超时后，当前窗口下载完即释放 BATCH_ZIP_SEM 名额给下一位，
|
||
保存进度进入暂停等待；重新打开网站（心跳恢复）后自动回到队列继续下载
|
||
- 音频入包即删，缓解磁盘占用
|
||
"""
|
||
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)
|