Archived
snowy-yang
0569b50737
为什么：用 AST 级引用分析（而非文本 grep——同名符号会互相掩盖）确认这些 符号全项目零引用。文本 grep 查不出的原因是它们都有「活着的同名兄弟」： services/downloads.py（共 6 个，93 行） _parse_cookie_str / _global_music_u / _vip_name / _detect_vip / _admin_client / _user_client —— 活的是 core.media.parse_cookie_str、web/routers/music.py、account.py 的版本 随之失效的 import：USER_COOKIE_FILE / GLOBAL_MUSIC_U_CACHE / VIP_PROBE_LOCK / account_info / detect_vip_by_probe / QRLoginClient services/api_service.py APIResponse 类（29 行）—— web/response.py 才是活的响应封装 services/_shared.py _rate_ok —— 逐字复制 web.ratelimit.rate_ok core/utils.py safe_int / safe_console_print —— 活的是 web/request_utils.safe_int core/repo.py all_settings / get_user_by_name —— auth_system.get_user_by_name 才是活的 core/db.py / web/auth_state.py / web/request_utils.py execute_many / _safe_register_credits + secrets_rand / dedup_ids 结果：生产代码死代码候选 13 → 0（余下 118 个全是 pytest 测试函数，属正常）。
235 lines
9.5 KiB
Python
235 lines
9.5 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""单曲下载任务 worker：下载 → 打包 → 交付 → 结算。
|
||
|
||
账号探测（VIP/音质）与二维码登录客户端均已迁出（前者只剩 web.routers.account
|
||
的异步实现，后者只剩 web.routers.admin 自己的客户端），本模块只负责单曲任务。
|
||
"""
|
||
|
||
import os
|
||
import shutil
|
||
import time
|
||
import traceback
|
||
import zipfile
|
||
from pathlib import Path
|
||
|
||
import auth_system
|
||
from core.state import (
|
||
DOWNLOAD_TASKS,
|
||
DOWNLOAD_TASKS_LOCK,
|
||
)
|
||
from core.utils import format_lyric as _format_lyric
|
||
from music_downloader import (
|
||
DownloadCanceled,
|
||
DownloadException,
|
||
fetch_cover_bytes,
|
||
)
|
||
|
||
from ._shared import (
|
||
_build_download_base,
|
||
_record_log,
|
||
)
|
||
from ._taskcore import (
|
||
_mark_task_terminal,
|
||
)
|
||
from .api_service import (
|
||
api_service,
|
||
)
|
||
|
||
|
||
def _task_delivery_copy(task_id: str, src: Path, filename: str) -> str:
|
||
"""把共享缓存文件复制/硬链接为任务专属交付文件（任务成果与共享缓存分离，
|
||
任何任务的清理都不会再删掉别的任务的成果）"""
|
||
task_dir = api_service.downloads_path / 'tasks' / task_id
|
||
task_dir.mkdir(parents=True, exist_ok=True)
|
||
dst = task_dir / filename
|
||
if dst.exists():
|
||
return str(dst)
|
||
try:
|
||
os.link(str(src), str(dst)) # 同卷硬链接：零拷贝
|
||
except OSError:
|
||
shutil.copy2(str(src), str(dst))
|
||
return str(dst)
|
||
|
||
|
||
# _cleanup_task_dir 死代码删除——worker 直接 rmtree、监控线程直接
|
||
# rmtree，本函数全文件零调用（审计确认）。
|
||
|
||
|
||
def _single_download_worker(task_id: str) -> None:
|
||
"""后台单曲下载：下载音频（可选 ZIP 打包），支持取消；完成后由 /download/result 交付，35 分钟保留期清理"""
|
||
import tempfile as _tempfile
|
||
|
||
file_path_str: str | None = None
|
||
zip_tmp_path: str | None = None
|
||
task_dir_path: str | None = None
|
||
try:
|
||
with DOWNLOAD_TASKS_LOCK:
|
||
task = DOWNLOAD_TASKS[task_id]
|
||
task['status'] = 'running'
|
||
task['msg'] = '下载中...'
|
||
params = task['params']
|
||
cancel_ev = task['cancel']
|
||
music_id = params['music_id']
|
||
quality = params['quality']
|
||
filename_format = params['filename_format']
|
||
write_meta = params['write_meta']
|
||
meta_fields = params['meta_fields']
|
||
lyric_format = params['lyric_format']
|
||
need_zip = params['need_zip']
|
||
cookies = params['cookies']
|
||
remote_ip = params.get('remote_ip') or '-'
|
||
|
||
# 一次预取歌曲完整信息（URL/详情/歌词三接口并行），
|
||
# 替代原先 name_v1 + url_v1 + 下载器内部 get_music_info 的重复串行调用
|
||
mi = api_service.downloader.get_music_info(music_id, quality, cookies)
|
||
music_info = {
|
||
'id': music_id,
|
||
'name': mi.name,
|
||
'artist_string': mi.artists,
|
||
'album': mi.album,
|
||
'pic_url': mi.pic_url,
|
||
'file_type': mi.file_type,
|
||
'file_size': mi.file_size,
|
||
'duration': mi.duration * 1000,
|
||
'download_url': mi.download_url,
|
||
}
|
||
base = _build_download_base(filename_format, music_info)
|
||
filename = f"{base}.{music_info['file_type']}"
|
||
|
||
# 字节级进度回调（限频 0.4s）：写进任务 live_percent，/api/batch_progress
|
||
# 对运行中的单曲任务优先返回它——原实现 done 按首计数，单曲进度条全程 0%
|
||
_cb_ts = {'t': 0.0}
|
||
|
||
def _progress_cb(written, total_bytes):
|
||
try:
|
||
now = time.time()
|
||
if now - _cb_ts['t'] < 0.4 and written < (total_bytes or 0):
|
||
return
|
||
_cb_ts['t'] = now
|
||
with DOWNLOAD_TASKS_LOCK:
|
||
t = DOWNLOAD_TASKS.get(task_id)
|
||
if not t or t.get('status') != 'running':
|
||
return
|
||
if total_bytes:
|
||
pct = min(99.0, written / total_bytes * 100)
|
||
t['live_percent'] = pct
|
||
t['msg'] = (f"下载中 {int(pct)}% · "
|
||
f"{api_service._format_file_size(written)}/{api_service._format_file_size(total_bytes)}")
|
||
else:
|
||
t['msg'] = f"下载中 · 已接收 {api_service._format_file_size(written)}"
|
||
except Exception:
|
||
pass # 进度上报故障绝不影响下载
|
||
|
||
download_result = api_service.downloader.download_music_file(
|
||
music_id, quality,
|
||
write_meta=write_meta,
|
||
meta_fields=meta_fields,
|
||
cookies=cookies,
|
||
cancel_event=cancel_ev,
|
||
music_info=mi,
|
||
progress_cb=_progress_cb,
|
||
)
|
||
if not download_result.success:
|
||
raise DownloadException(download_result.error_message or "下载失败")
|
||
# 下载完成后的每一阶段都可能已被取消，发布前统一校验
|
||
if cancel_ev.is_set():
|
||
raise DownloadCanceled("任务已取消")
|
||
assert download_result.file_path is not None
|
||
cache_path = Path(download_result.file_path)
|
||
# 交付文件用任务专属副本，共享缓存不进清理清单
|
||
file_path_str = _task_delivery_copy(task_id, cache_path, filename)
|
||
task_dir_path = str(Path(file_path_str).parent)
|
||
|
||
if need_zip:
|
||
fd, zip_tmp_path = _tempfile.mkstemp(prefix='single_zip_', suffix='.zip',
|
||
dir=str(Path(file_path_str).parent))
|
||
os.close(fd)
|
||
with zipfile.ZipFile(zip_tmp_path, 'w', zipfile.ZIP_DEFLATED) as zf:
|
||
zf.write(file_path_str, arcname=filename)
|
||
# 歌词已在预取时一并拿到，无需再请求一次
|
||
if mi.lyric:
|
||
try:
|
||
zf.writestr(f"{base}.{lyric_format}", _format_lyric(mi.lyric, lyric_format))
|
||
except Exception:
|
||
pass
|
||
# 封面走共享缓存（与元数据写入共用同一次网络请求）
|
||
cover = fetch_cover_bytes(mi.pic_url)
|
||
if cover:
|
||
zf.writestr(f"{base}.jpg", cover)
|
||
info_txt = (
|
||
f"歌曲: {music_info['name']}\n"
|
||
f"歌手: {music_info['artist_string']}\n"
|
||
f"专辑: {music_info['album']}\n"
|
||
f"音质: {api_service._get_quality_display_name(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/{music_info['file_type']}",
|
||
'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, music_info['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_path:
|
||
shutil.rmtree(task_dir_path, 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_path:
|
||
shutil.rmtree(task_dir_path, ignore_errors=True)
|
||
api_service.logger.error(f"单曲下载失败: {task_id} {e}")
|
||
except Exception as e:
|
||
# 非预期异常的 repr 可能含上游主机/连接细节——用户可见文案
|
||
# 泛化为固定文案，原文 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_path:
|
||
shutil.rmtree(task_dir_path, ignore_errors=True)
|