# -*- coding: utf-8 -*- """SQLite schema 迁移。 规则(DEVELOPMENT.md §6.3): * 使用 ``schema_version(version, applied_at)`` 记录已应用版本; * 每个版本包在单一事务里,失败整体回滚; * 只向前追加,不允许删库重建; * 升级已有数据库前自动创建 SQLite 备份。 """ from __future__ import annotations import sqlite3 import time from dataclasses import dataclass from pathlib import Path from typing import Sequence @dataclass(frozen=True) class Migration: version: int name: str statements: tuple[str, ...] MIGRATIONS: tuple[Migration, ...] = ( Migration( version=1, name="initial schema", statements=( """ CREATE TABLE video ( aid INTEGER PRIMARY KEY, bvid TEXT UNIQUE, title TEXT NOT NULL, up_mid INTEGER, up_name TEXT, pubdate INTEGER NOT NULL, duration INTEGER, cover TEXT, folder_id INTEGER, fav_time INTEGER, status TEXT NOT NULL, source TEXT, first_seen INTEGER NOT NULL, last_sync INTEGER, attr INTEGER ) """, "CREATE INDEX idx_video_status ON video(status, pubdate DESC)", "CREATE INDEX idx_video_up ON video(up_mid, pubdate DESC)", """ CREATE TABLE snapshot ( stat_date TEXT NOT NULL, aid INTEGER NOT NULL, view INTEGER, like INTEGER, favorite INTEGER, coin INTEGER, share INTEGER, reply INTEGER, danmaku INTEGER, captured_at INTEGER NOT NULL, ok INTEGER NOT NULL DEFAULT 1, err TEXT, src TEXT DEFAULT 'view', PRIMARY KEY (stat_date, aid) ) """, "CREATE INDEX idx_snap_aid ON snapshot(aid, stat_date)", """ CREATE TABLE folder ( media_id INTEGER PRIMARY KEY, year INTEGER NOT NULL, title TEXT, media_count INTEGER, last_ids_at INTEGER, last_full_at INTEGER, enabled INTEGER NOT NULL DEFAULT 1 ) """, """ CREATE TABLE folder_resource ( media_id INTEGER NOT NULL, aid INTEGER NOT NULL, first_seen INTEGER NOT NULL, active INTEGER NOT NULL DEFAULT 1, last_seen INTEGER, PRIMARY KEY (media_id, aid) ) """, "CREATE INDEX idx_folder_resource_aid ON folder_resource(aid, active)", """ CREATE TABLE up_blacklist ( mid INTEGER PRIMARY KEY, name TEXT, reason TEXT, created_at INTEGER NOT NULL ) """, """ CREATE TABLE up_stat ( stat_date TEXT NOT NULL, mid INTEGER NOT NULL, follower INTEGER, following INTEGER, captured_at INTEGER NOT NULL, ok INTEGER NOT NULL DEFAULT 1, PRIMARY KEY (stat_date, mid) ) """, """ CREATE TABLE candidate ( aid INTEGER PRIMARY KEY, bvid TEXT, title TEXT, up_mid INTEGER, up_name TEXT, pubdate INTEGER, cover TEXT, score REAL, channel TEXT, first_seen INTEGER NOT NULL, notified_at INTEGER, decision TEXT, decided_at INTEGER ) """, """ CREATE TABLE seen_aid ( aid INTEGER PRIMARY KEY, first_seen INTEGER NOT NULL, source TEXT ) """, """ CREATE TABLE notify ( aid INTEGER NOT NULL, channel TEXT NOT NULL, batch_id TEXT, sent_at INTEGER NOT NULL, ok INTEGER NOT NULL, err TEXT, PRIMARY KEY (aid, channel) ) """, """ CREATE TABLE fav_action ( token TEXT PRIMARY KEY, aid INTEGER NOT NULL, folder_id INTEGER NOT NULL, state TEXT NOT NULL, created_at INTEGER NOT NULL, expires_at INTEGER NOT NULL, done_at INTEGER, position INTEGER, err TEXT ) """, """ CREATE TABLE kv ( k TEXT PRIMARY KEY, v TEXT ) """, ), ), Migration( version=2, name="folder_resource fav_time", statements=( "ALTER TABLE folder_resource ADD COLUMN fav_time INTEGER", ), ), Migration( version=3, name="candidate stats", statements=( "ALTER TABLE candidate ADD COLUMN duration INTEGER", "ALTER TABLE candidate ADD COLUMN play INTEGER", "ALTER TABLE candidate ADD COLUMN like INTEGER", "ALTER TABLE candidate ADD COLUMN favorite INTEGER", "ALTER TABLE candidate ADD COLUMN danmaku INTEGER", "ALTER TABLE candidate ADD COLUMN review INTEGER", ), ), ) class MigrationError(RuntimeError): """迁移失败,原事务已回滚。""" def _schema_version_table(conn: sqlite3.Connection) -> None: conn.execute( "CREATE TABLE IF NOT EXISTS schema_version (" "version INTEGER PRIMARY KEY, applied_at INTEGER NOT NULL)" ) def current_version(conn: sqlite3.Connection) -> int: _schema_version_table(conn) row = conn.execute("SELECT MAX(version) FROM schema_version").fetchone() return int(row[0] or 0) if row else 0 def _user_tables(conn: sqlite3.Connection) -> list[str]: rows = conn.execute( "SELECT name FROM sqlite_master " "WHERE type = 'table' AND name NOT LIKE 'sqlite_%' AND name != 'schema_version'" ).fetchall() return [str(row[0]) for row in rows] def backup_database( conn: sqlite3.Connection, backup_dir: str | Path, *, version: int, now: float | None = None, ) -> Path | None: """用 SQLite backup API 生成一致性备份;内存库返回 None。""" try: row = conn.execute("PRAGMA database_list").fetchone() except sqlite3.Error: return None if not row or not row[2] or str(row[2]) == ":memory:": return None target_dir = Path(backup_dir) target_dir.mkdir(parents=True, exist_ok=True) stamp = time.strftime("%Y%m%d%H%M%S", time.localtime(now)) target = target_dir / f"app-pre-v{version}-{stamp}.db" suffix = 1 while target.exists(): target = target_dir / f"app-pre-v{version}-{stamp}-{suffix}.db" suffix += 1 dst = sqlite3.connect(target) try: conn.backup(dst) finally: dst.close() return target def prune_backups( backup_dir: str | Path, *, keep_days: int, now: float | None = None ) -> list[Path]: """删除超过保留期的迁移备份。""" target_dir = Path(backup_dir) if not target_dir.is_dir(): return [] cutoff = (time.time() if now is None else now) - keep_days * 86400 removed: list[Path] = [] for path in target_dir.glob("app-pre-v*.db"): try: if path.stat().st_mtime < cutoff: path.unlink() removed.append(path) except OSError: continue return removed def _validate_migrations(migrations: Sequence[Migration]) -> None: versions = [m.version for m in migrations] if versions != sorted(versions): raise ValueError("迁移版本必须按升序排列") if len(set(versions)) != len(versions): raise ValueError("迁移版本不能重复") if versions and versions[0] < 1: raise ValueError("迁移版本必须从 1 开始") def run_migrations( conn: sqlite3.Connection, *, backup_dir: str | Path | None = None, backup_keep_days: int = 14, now: float | None = None, migrations: Sequence[Migration] = MIGRATIONS, logger=None, ) -> list[int]: """应用所有待执行迁移,返回本次应用的版本列表。""" _validate_migrations(migrations) _schema_version_table(conn) version = current_version(conn) pending = [m for m in migrations if m.version > version] if not pending: return [] if backup_dir is not None and _user_tables(conn): backup = backup_database(conn, backup_dir, version=version, now=now) prune_backups(backup_dir, keep_days=backup_keep_days) if logger is not None and backup is not None: logger.info("db.backup", path=str(backup), version=version) applied: list[int] = [] for migration in pending: try: conn.execute("BEGIN IMMEDIATE") for statement in migration.statements: conn.execute(statement) conn.execute( "INSERT INTO schema_version(version, applied_at) VALUES (?, ?)", (migration.version, int(time.time() if now is None else now)), ) conn.execute("COMMIT") except Exception as exc: # noqa: BLE001 任何错误都必须回滚 try: conn.execute("ROLLBACK") except sqlite3.Error: pass raise MigrationError( f"迁移 v{migration.version} ({migration.name}) 失败,已回滚: {exc}" ) from exc applied.append(migration.version) if logger is not None: logger.info("db.migrated", version=migration.version, name=migration.name) return applied