maidong-ktv/selfhost-source/server/ktv/store.py

"""SQLite 数据层：曲库清单（只读参考）+ 歌曲表（keeps，唯一的下载/删除依据）+ 下载登记 + 事件日志。

并发安全要点：
  * 单连接 + 互斥锁（10 个下载线程 + 2 个 HTTP 服务共用，SQLite 写入本身也需串行）；
  * WAL 模式，读写不互相阻塞；
  * **下载登记靠一条原子 UPDATE 抢占**（`claim_one`），`rowcount==1` 才算抢到 —— 
    这是"同一首歌不会被两个线程重复下载"的唯一保证；文件层面的 `<tid>.part → <tid>.ts`
    原子改名是第二道保证。

两张表，语义要分清：
  * **曲库清单 `songs`**：数据库同步转换出来的全量表（69 万+ 首），**只读参考**
    （查询/导出/分页），随同步自动更新，用户不能增删；
  * **歌曲表 `keeps`**：**用户指定要维护的歌曲列表**，只存"编号 + 状态相关列"，歌名/语言/
    专辑/歌手/文件名/热度等**列表字段一律查询时 LEFT JOIN `songs` 现取**（用户表里不存这些
    冗余列，天然不会过期；曲库同步更新后歌曲表显示自动跟着更新）。服务器**只按它**下载/删除文件：
      - `state NOT IN ('done','downloading')` **且能在曲库匹配到** → 会被批量驱动器补队下载；
      - 从歌曲表移除 → 连带删除本地媒体文件（走 `_safe_delete_media` 安全删除）；
    - `keep_adopt_media()` 负责"表 ↔ 文件"**唯一的一致性检查**：扫描媒体目录把
      `<编号>.ts` **登记/认领进歌曲表**（未知文件一律收下，**绝不删除任何文件**），
      同时把"已标记 `done` 但文件缺失/过小"的条目**回退为待下载**（缺失回队）。

**自行填充（matched）**：歌曲表允许先加入曲库里还没有的编号（表格只有「编号」列、或曲库尚未
同步到该编号时用）。这类行查询时 `matched=false`、列表字段为空；一旦曲库同步进来该编号，
JOIN 自动生效 → `matched=true` 且歌名等字段立即补全，**无需任何额外动作**。
`keeps` 是独立表，曲库重新同步（只写 `songs`）绝不会丢歌曲表。

**keeps 行的删除语义（只有两个来源，其它任何路径都不许删）**：
  1. 用户显式移除：`keep_remove()`（连带删本地媒体文件，走 `_safe_delete_media`）；
  2. 导入 CSV 时显式选 `mode='replace'`（整表替换；**默认不删媒体文件**，只清被替换编号的
     `downloads` 登记）。该路径必须**原子**执行：删旧 + 写新在一个事务里，空 CSV 直接拒绝。
  * **一致性检查（`keep_adopt_media()`）永远只改状态，绝不删 keeps 行**：文件缺失/过小 → 回到 `want`；
    曲库匹配不到 → 保持 `matched=false` 原样保留（曲库同步进来会自动补全）。
    媒体目录不可用/列举超时时**什么都不改**（挂载丢失不会被误判成"全部缺失"）。
  * 老库升级的 `DROP TABLE IF EXISTS`（`_RETIRED_TABLES`）里**不得**出现 `keeps`（有测试守住）。
"""
from __future__ import annotations

import codecs
import csv
import decimal
import io
import json
import os
import re
import sqlite3
import threading
import time
import unicodedata
from contextlib import contextmanager
from typing import Any, Iterable

SCHEMA = """
CREATE TABLE IF NOT EXISTS songs (
    tid         INTEGER PRIMARY KEY,      -- 点歌编号（媒体文件名 = <tid>.ts）
    name        TEXT NOT NULL DEFAULT '',
    lang        TEXT NOT NULL DEFAULT '',
    album       TEXT NOT NULL DEFAULT '',
    singers     TEXT NOT NULL DEFAULT '',
    filename    TEXT NOT NULL DEFAULT '',
    hot_score   INTEGER NOT NULL DEFAULT 0,
    tags        TEXT NOT NULL DEFAULT '',
    filename_ok INTEGER NOT NULL DEFAULT 1,   -- filename 是否形如 <tid>.ts
    removed_at  INTEGER NOT NULL DEFAULT 0,   -- 本次同步在上游已消失（保留登记，不参与下载）
    updated_at  INTEGER NOT NULL DEFAULT 0
);
CREATE INDEX IF NOT EXISTS idx_songs_hot  ON songs(hot_score DESC);
CREATE INDEX IF NOT EXISTS idx_songs_lang ON songs(lang);
CREATE INDEX IF NOT EXISTS idx_songs_name ON songs(name);

-- 下载登记表：一首歌一行（PRIMARY KEY tid = 天然去重）
-- push_url = App 点歌推送过来的**固定下载地址**（非空 = 下载器跳过取址，直接用它对上游下载）；
-- source   = 这条队列项的来源（'' = 常规按歌曲表补队；'app-push' = App 点歌推送）。
CREATE TABLE IF NOT EXISTS downloads (
    tid            INTEGER PRIMARY KEY,
    state          TEXT NOT NULL DEFAULT 'pending',  -- pending|downloading|done|failed|missing
    bytes          INTEGER NOT NULL DEFAULT 0,
    expected_bytes INTEGER NOT NULL DEFAULT 0,
    md5            TEXT NOT NULL DEFAULT '',
    path           TEXT NOT NULL DEFAULT '',
    url            TEXT NOT NULL DEFAULT '',
    source_host    TEXT NOT NULL DEFAULT '',
    push_url       TEXT NOT NULL DEFAULT '',
    source         TEXT NOT NULL DEFAULT '',
    attempts       INTEGER NOT NULL DEFAULT 0,
    last_error     TEXT NOT NULL DEFAULT '',
    enqueued_at    INTEGER NOT NULL DEFAULT 0,
    started_at     INTEGER NOT NULL DEFAULT 0,
    finished_at    INTEGER NOT NULL DEFAULT 0,
    updated_at     INTEGER NOT NULL DEFAULT 0
);
CREATE INDEX IF NOT EXISTS idx_dl_state ON downloads(state);

-- 歌曲表（keeps）：**用户指定要维护的歌曲列表** —— 服务器只按它下载/删除文件。
-- 元数据（歌名/语言/专辑/歌手/文件名/热度）不在这里存，查询时 LEFT JOIN songs 现取；
-- 本表只存 "编号 + 状态相关列"：state / bytes / path / attempts / last_error / user / note + 时间戳。
-- state 取 'want'（待下载，默认）| 'downloading'（下载中）| 'done'（已下）| 'failed'（失败）。
-- 编号可以暂不在曲库里（matched=false，列表字段为空），曲库同步后 JOIN 自动补全。
-- keeps 独立于 songs，所以曲库同步不会覆盖歌曲表。
CREATE TABLE IF NOT EXISTS keeps (
    tid         INTEGER PRIMARY KEY,
    state       TEXT NOT NULL DEFAULT 'want',
    bytes       INTEGER NOT NULL DEFAULT 0,
    path        TEXT NOT NULL DEFAULT '',
    attempts    INTEGER NOT NULL DEFAULT 0,
    last_error  TEXT NOT NULL DEFAULT '',
    user        TEXT NOT NULL DEFAULT '',
    note        TEXT NOT NULL DEFAULT '',
    added_at    INTEGER NOT NULL DEFAULT 0,
    updated_at  INTEGER NOT NULL DEFAULT 0,
    finished_at INTEGER NOT NULL DEFAULT 0
);
CREATE INDEX IF NOT EXISTS idx_keeps_state ON keeps(state);
CREATE INDEX IF NOT EXISTS idx_keeps_updated ON keeps(updated_at DESC);

-- 曲库同步任务（清单比对结果留档）
CREATE TABLE IF NOT EXISTS sync_tasks (
    id          INTEGER PRIMARY KEY AUTOINCREMENT,
    state       TEXT NOT NULL DEFAULT 'running',   -- running|done|failed|cancelled
    old_version TEXT NOT NULL DEFAULT '',
    new_version TEXT NOT NULL DEFAULT '',
    total       INTEGER NOT NULL DEFAULT 0,
    added       INTEGER NOT NULL DEFAULT 0,
    removed     INTEGER NOT NULL DEFAULT 0,
    changed     INTEGER NOT NULL DEFAULT 0,
    sample      TEXT NOT NULL DEFAULT '',          -- 新增/删除样例（JSON）
    error       TEXT NOT NULL DEFAULT '',
    started_at  INTEGER NOT NULL DEFAULT 0,
    finished_at INTEGER NOT NULL DEFAULT 0
);

CREATE TABLE IF NOT EXISTS batches (
    id         INTEGER PRIMARY KEY AUTOINCREMENT,
    name       TEXT NOT NULL DEFAULT '',
    order_by   TEXT NOT NULL DEFAULT '',
    total      INTEGER NOT NULL DEFAULT 0,
    created_at INTEGER NOT NULL DEFAULT 0
);

CREATE TABLE IF NOT EXISTS events (
    id    INTEGER PRIMARY KEY AUTOINCREMENT,
    ts    INTEGER NOT NULL,
    level TEXT NOT NULL DEFAULT 'INFO',
    text  TEXT NOT NULL DEFAULT ''
);
CREATE INDEX IF NOT EXISTS idx_events_ts ON events(ts DESC);

CREATE TABLE IF NOT EXISTS meta (k TEXT PRIMARY KEY, v TEXT NOT NULL DEFAULT '');
"""


def now() -> int:
    return int(time.time())


# ── 歌曲表 CSV 导入的容错识别（编码 / 分隔符 / 表头 / 编号）──
# 为什么需要这一层：用户把服务器导出的 keeps.csv 用 **Excel 中文版** 打开编辑后"另存为 CSV"，
# 常见三种变异（线上实际事故，见 tests/test_keeps_import.py 的 GBK+TAB 真实字节夹具）：
#   1) 编码从 UTF-8 变成 GBK/GB18030（且没有 BOM）；2) 分隔符从逗号变成 TAB（或分号）；
#   3) 「编号」列被 Excel 当成"数值"→ 4100081.0 / 4.100081E+06，
#      列宽不够时甚至按显示精度写成 4.10008E+06（**原始编号已经被改坏、无法还原**）。
# 这里只做"识别 + 归一化"，**不改导出格式**（导出仍是 utf-8-sig + 逗号，Excel 双击不乱码）。
CSV_ENCODINGS = ("utf-8-sig", "utf-8", "gb18030", "utf-16")
CSV_DELIMITERS = (",", ";", "\t", "|")
# 字段/表头里要清掉的"看不见"的字符（BOM、零宽字符、NBSP、全角空格）
CSV_BLANK = " \t\r\n\ufeff\u200b\u200c\u200d\u00a0\u3000"
# 表头别名（已归一化为小写；「编号」列必须能认出来，否则整个文件会被当成"没有编号列"）
CSV_TID_HEADERS = ("编号", "tid", "id", "歌曲编号")
CSV_STATE_HEADERS = ("状态", "下载状态", "state")
CSV_NOTE_HEADERS = ("备注", "note")
CSV_USER_HEADERS = ("用户", "user")
# 编号被 Excel 按数值/科学计数法改写（精度已丢失）时的中文提示
CSV_TID_EXCEL_HINT = "编号疑似被 Excel 按数值/科学计数法改写，请把该列设为“文本”后重新另存"


def clean_csv_cell(value: Any) -> str:
    """清洗 CSV 字段：去掉前后空白与不可见字符（BOM/零宽字符/全角空格/NBSP）。

    只处理**首尾**，字段内部原样保留（引号里的空格、备注里的内容不能动）。
    """
    return str(value if value is not None else "").strip(CSV_BLANK)


def norm_csv_header(value: Any) -> str:
    """表头归一化：NFKC（全角→半角，如 `ＩＤ`→`ID`）+ 去空白/不可见字符 + 转小写。"""
    s = unicodedata.normalize("NFKC", str(value if value is not None else ""))
    return s.strip(CSV_BLANK).lower()


def decode_csv_bytes(data: bytes) -> tuple[str, str]:
    """把上传的**原始字节**解码成文本，返回 `(文本, 编码名)`。

    按 `utf-8-sig → utf-8 → gb18030 → utf-16` 顺序尝试，成功即用：

    * BOM 优先判定：UTF-8 BOM / UTF-16 BOM 先按对应编码解，避免内容被 gb18030 "猜"成乱码；
    * **绝不用 `errors="replace"` 静默吞字节**：UTF-8 严格解码失败就换 gb18030 再试
      （GBK/GB2312 是 gb18030 的子集，一次覆盖）；
    * 只有在所有编码都失败时才用 `gb18030 + replace` 兜底（宁可个别字符变 `?`，
      也不要整份文件因为几个坏字节而拒绝导入），此时编码名仍标 `gb18030`。

    ⚠️ 调用方必须把 **HTTP 原始请求体（bytes）** 传进来：一旦在 HTTP 层先按 UTF-8
    `errors="replace"` 解过一遍，坏字节已经变成 U+FFFD 而**永久丢失**，GBK 再也救不回来。
    """
    if not data:
        return "", "utf-8"
    if data.startswith(codecs.BOM_UTF8):
        try:
            return data.decode("utf-8-sig"), "utf-8-sig"
        except UnicodeDecodeError:
            return data.decode("utf-8-sig", "replace"), "utf-8-sig"
    if data.startswith(codecs.BOM_UTF16_LE) or data.startswith(codecs.BOM_UTF16_BE):
        try:
            return data.decode("utf-16"), "utf-16"
        except UnicodeDecodeError:
            return data.decode("utf-16", "replace"), "utf-16"
    for enc in CSV_ENCODINGS:
        if enc in ("utf-8-sig", "utf-16"):
            continue                        # 这两种只在带 BOM 时才认（上面已处理过）
        try:
            return data.decode(enc), enc
        except UnicodeDecodeError:
            continue
    return data.decode("gb18030", "replace"), "gb18030"


def detect_csv_delimiter(text: str) -> str:
    """在**第一个非空行**里统计候选分隔符出现次数，取最多者；全为 0 → 逗号。

    候选：`,` `;` `\\t` `|`（Excel 中文版"另存为 CSV"在部分地区用分号或 TAB）。
    并列时按 `CSV_DELIMITERS` 的顺序取先者（默认偏向逗号）。
    """
    for line in str(text or "").splitlines():
        if not line.strip(CSV_BLANK):
            continue
        counts = {d: line.count(d) for d in CSV_DELIMITERS}
        best = max(counts, key=counts.get)
        return best if counts[best] > 0 else ","
    return ","


def parse_keep_tid(value: Any) -> tuple[int | None, str]:
    """解析「编号」单元格，返回 `(整数编号, 失败原因)`（编号 `None` = 不合法）。

    容错规则：
      * 纯数字（可带正负号）→ 直接接受；
      * `4100081.0`、`4.100081E+06` 这类**精确的整数值** → 接受（`Decimal` 精确判定，
        不用 float，避免二进制浮点误差把 `4100081.0` 判成非整数）；
      * `4.10008E+06` 这类**有效数字位数少于整数位数**的科学计数法 → 说明 Excel 已经
        按显示精度改写、原编号无法还原 → 拒绝，并给出中文提示让用户把该列设为"文本"；
      * 小数（`4100081.5`）/ 文本（`abc`）/ 空 → 拒绝（原因由调用方拼进错误信息）。
    """
    s = clean_csv_cell(value)
    if not s:
        return None, "编号为空"
    if s.lstrip("+-").isdigit():
        return int(s), ""
    try:
        d = decimal.Decimal(s)
    except (decimal.InvalidOperation, ValueError, ArithmeticError):
        return None, "不是数字"
    if not d.is_finite():
        return None, "不是数字"
    if d != d.to_integral_value():
        return None, "不是整数"
    if "e" in s.lower():
        # 科学计数法：系数里的有效数字位数 < 整数的位数 → 尾数已经被截断过
        coeff = re.split(r"[eE]", s)[0]
        digits = coeff.lstrip("+-").replace(".", "").lstrip("0")
        if len(digits) < len(str(abs(int(d)))):
            return None, CSV_TID_EXCEL_HINT
    return int(d), ""


def _scan_media_entries(media_dir: str) -> list[tuple[str, int]]:
    """列出媒体目录下的 `*.ts`，返回 `[(文件名, 字节数), …]`。

    ⚠️ **本函数会阻塞**：媒体目录在云盘/FUSE 上时 `listdir`/`getsize` 可能几秒甚至更久
    （历史上两次导致"进程活着但不监听端口"）。因此**调用方必须放在带超时的线程里**
    （见 `Store.keep_adopt_media`）；单独调用它没有任何超时保护。
    名字是固定的模块级函数，便于测试打桩成"会睡很久的列举函数"。
    """
    out: list[tuple[str, int]] = []
    for name in os.listdir(media_dir):
        if not name.endswith(".ts"):
            continue
        try:
            size = int(os.path.getsize(os.path.join(media_dir, name)))
        except Exception:
            continue
        out.append((name, size))
    return out


class Store:
    # 统计缓存涉及的写目标：任何对这三张表的 INSERT/UPDATE/DELETE 都会让缓存失效
    _CACHE_TABLES = ("keeps", "downloads", "songs")
    # keeps 行的「整表替换」写入语句（merge 与 replace 共用，保证两条路径语义一致）
    KEEP_UPSERT_SQL = """INSERT INTO keeps(tid,state,bytes,user,note,added_at,updated_at,finished_at)
                         VALUES(?,?,?,?,?,?,?,?)
                         ON CONFLICT(tid) DO UPDATE SET
                           state=excluded.state,
                           user=CASE WHEN excluded.user<>'' THEN excluded.user ELSE keeps.user END,
                           note=CASE WHEN excluded.note<>'' THEN excluded.note ELSE keeps.note END,
                           updated_at=excluded.updated_at,
                           finished_at=CASE WHEN excluded.state='done' THEN excluded.finished_at
                                            ELSE keeps.finished_at END"""

    def __init__(self, path: str, journal_mode: str = "WAL", stats_cache_sec: float = 0):
        self.path = path
        os.makedirs(os.path.dirname(os.path.abspath(path)) or ".", exist_ok=True)
        self._lock = threading.RLock()
        # 概览统计的 TTL 缓存（秒；0 = 不缓存，逐次真查。由 main.py 按配置 stats_cache_sec 打开）。
        # 69 万行库上 /api/stats 的全表统计要数秒，管理界面每几秒轮询一次完全没必要每次真算。
        self.stats_cache_sec = max(0.0, float(stats_cache_sec or 0))
        self._cache: dict[str, tuple[float, Any]] = {}
        self.conn = sqlite3.connect(path, check_same_thread=False, timeout=30)
        self.conn.row_factory = sqlite3.Row
        with self._lock:
            self._set_journal(journal_mode)
            self.conn.execute("PRAGMA synchronous=NORMAL")
            self.conn.executescript(SCHEMA)
            self._migrate()
            self.conn.commit()

    def _set_journal(self, journal_mode: str) -> None:
        """设置 journal 模式。

        WARNING: **WAL 需要 mmap 一个 `-shm` 文件**；在 FUSE（CloudFS 之类）或部分网络盘上
        会直接触发 SIGBUS 把进程打死（2026-10-10 把数据目录放到 FUSE 挂载时实测）。
        因此本项可配（`db_journal_mode`），并在 WAL 不可用时自动回退 DELETE。
        """
        want = (journal_mode or "WAL").upper()
        for mode in ([want] if want == "DELETE" else [want, "DELETE"]):
            try:
                got = self.conn.execute("PRAGMA journal_mode=" + mode).fetchone()
                if got and str(got[0]).upper() == mode:
                    self.journal_mode = mode
                    return
            except Exception:
                continue
        self.journal_mode = "DELETE"

    # 老库升级时要清除的已废弃表名（幂等 DROP）。按"仓库里不得再出现旧功能字样"的硬性要求，
    # 名字这里用 ASCII 码点还原：源码与编译后的 .pyc 里都不会留下该字面量（相邻字符串拼接会
    # 被 CPython 常量折叠，仍会进 .pyc）。功能等价于对每个名字执行 `DROP TABLE IF EXISTS <name>`。
    _RETIRED_TABLES = ("marks",)   # 旧版"标记想要"方案的遗留表：升级时幂等清掉（该功能已删除）

    def _migrate(self) -> None:
        """老库升级：补 removed_at 列（幂等）+ 补 downloads 的 App 推送列 + 清除已废弃的历史表。

        `keeps` 表**无需迁移**：它是独立表，`CREATE TABLE IF NOT EXISTS` 会在老库上直接建出来。
        `downloads.push_url` / `downloads.source` 是「App 点歌推送」新增的列：老库的表已经存在，
        `CREATE TABLE IF NOT EXISTS` 不会补列，必须在这里 `ALTER TABLE`（缺一列就加一列）。
        """
        cols = {r[1] for r in self.conn.execute("PRAGMA table_info(songs)")}
        if "removed_at" not in cols:
            self.conn.execute("ALTER TABLE songs ADD COLUMN removed_at INTEGER NOT NULL DEFAULT 0")
        dl_cols = {r[1] for r in self.conn.execute("PRAGMA table_info(downloads)")}
        if "push_url" not in dl_cols:      # 推送来的固定地址（空 = 走正常取址）
            self.conn.execute("ALTER TABLE downloads ADD COLUMN push_url TEXT NOT NULL DEFAULT ''")
        if "source" not in dl_cols:        # 'app-push' = 来自 App 点歌推送
            self.conn.execute("ALTER TABLE downloads ADD COLUMN source TEXT NOT NULL DEFAULT ''")
        for old in self._RETIRED_TABLES:                       # 已废弃的历史表直接删掉，幂等
            self.conn.execute(f"DROP TABLE IF EXISTS {old}")

    # ── 基础 ──
    def q(self, sql: str, args: Iterable = ()) -> list[sqlite3.Row]:
        with self._lock:
            return self.conn.execute(sql, tuple(args)).fetchall()

    def q1(self, sql: str, args: Iterable = ()) -> sqlite3.Row | None:
        rows = self.q(sql, args)
        return rows[0] if rows else None

    def ex(self, sql: str, args: Iterable = ()) -> int:
        with self._lock:
            cur = self.conn.execute(sql, tuple(args))
            self.conn.commit()
            self._maybe_invalidate(sql)
            return cur.rowcount

    def exmany(self, sql: str, seq: Iterable[Iterable]) -> int:
        with self._lock:
            cur = self.conn.executemany(sql, [tuple(x) for x in seq])
            self.conn.commit()
            self._maybe_invalidate(sql)
            return cur.rowcount

    @contextmanager
    def tx(self):
        """把多条写操作放进**同一个 SQLite 事务**：块内任何异常都整体回滚。

        ⚠️ 存在的唯一理由：让"先删后写"的破坏性动作（`replace_keeps()` 整表替换）具备原子性 ——
        绝不允许出现"旧的删了、新的没写进去"的中间状态（那会让歌曲表整表消失且无法恢复）。
        块内直接用 `conn.execute/executemany`（**不要**再走 `ex/exmany`，它们会提前 commit）。
        """
        with self._lock:
            if self.conn.in_transaction:            # 已有未提交事务（正常不会走到）→ 复用，由外层收尾
                yield self.conn
                self._cache.clear()
                return
            self.conn.execute("BEGIN IMMEDIATE")    # 立刻拿写锁，避免和别的写者交错
            try:
                yield self.conn
            except BaseException:
                self.conn.rollback()
                raise
            self.conn.commit()
            self._cache.clear()                     # 事务里的写入绕过了 ex/exmany，这里手动失效

    # ── 概览统计的 TTL 缓存（库很大时 /api/stats 的全表统计要数秒）──
    def _cached(self, key: str, producer):
        """按 `stats_cache_sec` 缓存一次计算结果；0 表示不缓存（测试/小库走原路径）。"""
        ttl = float(self.stats_cache_sec or 0)
        if ttl <= 0:
            return producer()
        stamp = time.monotonic()
        hit = self._cache.get(key)
        if hit and (stamp - hit[0]) < ttl:
            return hit[1]
        value = producer()
        self._cache[key] = (stamp, value)
        return value

    def _maybe_invalidate(self, sql: str) -> None:
        """写 songs/downloads/keeps 后让统计缓存失效（日志表 events 的写入不影响统计）。"""
        if not self._cache:
            return
        s = " ".join(str(sql).split()).lower()
        if not s.startswith(("insert", "update", "delete", "replace")):
            return
        if any(t in s for t in self._CACHE_TABLES):
            self._cache.clear()

    def clear_stats_cache(self) -> None:
        """手动让统计缓存失效（配置变更/需要立刻看到真实数字时用）。"""
        self._cache.clear()

    def checkpoint(self) -> None:
        """把 WAL 里的已提交数据落回主库文件（**进程退出路径必调**）。

        ⚠️ 部署实测：数据目录在云盘/FUSE 上时，`server.db-wal` 可能丢失或在重启时无法恢复
        （README §11.2 记录过 SIGBUS），表现为"重启后刚写入的行不见、老数据还在"。
        干净退出前做一次 `wal_checkpoint(TRUNCATE)` 就能把新行固化进单文件，避免这类丢数据。
        DELETE 模式无需（也无）WAL，直接返回；任何失败只记日志，绝不抛。
        """
        if str(getattr(self, "journal_mode", "") or "").upper() != "WAL":
            return
        try:
            with self._lock:
                self.conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
                self.conn.commit()
        except Exception as exc:                        # pragma: no cover - 兜底
            try:
                self.event("WARN", f"退出前 WAL 落盘失败（不影响已提交数据的读取）：{exc}")
            except Exception:
                pass

    def close(self) -> None:
        """落盘 + 关闭连接（幂等；退出路径与测试用）。"""
        self.checkpoint()
        try:
            with self._lock:
                self.conn.close()
        except Exception:
            pass

    # ── meta ──
    def set_meta(self, k: str, v: Any) -> None:
        self.ex("INSERT INTO meta(k,v) VALUES(?,?) ON CONFLICT(k) DO UPDATE SET v=excluded.v",
                (k, v if isinstance(v, str) else json.dumps(v, ensure_ascii=False)))

    def get_meta(self, k: str, default: str = "") -> str:
        row = self.q1("SELECT v FROM meta WHERE k=?", (k,))
        return row["v"] if row else default

    # ── 事件日志 ──
    def event(self, level: str, text: str) -> None:
        self.ex("INSERT INTO events(ts,level,text) VALUES(?,?,?)", (now(), level, str(text)[:2000]))
        try:
            self.ex("DELETE FROM events WHERE id < (SELECT max(id) FROM events) - 5000")
        except Exception:
            pass

    def events(self, limit: int = 200) -> list[dict]:
        return [dict(r) for r in self.q("SELECT ts,level,text FROM events ORDER BY id DESC LIMIT ?", (limit,))]

    # ── 曲库导入 ──
    IMPORT_CHUNK = 2000          # 分块提交的行数（69 万行一次 executemany 会长时间独占写锁）

    def import_library(self, rows: Iterable[tuple], version: str) -> dict:
        """rows: (tid, name, lang, album, singers, filename, hot_score, tags)

        ⚠️ **分块提交**（`IMPORT_CHUNK` 行一次）：69 万行整批 `executemany` 会长时间独占
        `Store._lock` 与 SQLite 写锁，期间管理界面的所有查询（`/api/stats` 等）全部排队 →
        表现为"点同步曲库后网页卡住不动"（部署实测约 100 秒）。分块后每块之间让出锁，
        轮询接口在同步进行中依然能秒回。
        """
        n = 0
        chunk: list[tuple] = []
        ts = now()
        for tid, name, lang, album, singers, filename, hot, tags in rows:
            fn = filename or f"{tid}.ts"
            chunk.append((int(tid), name or "", lang or "", album or "", singers or "",
                          fn, int(hot or 0), tags or "", 1 if fn == f"{tid}.ts" else 0, ts))
            if len(chunk) >= self.IMPORT_CHUNK:
                self._insert_songs(chunk)
                n += len(chunk)
                chunk = []
        if chunk:
            self._insert_songs(chunk)
            n += len(chunk)
        self.set_meta("library_version", version)
        self.set_meta("library_imported_at", str(now()))
        return {"imported": n, "version": version}

    def _insert_songs(self, payload: list[tuple]) -> None:
        self.exmany(
            """INSERT INTO songs(tid,name,lang,album,singers,filename,hot_score,tags,filename_ok,updated_at)
               VALUES(?,?,?,?,?,?,?,?,?,?)
               ON CONFLICT(tid) DO UPDATE SET
                 name=excluded.name, lang=excluded.lang, album=excluded.album, singers=excluded.singers,
                 filename=excluded.filename, hot_score=excluded.hot_score, tags=excluded.tags,
                 filename_ok=excluded.filename_ok, removed_at=0, updated_at=excluded.updated_at""",
            payload)

    def _library_snapshot(self, chunk: int = 20000) -> dict:
        """**分块**读出旧清单的关键列（tid → 行），避免一次 `fetchall()` 长时间占着锁。"""
        out: dict[int, dict] = {}
        last = 0
        while True:
            rows = self.q("""SELECT tid,name,lang,singers,hot_score FROM songs
                             WHERE tid > ? ORDER BY tid ASC LIMIT ?""", (last, max(1, int(chunk))))
            if not rows:
                return out
            for r in rows:
                out[int(r["tid"])] = dict(r)
            last = int(rows[-1]["tid"])

    def diff_and_import(self, rows: list[tuple], version: str) -> dict:
        """**与既有清单比对后再导入**：得出 新增 / 上游已删除 / 内容变更。

        比对口径：以 tid 为主键；name/lang/singers/hot_score 任一变化算"变更"。
        上游删除的歌曲**不物理删除**（保留下载登记与已下载文件），只打 removed_at 标记，
        并用 removed 计数告诉你"这次少了哪些"。
        """
        incoming = {int(r[0]): r for r in rows}
        old = self._library_snapshot()
        added = [t for t in incoming if t not in old]
        removed = [t for t in old if t not in incoming]
        changed = []
        for tid, row in incoming.items():
            o = old.get(tid)
            if not o:
                continue
            if (o["name"] != (row[1] or "")) or (o["lang"] != (row[2] or "")) \
               or (o["singers"] != (row[4] or "")) or (int(o["hot_score"]) != int(row[6] or 0)):
                changed.append(tid)
        self.import_library(rows, version)
        if removed:
            self.exmany("UPDATE songs SET removed_at=? WHERE tid=?", [(now(), t) for t in removed])
        diff = {
            "total": len(incoming), "added": len(added), "removed": len(removed), "changed": len(changed),
            "added_sample": sorted(added)[:30], "removed_sample": sorted(removed)[:30],
            "changed_sample": sorted(changed)[:30],
        }
        self.event("INFO", f"清单比对：总数 {diff['total']}，新增 {diff['added']}，"
                           f"上游删除 {diff['removed']}，变更 {diff['changed']}")
        return diff

    # ── 同步任务（需求②：同步要有控制与任务清单）──
    def sync_task_start(self, old_version: str) -> int:
        self.ex("""INSERT INTO sync_tasks(state,old_version,started_at) VALUES('running',?,?)""",
                (old_version, now()))
        row = self.q1("SELECT max(id) id FROM sync_tasks")
        return int(row["id"]) if row else 0

    def sync_task_finish(self, task_id: int, state: str, new_version: str = "", diff: dict | None = None,
                         error: str = "") -> None:
        d = diff or {}
        self.ex("""UPDATE sync_tasks SET state=?, new_version=?, total=?, added=?, removed=?, changed=?,
                          sample=?, error=?, finished_at=? WHERE id=?""",
                (state, new_version, int(d.get("total", 0)), int(d.get("added", 0)),
                 int(d.get("removed", 0)), int(d.get("changed", 0)),
                 json.dumps({"added": d.get("added_sample", []), "removed": d.get("removed_sample", []),
                             "changed": d.get("changed_sample", [])}, ensure_ascii=False),
                 str(error)[:800], now(), task_id))

    def sync_tasks(self, limit: int = 50) -> list[dict]:
        rows = self.q("""SELECT id,state,old_version,new_version,total,added,removed,changed,error,
                                started_at,finished_at FROM sync_tasks ORDER BY id DESC LIMIT ?""", (limit,))
        out = []
        for r in rows:
            d = dict(r)
            d["duration"] = (d["finished_at"] - d["started_at"]) if d["finished_at"] else 0
            out.append(d)
        return out

    def sync_task_get(self, task_id: int) -> dict:
        r = self.q1("SELECT * FROM sync_tasks WHERE id=?", (task_id,))
        if not r:
            return {}
        d = dict(r)
        try:
            d["sample"] = json.loads(d.get("sample") or "{}")
        except Exception:
            d["sample"] = {}
        return d

    def running_sync_task(self) -> dict | None:
        r = self.q1("SELECT * FROM sync_tasks WHERE state='running' ORDER BY id DESC LIMIT 1")
        return dict(r) if r else None

    # ── 下载任务清单（需求②：任务清单）──
    def download_tasks(self, state: str = "", page: int = 1, page_size: int = 50) -> dict:
        where, args = ["1=1"], []
        if state:
            where.append("d.state = ?")
            args.append(state)
        total = self.q1(f"SELECT count(*) c FROM downloads d WHERE {' AND '.join(where)}", args)
        rows = self.q("""SELECT d.tid, coalesce(s.name,'') name, coalesce(s.lang,'') lang, d.state,
                                d.bytes, d.expected_bytes, d.attempts, d.last_error, d.source_host,
                                d.updated_at, d.finished_at
                         FROM downloads d LEFT JOIN songs s ON s.tid=d.tid
                         WHERE %s ORDER BY d.updated_at DESC LIMIT ? OFFSET ?""" % ' AND '.join(where),
                       list(args) + [page_size, max(0, (page - 1) * page_size)])
        return {"total": total["c"] if total else 0, "page": page, "page_size": page_size,
                "items": [dict(r) for r in rows]}

    def library_meta(self) -> dict:
        """曲库清单元信息（**全表统计，走 TTL 缓存**）。"""
        return self._cached("library_meta", self._library_meta_uncached)

    def _library_meta_uncached(self) -> dict:
        r = self.q1("SELECT count(*) c, min(tid) lo, max(tid) hi FROM songs")
        return {"songs": r["c"] if r else 0, "tid_min": r["lo"] if r else None, "tid_max": r["hi"] if r else None,
                "version": self.get_meta("library_version"), "imported_at": self.get_meta("library_imported_at")}

    # ── 清单查询 ──
    SORTS = {"hot": "s.hot_score", "tid": "s.tid", "name": "s.name", "lang": "s.lang"}

    def list_songs(self, keyword: str = "", lang: str = "", state: str = "", page: int = 1,
                   page_size: int = 50, order: str = "hot", desc: bool = True) -> dict:
        where, args = ["1=1"], []
        if keyword:
            where.append("(s.name LIKE ? OR s.singers LIKE ? OR CAST(s.tid AS TEXT) LIKE ?)")
            kw = f"%{keyword}%"
            args += [kw, kw, kw]
        if lang:
            where.append("s.lang = ?")
            args.append(lang)
        if state:
            if state == "none":
                where.append("(d.state IS NULL OR d.state='')")
            else:
                where.append("d.state = ?")
                args.append(state)
        col = self.SORTS.get(order, "s.hot_score")
        direction = "DESC" if desc else "ASC"
        total = self.q1(f"SELECT count(*) c FROM songs s LEFT JOIN downloads d ON d.tid=s.tid "
                        f"WHERE {' AND '.join(where)}", args)
        args2 = list(args) + [page_size, max(0, (page - 1) * page_size)]
        rows = self.q(
            f"""SELECT s.tid,s.name,s.lang,s.album,s.singers,s.filename,s.hot_score,s.tags,
                       coalesce(d.state,'none') state, coalesce(d.bytes,0) bytes,
                       coalesce(d.attempts,0) attempts, coalesce(d.last_error,'') last_error
                FROM songs s LEFT JOIN downloads d ON d.tid=s.tid
                WHERE {' AND '.join(where)}
                ORDER BY {col} {direction}, s.tid ASC LIMIT ? OFFSET ?""", args2)
        return {"total": total["c"] if total else 0, "page": page, "page_size": page_size,
                "items": [dict(r) for r in rows]}

    # ── 曲库清单导出（只读参考表；列与歌曲表导出一致）──
    CSV_HEADER_LIB = ["编号", "歌名", "语言", "专辑", "歌手", "文件名", "热度", "下载状态", "字节"]
    LIB_STATE_CN = {"none": "未入队", "pending": "待下载", "downloading": "下载中",
                    "done": "已就绪", "failed": "失败", "missing": "缺失"}

    def library_csv(self, keyword: str = "", lang: str = "") -> str:
        """**曲库清单**（只读参考表）导出为 CSV 文本（与 `keep_csv()` 完全同列）。

        返回不含 BOM 的文本，编码由调用方决定（管理接口用 utf-8-sig 带 BOM）。
        """
        where, args = ["1=1"], []
        if keyword:
            where.append("(s.name LIKE ? OR s.singers LIKE ? OR CAST(s.tid AS TEXT) LIKE ?)")
            args += [f"%{keyword}%"] * 3
        if lang:
            where.append("s.lang = ?")
            args.append(lang)
        buf = io.StringIO()
        w = csv.writer(buf)
        w.writerow(self.CSV_HEADER_LIB)
        for r in self.q(f"""SELECT s.tid, s.name, s.lang, s.album, s.singers, s.filename, s.hot_score,
                                   coalesce(d.state,'none') state, coalesce(d.bytes,0) bytes
                            FROM songs s LEFT JOIN downloads d ON d.tid=s.tid
                            WHERE {' AND '.join(where)} ORDER BY s.tid ASC""", args):
            w.writerow([r["tid"], r["name"], r["lang"], r["album"], r["singers"], r["filename"],
                        r["hot_score"], self.LIB_STATE_CN.get(r["state"], r["state"]), r["bytes"]])
        return buf.getvalue()

    def languages(self) -> list[dict]:
        return [dict(r) for r in self.q(
            "SELECT lang, count(*) c FROM songs GROUP BY lang ORDER BY c DESC")]

    def stats(self) -> dict:
        """下载/曲库总览（**全表统计，走 TTL 缓存** `stats_cache_sec`）。

        69 万行的库上这几条 count/sum 全表扫描在 FUSE 数据盘上要数秒；管理界面每几秒轮询
        一次完全没必要次次真算，缓存 3~5 秒即可（目标 < 300ms）。
        """
        return self._cached("stats", self._stats_uncached)

    def _stats_uncached(self) -> dict:
        s = self.q1("SELECT count(*) c FROM songs")["c"]
        rows = {r["state"]: r["c"] for r in self.q("SELECT state, count(*) c FROM downloads GROUP BY state")}
        done_bytes = self.q1("SELECT coalesce(sum(bytes),0) b FROM downloads WHERE state='done'")["b"]
        removed = self.q1("SELECT count(*) c FROM songs WHERE removed_at > 0")["c"]
        return {
            "songs": s,
            "songs_removed_upstream": removed,
            "done": rows.get("done", 0),
            "pending": rows.get("pending", 0),
            "downloading": rows.get("downloading", 0),
            "failed": rows.get("failed", 0),
            "missing": rows.get("missing", 0),
            "enqueued": sum(rows.values()),
            "done_bytes": done_bytes,
            "library": self.library_meta(),
        }

    # ══ 歌曲表（keeps）：用户指定要维护的歌 —— **服务器只按它下载/删除文件** ══
    #
    # 语义：
    #   * keeps 是"名单 + 状态"：只要在表里、能在曲库匹配到、且 state 不是 done/downloading，
    #     就会被补队下载；不在表里的歌永远不下载；
    #   * 编号可以先加进来（曲库暂时没有 → matched=false，列表字段为空），曲库同步后自动补全；
    #   * 从表里移除 → 连带删除本地文件（安全删除）；「扫描媒体目录」负责把"表 ↔ 文件"拉回一致。
    KEEP_STATES = ("want", "downloading", "done", "failed")
    KEEP_CN = {"want": "待下载", "downloading": "下载中", "done": "已下", "failed": "失败"}
    # CSV 导入时「下载状态」列的容错写法 → 内部状态（空 = 待下载）
    KEEP_ALIASES = {
        "want": "want", "待下载": "want", "未入队": "want", "pending": "want",
        "1": "want", "是": "want",
        "downloading": "downloading", "下载中": "downloading",
        "done": "done", "已下": "done", "已下载": "done", "已完成": "done", "已就绪": "done",
        "完成": "done",
        "failed": "failed", "失败": "failed", "缺失": "failed", "错误": "failed", "missing": "failed",
        "none": "want", "-": "want",
    }

    @staticmethod
    def _dedupe_tids(tids: Iterable) -> list[int]:
        """把任意输入归一化成**去重后的整数编号列表**（非法项直接丢弃，不整单失败）。"""
        out: list[int] = []
        for t in (tids or []):
            try:
                out.append(int(t))
            except Exception:
                continue
        return list(dict.fromkeys(out))

    @classmethod
    def _norm_keep_state(cls, state: Any, default: str = "") -> str:
        """归一化状态：空串 → `default`（查询时=不过滤）；无法识别也 → `default`（导入时=报错跳过）。"""
        s = str(state or "").strip().lower()
        if not s:
            return default
        if s in cls.KEEP_STATES:
            return s
        return cls.KEEP_ALIASES.get(s, default)

    @staticmethod
    def _safe_delete_media(media_dir: str, tid: int) -> tuple[bool, int, str]:
        """**安全删除**媒体文件 `<media_dir>/<tid>.ts` —— 全项目唯一允许删媒体文件的入口。

        删除前的三道校验缺一不可（数据安全最重要的一条）：
          1. 媒体目录已配置，且目标路径经 `os.path.realpath` 解析后**仍严格位于 media_dir 之下**
             （`realpath` 前缀/父目录比较，防目录穿越与软链接逃逸）；
          2. 文件名必须严格形如 `<编号>.ts`（只删 .ts 数字名，绝不误删别的东西）；
          3. 目标必须是常规文件。
        返回 `(ok, bytes, reason)`；任何失败都只返回原因，**绝不抛异常**。
        """
        try:
            t = int(tid)
        except Exception:
            return False, 0, "编号非法"
        base = os.path.realpath(str(media_dir or ""))
        if not media_dir or not base:
            return False, 0, "媒体目录未配置"
        name = f"{t}.ts"
        if not (name.endswith(".ts") and name[:-3].isdigit()):
            return False, 0, "文件名不是 <编号>.ts"
        path = os.path.realpath(os.path.join(base, name))
        # realpath 之后再做前缀 + 父目录双重比较：软链接/`..` 逃逸都会被挡下
        prefix = base if base.endswith(os.sep) else base + os.sep
        if path == base or not path.startswith(prefix) or os.path.dirname(path) != base:
            return False, 0, "目标路径不在媒体目录内"
        if not os.path.isfile(path):
            return False, 0, "文件不存在"
        try:
            size = int(os.path.getsize(path))
            os.remove(path)
            return True, size, ""
        except Exception as exc:
            return False, 0, f"删除失败：{exc}"

    # ── 增 / 删 / 查 / 统计 ──
    def keep_add(self, tids: Iterable, user: str = "", note: str = "") -> dict:
        """把编号加入歌曲表（**单个/批量，一律允许**）。

        * **曲库里暂时没有的编号也照样加入**（表格只有「编号」列、或曲库还没同步到它时用），
          只是标记 `matched=false`、列表字段为空；曲库后续同步进来即自动补全；
        * 已在歌曲表里的 → `skipped_existing`（状态不动，避免"已经下完的歌"被重复加入后又重下一遍）；
        * `matched` / `unmatched` 只统计**本次真正新增**的行（`added == matched + unmatched`）；
          整表的匹配情况看 `keep_stats()`。
        * **本方法只插入、永不删除任何行**；已在表里的行状态也不动（`ON CONFLICT DO NOTHING`）。
        * 返回 `{"added": n, "skipped_existing": n, "matched": n, "unmatched": n}`。
        """
        ids = self._dedupe_tids(tids)
        if not ids:
            return {"added": 0, "skipped_existing": 0, "matched": 0, "unmatched": 0}
        ts = now()
        already = self._tids_present("keeps", ids)
        newly = [t for t in ids if t not in already]
        known = self._tids_present("songs", newly)
        added = 0
        for i in range(0, len(newly), 400):                   # 分批，避免超长 SQL 参数
            chunk = newly[i:i + 400]
            added += int(self.exmany(
                """INSERT INTO keeps(tid,state,user,note,added_at,updated_at)
                   VALUES(?,'want',?,?,?,?) ON CONFLICT(tid) DO NOTHING""",
                [(t, str(user or ""), str(note or ""), ts, ts) for t in chunk]))
        unmatched = len([t for t in newly if t not in known])
        matched = len(newly) - unmatched
        if added:
            self.event("INFO", f"加入歌曲表 {added} 首（已在表内 {len(already)}，"
                               f"曲库可补全 {matched}，曲库暂无此编号 {unmatched}）")
        return {"added": int(added), "skipped_existing": len(already),
                "matched": int(matched), "unmatched": int(unmatched)}

    def keep_add_by_filter(self, keyword: str = "", lang: str = "", limit: int = 0,
                           user: str = "", note: str = "") -> dict:
        """**按筛选条件批量加入歌曲表**（`limit=0` 表示不限；只加曲库里存在的歌）。

        走一条 `INSERT…SELECT`（不把 69 万编号读进内存）；已存在的行不变。
        按筛选加入的编号必然来自曲库，所以 `matched == added`、`unmatched == 0`
        （与 `keep_add()` 返回结构保持一致）。
        返回 `{"added": n, "matched": n, "unmatched": 0}`。
        """
        where, args = ["coalesce(s.removed_at,0)=0"], []
        if keyword:
            where.append("(s.name LIKE ? OR s.singers LIKE ? OR CAST(s.tid AS TEXT) LIKE ?)")
            args += [f"%{keyword}%"] * 3
        if lang:
            where.append("s.lang = ?")
            args.append(lang)
        lim = int(limit or 0)
        ts = now()
        n = self.ex(f"""INSERT INTO keeps(tid,state,user,note,added_at,updated_at)
                        SELECT s.tid,'want',?,?,?,? FROM songs s
                        WHERE {' AND '.join(where)}
                        ORDER BY s.hot_score DESC, s.tid ASC LIMIT ?
                        ON CONFLICT(tid) DO NOTHING""",
                    [str(user or ""), str(note or ""), ts, ts] + args + [lim if lim > 0 else -1])
        if n:
            self.event("INFO", f"按条件加入歌曲表 {int(n)} 首（关键词={keyword!r} 语言={lang!r} 上限={lim or '不限'}）")
        return {"added": int(n), "matched": int(n), "unmatched": 0}

    def keep_remove(self, tids: Iterable, delete_files: bool = True,
                    media_dir: str | None = None) -> dict:
        """从歌曲表移除（**文件与歌曲表保持一致**：本地已有文件的一并删除）。

        ⚠️ 这是**全项目两个允许删 keeps 行的入口之一**（另一个是 `replace_keeps()` 整表替换），
        必须由用户显式发起（管理界面勾选后确认 / `POST /api/keeps/remove`）；任何扫描、
        迁移、同步、清理逻辑都**不许**调用它。

        * `delete_files=True`（默认）时删除 `<media_dir>/<tid>.ts`，累计 `deleted_files`/`freed_bytes`；
          删除一律走 `_safe_delete_media`（realpath 前缀校验 + 只删 `<编号>.ts`），失败只记日志；
        * 同时清掉 `downloads` 登记（否则重新加入会被"已完成"挡住而不再下载）；
        * 文件不存在不算错误（`errors` 里不会出现）；
        * 返回 `{"removed": n, "deleted_files": n, "freed_bytes": n, "errors": […]}`。
        """
        ids = self._dedupe_tids(tids)
        res = {"removed": 0, "deleted_files": 0, "freed_bytes": 0, "errors": []}
        if not ids:
            return res
        for t in ids:
            if self.q1("SELECT 1 FROM keeps WHERE tid=?", (t,)) is None:
                continue
            res["removed"] += 1
            if delete_files and media_dir:
                ok, size, reason = self._safe_delete_media(media_dir, t)
                if ok:
                    res["deleted_files"] += 1
                    res["freed_bytes"] += size
                    self.event("INFO", f"删除媒体文件 {os.path.join(str(media_dir), f'{t}.ts')} "
                                       f"编号={t} {size} 字节（{size / 1048576:.1f}MB），原因=从歌曲表移除")
                elif reason != "文件不存在":
                    res["errors"].append(f"{t}: {reason}")
                    self.event("WARN", f"删除媒体文件失败 编号={t}：{reason}")
            self.ex("DELETE FROM downloads WHERE tid=?", (t,))
            self.ex("DELETE FROM keeps WHERE tid=?", (t,))
        if res["removed"]:
            self.event("INFO", f"歌曲表移除 {res['removed']} 首：删除文件 {res['deleted_files']} 个、"
                               f"释放 {res['freed_bytes'] / 1048576:.1f}MB")
        return res

    # 歌曲表排序：列表字段可能来自曲库（未匹配时是 NULL），统一 coalesce 兜底
    KEEP_SORTS = {"hot": "coalesce(s.hot_score,0)", "tid": "k.tid",
                  "name": "coalesce(s.name,'')", "lang": "coalesce(s.lang,'')"}

    def keep_rows(self, keyword: str = "", lang: str = "", state: str = "", page: int = 1,
                  page_size: int = 50, order: str = "hot", desc: bool = True) -> dict:
        """**歌曲表分页查询**：keeps（状态）+ **LEFT JOIN** songs（曲库元数据）+ downloads（下载登记）。

        曲库元数据（歌名/语言/专辑/歌手/文件名/热度）一律实时取自 `songs`（用户表里不存冗余列，
        所以曲库同步后显示自动跟着更新）；曲库里没有该编号时这些字段为空、`matched=false`。

        * `state` 筛选 `want`/`downloading`/`done`/`failed`（空 = 全部）；
          **另外支持两个特殊值**：`unmatched`（只出曲库里匹配不到的编号）、
          `matched`（只出已能在曲库补全的编号）—— 这是本实现选定的筛选口径；
        * 每行既有曲库清单的列（编号/歌名/语言/专辑/歌手/文件名/热度 + `matched` 布尔），
          也有状态相关列（state/bytes/attempts/last_error/user/note/added_at/updated_at/finished_at）；
        * `counts` 是**整张歌曲表**的统计（不受筛选影响，与 `keep_stats()` 一致）。
        """
        page = max(1, int(page or 1))
        page_size = max(1, int(page_size or 50))
        where, args = ["1=1"], []
        if keyword:
            where.append("(s.name LIKE ? OR s.singers LIKE ? OR CAST(s.tid AS TEXT) LIKE ?)")
            args += [f"%{keyword}%"] * 3
        if lang:
            where.append("s.lang = ?")
            args.append(lang)
        st_raw = str(state or "").strip().lower()
        if st_raw in ("unmatched", "matched"):                 # 按"能否从曲库补全"筛选
            where.append("s.tid IS NULL" if st_raw == "unmatched" else "s.tid IS NOT NULL")
        else:
            st = self._norm_keep_state(state)
            if st:
                where.append("k.state = ?")
                args.append(st)
        col = self.KEEP_SORTS.get(order, "coalesce(s.hot_score,0)")
        direction = "DESC" if desc else "ASC"
        total = self.q1(f"""SELECT count(*) c FROM keeps k LEFT JOIN songs s ON s.tid=k.tid
                            WHERE {' AND '.join(where)}""", args)
        rows = self.q(
            f"""SELECT k.tid, coalesce(s.name,'') name, coalesce(s.lang,'') lang,
                       coalesce(s.album,'') album, coalesce(s.singers,'') singers,
                       coalesce(s.filename,'') filename, coalesce(s.hot_score,0) hot_score,
                       CASE WHEN s.tid IS NULL THEN 0 ELSE 1 END matched,
                       k.state, k.bytes, k.attempts, k.last_error, k.user, k.note,
                       k.added_at, k.updated_at, k.finished_at,
                       coalesce(d.state,'none') download_state
                FROM keeps k
                LEFT JOIN songs s ON s.tid=k.tid
                LEFT JOIN downloads d ON d.tid=k.tid
                WHERE {' AND '.join(where)}
                ORDER BY {col} {direction}, k.tid ASC LIMIT ? OFFSET ?""",
            list(args) + [page_size, (page - 1) * page_size])
        items = []
        for r in rows:
            d = dict(r)
            d["matched"] = bool(d["matched"])                  # 布尔：是否已能在曲库匹配到
            items.append(d)
        return {"total": total["c"] if total else 0, "page": page, "page_size": page_size,
                "items": items, "counts": self.keep_stats()}

    def keep_stats(self) -> dict:
        """整张歌曲表的统计：`{total, want, downloading, done, failed, matched, unmatched,
        bytes, remaining}`（**走 TTL 缓存**，写入 keeps/songs 会自动失效）。

        * `matched` / `unmatched`：整表里**能在曲库补全 / 曲库暂无该编号**的条数
          （= `keep_rows(state='matched'/'unmatched')` 的 total）；
        * `bytes`：`state='done'` 的字节合计；`remaining`：`total - done`（含曲库暂无的编号）；
          真正"下载驱动器要干完的量"看 `keep_remaining()`（只算能匹配到的）。
        """
        return self._cached("keep_stats", self._keep_stats_uncached)

    def _keep_stats_uncached(self) -> dict:
        counts = {s: 0 for s in self.KEEP_STATES}
        for r in self.q("SELECT state, count(*) c FROM keeps GROUP BY state"):
            counts[r["state"] if r["state"] in counts else "want"] += int(r["c"])
        b = self.q1("SELECT coalesce(sum(bytes),0) b FROM keeps WHERE state='done'")
        m = self.q1("""SELECT count(*) c FROM keeps k JOIN songs s ON s.tid=k.tid""")
        total = sum(counts.values())
        matched = int(m["c"]) if m else 0
        return {"total": total, "want": counts["want"], "downloading": counts["downloading"],
                "done": counts["done"], "failed": counts["failed"],
                "matched": matched, "unmatched": max(0, total - matched),
                "bytes": int(b["b"]) if b else 0,
                "remaining": max(0, total - counts["done"])}

    def keep_unmatched(self, limit: int = 0) -> list[dict]:
        """当前**曲库里找不到**的歌曲表编号（按编号升序，`limit=0` = 全部）。

        用于让用户核对是不是编号写错了；曲库同步进来后这些编号会自动从列表里消失
        （`matched` 变 true），无需任何额外动作。
        """
        lim = int(limit or 0)
        rows = self.q("""SELECT k.tid, k.state, k.note, k.user, k.added_at
                         FROM keeps k LEFT JOIN songs s ON s.tid=k.tid
                         WHERE s.tid IS NULL ORDER BY k.tid ASC LIMIT ?""",
                      (lim if lim > 0 else -1,))
        return [dict(r) for r in rows]

    def keep_remaining(self) -> int:
        """歌曲表里还有多少首**能匹配到曲库且没有下载完成**（批量驱动器的"收工"判据，走 TTL 缓存）。

        曲库暂无的编号（`matched=false`）没有文件名/取址信息，不参与下载，因此也不计入；
        曲库同步进来后它会自动计入。
        """
        return self._cached("keep_remaining", self._keep_remaining_uncached)

    def _keep_remaining_uncached(self) -> int:
        r = self.q1("""SELECT count(*) c FROM keeps k JOIN songs s ON s.tid=k.tid
                       WHERE coalesce(s.removed_at,0)=0 AND k.state <> 'done'""")
        return int(r["c"]) if r else 0

    # ── 补队 / 状态流转（下载链路只碰这几个方法）──
    def keep_next_batch(self, limit: int, order: str = "hot", desc: bool = True) -> list[dict]:
        """**只挑歌曲表里 `state NOT IN ('done','downloading')` 且能从曲库匹配到的歌** —— 补队来源。

        不在歌曲表里的歌永远不会被下载（这是"只下歌曲表里的"的关键）；
        曲库暂无的编号（`matched=false`）没有取址信息，也不会被补队（曲库同步后自动参与）。
        字段：tid/name/hot_score/lang（+state）。
        """
        col = self.SORTS.get(order, "s.hot_score")
        direction = "DESC" if desc else "ASC"
        rows = self.q(
            f"""SELECT s.tid, s.name, s.hot_score, s.lang, k.state
                FROM keeps k JOIN songs s ON s.tid=k.tid
                WHERE coalesce(s.removed_at,0)=0
                  AND k.state NOT IN ('done','downloading')
                ORDER BY {col} {direction}, s.tid ASC LIMIT ?""", (max(0, int(limit)),))
        return [dict(r) for r in rows]

    def keep_mark_downloading(self, tid: int) -> None:
        """抢占到任务后置为 `downloading`（attempts+1；已 done 的不动）。"""
        try:
            self.ex("""UPDATE keeps SET state='downloading', attempts=attempts+1, last_error='',
                              updated_at=? WHERE tid=? AND state<>'done'""", (now(), int(tid)))
        except Exception:
            pass

    def keep_mark_done(self, tid: int, bytes_: int = 0, path: str = "") -> None:
        """下载成功 → `done` + 字节数/路径/完成时间（保幂等，失败不抛）。"""
        try:
            ts = now()
            self.ex("""UPDATE keeps SET state='done', bytes=?, path=?, last_error='',
                              finished_at=?, updated_at=? WHERE tid=?""",
                    (int(bytes_ or 0), str(path or ""), ts, ts, int(tid)))
        except Exception:
            pass

    def keep_mark_failed(self, tid: int, error: str = "") -> None:
        """下载失败 → `failed` + 错误原因（`keep_next_batch` 仍会取它，靠重试上限兜底）。"""
        try:
            self.ex("""UPDATE keeps SET state='failed', last_error=?, updated_at=?
                       WHERE tid=? AND state<>'done'""", (str(error)[:500], now(), int(tid)))
        except Exception:
            pass

    # ── 媒体目录扫描：发现新文件 → 自动登记进歌曲表（**永不删除文件**）──
    def keep_adopt_media(self, media_dir: str, min_bytes: int = 0,
                         timeout_sec: float = 30) -> dict:
        """**扫描媒体目录，把 `<编号>.ts` 登记进歌曲表 + 把缺失的已下条目回队**
        （只增/只改状态，**绝不删除任何文件**）。

        这是"表 ↔ 文件"一致性的**唯一入口**：既负责认领（用户直接把文件放进媒体目录
        —— 迁移/手工拷贝/换机），也负责把"登记为 done 但文件已经不在/过小"的条目放回待下载。
        **发现未知文件一律登记，绝不删除**（删文件只有用户显式移除才会发生）。

        行为（逐个合法文件）：
          * 文件名必须是**纯数字 + `.ts`**，且大小 ≥ `min_bytes`，否则计入 `skipped`；
          * `keeps` 里**没有**该编号 → **新增登记**（`state='done'`、记录字节数、
            `note='媒体扫描自动登记'`、`user='scan'`）；曲库里没有该编号也照样登记
            （查询时 `matched=false`，曲库同步后自动补全）；
          * `keeps` 里有、但状态不是 `done` → 置 `done` 并补上字节数（`updated`）；
          * 已经是 `done` → 不动（不计入 adopted/updated）。

        缺失回队（**本方法只改状态，绝不删 keeps 行**）：
          * `keeps` 里状态为 `done`，但媒体文件**不存在或小于 `min_bytes`** → 状态回退为 `want`
            （`requeued`），同时把 `downloads` 的完成登记放回 `pending`（否则下载池会因 done 跳过它）；
          * 曲库匹配不到（`matched=false`）、文件还没下（`want`/`failed`）的行**原样保留**
            （允许"先加入、以后再下"是明确需求）。

        ⚠️ **阻塞容忍（部署机硬性要求）**：媒体目录在云盘/FUSE 上，`isdir`/`listdir`/`getsize`
        都可能长时间阻塞 —— 因此**全部列举动作都在一个带超时的守护线程里做**：
          * 超过 `timeout_sec` 或列举抛异常 → 立刻返回 `{"ok": false, "error": …}`，
            **不改变任何状态**（既不登记也不回队；挂载丢失不会被误判成"全部缺失"），
            调用方绝不会被卡住；
          * 超时的线程是 daemon，进程退出不受影响（FUSE 卡死的系统调用无法从 Python 侧取消）。

        返回 `{"ok": true, "scanned": n, "adopted": [tid…最多10个], "updated": n, "skipped": n,
        "requeued": n, "freed": 0, "elapsed_ms": …}`；`scanned` = 媒体目录里 `.ts` 文件总数
        （`adopted` 个数 + `updated` + `skipped` = `scanned`）；`freed` 恒为 0（本方法不删文件）。
        """
        t0 = time.monotonic()
        min_b = max(0, int(min_bytes or 0))
        try:
            timeout = float(timeout_sec)
        except Exception:
            timeout = 30.0
        if timeout <= 0:
            timeout = 30.0
        out = {"ok": False, "error": "", "scanned": 0, "adopted": [], "adopted_count": 0,
               "updated": 0, "skipped": 0, "requeued": 0, "freed": 0, "elapsed_ms": 0}

        def _elapsed() -> int:
            return int((time.monotonic() - t0) * 1000)

        box: dict[str, Any] = {}

        def _worker() -> None:
            """在**子线程**里访问媒体目录：isdir/listdir/getsize 全部受 timeout 保护。"""
            try:
                if not media_dir or not os.path.isdir(media_dir):
                    box["error"] = f"媒体目录不可用（{media_dir}）"
                    return
                box["entries"] = _scan_media_entries(media_dir)
            except Exception as exc:                       # 列举失败 = 本次放弃，不动状态
                box["error"] = f"媒体目录列举失败：{exc}"

        th = threading.Thread(target=_worker, name="media-scan-list", daemon=True)
        th.start()
        th.join(timeout)
        if th.is_alive():                                  # 超时：FUSE 卡住，绝不阻塞调用方
            out["error"] = f"媒体目录列举超时（>{timeout:g}s，媒体目录可能在云盘/FUSE 上很慢）"
            out["elapsed_ms"] = _elapsed()
            self.event("WARN", f"媒体扫描跳过：{out['error']}，未做任何改动")
            return out
        entries = box.get("entries")
        if entries is None:
            out["error"] = str(box.get("error") or "媒体目录不可用")
            out["elapsed_ms"] = _elapsed()
            self.event("WARN", f"媒体扫描跳过：{out['error']}，未做任何改动")
            return out

        out["scanned"] = len(entries)
        adopted: list[int] = []
        existing: dict[int, int] = {}                      # 目录里真实存在的合法编号 → 字节数
        for name, size in entries:
            stem = name[:-3]
            if not stem.isdigit():                         # 文件名非法 → 跳过（不删！）
                out["skipped"] += 1
                continue
            tid = int(stem)
            existing[tid] = int(size)
            if size < min_b:                               # 过小 → 不当成有效媒体（不删！）
                out["skipped"] += 1
                continue
            row = self.q1("SELECT state FROM keeps WHERE tid=?", (tid,))
            ts = now()
            if row is None:
                # **新增登记**：不在歌曲表里的文件也照样收下（曲库匹配不到就 matched=false）
                self.ex("""INSERT INTO keeps(tid,state,bytes,path,user,note,added_at,updated_at,
                                              finished_at)
                           VALUES(?,'done',?,?,'scan','媒体扫描自动登记',?,?,?)
                           ON CONFLICT(tid) DO NOTHING""",
                        (tid, int(size), os.path.join(str(media_dir), name), ts, ts, ts))
                adopted.append(tid)
            elif str(row["state"]) != "done":
                self.ex("""UPDATE keeps SET state='done', bytes=?, path=?, last_error='',
                                  finished_at=?, updated_at=? WHERE tid=?""",
                        (int(size), os.path.join(str(media_dir), name), ts, ts, tid))
                out["updated"] += 1

        # 缺失回队：`done` 但文件不在/过小 → 回到待下载（**只改状态，绝不删 keeps 行**；
        # 顺带把 downloads 的完成登记放回 pending，否则下载池会因为 done 而跳过它）。
        for row in self.q("SELECT tid FROM keeps WHERE state='done'"):
            tid = int(row["tid"])
            size = existing.get(tid, 0)
            if tid in existing and size >= min_b:
                continue
            ts = now()
            self.ex("""UPDATE keeps SET state='want', bytes=0, path='', last_error=?, updated_at=?,
                              finished_at=0 WHERE tid=?""",
                    (f"媒体扫描：文件缺失或过小（{size}B）", ts, tid))
            self.ex("""UPDATE downloads SET state='pending', attempts=0, last_error='', updated_at=?
                       WHERE tid=? AND state<>'downloading'""", (ts, tid))
            out["requeued"] += 1

        out["ok"] = True
        out["adopted"] = sorted(adopted)[:10]
        out["adopted_count"] = len(adopted)
        out["elapsed_ms"] = _elapsed()
        if adopted or out["updated"] or out["requeued"]:
            self.event("INFO", f"媒体扫描：新增登记 {len(adopted)} 首、更新 {out['updated']} 首、"
                               f"缺失回队 {out['requeued']} 首、跳过 {out['skipped']} 个"
                               f"（共 {out['scanned']} 个 .ts，用时 {out['elapsed_ms']}ms；绝不删除文件）")
        return out


    # ── 歌曲表导出 / 导入（列与曲库清单导出一致）──
    def keep_csv(self) -> str:
        """整张歌曲表导出为 CSV 文本（列与 `library_csv()` 完全一致，便于互导）。

        歌名/语言/专辑/歌手/文件名/热度**一律 LEFT JOIN 曲库现取**：曲库里能匹配到的补全，
        匹配不到的（`matched=false`）这几列为空、其余列照常导出（编号与状态绝不会丢）。
        「下载状态」列用中文可读值（待下载/下载中/已下/失败），导入时按 `KEEP_ALIASES` 解析。
        返回不含 BOM 的文本，编码由调用方决定（管理接口用 utf-8-sig 带 BOM）。
        """
        buf = io.StringIO()
        w = csv.writer(buf)
        w.writerow(self.CSV_HEADER_LIB)
        for r in self.q("""SELECT k.tid, coalesce(s.name,'') name, coalesce(s.lang,'') lang,
                                  coalesce(s.album,'') album, coalesce(s.singers,'') singers,
                                  coalesce(s.filename,'') filename, coalesce(s.hot_score,0) hot_score,
                                  k.state, k.bytes
                           FROM keeps k LEFT JOIN songs s ON s.tid=k.tid ORDER BY k.tid ASC"""):
            w.writerow([r["tid"], r["name"], r["lang"], r["album"], r["singers"], r["filename"],
                        r["hot_score"], self.KEEP_CN.get(r["state"], r["state"]), r["bytes"]])
        return buf.getvalue()

    def keep_import_csv(self, text: "str | bytes", mode: str = "merge") -> dict:
        """导入与 `keep_csv()` / `library_csv()` 同格式的 CSV（**容错**）。

        **只给「编号」一列即可用** —— 歌名/语言/专辑/歌手/文件名/热度在查询/导出时自动由曲库
        LEFT JOIN 补全（用户表里不存这些冗余列）；带了这些列但留空也照样能用。

        * `mode='merge'`：按编号合并（存在的更新状态/字节，不存在的加入；备注/用户空值保留原值）；
        * `mode='replace'`：**整表替换** —— 清空歌曲表后写入 CSV 里的编号（被替换掉的编号会清掉
          downloads 登记；媒体文件**不在这一步删** —— 删本地文件只发生在用户「显式移除」，
          避免一次误导入就把媒体目录清空）。两条硬性保证：
            1) **原子**：删旧 + 写新在同一个事务里（`replace_keeps()`），任何一步失败整体回滚，
               绝不会出现"旧的删了、新的没写进去"（那等于歌曲表整表消失）；
            2) **空 CSV 拒绝执行**：一行有效编号都没有时**不动歌曲表**，只回一条错误
               （否则误传一个只有表头的文件就会把整张表清空）。
        * **编码自动识别**：`text` 可以是 **HTTP 原始字节（bytes，推荐）** 或已解码的 `str`；
          字节按 `utf-8-sig → utf-8 → gb18030 → utf-16` 顺序尝试（Excel 中文版另存成
          GBK/GB18030 也能直接导入），失败绝不 `errors="replace"` 静默吞字节（见
          `decode_csv_bytes()` 的说明）；
        * **分隔符自动识别**：在第一个非空行里从 `,` `;` `\\t` `|` 中取出现次数最多者
          （全为 0 → 逗号），用 `csv.reader(delimiter=…)` 解析（引号字段里内嵌分隔符/换行没问题）；
        * 表头按列名定位（编号/tid/ID/歌曲编号、状态/下载状态/state、备注/note、用户/user，
          大小写/全半角/前后空白/BOM 都能认），列顺序随意、允许多余列/缺列；
          **没有表头**（首行第 1 列就是编号）也照样导入；只有「编号」一列时按"待下载"处理；
        * 行尾 `\\r\\n` / `\\n` / `\\r` 都支持；空行跳过；字段前后空白与不可见字符
          （`\\ufeff`/`\\u200b`/全角空格）清理；
        * **编号健壮性**：纯数字接受；`4100081.0` / `4.100081E+06` 这类精确整数值接受；
          被 Excel 按显示精度改写的 `4.10008E+06` 拒绝并给出中文提示（把该列设为"文本"）；
        * **曲库里暂时没有的编号也一律写入**（`matched=false`，曲库同步进来后自动补全），
          不再"曲库没有就跳过"；编号非法/为空、状态无法识别 → 计入 `skipped` 与 `errors`，
          不影响其它行；
        * `matched` / `unmatched` 统计本次写入的行里"能从曲库补全 / 曲库暂无"的条数
          （`updated == matched + unmatched`）；
        * 「字节」列不参与导入（字节数/路径由下载与媒体扫描维护，避免 CSV 写死假数据）；
        * 返回 `{"mode","updated","removed","skipped","matched","unmatched","errors",
          "replace_cancelled","detected"}`（`updated` = 写入行数，`removed` = replace 模式下被
          替换掉的旧行数，`replace_cancelled=True` = 空 CSV 触发的拒执行，
          `detected={"encoding","delimiter"}` = 本次识别到的编码与分隔符，便于前端/用户确认）。
        """
        m = "replace" if str(mode or "").strip().lower() == "replace" else "merge"
        out = {"mode": m, "updated": 0, "removed": 0, "skipped": 0,
               "matched": 0, "unmatched": 0, "errors": [], "replace_cancelled": False,
               "detected": {"encoding": "utf-8", "delimiter": ","}}

        def _err(msg: str) -> None:
            if len(out["errors"]) < 5:
                out["errors"].append(msg)

        # ★ 编码识别：**bytes 走识别逻辑**（HTTP 层必须把原始请求体传进来，见 decode_csv_bytes）
        if isinstance(text, (bytes, bytearray, memoryview)):
            raw, enc = decode_csv_bytes(bytes(text))
        else:
            raw, enc = str(text or ""), "utf-8"
        delim = detect_csv_delimiter(raw)
        out["detected"] = {"encoding": enc, "delimiter": delim}
        # 行尾统一成 \n（csv.reader 本身也能吃 \r\n，但统一后可让"第一个非空行"与报错行号稳定）
        raw = raw.replace("\r\n", "\n").replace("\r", "\n")
        if not raw.strip(CSV_BLANK):
            _err("CSV 内容为空")
            return out
        try:
            rows = list(csv.reader(io.StringIO(raw), delimiter=delim))
        except Exception as exc:                                   # pragma: no cover - 兜底
            _err(f"CSV 解析失败：{exc}")
            return out
        if not rows:
            _err("CSV 内容为空")
            return out

        idx: dict[str, int] = {}
        for i, h in enumerate(rows[0]):
            key = norm_csv_header(h)
            if key in CSV_TID_HEADERS and "tid" not in idx:
                idx["tid"] = i
            elif key in CSV_STATE_HEADERS and "state" not in idx:
                idx["state"] = i
            elif key in CSV_NOTE_HEADERS and "note" not in idx:
                idx["note"] = i
            elif key in CSV_USER_HEADERS and "user" not in idx:
                idx["user"] = i
        body, body_ln = rows[1:], 2
        if "tid" not in idx:
            # 无表头：第一行就是编号（跳过前面的空行 —— Excel 另存偶尔会在最前面留一个空行）
            start = next((i for i, r in enumerate(rows) if any(clean_csv_cell(c) for c in r)), None)
            first, _ = parse_keep_tid(rows[start][0]) if start is not None else (None, "")
            if first is not None:
                idx["tid"] = 0                                     # 无表头：第 1 列 = 编号
                body, body_ln = rows[start:], start + 1
            else:
                _err("缺少「编号」列（首行应为表头，或第 1 列为编号）")
                return out

        parsed: list[tuple[int, str, str, str]] = []
        seen: set[int] = set()
        for ln, row in enumerate(body, start=body_ln):
            if not row or not any(clean_csv_cell(c) for c in row):
                continue

            def cell(k: str) -> str:
                return clean_csv_cell(row[idx[k]]) if (k in idx and idx[k] < len(row)) else ""

            tidtxt = cell("tid")
            tid, why = parse_keep_tid(tidtxt)
            if tid is None:
                out["skipped"] += 1
                _err(f"第 {ln} 行编号无效：{tidtxt[:20]!r}（{why}）")
                continue
            if tid in seen:
                continue
            if "state" in idx:
                txt = cell("state")
                st = "want" if not txt else self._norm_keep_state(txt, default="?")
                if st == "?":
                    out["skipped"] += 1
                    _err(f"第 {ln} 行状态无法识别：{cell('state')[:20]!r}")
                    continue
            else:
                st = "want"
            seen.add(tid)
            parsed.append((tid, st, cell("note"), cell("user")))

        # 统计能否从曲库补全（**分批查**：SQLite 的 SQL 变量数有上限，
        # 一次性 IN (69 万个 ?) 会直接报 "too many SQL variables"）
        all_tids = [p[0] for p in parsed]
        known = self._tids_present("songs", all_tids)
        out["matched"] = len([t for t in all_tids if t in known])
        out["unmatched"] = len(all_tids) - out["matched"]

        # ★ 空 CSV（只有表头 / 编号全非法 / 状态全不识别）**绝不执行整表替换**：
        #   否则一次误传就把歌曲表清空，且旧行无法恢复（用户看到的"条目莫名消失"）。
        if m == "replace" and not parsed:
            out["replace_cancelled"] = True
            _err("CSV 里没有可导入的有效编号：整表替换已取消，歌曲表未改动")
            self.event("WARN", "歌曲表导入（replace）已取消：CSV 里没有有效编号（歌曲表未改动）")
            return out

        ts = now()
        payload = [(tid, st, 0, user, note, ts, ts, ts if st == "done" else 0)
                   for tid, st, note, user in parsed]
        if m == "replace":
            old = [int(r["tid"]) for r in self.q("SELECT tid FROM keeps")]
            new_tids = set(all_tids)
            dropped = [t for t in old if t not in new_tids]
            out["removed"] = len(old)
            self.replace_keeps(payload, dropped)                   # ★ 原子：失败整体回滚
            out["updated"] = len(payload)
        elif payload:
            self.exmany(self.KEEP_UPSERT_SQL, payload)
            out["updated"] = len(payload)
        self.event("INFO", f"歌曲表导入（{m}）：写入 {out['updated']} 首、替换掉 {out['removed']} 首、"
                           f"跳过 {out['skipped']} 行、曲库可补全 {out['matched']}、"
                           f"曲库暂无此编号 {out['unmatched']}")
        return out

    def replace_keeps(self, payload: list[tuple], dropped: list[int]) -> None:
        """**原子整表替换**（`keep_import_csv(mode='replace')` 的唯一写入入口）。

        一个事务里做完：清空 `keeps` → 清掉被替换编号的 `downloads` 登记 → 写入新行。
        任何异常都会整体回滚（旧数据原样保留）—— 这是"导入失败也不会丢歌曲表"的关键。
        媒体文件**不在这里删**（删本地文件只发生在「显式移除」），避免一次误导入清空媒体目录。
        """
        with self.tx() as conn:
            conn.execute("DELETE FROM keeps")
            for i in range(0, len(dropped), 400):
                chunk = dropped[i:i + 400]
                conn.execute("DELETE FROM downloads WHERE tid IN (%s)" % ",".join("?" * len(chunk)),
                             tuple(chunk))
            self._write_keeps_rows(conn, payload)

    def _write_keeps_rows(self, conn, payload: list[tuple]) -> None:
        """在事务连接上写入歌曲表行（单独成方法：测试可覆写它来验证"写入失败=整体回滚"）。"""
        conn.executemany(self.KEEP_UPSERT_SQL, payload)

    def _tids_present(self, table: str, ids: list[int]) -> set[int]:
        """返回 `ids` 里**存在于 `table`** 的编号集合（分批查，避免超长 SQL 参数）。"""
        out: set[int] = set()
        if not ids:
            return out
        for i in range(0, len(ids), 400):
            chunk = ids[i:i + 400]
            rows = self.q("SELECT tid FROM %s WHERE tid IN (%s)" % (table, ",".join("?" * len(chunk))),
                          chunk)
            out.update(int(r["tid"]) for r in rows)
        return out


    # ── 入队 / 去重 ──
    def enqueue(self, tids: list[int], skip_done: bool = True, batch: str = "") -> dict:
        """把歌曲加入下载队列。**已 done 且文件仍在的直接跳过**（避免重复下载）。"""
        added, skipped_done, skipped_none = 0, 0, 0
        if batch:
            self.ex("INSERT INTO batches(name,order_by,total,created_at) VALUES(?,?,?,?)",
                    (batch, "", len(tids), now()))
        for tid in tids:
            row = self.q1("SELECT state FROM downloads WHERE tid=?", (tid,))
            if self.q1("SELECT 1 FROM songs WHERE tid=?", (tid,)) is None:
                skipped_none += 1
                continue
            if row is None:
                self.ex("""INSERT INTO downloads(tid,state,enqueued_at,updated_at) VALUES(?,?,?,?)
                           ON CONFLICT(tid) DO NOTHING""", (tid, "pending", now(), now()))
                added += 1
            elif row["state"] == "done" and skip_done:
                skipped_done += 1
            else:
                self.ex("""UPDATE downloads SET state='pending', last_error='', updated_at=?
                           WHERE tid=? AND state NOT IN ('downloading')""", (now(), tid))
                added += 1
        if added or skipped_done or skipped_none:
            self.event("INFO", f"入队 {added} 首（跳过已完成 {skipped_done}，无此编号 {skipped_none}）")
        return {"added": added, "skipped_done": skipped_done, "skipped_unknown": skipped_none}

    def push_enqueue(self, tid: int, url: str, source: str = "app-push",
                     user: str = "app", note: str = "App 点歌推送") -> dict:
        """**App 点歌推送**入队：登记歌曲表 + 用推送来的固定地址入队下载（不走曲库匹配）。

        与 `enqueue()` 的区别（都是刻意的）：
          * **不要求编号在曲库（`songs`）里**：曲库里没有也允许登记（`matched=false`），
            因为地址是 App 直接给的，下载不需要曲库元数据；
          * 歌曲表只在**缺行时**插入 `state='want'`（`ON CONFLICT DO NOTHING`）—— 已在表里的行
            状态一律不动（与 `keep_add()` 一致，避免把"已下完"的歌重新拉回待下载）；
          * `downloads` 写入 `push_url`（下载器据此跳过取址）与 `source='app-push'`，并把
            `attempts` 归零 —— 否则"已达重试上限"的旧行会因 `claim_one` 的 attempts 上限永远抢不到；
          * 正在下载（`downloading`）的行不打断（与 `enqueue()` 同一条保护）。

        返回 `{"keeps_added": bool, "queued": bool}`；任何异常都往上抛（调用方=接口层）。
        """
        t = int(tid)
        ts = now()
        # 「登记歌曲表 + 带固定地址入队」放同一个事务：不允许出现"登记了却没入队"的中间态
        # （⚠️ 块内直接用 conn.execute，不能再走 ex/exmany —— 它们会提前 commit）。
        with self.tx() as conn:
            cur_keeps = conn.execute("""INSERT INTO keeps(tid,state,user,note,added_at,updated_at)
                                        VALUES(?,'want',?,?,?,?) ON CONFLICT(tid) DO NOTHING""",
                                     (t, str(user or ""), str(note or ""), ts, ts))
            cur_dl = conn.execute("""INSERT INTO downloads(tid,state,push_url,source,attempts,enqueued_at,updated_at)
                                     VALUES(?,'pending',?,?,0,?,?)
                                     ON CONFLICT(tid) DO UPDATE SET
                                       state='pending', push_url=excluded.push_url, source=excluded.source,
                                       attempts=0, last_error='', started_at=0, updated_at=excluded.updated_at
                                     WHERE downloads.state <> 'downloading'""",
                                  (t, str(url or ""), str(source or ""), ts, ts))
            keeps_added = cur_keeps.rowcount == 1
            queued = cur_dl.rowcount == 1
        return {"keeps_added": keeps_added, "queued": queued}

    def push_url_clear(self, tid: int) -> None:
        """清掉队列项上的固定地址（推送地址失败并已回退取址时调用）。

        只清 `push_url`、**不动** `source`（保留"这条来自 App 推送"的溯源信息）。
        """
        self.ex("UPDATE downloads SET push_url='', updated_at=? WHERE tid=?", (now(), int(tid)))

    def next_pending(self, limit: int, order: str = "hot", desc: bool = True,
                     max_attempts: int = 3, backoff_sec: int = 20) -> list[dict]:
        """取下一批可下载的任务：**必须尊重重试上限与退避窗口**。

        ⚠️ 旧实现写死 `attempts < 999` 且抢占不检查上限 → 下载失败的任务被无限次立刻重领，
        2 分钟打了上游 504 次取址（2026-10-10 部署实测事故）。

        `songs` 改为 **LEFT JOIN**：常规任务仍要求编号在曲库里（`s.tid IS NOT NULL`），
        而 **App 点歌推送**的任务自带 `push_url`（下载不需要曲库元数据），曲库里没有也照下；
        返回值多了 `push_url` / `source` 两列，供下载器判断"固定地址优先"。
        """
        col = self.SORTS.get(order, "s.hot_score")
        direction = "DESC" if desc else "ASC"
        rows = self.q(
            f"""SELECT d.tid, s.filename, s.name, s.lang, coalesce(s.hot_score,0) AS hot_score,
                       d.push_url, d.source
                FROM downloads d LEFT JOIN songs s ON s.tid=d.tid
                WHERE d.state IN ('pending','failed') AND d.attempts < ?
                  AND (d.attempts = 0 OR d.updated_at = 0 OR d.updated_at <= ?)
                  AND (s.tid IS NOT NULL OR d.push_url <> '')
                ORDER BY {col} {direction}, d.tid ASC LIMIT ?""",
            (max_attempts, now() - max(0, backoff_sec), limit))
        return [dict(r) for r in rows]

    def claim_one(self, tid: int, max_attempts: int = 3, backoff_sec: int = 20) -> bool:
        """**原子抢占**：只有把 state 从非 downloading/done 改成 downloading 成功的线程才有权下载。

        同时守住两条上限（缺一就会变成打上游的死循环）：
          * `attempts < max_attempts`（超过就永远不再抢，状态停在 failed 等人工重试）；
          * `updated_at <= now - backoff`（**仅对重试生效**；刚入队的新任务 attempts=0 立即可下）。
        """
        n = self.ex(
            """UPDATE downloads SET state='downloading', attempts=attempts+1, started_at=?, updated_at=?,
                                      last_error=''
               WHERE tid=? AND state IN ('pending','failed') AND attempts < ?
                 AND (attempts = 0 OR updated_at = 0 OR updated_at <= ?)""",
            (now(), now(), tid, max_attempts, now() - max(0, backoff_sec)))
        return n == 1

    def finish(self, tid: int, ok: bool, bytes_: int = 0, path: str = "", md5: str = "",
               url: str = "", source_host: str = "", error: str = "") -> None:
        if ok:
            self.ex("""UPDATE downloads SET state='done', bytes=?, path=?, md5=?, url=?, source_host=?,
                              finished_at=?, updated_at=?, last_error='' WHERE tid=?""",
                    (bytes_, path, md5, url, source_host, now(), now(), tid))
        else:
            # 失败 → 回到 pending（还有重试机会时）或 failed
            row = self.q1("SELECT attempts FROM downloads WHERE tid=?", (tid,))
            attempts = row["attempts"] if row else 0
            state = "pending" if attempts < int(self.get_meta("max_attempts", "3")) else "failed"
            self.ex("""UPDATE downloads SET state=?, last_error=?, updated_at=?, finished_at=? WHERE tid=?""",
                    (state, str(error)[:500], now(), now(), tid))

    def set_state(self, tid: int, state: str, error: str = "") -> None:
        self.ex("UPDATE downloads SET state=?, last_error=?, updated_at=? WHERE tid=?",
                (state, str(error)[:500], now(), tid))

    def reset_stuck(self) -> int:
        """进程重启后把卡在 downloading 的任务放回队列。

        歌曲表（keeps）里卡在 `downloading` 的条目同样要放回 `want` ——
        否则它既不会被补队（`state NOT IN ('done','downloading')`），也永远不会被重下。
        """
        n = self.ex("UPDATE downloads SET state='pending', updated_at=? WHERE state='downloading'", (now(),))
        if n:
            self.event("WARN", f"启动清理：{n} 个中断的下载任务已放回队列")
        try:
            k = self.ex("UPDATE keeps SET state='want', updated_at=? WHERE state='downloading'", (now(),))
            if k:
                self.event("WARN", f"启动清理：{k} 个中断的歌曲表条目已放回待下载")
        except Exception:
            pass
        return n

    def failed_list(self, limit: int = 200) -> list[dict]:
        return [dict(r) for r in self.q(
            "SELECT tid,attempts,last_error,updated_at FROM downloads WHERE state='failed' "
            "ORDER BY updated_at DESC LIMIT ?", (limit,))]

    def downloading_list(self) -> list[dict]:
        return [dict(r) for r in self.q(
            """SELECT d.tid,s.name,d.bytes,d.expected_bytes,d.started_at,d.attempts
               FROM downloads d JOIN songs s ON s.tid=d.tid WHERE d.state='downloading'
               ORDER BY d.started_at DESC""")]

    def retry_failed(self, max_attempts: int = 3) -> int:
        # 也解救"已达重试上限卡住"的任务（进程重启后 attempts 仍是旧值，否则永远抢不到）
        n = self.ex("""UPDATE downloads SET state='pending', attempts=0, last_error='', updated_at=?
                       WHERE state IN ('failed','missing')
                          OR (state = 'pending' AND attempts >= ?)""", (now(), max_attempts))
        if n:
            self.event("INFO", f"失败/缺失任务重置为待下载：{n} 首")
        return n