zijunchiang
87aa979abd
版本号：
- 1.5.11 越界（规则 3 规定 patch ≤ 10）→ 修正为 1.6.0（进位到 minor、patch 归 0）
- versionCode 43 不变：43 从未对外发布，本次修正只是把命名拉回约定
合并支线（原在工作区根目录、App 仓库之外，DESIGN.md 自称"支线"）：
- selfhost-source/server/ 自建歌曲源服务器（纯 Python 标准库、零 pip 依赖）
- selfhost-source/deploy/、DESIGN.md、LIBRARY.md、build_inventory.py
- selfhost-source/inventory/ 小清单（playlists.csv、song-stats.md 等）
- ⚠️ inventory 生成物（songs-*.csv / singers-top.csv，共 84MB）不入库，
已写入 .gitignore；需要时用 build_inventory.py 依据 muse.db 重新生成
同时入库的 App 侧支线改动（此前为工作树未提交状态）：
- Whoami.kt（whoami 设备心跳）、DeviceIdentity.kt（设备身份伪装）
- PushDownload.kt / PushDownloadClient.kt（点歌推送）
- AdVideoFilter 带 query/fragment 的地址闸门；KTV_BRIDGE_API 11 → 13
- 修 ktv_api.js 的 global.KtvBridgeApiVersion（12 → 13，与注释及 Kotlin 两处校验一致）
- 对应单测：WhoamiTest / DeviceIdentityTest / PushDownloadTest / AdVideoFilterTest（共 39 例）
文档：
- DESIGN.md 的"在 App 仓库之外"约定改为"已并入仓库统一版本化"
- AGENTS.md 版本号行更新为 1.6.0/43，并新增支线并入说明
验证：assembleRelease + testReleaseUnitTest 全过（39 例）
656 lines
30 KiB
Python
656 lines
30 KiB
Python
"""下载管理：**并发上限 10** 的工作池 + 取址（App 方式）+ 断点续传 + 下载登记（避免重复下载）。
|
||
|
||
去重的三道保证：
|
||
1. `downloads.tid` 是主键 —— 一首歌只有一行；
|
||
2. `Store.claim_one()` 用一条原子 UPDATE 抢占（`rowcount==1` 才有权下载），
|
||
所以 10 个线程也不会同时下同一首；
|
||
3. 落盘用 `<tid>.part` 写、完成后 `os.replace` 成 `<tid>.ts`；
|
||
完成后写 `state='done'` + 字节数 + md5 → 再次入队时 `enqueue(skip_done=True)` 直接跳过。
|
||
|
||
歌曲表（keeps）同步：下载成功/失败/抢占都会把歌曲表状态改成 `done/failed/downloading`
|
||
（`_note_done` / `_note_failed` / `_note_downloading`，一律 `getattr` 防呆）——
|
||
文件与歌曲表保持一致，服务器只按歌曲表下载。
|
||
|
||
**固定 URL（App 点歌推送）**：队列项带 `downloads.push_url` 时**跳过取址**，直接用该地址下载
|
||
（省一次上游请求，也避开取址限流）；该地址失败（过期/403/非 TS/过小）→ **回退到正常取址重试一次**
|
||
并把回退写进事件。它**不省任何校验**：地址闸门与 TS/大小校验与取址路径完全同一套，见 `_download()`。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import os
|
||
import shutil
|
||
import threading
|
||
import time
|
||
import urllib.error
|
||
import urllib.request
|
||
from typing import Callable
|
||
|
||
from . import resolver as R
|
||
|
||
|
||
class RateLimiter:
|
||
"""全局令牌桶：限制**媒体下载的聚合速率**（MB/s）。
|
||
|
||
为什么必须限速：媒体目录在云盘/FUSE 上时，写入先落到本地缓冲、再由云盘客户端异步上传。
|
||
如果下载速度长期高于上传速度，本地盘迟早被缓冲撑满 → 任务失败甚至影响整机。
|
||
"""
|
||
|
||
def __init__(self, mb_per_sec: float = 0.0):
|
||
self._lock = threading.Lock()
|
||
self.rate = 0.0
|
||
self._next = time.time()
|
||
self.set_rate(mb_per_sec)
|
||
|
||
def set_rate(self, mb_per_sec: float) -> None:
|
||
with self._lock:
|
||
self.rate = max(0.0, float(mb_per_sec or 0))
|
||
self._next = time.time()
|
||
|
||
def acquire(self, nbytes: int) -> None:
|
||
"""按字节数预留带宽；必要时睡眠（虚拟调度，长期平均速率≈rate）。"""
|
||
if nbytes <= 0:
|
||
return
|
||
with self._lock:
|
||
if self.rate <= 0:
|
||
return
|
||
now = time.time()
|
||
need = nbytes / (self.rate * 1048576.0)
|
||
start = max(self._next, now)
|
||
self._next = start + need
|
||
delay = start - now
|
||
if delay > 0:
|
||
time.sleep(delay)
|
||
|
||
|
||
class FailedRetry:
|
||
"""下载队列**失败任务自动重试**：每 `retry_failed_interval_sec` 秒把 failed/missing 重置为待下载。
|
||
|
||
参考 `StorageGuard` 的写法：独立**守护线程** + 事件日志 + 任何异常只记日志。
|
||
⚠️ **必须在端口监听之后 `start()`**（见 main.py）：这台机器的部署铁律是"任何可能出问题的
|
||
启动动作都不要挡在监听之前"（历史上两次 FUSE 阻塞导致"进程活着但不监听端口"）。
|
||
不过本线程只做数据库小事务、不碰媒体目录，风险很低。
|
||
|
||
跳过条件（避免瞎折腾 / 白跑）：
|
||
* `retry_failed_enabled=False` → 完全不跑；
|
||
* 下载池被**人工暂停**（`paused_by_user`）→ 不重试（用户就是要它停着）；
|
||
* **存储守卫**判定水位不足 → 不重试（重试了也下不动，还会白占空间）；
|
||
* 重置数量 `N == 0` → **不写日志**（否则每 10 分钟刷一条无意义日志）；
|
||
* 下载池已暂停（非人工，例如空间原因）→ 不重试。
|
||
手工验证用 `POST /api/download/retry_failed_now`（可选 `{"force": true}` 绕过跳过条件）。
|
||
"""
|
||
|
||
def __init__(self, cfg, store, downloader=None, guard=None):
|
||
self.cfg = cfg
|
||
self.store = store
|
||
self.downloader = downloader
|
||
self.guard = guard
|
||
self._stop = threading.Event()
|
||
self._thread: threading.Thread | None = None
|
||
self._run_lock = threading.Lock() # 防重入：定时线程与手工接口可能同时触发
|
||
self.state = {"enabled": bool(cfg.get("retry_failed_enabled", True)),
|
||
"interval_sec": self._interval(),
|
||
"last_at": 0, "last_count": 0, "total_count": 0,
|
||
"skipped_reason": "", "running": False}
|
||
|
||
# ── 配置（每次都现读，改设置后下一轮即生效）──
|
||
def _interval(self) -> int:
|
||
try:
|
||
return max(0, int(self.cfg.get("retry_failed_interval_sec") or 0))
|
||
except Exception:
|
||
return 600
|
||
|
||
def _enabled(self) -> bool:
|
||
return bool(self.cfg.get("retry_failed_enabled", True))
|
||
|
||
def _max_attempts(self) -> int:
|
||
"""自动重试用的重试上限：配置为 0 时跟随 `max_attempts`（见 Config.retry_failed_max_attempts）。"""
|
||
fn = getattr(self.cfg, "retry_failed_max_attempts", None)
|
||
if callable(fn):
|
||
try:
|
||
return int(fn())
|
||
except Exception:
|
||
pass
|
||
try:
|
||
return max(1, int(self.cfg.get("max_attempts") or 3))
|
||
except Exception:
|
||
return 3
|
||
|
||
# ── 跳过条件 ──
|
||
def skip_reason(self) -> str:
|
||
"""返回本次不该重试的原因（空串 = 可以重试）。"""
|
||
if not self._enabled():
|
||
return "自动重试已关闭（retry_failed_enabled=false）"
|
||
if self.downloader is not None and getattr(self.downloader, "paused_by_user", False):
|
||
return "下载池已被人工暂停"
|
||
st = getattr(self.guard, "state", None) if self.guard is not None else None
|
||
if isinstance(st, dict) and not st.get("ok", True):
|
||
return f"存储水位不足：{st.get('reason') or '空间不足'}"
|
||
if self.downloader is not None and getattr(self.downloader, "paused", False):
|
||
by = getattr(self.downloader, "_paused_by", "") or "未知"
|
||
return f"下载池已暂停（{by}）"
|
||
return ""
|
||
|
||
# ── 执行一次 ──
|
||
def run_once(self, force: bool = False) -> dict:
|
||
"""把 failed/missing 重置为待下载一次；返回结果（含 reset / skipped_reason / auto_retry）。"""
|
||
if not self._run_lock.acquire(blocking=False):
|
||
return {"ok": True, "reset": 0, "skipped": True,
|
||
"skipped_reason": "上一次重试还在进行中", "auto_retry": self.status()}
|
||
try:
|
||
if not force:
|
||
reason = self.skip_reason()
|
||
if reason:
|
||
self.state["skipped_reason"] = reason
|
||
return {"ok": True, "reset": 0, "skipped": True, "skipped_reason": reason,
|
||
"auto_retry": self.status()}
|
||
max_attempts = self._max_attempts()
|
||
n = int(self.store.retry_failed(max_attempts) or 0)
|
||
self.state.update({"last_at": int(time.time()), "last_count": n,
|
||
"total_count": int(self.state.get("total_count", 0)) + n,
|
||
"skipped_reason": "", "running": False})
|
||
if n > 0: # N==0 不写日志（避免刷屏）
|
||
self.store.event("INFO", f"下载队列自动重试：重置 {n} 首失败任务"
|
||
f"（间隔 {self._interval()}s）")
|
||
return {"ok": True, "reset": n, "skipped": False, "max_attempts": max_attempts,
|
||
"auto_retry": self.status()}
|
||
finally:
|
||
self._run_lock.release()
|
||
|
||
def loop(self) -> None:
|
||
while not self._stop.is_set():
|
||
interval = self._interval()
|
||
if interval <= 0 or not self._enabled():
|
||
return # 关掉定时重试（手工接口仍可用）
|
||
if self._stop.wait(interval):
|
||
return
|
||
try:
|
||
self.run_once()
|
||
except Exception as exc: # pragma: no cover - 兜底
|
||
self.store.event("WARN", f"下载队列自动重试失败：{exc}")
|
||
|
||
def start(self) -> None:
|
||
if self._thread is not None:
|
||
return
|
||
self._stop.clear()
|
||
self._thread = threading.Thread(target=self.loop, name="retry-failed", daemon=True)
|
||
self._thread.start()
|
||
|
||
def stop(self) -> None:
|
||
self._stop.set()
|
||
|
||
def status(self) -> dict:
|
||
"""给管理界面：开关 / 间隔 / 上次执行时间与数量 / 累计 / 上次跳过原因。"""
|
||
st = dict(self.state)
|
||
st.update({"enabled": self._enabled(), "interval_sec": self._interval(),
|
||
"max_attempts": self._max_attempts(),
|
||
"running": bool(self._thread is not None and self._thread.is_alive())})
|
||
return st
|
||
|
||
|
||
class StorageGuard:
|
||
"""存储水位守卫：周期检查**本地盘**与**媒体所在文件系统**的剩余空间。
|
||
|
||
- 低于阈值 → 暂停下载池（标记 by='space'，不会覆盖用户手动暂停）；
|
||
- 回到「阈值 + 余量」→ 自动恢复（迟滞，避免在边界反复启停）。
|
||
"""
|
||
|
||
def __init__(self, cfg, store, downloader=None):
|
||
self.cfg = cfg
|
||
self.store = store
|
||
self.downloader = downloader
|
||
self._stop = threading.Event()
|
||
self.state = {"ok": True, "local_free": 0, "media_free": 0, "checked_at": 0,
|
||
"local_path": str(cfg.get("local_space_path")), "media_path": cfg.media_dir,
|
||
"reason": "尚未检查"}
|
||
|
||
def sample(self) -> dict:
|
||
local_path = str(self.cfg.get("local_space_path") or "/")
|
||
try:
|
||
local_free = shutil.disk_usage(local_path).free
|
||
except Exception:
|
||
local_free = -1
|
||
try:
|
||
media_free = shutil.disk_usage(self.cfg.media_dir).free
|
||
except Exception:
|
||
media_free = -1
|
||
min_local = int(self.cfg.get("min_free_local_bytes") or 0)
|
||
min_media = int(self.cfg.get("min_free_media_bytes") or 0)
|
||
margin = int(self.cfg.get("resume_free_margin_bytes") or 0)
|
||
low = []
|
||
if 0 <= local_free < min_local:
|
||
low.append(f"本地盘剩余 {local_free / 1e9:.1f}GB < {min_local / 1e9:.1f}GB")
|
||
if 0 <= media_free < min_media:
|
||
low.append(f"云盘剩余 {media_free / 1e9:.1f}GB < {min_media / 1e9:.1f}GB")
|
||
ok = not low
|
||
recover = (local_free < 0 or local_free >= min_local + margin) and \
|
||
(media_free < 0 or media_free >= min_media + margin)
|
||
return {"ok": ok, "recover": recover, "local_free": local_free, "media_free": media_free,
|
||
"local_path": local_path, "media_path": self.cfg.media_dir,
|
||
"checked_at": int(time.time()), "reason": "；".join(low) or "空间充足"}
|
||
|
||
def loop(self) -> None:
|
||
interval = max(5, int(self.cfg.get("storage_check_interval_sec") or 15))
|
||
first = True
|
||
while not self._stop.wait(0 if first else interval):
|
||
first = False
|
||
try:
|
||
st = self.sample()
|
||
self.state = st
|
||
if self.downloader is None:
|
||
continue
|
||
if not st["ok"] and bool(self.cfg.get("pause_on_low_space")):
|
||
if not self.downloader.paused_by_space:
|
||
self.downloader.pause(by="space")
|
||
self.store.event("WARN", f"空间不足，已暂停下载：{st['reason']}")
|
||
elif st["recover"] and self.downloader.paused_by_space:
|
||
self.downloader.resume(by="space")
|
||
self.store.event("INFO", "空间已恢复，继续下载")
|
||
except Exception as exc: # pragma: no cover
|
||
self.store.event("WARN", f"空间检查异常：{exc}")
|
||
|
||
def start(self) -> None:
|
||
threading.Thread(target=self.loop, name="storage-guard", daemon=True).start()
|
||
|
||
def stop(self) -> None:
|
||
self._stop.set()
|
||
|
||
|
||
class Downloader:
|
||
def __init__(self, cfg, store, on_event: Callable[[str, str], None] | None = None):
|
||
self.cfg = cfg
|
||
self.store = store
|
||
self.resolver = R.Resolver(cfg, store)
|
||
self.on_event = on_event
|
||
self._threads: list[threading.Thread] = []
|
||
self._stop = threading.Event()
|
||
self._pause = threading.Event()
|
||
self._active: dict[int, dict] = {}
|
||
self._lock = threading.Lock()
|
||
self._last_error = ""
|
||
self._started_at = 0
|
||
self._paused_by = ""
|
||
self.limiter = RateLimiter(0)
|
||
self.guard: "StorageGuard | None" = None
|
||
# 失败任务自动重试调度器（main.py 在端口监听之后装配并 start；管理接口通过它触发手工重试）
|
||
self.retry: "FailedRetry | None" = None
|
||
|
||
# ── 生命周期 ──
|
||
@property
|
||
def concurrency(self) -> int:
|
||
try:
|
||
n = int(self.cfg.get("concurrency"))
|
||
except Exception:
|
||
n = 10
|
||
return max(1, min(10, n)) # 硬上限 10（需求）
|
||
|
||
def start(self) -> None:
|
||
if self._threads:
|
||
return
|
||
self._stop.clear()
|
||
self.store.set_meta("max_attempts", str(self.cfg.get("max_attempts")))
|
||
for i in range(self.concurrency):
|
||
t = threading.Thread(target=self._worker, name=f"dl-{i}", daemon=True)
|
||
t.start()
|
||
self._threads.append(t)
|
||
self._started_at = int(time.time())
|
||
self.apply_limits()
|
||
self.store.event("INFO", f"下载池已启动，并发 {self.concurrency}（上限 10）"
|
||
f"，媒体限速 {self.limiter.rate or '不限'} MB/s")
|
||
|
||
def stop(self, wait: bool = True) -> None:
|
||
self._stop.set()
|
||
if wait:
|
||
for t in self._threads:
|
||
t.join(timeout=5)
|
||
self._threads.clear()
|
||
|
||
def pause(self, by: str = "user") -> None:
|
||
self._pause.set()
|
||
self._paused_by = by
|
||
if by == "user":
|
||
self.store.event("INFO", "下载已暂停（活动任务会跑完当前分片）")
|
||
|
||
def resume(self, by: str = "user") -> None:
|
||
# 双向保护：手动暂停不被"空间恢复"解除；空间保护暂停也不被手动"继续"解除
|
||
# （否则用户点一下继续，15 秒后又被守卫暂停，来回抖动；修空间或调低阈值后守卫会自动恢复）
|
||
if self._paused_by and self._paused_by != by:
|
||
self.store.event("INFO", f"忽略来自 {by} 的恢复请求：当前暂停原因是 {self._paused_by}")
|
||
return
|
||
self._pause.clear()
|
||
self._paused_by = ""
|
||
if by == "user":
|
||
self.store.event("INFO", "下载已继续")
|
||
|
||
@property
|
||
def paused(self) -> bool:
|
||
return self._pause.is_set()
|
||
|
||
@property
|
||
def paused_by_space(self) -> bool:
|
||
return self._paused_by == "space"
|
||
|
||
@property
|
||
def paused_by_user(self) -> bool:
|
||
"""是否由**用户手动**暂停（自动重试调度器据此跳过重试）。"""
|
||
return self._paused_by == "user"
|
||
|
||
def apply_limits(self) -> None:
|
||
"""把配置里的媒体限速应用到令牌桶（改设置后调用即可生效）。"""
|
||
try:
|
||
self.limiter.set_rate(float(self.cfg.get("media_rate_limit_mb_s") or 0))
|
||
except Exception:
|
||
self.limiter.set_rate(0)
|
||
|
||
def rebuild(self) -> None:
|
||
"""并发数改了之后重建线程池。"""
|
||
was_running = bool(self._threads)
|
||
self.stop()
|
||
if was_running:
|
||
self.start()
|
||
|
||
# ── 状态（给管理界面） ──
|
||
def health(self) -> dict:
|
||
"""给 `/healthz` 的**纯内存**快照：不查库、不碰媒体目录、绝不阻塞。
|
||
|
||
探活要求"必须快速返回"（见 ktv/health.py）：这里只读几个内存字段，
|
||
刻意不调 `status()`（那个会查 pending/failed 统计，库大时是几秒的全表扫描）。
|
||
"""
|
||
try:
|
||
with self._lock:
|
||
active = len(self._active)
|
||
return {"running": bool(self._threads), "paused": self.paused,
|
||
"paused_by": self._paused_by, "concurrency": self.concurrency,
|
||
"active": active, "rate_limit_mb_s": self.limiter.rate}
|
||
except Exception: # pragma: no cover - 兜底
|
||
return {}
|
||
|
||
def status(self) -> dict:
|
||
with self._lock:
|
||
active = [dict(v) for v in self._active.values()]
|
||
return {"running": bool(self._threads), "paused": self.paused,
|
||
"paused_by": self._paused_by, "paused_by_space": self.paused_by_space,
|
||
"rate_limit_mb_s": self.limiter.rate,
|
||
"storage": (self.guard.state if self.guard else {}),
|
||
"concurrency": self.concurrency,
|
||
"active": active, "active_count": len(active),
|
||
"last_error": self._last_error, "started_at": self._started_at,
|
||
"pending": self.store.stats()["pending"], "failed": self.store.stats()["failed"]}
|
||
|
||
# ── 取任务 ──
|
||
def _fetch_one(self) -> dict | None:
|
||
order = str(self.cfg.get("download_order"))
|
||
desc = bool(self.cfg.get("download_order_desc"))
|
||
max_attempts = int(self.cfg.get("max_attempts") or 3)
|
||
backoff = int(self.cfg.get("retry_backoff_sec") or 20)
|
||
for cand in self.store.next_pending(20, order, desc, max_attempts, backoff):
|
||
if self.store.claim_one(cand["tid"], max_attempts, backoff):
|
||
return cand
|
||
return None
|
||
|
||
# ── 工作线程 ──
|
||
def _worker(self) -> None:
|
||
while not self._stop.is_set():
|
||
if self._pause.is_set():
|
||
time.sleep(1.0)
|
||
continue
|
||
if self.guard is not None and not self.guard.state.get("ok", True):
|
||
time.sleep(1.0) # 空间不足：不领新任务（守卫会负责暂停/恢复）
|
||
continue
|
||
item = self._fetch_one()
|
||
if not item:
|
||
time.sleep(1.0)
|
||
continue
|
||
tid = int(item["tid"])
|
||
with self._lock:
|
||
self._active[tid] = {"tid": tid, "name": item["name"], "bytes": 0, "total": 0,
|
||
"started": int(time.time())}
|
||
self._note_downloading(tid) # 歌曲表置「下载中」
|
||
try:
|
||
self._download(item)
|
||
except Exception as exc: # pragma: no cover - 兜底
|
||
self._last_error = f"{tid}: {exc}"
|
||
self.store.finish(tid, False, error=str(exc))
|
||
self._note_failed(tid, str(exc))
|
||
finally:
|
||
with self._lock:
|
||
self._active.pop(tid, None)
|
||
time.sleep(0.05)
|
||
|
||
def _media_path(self, tid: int) -> str:
|
||
return os.path.join(str(self.cfg.get("media_dir")), f"{tid}.ts")
|
||
|
||
def _tmp_path(self, tid: int) -> str:
|
||
"""分片写到**本地盘**（tmp_dir），完成后再原子改名进媒体目录（云盘/FUSE）。
|
||
|
||
这样 .part 不会被云盘上传，且避免在 FUSE 上做随机写（2026-10-10 修正）。
|
||
"""
|
||
base = str(self.cfg.get("tmp_dir") or os.path.join(str(self.cfg.get("media_dir")), "tmp"))
|
||
return os.path.join(base, f"{tid}.ts.part")
|
||
|
||
def _download(self, item: dict) -> None:
|
||
tid = int(item["tid"])
|
||
final = self._media_path(tid)
|
||
tmp = self._tmp_path(tid)
|
||
min_bytes = int(self.cfg.get("min_file_bytes"))
|
||
timeout = int(self.cfg.get("timeout_sec"))
|
||
|
||
# 已经登记为 done 且文件完好 → 直接结束（避免重复下载）
|
||
row = self.store.q1("SELECT state, bytes FROM downloads WHERE tid=?", (tid,))
|
||
if row and row["state"] == "done" and os.path.isfile(final) and os.path.getsize(final) >= min_bytes:
|
||
size = os.path.getsize(final)
|
||
self.store.finish(tid, True, bytes_=size, path=final)
|
||
self._note_done(tid, size, final)
|
||
return
|
||
|
||
# ── 地址来源：App 点歌推送的**固定地址**优先，其余走正常取址 ──
|
||
# ⚠️ 固定地址**不省任何校验**：入队前闸门已判过一次，这里下载前再判一次（防止库里的行被改动）；
|
||
# 下载完成后仍然走与取址路径**完全相同**的大小/TS 校验（下面逐条判定）。
|
||
push_url = str(item.get("push_url") or "").strip()
|
||
plan: list[str | None] = []
|
||
if push_url:
|
||
ok, why = R.is_trusted_url(push_url, tid)
|
||
if ok:
|
||
plan.append(push_url)
|
||
else:
|
||
# 闸门已经判过：库里这条固定地址不可信（可能被手工改过）→ 直接丢掉，走正常取址
|
||
self.store.push_url_clear(tid)
|
||
self.store.event("WARN", f"App 推送地址未过闸门 编号={tid} 原因={why}，改为正常取址")
|
||
plan.append(None) # None = 正常取址（也是固定地址的回退目标）
|
||
|
||
last_reason = ""
|
||
for index, fixed in enumerate(plan):
|
||
if fixed is None:
|
||
url, host, detail = self.resolver.resolve(tid)
|
||
if not url:
|
||
self.store.finish(tid, False, error=f"取址失败 {detail}")
|
||
self._note_failed(tid, f"取址失败 {detail}")
|
||
return
|
||
if index > 0:
|
||
# 固定地址失败后的回退（事件用于发现"App 推送的地址是否普遍失效"）
|
||
self.store.event("WARN", f"App 推送地址回退取址 编号={tid} 源={host} 原因={last_reason[:80]}")
|
||
else:
|
||
url, host = fixed, "app-push"
|
||
self.store.event("INFO", f"开始下载 编号={tid} 源={host} 地址={(url.split('?')[0])[-60:]}")
|
||
expected = self._head_size(url, timeout)
|
||
reason = ""
|
||
try:
|
||
got = self._fetch(url, tmp, timeout, tid, expected)
|
||
except Exception as exc: # 403/过期/断流：固定地址也要能回退，不能直接抛
|
||
got, reason = 0, f"下载异常 {type(exc).__name__}: {exc}"
|
||
|
||
if not reason:
|
||
if got < min_bytes:
|
||
reason = f"文件过小 {got}B（阈值 {min_bytes}B）"
|
||
else:
|
||
ts_ok, ts_offset = self._looks_like_ts(tmp)
|
||
if not ts_ok:
|
||
reason = f"不是 TS 流（188 字节包同步未找到，文件头 {self._head_hex(tmp)}）"
|
||
elif expected and got < expected:
|
||
reason = f"下载不完整 {got}/{expected}B"
|
||
elif ts_offset:
|
||
# ⚠️ 上游 CDN（download.origjoy.com）的流是「512 字节私有头 + TS」，包同步不在偏移 0。
|
||
# 这正是 App 能播、我们一开始判成"坏文件"的原因 —— 内容照原样保留，只把校验改成扫描式。
|
||
self.store.event("INFO", f"编号={tid} 媒体含 {ts_offset} 字节私有头"
|
||
f"（TS 从该偏移开始），按原样落盘")
|
||
|
||
if reason:
|
||
if fixed is not None and len(plan) > 1:
|
||
# 固定地址失败 → 清掉它（免得以后重试又先撞一次死地址）+ 删掉可能已写了一半的分片，
|
||
# 然后回退到正常取址重试一次（同一套闸门/TS 校验，不绕过任何一道）。
|
||
self.store.push_url_clear(tid)
|
||
self._remove_tmp(tmp)
|
||
self.store.event("WARN", f"App 推送地址下载失败 编号={tid} 原因={reason[:120]}，"
|
||
f"回退正常取址重试一次")
|
||
last_reason = reason
|
||
continue
|
||
self._fail(tid, reason, url, host)
|
||
return
|
||
|
||
digest = ""
|
||
if bool(self.cfg.get("verify_md5")):
|
||
digest = self._md5(tmp)
|
||
os.makedirs(os.path.dirname(final), exist_ok=True)
|
||
try:
|
||
os.replace(tmp, final) # 同一文件系统：原子改名
|
||
except OSError:
|
||
shutil.copyfile(tmp, final) # 跨文件系统（本地盘 → 云盘）：复制后再删
|
||
os.remove(tmp)
|
||
self.store.finish(tid, True, bytes_=got, path=final, md5=digest, url=url, source_host=host)
|
||
self._note_done(tid, got, final)
|
||
self.store.event("INFO", f"下载完成 编号={tid} {got // 1048576}MB 源={host}")
|
||
return
|
||
|
||
@staticmethod
|
||
def _remove_tmp(path: str) -> None:
|
||
"""删掉可能残留的分片（换地址重试前必须删：断点续传的 Range 不能跨地址复用）。"""
|
||
try:
|
||
if os.path.isfile(path):
|
||
os.remove(path)
|
||
except Exception:
|
||
pass
|
||
|
||
def _note_done(self, tid: int, bytes_: int = 0, path: str = "") -> None:
|
||
"""下载成功 → 歌曲表（keeps）该条目置 `done` + 字节数/路径。
|
||
|
||
**文件与歌曲表保持一致**：文件落盘后必须同步歌曲表状态，否则媒体扫描会判定
|
||
"done 但文件缺失"而把它放回待下载（白下一遍）。
|
||
用 `getattr` 防呆：测试里的假 store / 老库可能没有该方法，绝不能因此让下载流程失败。
|
||
"""
|
||
self._keep("keep_mark_done", int(tid), int(bytes_ or 0), str(path or ""))
|
||
|
||
def _note_failed(self, tid: int, error: str = "") -> None:
|
||
"""下载失败 → 歌曲表该条目置 `failed` + 原因（同样 `getattr` 防呆）。"""
|
||
self._keep("keep_mark_failed", int(tid), str(error))
|
||
|
||
def _note_downloading(self, tid: int) -> None:
|
||
"""抢占到任务 → 歌曲表该条目置 `downloading`（attempts+1），便于界面看"在下的"。"""
|
||
self._keep("keep_mark_downloading", int(tid))
|
||
|
||
def _keep(self, method: str, *args) -> None:
|
||
"""调用歌曲表状态方法：方法不存在或抛异常都只记日志，不影响下载主流程。"""
|
||
fn = getattr(self.store, method, None)
|
||
if not callable(fn):
|
||
return
|
||
try:
|
||
fn(*args)
|
||
except Exception as exc: # pragma: no cover - 兜底
|
||
try:
|
||
self.store.event("WARN", f"歌曲表状态更新失败（{method}）：{exc}")
|
||
except Exception:
|
||
pass
|
||
|
||
# ── HTTP 细节 ──
|
||
def _head_size(self, url: str, timeout: int) -> int:
|
||
try:
|
||
req = urllib.request.Request(url, method="HEAD",
|
||
headers={"User-Agent": str(self.cfg.get("user_agent"))})
|
||
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
||
return int(resp.headers.get("Content-Length") or 0)
|
||
except Exception:
|
||
return 0
|
||
|
||
def _fetch(self, url: str, tmp: str, timeout: int, tid: int, expected: int) -> int:
|
||
"""带断点续传的下载；返回落盘字节数。"""
|
||
os.makedirs(os.path.dirname(tmp), exist_ok=True)
|
||
have = os.path.getsize(tmp) if os.path.isfile(tmp) else 0
|
||
headers = {"User-Agent": str(self.cfg.get("user_agent")), "Accept": "*/*"}
|
||
if have > 0:
|
||
headers["Range"] = f"bytes={have}-"
|
||
req = urllib.request.Request(url, headers=headers)
|
||
try:
|
||
resp = urllib.request.urlopen(req, timeout=timeout)
|
||
except urllib.error.HTTPError as exc:
|
||
if exc.code == 416 and have > 0: # 已经下完
|
||
return have
|
||
raise
|
||
mode = "ab" if (have > 0 and resp.status == 206) else "wb"
|
||
if mode == "wb":
|
||
have = 0
|
||
written = have
|
||
with resp, open(tmp, mode) as fh:
|
||
last_report = 0
|
||
while True:
|
||
chunk = resp.read(1 << 18)
|
||
if not chunk:
|
||
break
|
||
self.limiter.acquire(len(chunk)) # 聚合限速：不超过云盘上传速度
|
||
fh.write(chunk)
|
||
written += len(chunk)
|
||
now_ms = time.time()
|
||
if now_ms - last_report > 1.0:
|
||
last_report = now_ms
|
||
with self._lock:
|
||
if tid in self._active:
|
||
self._active[tid]["bytes"] = written
|
||
self._active[tid]["total"] = expected
|
||
if self._stop.is_set():
|
||
break
|
||
return written
|
||
|
||
def _fail(self, tid: int, reason: str, url: str = "", host: str = "") -> None:
|
||
attempts = self.store.q1("SELECT attempts FROM downloads WHERE tid=?", (tid,))
|
||
n = attempts["attempts"] if attempts else 0
|
||
self.store.event("WARN", f"下载失败 编号={tid} 第{n}次 原因={reason[:120]}")
|
||
self.store.finish(tid, False, error=reason, url=url, source_host=host)
|
||
self._note_failed(tid, reason)
|
||
|
||
@staticmethod
|
||
def _head_hex(path: str, n: int = 16) -> str:
|
||
try:
|
||
with open(path, "rb") as fh:
|
||
return fh.read(n).hex()
|
||
except Exception:
|
||
return "?"
|
||
|
||
@staticmethod
|
||
def _md5(path: str) -> str:
|
||
h = hashlib.md5()
|
||
with open(path, "rb") as fh:
|
||
for chunk in iter(lambda: fh.read(1 << 20), b""):
|
||
h.update(chunk)
|
||
return h.hexdigest()
|
||
|
||
@staticmethod
|
||
def _looks_like_ts(path: str, probe: int = 8192) -> tuple[bool, int]:
|
||
"""TS 判定：**扫描**头部若干字节，找连续 3 个 188 字节包同步点。
|
||
|
||
返回 (是否 TS, 包起始偏移)。之所以要扫描而不是只看偏移 0：
|
||
上游 `download.origjoy.com` 下发的流以 **512 字节私有头** 开头（实测），
|
||
只检查 offset 0 会把合法流误判为坏文件（App 的校验同样是扫描式）。
|
||
|
||
⚠️ 入参是**本地落盘文件路径**（不是 URL），因此不存在 query/fragment 问题；
|
||
「URL 里的 `.ts` 文件名是否等于编号」那条闸门在 `resolver.is_trusted_url`
|
||
（已按剥离 query 后的 path 判定）。
|
||
"""
|
||
try:
|
||
with open(path, "rb") as fh:
|
||
head = fh.read(probe)
|
||
except Exception:
|
||
return False, 0
|
||
limit = min(4096, max(0, len(head) - 376))
|
||
for off in range(0, limit):
|
||
if head[off] == 0x47 and head[off + 188] == 0x47 and head[off + 376] == 0x47:
|
||
return True, off
|
||
return False, 0
|