feat(database): add managed backup and offline restore (#6359)

Author: InfinityPacer

app/adapters/system/backup/__init__.py

@@ -0,0 +1 @@
+"""数据库备份的文件系统与数据库技术适配器命名空间。"""

app/adapters/system/backup/database.py

@@ -0,0 +1,209 @@
+"""基于活动 SQLAlchemy 引擎的 SQLite 与 PostgreSQL 备份实现。"""
+
+from __future__ import annotations
+
+import os
+import shutil
+import sqlite3
+import subprocess
+from contextlib import closing
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Callable, Mapping, Protocol, Sequence
+
+from sqlalchemy.engine import Engine
+
+
+@dataclass(frozen=True, slots=True)
+class DatabaseBackupCheck:
+    """数据库适配器返回的基础校验结果。"""
+
+    valid: bool
+    method: str
+    detail: str | None = None
+
+
+class ProcessResult(Protocol):
+    """数据库命令执行结果的最小合同。"""
+
+    returncode: int
+    stdout: str
+    stderr: str
+
+
+class ProcessRunner(Protocol):
+    """可替换的数据库命令执行边界。"""
+
+    def __call__(
+        self,
+        command: Sequence[str],
+        *,
+        env: Mapping[str, str],
+        capture_output: bool,
+        text: bool,
+        check: bool,
+    ) -> ProcessResult:
+        """执行命令并返回结果。"""
+
+
+class SQLiteBackupBackend:
+    """使用 SQLite 在线备份 API 管理活动文件数据库。"""
+
+    db_type = "sqlite"
+    suffix = ".db"
+
+    def __init__(self, engine: Engine) -> None:
+        self._engine = engine
+        database = engine.url.database
+        if not database or database == ":memory:":
+            raise ValueError("SQLite 内存数据库不支持文件备份")
+        self._database = Path(database)
+
+    def create(self, destination: Path) -> None:
+        """从活动引擎指向的 SQLite 文件创建一致快照。"""
+        source = self._engine.raw_connection()
+        try:
+            with closing(sqlite3.connect(destination)) as target:
+                source.driver_connection.backup(target)
+                target.commit()
+        finally:
+            source.close()
+
+    def verify(self, artifact: Path) -> DatabaseBackupCheck:
+        """通过 SQLite integrity_check 校验备份内容。"""
+        method = "PRAGMA integrity_check"
+        try:
+            # 已发布前的临时快照不会再变化；immutable 避免 WAL 模式为只读校验创建旁路文件。
+            uri = f"{artifact.resolve().as_uri()}?mode=ro&immutable=1"
+            with closing(sqlite3.connect(uri, uri=True)) as connection:
+                rows = connection.execute("PRAGMA integrity_check").fetchall()
+        except sqlite3.Error as error:
+            return DatabaseBackupCheck(False, method, str(error))
+        valid = bool(rows) and all(row[0] == "ok" for row in rows)
+        detail = None if valid else "; ".join(str(row[0]) for row in rows)
+        return DatabaseBackupCheck(valid, method, detail)
+
+    def restore(self, artifact: Path) -> None:
+        """在 CLI 离线进程中原子替换活动 SQLite 文件。"""
+        temporary = self._database.with_name(f".{self._database.name}.restore")
+        self._engine.dispose()
+        try:
+            shutil.copy2(artifact, temporary)
+            temporary.chmod(0o600)
+            self._database.with_name(f"{self._database.name}-wal").unlink(missing_ok=True)
+            self._database.with_name(f"{self._database.name}-shm").unlink(missing_ok=True)
+            os.replace(temporary, self._database)
+        finally:
+            temporary.unlink(missing_ok=True)
+
+
+class PostgreSQLBackupBackend:
+    """使用 pg_dump 与 pg_restore 管理活动 PostgreSQL 数据库。"""
+
+    db_type = "postgresql"
+    suffix = ".dump"
+
+    def __init__(
+        self,
+        engine: Engine,
+        *,
+        runner: ProcessRunner = subprocess.run,
+        tool_resolver: Callable[[str], str | None] = shutil.which,
+        pg_dump: str = "pg_dump",
+        pg_restore: str = "pg_restore",
+    ) -> None:
+        self._engine = engine
+        self._runner = runner
+        self._tool_resolver = tool_resolver
+        self._pg_dump = pg_dump
+        self._pg_restore = pg_restore
+
+    def create(self, destination: Path) -> None:
+        """创建 PostgreSQL custom-format 在线备份。"""
+        command = [
+            self._require_tool(self._pg_dump),
+            "--format=custom",
+            "--no-owner",
+            "--no-acl",
+            "--file",
+            str(destination),
+            *self._connection_arguments(),
+        ]
+        result = self._run(command, include_password=True)
+        if result.returncode != 0:
+            raise RuntimeError(f"pg_dump 执行失败，退出码 {result.returncode}")
+        if not destination.is_file() or destination.stat().st_size == 0:
+            raise RuntimeError("pg_dump 未生成有效的备份文件")
+
+    def verify(self, artifact: Path) -> DatabaseBackupCheck:
+        """通过 pg_restore 目录读取校验 custom-format 归档。"""
+        method = "pg_restore --list"
+        result = self._run(
+            [self._require_tool(self._pg_restore), "--list", str(artifact)],
+            include_password=False,
+        )
+        valid = result.returncode == 0 and bool(result.stdout.strip())
+        detail = None if valid else f"pg_restore 退出码 {result.returncode}"
+        return DatabaseBackupCheck(valid, method, detail)
+
+    def restore(self, artifact: Path) -> None:
+        """在 CLI 离线进程中覆盖当前 PostgreSQL 数据库内容。"""
+        command = [
+            self._require_tool(self._pg_restore),
+            "--clean",
+            "--if-exists",
+            "--no-owner",
+            "--no-acl",
+            "--single-transaction",
+            "--exit-on-error",
+            *self._connection_arguments(),
+            str(artifact),
+        ]
+        result = self._run(command, include_password=True)
+        if result.returncode != 0:
+            raise RuntimeError(f"pg_restore 执行失败，退出码 {result.returncode}")
+
+    def _connection_arguments(self) -> list[str]:
+        url = self._engine.url
+        host = str(url.query.get("host") or url.host or "")
+        port = str(url.query.get("port") or url.port or "")
+        arguments = [
+            "--username",
+            str(url.username or ""),
+            "--dbname",
+            str(url.database or ""),
+        ]
+        if host:
+            arguments.extend(["--host", host])
+        if port:
+            arguments.extend(["--port", port])
+        return arguments
+
+    def _run(self, command: Sequence[str], *, include_password: bool) -> ProcessResult:
+        return self._runner(
+            command,
+            env=self._environment(include_password=include_password),
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+
+    def _require_tool(self, executable: str) -> str:
+        resolved = self._tool_resolver(executable)
+        if resolved is None:
+            raise RuntimeError(
+                f"未找到 {executable}，请安装与服务端同主版本或更高的 "
+                "PostgreSQL client 并加入 PATH"
+            )
+        return resolved
+
+    def _environment(self, *, include_password: bool) -> dict[str, str]:
+        environment = dict(os.environ)
+        environment.pop("PGPASSWORD", None)
+        environment.pop("PGSSLMODE", None)
+        if include_password and self._engine.url.password:
+            environment["PGPASSWORD"] = str(self._engine.url.password)
+        sslmode = self._engine.url.query.get("sslmode")
+        if sslmode:
+            environment["PGSSLMODE"] = str(sslmode)
+        return environment

app/adapters/system/backup/files.py

@@ -0,0 +1,116 @@
+"""数据库备份单文件的受限文件系统操作。"""
+
+from __future__ import annotations
+
+import os
+import re
+import tempfile
+from datetime import datetime
+from pathlib import Path
+
+
+_BACKUP_NAME = re.compile(
+    r"^(?P<db_type>sqlite|postgresql)_"
+    r"(?P<timestamp>\d{8}_\d{6})"
+    r"(?:_(?P<sequence>\d+))?"
+    r"(?P<suffix>\.db|\.dump)$"
+)
+
+
+class BackupFiles:
+    """把备份文件操作限制在一个私有根目录内。"""
+
+    def __init__(self, root: Path) -> None:
+        self.root = Path(root)
+
+    def create_temporary(self, suffix: str) -> Path:
+        """在最终目录内创建私有临时文件，保证发布可使用原子替换。"""
+        self._ensure_root()
+        descriptor, filename = tempfile.mkstemp(
+            prefix=".database-",
+            suffix=f"{suffix}.partial",
+            dir=self.root,
+        )
+        os.close(descriptor)
+        path = Path(filename)
+        path.chmod(0o600)
+        return path
+
+    def publish(self, temporary: Path, name: str) -> Path:
+        """把已校验临时文件发布为正式备份文件。"""
+        destination = self._resolve_name(name, require_exists=False)
+        os.replace(temporary, destination)
+        destination.chmod(0o600)
+        return destination
+
+    def discard(self, temporary: Path) -> None:
+        """清理本次操作拥有的未发布临时文件。"""
+        path = Path(temporary)
+        if path.parent == self.root and path.name.startswith(".database-"):
+            path.unlink(missing_ok=True)
+
+    def list(self) -> list[Path]:
+        """返回当前根目录内格式合法的正式备份文件。"""
+        if not self.root.is_dir():
+            return []
+        paths = [
+            path
+            for path in self.root.iterdir()
+            if path.is_file() and _BACKUP_NAME.fullmatch(path.name)
+        ]
+        return sorted(paths, key=lambda path: (self.created_at(path.name), path.name), reverse=True)
+
+    def resolve(self, name: str) -> Path:
+        """按受限文件名解析一个必须存在的备份文件。"""
+        return self._resolve_name(name, require_exists=True)
+
+    def delete(self, name: str) -> None:
+        """删除一个已通过名称约束的备份文件。"""
+        self.resolve(name).unlink()
+
+    def available_name(self, *, db_type: str, created_at: datetime, suffix: str) -> str:
+        """生成包含数据库类型和秒级时间的简短可读文件名。"""
+        timestamp = created_at.strftime("%Y%m%d_%H%M%S")
+        base = f"{db_type}_{timestamp}"
+        candidate = f"{base}{suffix}"
+        sequence = 1
+        while (self.root / candidate).exists():
+            candidate = f"{base}_{sequence}{suffix}"
+            sequence += 1
+        if not _BACKUP_NAME.fullmatch(candidate):
+            raise ValueError("数据库备份文件名无效")
+        return candidate
+
+    @staticmethod
+    def database_type(name: str) -> str:
+        """从受管文件名读取数据库类型。"""
+        return BackupFiles._match(name).group("db_type")
+
+    @staticmethod
+    def created_at(name: str) -> datetime:
+        """从受管文件名读取本地创建时间。"""
+        return datetime.strptime(
+            BackupFiles._match(name).group("timestamp"),
+            "%Y%m%d_%H%M%S",
+        )
+
+    def _ensure_root(self) -> None:
+        self.root.mkdir(parents=True, exist_ok=True, mode=0o700)
+        self.root.chmod(0o700)
+
+    def _resolve_name(self, name: str, *, require_exists: bool) -> Path:
+        normalized = str(name).strip()
+        self._match(normalized)
+        if Path(normalized).name != normalized:
+            raise ValueError("数据库备份文件名不能包含路径")
+        path = self.root / normalized
+        if require_exists and not path.is_file():
+            raise FileNotFoundError(normalized)
+        return path
+
+    @staticmethod
+    def _match(name: str) -> re.Match[str]:
+        matched = _BACKUP_NAME.fullmatch(str(name))
+        if matched is None:
+            raise ValueError("数据库备份文件名无效")
+        return matched

app/api/endpoints/system.py

@@ -53,6 +53,7 @@
 )
 from app.application.messaging.message import MessageHelper
 from app.runtime.progress import AsyncProgressHelper
+from app.runtime.scheduling import TimerUtils
 from app.application.rules import RuleHelper
 from app.adapters.external.server import MoviePilotServerHelper
 from app.runtime.state import SystemHelper
@@ -86,6 +87,13 @@
 _PUBLIC_SETTINGS_KEYS = {"PLUGIN_MARKET"}
 _LOG_DOWNLOAD_LIMIT = 10
 _LOG_DOWNLOAD_NAME_PATTERN = re.compile(r"^[A-Za-z0-9_-]+$")
+_DATABASE_BACKUP_SETTING_KEYS = {
+    "DB_BACKUP_ENABLE",
+    "DB_BACKUP_CRON",
+    "DB_BACKUP_PATH",
+    "DB_BACKUP_RETENTION_DAYS",
+    "DB_BACKUP_MAX_COUNT",
+}
 
 
 def _validate_llm_server_tool_config(env: dict) -> Optional[str]:
@@ -129,6 +137,38 @@ def _validate_llm_server_tool_config(env: dict) -> Optional[str]:
     )
 
 
+def _validate_database_backup_config(env: dict) -> Optional[str]:
+    """在批量写入前校验数据库备份策略，避免只保存部分字段。"""
+    if not _DATABASE_BACKUP_SETTING_KEYS.intersection(env):
+        return None
+
+    cron = str(env.get("DB_BACKUP_CRON", settings.DB_BACKUP_CRON) or "").strip()
+    if cron:
+        try:
+            TimerUtils.normalize_schedule_trigger("cron", cron, settings.TZ)
+        except (TypeError, ValueError):
+            return "数据库备份周期格式不正确"
+
+    backup_path = env.get("DB_BACKUP_PATH", settings.DB_BACKUP_PATH)
+    if backup_path is not None and not isinstance(backup_path, str):
+        return "数据库备份目录必须是路径字符串"
+
+    for key, label in (
+        ("DB_BACKUP_RETENTION_DAYS", "数据库备份过期天数"),
+        ("DB_BACKUP_MAX_COUNT", "数据库备份最大保留份数"),
+    ):
+        value = env.get(key, getattr(settings, key))
+        if isinstance(value, bool):
+            return f"{label}必须是大于等于 0 的整数"
+        try:
+            converted = int(value)
+        except (TypeError, ValueError):
+            return f"{label}必须是大于等于 0 的整数"
+        if converted < 0 or str(value).strip() != str(converted):
+            return f"{label}必须是大于等于 0 的整数"
+    return None
+
+
 def _is_allowed_plugin_market_wiki_url(wiki_url: str) -> bool:
     """
     校验插件市场 Wiki 地址是否属于固定文档源。
@@ -796,6 +836,9 @@ async def set_env_setting(
     更新系统环境变量（仅管理员）
     """
     validation_error = _validate_llm_server_tool_config(env)
+    if validation_error:
+        return _SchemaResponse(success=False, message=validation_error)
+    validation_error = _validate_database_backup_config(env)
     if validation_error:
         return _SchemaResponse(success=False, message=validation_error)