document-management/backend/dms/services/backup_service.py

"""备份引擎（契约1.21）：链式增量备份 + 自有格式数据库导出 + LOCAL/RSYNC 目标。

对齐 Java 版 com.dms.backup 包（BackupConfig/BackupManifest/BackupTarget/
LocalBackupTarget/RsyncBackupTarget/SqlDataDumper/BackupEngine），两端读写同一
份 dms-backup-config.json（snake_case 字段）与同一备份目录结构，互为可恢复：

- 数据库导出为自有格式数据脚本 database/dms-data.sql（sql_dump 模块，
  驱动直连零外部工具；Java 版 com.dms.backup.SqlDataDumper 逐条对齐）；

- 备份包先构建于存储根下 backup-staging 暂存目录，完成后经目标送达；
- 增量 = 当前存储全清单 − 上一备份（链尾）manifest 的全清单，只传新增文件
  （内容寻址存储 hash 命名、写入后不可变，清单对比无需哈希校验）；
- manifest.json 完成时最后原子写入（无 manifest 的目录视为中断残留）；
- 历史摘要 index.json 存于目标根；保留策略按全量链整链清理；
- 定时器为后台守护线程（30 秒轮询配置，界面修改即时生效）。
"""

from __future__ import annotations

import json
import logging
import os
import re
import shutil
import subprocess
import sys
import tempfile
import threading
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any, Callable
from urllib.parse import unquote, urlsplit
from zoneinfo import ZoneInfo

from dms.common.errors import (
    BackupConfigInvalidError,
    BackupFailedError,
    BackupRunningError,
    BackupTargetUnavailableError,
    BackupToolMissingError,
)
from dms.services import sql_dump

logger = logging.getLogger(__name__)

INDEX_FILE = "index.json"
TRIGGER_MANUAL = "MANUAL"
TRIGGER_SCHEDULE = "SCHEDULE"
TYPE_FULL = "FULL"
TYPE_INCREMENTAL = "INCREMENTAL"

# 六目录全备（preview 可重建但按要求一并备份），仅排除 temporary
BACKUP_SUBDIRECTORIES = (
    "original",
    "preview",
    "extracted",
    "quarantine",
    "recycle",
    "voucher",
)

_TIME_FORMAT = "%Y-%m-%d %H:%M:%S"
_ID_FORMAT = "%Y%m%d-%H%M%S"
_SCHEDULE_TICK_SECONDS = 30
_SCHEDULE_RETRY_BACKOFF_MINUTES = 30


# ==================================================================== 配置


@dataclass
class BackupConfig:
    """部署级备份配置（JSON 文件持久化，与 .env 同级；双后端共用同一份）。"""

    target_mode: str = "LOCAL"
    local_path: str = ""
    rsync_url: str = ""
    rsync_password: str = ""
    schedule_enabled: bool = False
    schedule_time: str = "02:00"
    full_interval_days: int = 7
    retain_full_count: int = 2

    _TIME_PATTERN = re.compile(r"^([01]\d|2[0-3]):[0-5]\d$")

    @staticmethod
    def load(path: Path) -> "BackupConfig":
        """文件缺失或损坏返回默认配置（触发时再报配置错误）。"""
        if not path.is_file():
            return BackupConfig()
        try:
            data = json.loads(path.read_text(encoding="utf-8"))
            if not isinstance(data, dict):
                return BackupConfig()
            config = BackupConfig()
            config.target_mode = str(data.get("target_mode", config.target_mode))
            config.local_path = str(data.get("local_path", "") or "")
            config.rsync_url = str(data.get("rsync_url", "") or "")
            config.rsync_password = str(data.get("rsync_password", "") or "")
            config.schedule_enabled = bool(data.get("schedule_enabled", False))
            config.schedule_time = str(data.get("schedule_time", config.schedule_time))
            config.full_interval_days = int(data.get("full_interval_days", 7))
            config.retain_full_count = int(data.get("retain_full_count", 2))
            return config
        except (OSError, ValueError):
            return BackupConfig()

    def save(self, path: Path) -> None:
        self.validate()
        path.parent.mkdir(parents=True, exist_ok=True)
        temp = path.with_name(path.name + ".tmp")
        temp.write_text(
            json.dumps(self.to_dict(), ensure_ascii=False), encoding="utf-8"
        )
        os.replace(temp, path)

    def validate(self) -> None:
        if self.target_mode not in ("LOCAL", "RSYNC"):
            raise BackupConfigInvalidError("target_mode 必须为 LOCAL 或 RSYNC")
        if self.target_mode == "LOCAL" and not self.local_path.strip():
            raise BackupConfigInvalidError("LOCAL 模式必须配置 local_path 备份根目录")
        if self.target_mode == "RSYNC":
            if not self.rsync_url.startswith("rsync://"):
                raise BackupConfigInvalidError(
                    "RSYNC 模式必须配置 rsync://user@host:873/module/ 形式的 rsync_url"
                )
            if "?" in self.rsync_url or " " in self.rsync_url:
                raise BackupConfigInvalidError("rsync_url 不支持携带查询参数或空白字符")
        if not self._TIME_PATTERN.match(self.schedule_time or ""):
            raise BackupConfigInvalidError("schedule_time 必须为 HH:MM（24小时制）")
        if not 1 <= self.full_interval_days <= 365:
            raise BackupConfigInvalidError("full_interval_days 必须在 1~365 之间")
        if not 1 <= self.retain_full_count <= 100:
            raise BackupConfigInvalidError("retain_full_count 必须在 1~100 之间")

    def to_dict(self) -> dict[str, Any]:
        return {
            "target_mode": self.target_mode,
            "local_path": self.local_path,
            "rsync_url": self.rsync_url,
            "rsync_password": self.rsync_password,
            "schedule_enabled": self.schedule_enabled,
            "schedule_time": self.schedule_time,
            "full_interval_days": self.full_interval_days,
            "retain_full_count": self.retain_full_count,
        }

    def public_view(self) -> dict[str, Any]:
        """接口回显视图：密码永不回明文，仅回是否已设置。"""
        view = self.to_dict()
        del view["rsync_password"]
        view["rsync_password_set"] = bool(self.rsync_password)
        return view


# ==================================================================== 清单


@dataclass
class BackupManifest:
    """备份清单（manifest.json）：files 为合并后的全量清单。"""

    id: str = ""
    type: str = TYPE_FULL
    parent: str | None = None
    started_at: str = ""
    finished_at: str = ""
    database_file: str = "database/dms-data.sql"
    database_size: int = 0
    transferred_files: int = 0
    transferred_bytes: int = 0
    storage_file_count: int = 0
    storage_total_bytes: int = 0
    files: list[str] = field(default_factory=list)

    @staticmethod
    def parse(data: dict[str, Any]) -> "BackupManifest":
        return BackupManifest(
            id=str(data.get("id", "")),
            type=str(data.get("type", TYPE_FULL)),
            parent=data.get("parent"),
            started_at=str(data.get("started_at", "")),
            finished_at=str(data.get("finished_at", "")),
            database_file=str(data.get("database_file", "database/dms-data.sql")),
            database_size=int(data.get("database_size", 0)),
            transferred_files=int(data.get("transferred_files", 0)),
            transferred_bytes=int(data.get("transferred_bytes", 0)),
            storage_file_count=int(data.get("storage_file_count", 0)),
            storage_total_bytes=int(data.get("storage_total_bytes", 0)),
            files=[str(item) for item in data.get("files", [])],
        )

    def to_dict(self, include_files: bool = True) -> dict[str, Any]:
        data: dict[str, Any] = {
            "id": self.id,
            "type": self.type,
            "parent": self.parent,
            "started_at": self.started_at,
            "finished_at": self.finished_at,
            "database_file": self.database_file,
            "database_size": self.database_size,
            "transferred_files": self.transferred_files,
            "transferred_bytes": self.transferred_bytes,
            "storage_file_count": self.storage_file_count,
            "storage_total_bytes": self.storage_total_bytes,
        }
        if include_files:
            data["files"] = self.files
        return data

    def write(self, path: Path) -> None:
        """原子写入：目录存在但无 manifest 即中断残留的判定依据。"""
        temp = path.with_name(path.name + ".tmp")
        temp.write_text(
            json.dumps(self.to_dict(), ensure_ascii=False), encoding="utf-8"
        )
        os.replace(temp, path)


# ==================================================================== 目标


class BackupTarget:
    """备份目标抽象：LOCAL 本机目录 / RSYNC 异机专用协议。"""

    def check_writable(self) -> None:
        raise NotImplementedError

    def push_directory(
        self, local_directory: Path, name: str,
        reporter: Callable[[int], None] | None = None,
    ) -> None:
        raise NotImplementedError

    def fetch_file(self, remote_relative_path: str) -> bytes | None:
        raise NotImplementedError

    def fetch_directory(self, name: str, local_directory: Path) -> None:
        raise NotImplementedError

    def push_file(self, local_file: Path, remote_relative_path: str) -> None:
        raise NotImplementedError

    def list_backup_directories(self) -> list[str]:
        raise NotImplementedError

    def delete_backup_directory(self, name: str) -> None:
        raise NotImplementedError

    def describe(self) -> str:
        raise NotImplementedError

    def interrupt(self) -> None:
        """中断当前传输（备份取消）；无进程语义的目标为空操作。"""


class LocalBackupTarget(BackupTarget):
    """LOCAL 模式：直接写本机目录（开发/测试/单机）。"""

    def __init__(self, root: Path) -> None:
        self.root = root.expanduser().resolve()

    def _resolve_safe(self, relative: str) -> Path:
        if not relative or "\x00" in relative:
            raise BackupConfigInvalidError("备份目标相对路径无效")
        normalized = relative.replace("\\", "/")
        for part in normalized.split("/"):
            if ":" in part or part == "..":
                raise BackupConfigInvalidError("备份目标相对路径包含非法段")
        candidate = (self.root / normalized).resolve()
        if not candidate.is_relative_to(self.root):
            raise BackupConfigInvalidError("备份目标相对路径超出根目录")
        return candidate

    def check_writable(self) -> None:
        self.root.mkdir(parents=True, exist_ok=True)
        probe = self.root / ".dms-writable-probe"
        probe.write_bytes(b"")
        probe.unlink(missing_ok=True)

    def push_directory(
        self, local_directory: Path, name: str,
        reporter: Callable[[int], None] | None = None,
    ) -> None:
        destination = self._resolve_safe(name)
        destination.mkdir(parents=True, exist_ok=True)
        shutil.copytree(local_directory, destination, dirs_exist_ok=True)
        if reporter is not None:
            reporter(100)

    def fetch_file(self, remote_relative_path: str) -> bytes | None:
        file = self._resolve_safe(remote_relative_path)
        return file.read_bytes() if file.is_file() else None

    def fetch_directory(self, name: str, local_directory: Path) -> None:
        source = self._resolve_safe(name)
        if not source.is_dir():
            raise BackupTargetUnavailableError(f"备份目录不存在：{name}")
        local_directory.mkdir(parents=True, exist_ok=True)
        shutil.copytree(source, local_directory, dirs_exist_ok=True)

    def push_file(self, local_file: Path, remote_relative_path: str) -> None:
        destination = self._resolve_safe(remote_relative_path)
        destination.parent.mkdir(parents=True, exist_ok=True)
        shutil.copyfile(local_file, destination)

    def list_backup_directories(self) -> list[str]:
        if not self.root.is_dir():
            return []
        return sorted(
            item.name for item in self.root.iterdir() if item.is_dir()
        )

    def delete_backup_directory(self, name: str) -> None:
        directory = self._resolve_safe(name)
        if directory.is_dir():
            shutil.rmtree(directory)

    def describe(self) -> str:
        return str(self.root)


_PERCENT_PATTERN = re.compile(r"(\d+)%")
_LIST_DIRECTORY_PATTERN = re.compile(r"^d[-xrwSsTt]{9,}.*$")


class RsyncBackupTarget(BackupTarget):
    """RSYNC 模式：rsync:// 专用备份协议推送异机（生产）。

    rsync daemon 模块形态，密码经 RSYNC_PASSWORD 环境变量传递（不进命令行）；
    -a --partial 提供增量传输与断点续传；进度解析 --info=progress2 输出。
    """

    def __init__(self, executable: str, url: str, password: str) -> None:
        self.executable = executable
        self.url = url if url.endswith("/") else url + "/"
        self.password = password or ""
        self._current_process: subprocess.Popen[bytes] | None = None

    @staticmethod
    def locate_executable() -> str | None:
        configured = os.environ.get("DMS_RSYNC_EXECUTABLE", "").strip()
        if configured:
            return configured
        found = shutil.which("rsync")
        if found:
            return found
        # 发行包内置工具：exe 同目录与工作目录下的 tools/rsync/（Windows 随包携带
        # cwRsync，Linux 可放置静态二进制；run 脚本 cd 到 exe 目录后两者同址）。
        for root in (Path(sys.executable).resolve().parent, Path.cwd()):
            for name in ("rsync.exe", "rsync"):
                candidate = root / "tools" / "rsync" / name
                if candidate.is_file():
                    return str(candidate)
        return None

    @staticmethod
    def _local_path(path: Path) -> str:
        """本地路径转 rsync 可用形态：Windows 盘符路径（C:\\x 或 C:/x）转为 Cygwin
        /cygdrive/<盘>/x——发行包内置 cwRsync 为 Cygwin 构建，盘符冒号会被 rsync 按
        远端 host:path 语法解析，报 "source and destination cannot both be remote"
        （退出码 1 用法错误，main.c）；Linux 原生路径原样返回。"""
        raw = str(path)
        if len(raw) >= 2 and raw[1] == ":" and raw[0].lower() in "abcdefghijklmnopqrstuvwxyz":
            return "/cygdrive/" + raw[0].lower() + raw[2:].replace("\\", "/")
        return raw

    def check_writable(self) -> None:
        # 可达性：--list-only 模块根；可写性：覆盖推送固定探测文件
        self.run([self.executable, "--list-only", self.url])
        probe = Path(tempfile.mkstemp(prefix="dms-rsync-probe-")[1])
        try:
            self.push_file(probe, ".dms-writable-probe")
        finally:
            probe.unlink(missing_ok=True)

    def push_directory(
        self, local_directory: Path, name: str,
        reporter: Callable[[int], None] | None = None,
    ) -> None:
        # 尾斜杠 = 推目录内容；--partial 断点续传；progress2 整体进度
        self.run(
            [
                self.executable, "-a", "--partial", "--info=progress2",
                self._local_path(local_directory) + "/", self.url + name + "/",
            ],
            reporter=reporter,
        )

    def fetch_file(self, remote_relative_path: str) -> bytes | None:
        fd, temp_name = tempfile.mkstemp(prefix="dms-rsync-fetch-")
        os.close(fd)
        temp = Path(temp_name)
        try:
            code = self.run(
                [self.executable, self.url + remote_relative_path, self._local_path(temp)],
                check=False,
            )
            if code in (23, 3):
                # 23=部分传输（源不存在），3=协议/路径错误：统一按不存在处理
                return None
            if code != 0:
                raise BackupTargetUnavailableError(
                    f"拉取 {remote_relative_path} 失败（rsync 退出码 {code}）"
                )
            return temp.read_bytes() if temp.is_file() else None
        finally:
            temp.unlink(missing_ok=True)

    def fetch_directory(self, name: str, local_directory: Path) -> None:
        local_directory.mkdir(parents=True, exist_ok=True)
        self.run(
            [
                self.executable, "-a", "--partial",
                self.url + name + "/", self._local_path(local_directory.resolve()) + "/",
            ]
        )

    def push_file(self, local_file: Path, remote_relative_path: str) -> None:
        self.run([self.executable, self._local_path(local_file), self.url + remote_relative_path])

    def list_backup_directories(self) -> list[str]:
        output = self.run(
            [self.executable, "--list-only", self.url], capture=True
        )
        names: list[str] = []
        for raw_line in output.splitlines():
            line = raw_line.strip()
            if not line or not _LIST_DIRECTORY_PATTERN.match(line):
                continue
            name = line[line.rfind(" ") + 1:]
            if name and name not in (".", ".."):
                names.append(name)
        return names

    def delete_backup_directory(self, name: str) -> None:
        # rsync 无直接删除：空目录 --delete 镜像即可清空远端目录内容
        empty = Path(tempfile.mkdtemp(prefix="dms-rsync-empty-"))
        try:
            self.run(
                [self.executable, "-a", "--delete", self._local_path(empty) + "/", self.url + name + "/"]
            )
        finally:
            shutil.rmtree(empty, ignore_errors=True)

    def describe(self) -> str:
        return self.url

    def interrupt(self) -> None:
        process = self._current_process
        if process is not None and process.poll() is None:
            process.terminate()

    def run(
        self,
        args: list[str],
        *,
        reporter: Callable[[int], None] | None = None,
        capture: bool = False,
        check: bool = True,
    ) -> str | int:
        """执行 rsync：合流读输出防管道阻塞，按行喂进度解析，返回输出或退出码。"""
        if self.executable is None:
            raise BackupToolMissingError(
                "未找到 rsync 可执行文件（RSYNC 模式需安装 rsync 或配置 DMS_RSYNC_EXECUTABLE）"
            )
        env = dict(os.environ)
        if self.password:
            env["RSYNC_PASSWORD"] = self.password
        try:
            process = subprocess.Popen(
                args,
                stdout=subprocess.PIPE,
                stderr=subprocess.STDOUT,
                env=env,
            )
        except OSError as exc:
            raise BackupToolMissingError(f"启动 rsync 失败：{exc}") from exc
        self._current_process = process
        captured: list[bytes] = []
        tail = ""
        assert process.stdout is not None
        try:
            for raw_line in process.stdout:
                line = raw_line.decode("utf-8", errors="replace").strip()
                if not line:
                    continue
                if capture:
                    captured.append(raw_line)
                if reporter is not None:
                    match = _PERCENT_PATTERN.search(line)
                    if match:
                        reporter(int(match.group(1)))
                tail = line
        finally:
            self._current_process = None
        try:
            code = process.wait()
        except KeyboardInterrupt:
            process.terminate()
            raise
        if check and code != 0:
            raise BackupTargetUnavailableError(
                f"rsync 传输失败（退出码 {code}）：{tail[-300:]}"
            )
        if capture:
            return b"".join(captured).decode("utf-8", errors="replace")
        return code


# ==================================================================== 数据库导出


# 自有格式数据库备份（dms-data.sql 数据脚本）实现于 sql_dump 模块；
# DmBackupTool（dexp/dimp 外部工具封装）已移除——导出/恢复全部走驱动直连，
# 不再依赖达梦客户端工具。保留 BackupToolMissingError 供 rsync 目标使用。


# ==================================================================== 引擎


def _dm_userid(database_url: str) -> tuple[str, str]:
    """解析达梦连接串为 (userid, schema)：dm+dmPython://user:pwd@host:port。"""
    parts = urlsplit(database_url)
    if parts.username is None or parts.hostname is None:
        raise BackupConfigInvalidError(
            f"无法从数据库连接串解析达梦凭据：{database_url}"
        )
    username = unquote(parts.username)
    password = unquote(parts.password or "")
    return f"{username}/{password}@{parts.hostname}:{parts.port}", username


@dataclass(frozen=True)
class BackupOutcome:
    """备份任务终态快照（完成回调入参；backup_id 在早期失败时为 None）。"""

    success: bool
    type: str
    trigger: str
    backup_id: str | None
    error: str | None


class BackupEngine:
    """链式增量备份引擎：单任务互斥、进度快照、定时轮询。"""

    def __init__(
        self,
        *,
        storage_root: Path,
        business_timezone: str,
        database_url: str,
        backup_config_path: Path,
        env_path: Path,
    ) -> None:
        self.storage_root = storage_root.expanduser().resolve()
        self.timezone = ZoneInfo(business_timezone)
        self.database_url = database_url
        self.backup_config_path = backup_config_path
        self.env_path = env_path
        self._running = threading.Event()
        self._lock = threading.Lock()
        self._cancel_requested = False
        self._active_target: BackupTarget | None = None
        # 进度快照恒为全量12字段形状（从未运行时业务字段为null/0），与Java版对齐
        self._progress: dict[str, Any] = {
            "backupId": None,
            "type": None,
            "trigger": None,
            "phase": None,
            "bytesTotal": 0,
            "bytesCopied": 0,
            "filesTotal": 0,
            "filesCopied": 0,
            "pushPercent": None,
            "error": None,
            "startedAt": None,
            "finishedAt": None,
        }
        self._scheduler_stop = threading.Event()

    # ------------------------------------------------------------ 公共 API

    @property
    def is_running(self) -> bool:
        return self._running.is_set()

    def progress_snapshot(self) -> dict[str, Any]:
        with self._lock:
            return dict(self._progress)

    def load_config(self) -> BackupConfig:
        return BackupConfig.load(self.backup_config_path)

    def start_async(
        self,
        backup_type: str,
        trigger: str,
        on_complete: Callable[[BackupOutcome], None] | None = None,
    ) -> None:
        """异步触发备份；on_complete 在任务终态恰调一次（先于 running 复位）。"""
        if self._running.is_set():
            raise BackupRunningError()
        self._running.set()
        self._cancel_requested = False
        thread = threading.Thread(
            target=self._run_guarded,
            args=(backup_type, trigger, on_complete),
            name="dms-backup",
            daemon=True,
        )
        thread.start()

    def _run_guarded(
        self,
        backup_type: str,
        trigger: str,
        on_complete: Callable[[BackupOutcome], None] | None,
    ) -> None:
        try:
            outcome = self._run_backup(backup_type, trigger)
        except Exception as exc:  # noqa: BLE001 - 后台线程兜底
            logger.exception("备份任务异常终止")
            self._finish("FAILED", str(exc))
            outcome = BackupOutcome(False, backup_type, trigger, None, str(exc))
        try:
            if on_complete is not None:
                on_complete(outcome)
        except Exception:  # noqa: BLE001 - 回调失败不否定备份结果
            logger.warning("备份完成回调执行失败", exc_info=True)
        finally:
            self._running.clear()

    def cancel(self) -> None:
        if not self._running.is_set():
            return
        self._cancel_requested = True
        target = self._active_target
        if target is not None:
            target.interrupt()

    def list_history(self, config: BackupConfig) -> list[dict[str, Any]]:
        target = self.build_target(config)
        try:
            return self._sorted_index(target)
        except (BackupFailedError, BackupTargetUnavailableError, BackupToolMissingError):
            raise
        except Exception as exc:
            raise BackupTargetUnavailableError(f"读取备份历史失败：{exc}") from exc

    def decide_scheduled_type(self, config: BackupConfig) -> str:
        """距最近全量 ≥ full_interval_days 天（或无历史）→ FULL。"""
        target = self.build_target(config)
        entries = self._sorted_index(target)
        last_full = max(
            (entry["id"] for entry in entries if entry["id"].endswith("-F")),
            default=None,
        )
        if last_full is None:
            return TYPE_FULL
        last_full_date = datetime.strptime(
            last_full.removeprefix("dms-bak-")[:8], "%Y%m%d"
        ).date()
        threshold = (
            datetime.now(self.timezone) - timedelta(days=config.full_interval_days)
        ).date()
        return TYPE_FULL if last_full_date < threshold else TYPE_INCREMENTAL

    def build_target(self, config: BackupConfig) -> BackupTarget:
        if config.target_mode == "RSYNC":
            rsync = RsyncBackupTarget.locate_executable()
            if rsync is None:
                raise BackupToolMissingError(
                    "未找到 rsync（RSYNC 模式需安装 rsync 或配置 DMS_RSYNC_EXECUTABLE）"
                )
            return RsyncBackupTarget(rsync, config.rsync_url, config.rsync_password)
        return LocalBackupTarget(Path(config.local_path))

    # ------------------------------------------------------------ 定时器

    def start_scheduler(
        self, audit_writer: Callable[[BackupOutcome], None]
    ) -> None:
        """启动定时轮询守护线程（audit_writer 在任务终态写审计，入参 BackupOutcome）。"""

        def loop() -> None:
            triggered_date = None
            last_attempt: datetime | None = None
            while not self._scheduler_stop.wait(_SCHEDULE_TICK_SECONDS):
                try:
                    now = datetime.now(self.timezone)
                    today = now.date()
                    if today == triggered_date:
                        continue
                    config = self.load_config()
                    if not config.schedule_enabled:
                        continue
                    if now.time() < datetime.strptime(
                        config.schedule_time, "%H:%M"
                    ).time():
                        continue
                    if (
                        last_attempt is not None
                        and last_attempt
                        + timedelta(minutes=_SCHEDULE_RETRY_BACKOFF_MINUTES)
                        > now
                    ):
                        continue
                    last_attempt = now
                    backup_type = self.decide_scheduled_type(config)
                    self.start_async(
                        backup_type, TRIGGER_SCHEDULE, on_complete=audit_writer
                    )
                    triggered_date = today
                    logger.info(
                        "定时备份已触发：type=%s schedule_time=%s",
                        backup_type,
                        config.schedule_time,
                    )
                except BackupRunningError:
                    # 手动任务占用：不退避，下一轮再试
                    logger.info("定时备份等待手动任务结束")
                except Exception as exc:  # noqa: BLE001 - 定时轮询兜底
                    logger.error(
                        "定时备份触发失败（%d分钟后重试）：%s",
                        _SCHEDULE_RETRY_BACKOFF_MINUTES,
                        exc,
                    )

        thread = threading.Thread(target=loop, name="dms-backup-scheduler", daemon=True)
        thread.start()

    def stop_scheduler(self) -> None:
        self._scheduler_stop.set()

    # ------------------------------------------------------------ 核心流程

    def _run_backup(self, requested_type: str, trigger: str) -> BackupOutcome:
        now = datetime.now(self.timezone)
        config = self.load_config()
        config.validate()

        target = self.build_target(config)
        self._active_target = target
        try:
            target.check_writable()
            index = self._read_index(target)
        except (BackupFailedError, BackupTargetUnavailableError, BackupToolMissingError):
            self._finish("FAILED", "备份目标不可用")
            return BackupOutcome(
                False, requested_type, trigger, None, "备份目标不可用"
            )
        except Exception as exc:
            message = f"备份目标不可用：{exc}"
            self._finish("FAILED", message)
            return BackupOutcome(False, requested_type, trigger, None, message)

        parent_manifest: BackupManifest | None = None
        if (
            requested_type == TYPE_INCREMENTAL
            and index
        ):
            manifest_data = self._fetch_manifest(target, index[-1]["id"])
            if manifest_data is not None:
                parent_manifest = BackupManifest.parse(manifest_data)

        # 无历史时增量自动升级全量（增量必须有父）
        backup_type = (
            TYPE_INCREMENTAL
            if requested_type == TYPE_INCREMENTAL and parent_manifest is not None
            else TYPE_FULL
        )
        backup_id = (
            "dms-bak-"
            + now.strftime(_ID_FORMAT)
            + ("-F" if backup_type == TYPE_FULL else "-I")
        )

        with self._lock:
            self._progress = {
                "backupId": backup_id,
                "type": backup_type,
                "trigger": trigger,
                "phase": "SCANNING",
                "bytesTotal": 0,
                "bytesCopied": 0,
                "filesTotal": 0,
                "filesCopied": 0,
                "pushPercent": None,
                "error": None,
                "startedAt": now.strftime(_TIME_FORMAT),
                "finishedAt": None,
            }
        logger.info(
            "备份开始：id=%s type=%s trigger=%s 目标=%s",
            backup_id, backup_type, trigger, target.describe(),
        )

        staging_root = self.storage_root / "backup-staging"
        try:
            shutil.rmtree(staging_root, ignore_errors=True)
            staging = staging_root / backup_id
            database_dir = staging / "database"
            env_dir = staging / "env"
            (staging / "storage").mkdir(parents=True, exist_ok=True)
            database_dir.mkdir(parents=True, exist_ok=True)
            env_dir.mkdir(parents=True, exist_ok=True)

            # 1. 扫描存储：合并全清单（相对路径→字节，有序）
            current = self._scan_storage()

            # 2. 数据库全量导出（自有格式数据脚本）+ .env 副本（best-effort）
            self._set_phase("EXPORTING_DB")
            _, schema = _dm_userid(self.database_url)
            dump_file = sql_dump.export_dump(self.database_url, database_dir, schema)
            self._copy_env_best_effort(env_dir)

            # 3. 拷贝文件到暂存（全量=全部；增量=相对父清单新增）
            self._set_phase("COPYING")
            parent_files = set(parent_manifest.files) if parent_manifest else set()
            transfer_list = [
                relative for relative in current if relative not in parent_files
            ]
            total_bytes = sum(current[relative] for relative in transfer_list)
            with self._lock:
                self._progress["filesTotal"] = len(transfer_list)
                self._progress["bytesTotal"] = total_bytes
            copied_bytes = 0
            copied_files = 0
            for relative in transfer_list:
                self._check_cancelled()
                source = self.storage_root / relative
                if not source.is_file():
                    # 扫描后被删除的文件：跳过（内容寻址下该记录不会指向它）
                    logger.warning("备份拷贝跳过已消失文件：%s", relative)
                    continue
                destination = staging / "storage" / relative
                destination.parent.mkdir(parents=True, exist_ok=True)
                shutil.copyfile(source, destination)
                copied_bytes += current[relative]
                copied_files += 1
                with self._lock:
                    self._progress["bytesCopied"] = copied_bytes
                    self._progress["filesCopied"] = copied_files

            # 4. manifest 最后原子写入（无 manifest 即中断残留）
            manifest = BackupManifest(
                id=backup_id,
                type=backup_type,
                parent=parent_manifest.id if parent_manifest else None,
                started_at=self._progress["startedAt"],
                finished_at=datetime.now(self.timezone).strftime(_TIME_FORMAT),
                database_size=dump_file.stat().st_size,
                transferred_files=copied_files,
                transferred_bytes=copied_bytes,
                storage_file_count=len(current),
                storage_total_bytes=sum(current.values()),
                files=list(current.keys()),
            )
            manifest.write(staging / "manifest.json")

            # 5. 推送整个备份目录
            self._set_phase("PUSHING")
            target.push_directory(
                staging,
                backup_id,
                reporter=lambda percent: self._set_progress("pushPercent", percent),
            )

            # 6. 更新历史索引 + 保留链清理 + 残留清理
            self._set_phase("CLEANING")
            index.append(manifest.to_dict(include_files=False))
            self._write_index(target, index)
            self._cleanup_retention_policy(target, index, config.retain_full_count)
        except (BackupFailedError, BackupTargetUnavailableError, BackupToolMissingError) as exc:
            logger.error("备份失败：id=%s %s", backup_id, exc)
            self._finish("FAILED", str(exc))
            shutil.rmtree(staging_root, ignore_errors=True)
            return BackupOutcome(False, backup_type, trigger, backup_id, str(exc))
        except Exception as exc:  # noqa: BLE001 - 后台线程兜底
            logger.exception("备份失败：id=%s", backup_id)
            message = f"备份执行异常：{exc}"
            self._finish("FAILED", message)
            shutil.rmtree(staging_root, ignore_errors=True)
            return BackupOutcome(False, backup_type, trigger, backup_id, message)
        shutil.rmtree(staging_root, ignore_errors=True)
        with self._lock:
            self._progress["phase"] = "SUCCESS"
            self._progress["finishedAt"] = datetime.now(self.timezone).strftime(
                _TIME_FORMAT
            )
        logger.info("备份完成：id=%s", backup_id)
        return BackupOutcome(True, backup_type, trigger, backup_id, None)

    def _scan_storage(self) -> dict[str, int]:
        """扫描六个备份目录，返回 相对路径→字节（有序保证清单顺序稳定）。"""
        files: dict[str, int] = {}
        for directory in BACKUP_SUBDIRECTORIES:
            base = self.storage_root / directory
            if not base.is_dir():
                continue
            for path in base.rglob("*"):
                if path.is_file():
                    relative = path.relative_to(self.storage_root).as_posix()
                    files[relative] = path.stat().st_size
        return dict(sorted(files.items()))

    def _copy_env_best_effort(self, env_dir: Path) -> None:
        try:
            if self.env_path.is_file():
                shutil.copyfile(self.env_path, env_dir / ".env")
        except OSError as exc:
            logger.warning("备份 .env 副本跳过：%s", exc)

    def _read_index(self, target: BackupTarget) -> list[dict[str, Any]]:
        content = target.fetch_file(INDEX_FILE)
        if content is None:
            return []
        data = json.loads(content.decode("utf-8"))
        return data if isinstance(data, list) else []

    def _sorted_index(self, target: BackupTarget) -> list[dict[str, Any]]:
        entries = self._read_index(target)
        entries.sort(key=lambda entry: str(entry.get("id", "")))
        return entries

    def _fetch_manifest(self, target: BackupTarget, backup_id: str) -> dict[str, Any] | None:
        content = target.fetch_file(f"{backup_id}/manifest.json")
        if content is None:
            raise BackupFailedError("上一备份缺少 manifest，无法执行增量")
        return json.loads(content.decode("utf-8"))

    def _write_index(self, target: BackupTarget, entries: list[dict[str, Any]]) -> None:
        fd, temp_name = tempfile.mkstemp(prefix="dms-backup-index-")
        os.close(fd)
        temp = Path(temp_name)
        try:
            temp.write_text(
                json.dumps(entries, ensure_ascii=False), encoding="utf-8"
            )
            target.push_file(temp, INDEX_FILE)
        finally:
            temp.unlink(missing_ok=True)

    def _cleanup_retention_policy(
        self,
        target: BackupTarget,
        index: list[dict[str, Any]],
        retain_full_count: int,
    ) -> None:
        """保留最近 retain_full_count 条全量链；目标上不在 index 内的目录为残留。"""
        try:
            # index 按时间升序：遇 FULL 链号+1，其后增量归属当前链
            total_chains = sum(
                1 for entry in index if entry.get("type") == TYPE_FULL
            )
            excess_chains = total_chains - retain_full_count
            if excess_chains > 0:
                chain_number = -1
                removed: list[str] = []
                for entry in index:
                    if entry.get("type") == TYPE_FULL:
                        chain_number += 1
                    if chain_number < excess_chains:
                        removed.append(entry["id"])
                for backup_id in removed:
                    target.delete_backup_directory(backup_id)
                    logger.info("保留策略删除过期备份：%s", backup_id)
                index[:] = [entry for entry in index if entry["id"] not in removed]

            # 残留清理：目录存在但不在 index（中断备份的半成品目录）
            index_ids = {entry["id"] for entry in index}
            for directory in target.list_backup_directories():
                if directory not in index_ids:
                    target.delete_backup_directory(directory)
                    logger.info("清理中断备份残留目录：%s", directory)
        except Exception as exc:  # noqa: BLE001 - 清理失败不否定已成功的备份
            logger.warning("备份保留策略清理失败（不影响本次备份）：%s", exc)

    def _check_cancelled(self) -> None:
        if self._cancel_requested:
            raise BackupFailedError("备份已取消")

    def _set_phase(self, phase: str) -> None:
        with self._lock:
            self._progress["phase"] = phase

    def _set_progress(self, key: str, value: Any) -> None:
        with self._lock:
            self._progress[key] = value

    def _finish(self, phase: str, error: str) -> None:
        with self._lock:
            self._progress["phase"] = phase
            self._progress["error"] = error
            self._progress["finishedAt"] = datetime.now(self.timezone).strftime(
                _TIME_FORMAT
            )


# ==================================================================== 恢复（离线 CLI）


def run_restore(args: list[str]) -> int:
    """dms-server restore-backup [--backup-id dms-bak-xxx] [--yes]。

    沿 parent 回溯备份链拉取到本地临时目录 → 链尾数据库脚本驱动直连导入 →
    清空存储六目录按链序叠加还原文件。退出码：0 成功 / 1 取消 / 2 失败。
    """
    backup_id: str | None = None
    assume_yes = False
    remaining = list(args[1:])
    while remaining:
        item = remaining.pop(0)
        if item == "--backup-id" and remaining:
            backup_id = remaining.pop(0)
        elif item == "--yes":
            assume_yes = True
        else:
            print(f"错误：无法识别的参数 {item}")
            print("用法：dms-server restore-backup [--backup-id dms-bak-xxx] [--yes]")
            return 1

    engine = _build_standalone_engine()
    config = engine.load_config()
    try:
        config.validate()
    except BackupConfigInvalidError as exc:
        print(f"错误：备份配置无效（{engine.backup_config_path}）：{exc}")
        return 2

    try:
        target = engine.build_target(config)
        index = engine._sorted_index(target)
    except Exception as exc:  # noqa: BLE001 - CLI 汇总报错
        print(f"错误：读取备份历史失败：{exc}")
        return 2
    if not index:
        print("错误：备份目标上没有任何历史备份。")
        return 2

    if backup_id is None:
        backup_id = max(entry["id"] for entry in index)
        print(f"未指定 --backup-id，恢复最新备份：{backup_id}")

    # 沿 parent 回溯整条链（新→旧收集，再反转为 旧→新）
    chain: list[BackupManifest] = []
    try:
        cursor: str | None = backup_id
        while cursor is not None and len(chain) < 1000:
            content = target.fetch_file(f"{cursor}/manifest.json")
            if content is None:
                print(f"错误：备份 {cursor} 缺少 manifest，链不完整。")
                return 2
            manifest = BackupManifest.parse(json.loads(content.decode("utf-8")))
            if manifest.id != cursor:
                print(f"错误：备份 {cursor} 的 manifest 内容与目录不符。")
                return 2
            chain.append(manifest)
            cursor = manifest.parent
    except Exception as exc:  # noqa: BLE001 - CLI 汇总报错
        print(f"错误：读取备份链失败：{exc}")
        return 2
    chain.reverse()
    tail = chain[-1]

    print(f"备份目标：{target.describe()}")
    print(f"恢复计划（{len(chain)} 个备份，链头→链尾）：")
    for member in chain:
        print(
            f"  - {member.id} {member.type}"
            f" 文件 {member.transferred_files} 个 / {member.transferred_bytes} 字节，"
            f"数据库导出 {member.database_size} 字节"
        )
    print(f"链尾合并清单：{tail.storage_file_count} 个文件 / {tail.storage_total_bytes} 字节")
    print("恢复将执行：")
    print(f"  1. 导入链尾数据库脚本 {tail.id}/database/{sql_dump.DUMP_FILE_NAME}"
          f"（禁用外键→清空业务表→导入→恢复外键；目标库为空时先执行 alembic 迁移建库）")
    print(f"  2. 清空并重建存储目录：{'、'.join(BACKUP_SUBDIRECTORIES)}")
    print(f"  3. 按链序叠加还原 {tail.storage_file_count} 个文件至 {engine.storage_root}")

    if not assume_yes:
        answer = input("确认执行恢复？(y/N)：").strip().lower()
        if answer not in ("y", "yes"):
            print("已取消。")
            return 1

    workspace = Path(tempfile.mkdtemp(prefix="dms-restore-"))
    try:
        # 1. 拉取整条链
        print("拉取备份链……")
        for member in chain:
            print(f"  拉取 {member.id} ……")
            target.fetch_directory(member.id, workspace / member.id)

        # 2. 数据库恢复：链尾数据脚本即最新状态（空库自动 alembic 建库）
        print("导入数据库……")
        _, schema = _dm_userid(engine.database_url)

        def ensure_schema_ready() -> None:
            from dms.db_upgrade import run_upgrade_silent

            print("目标库为空，先执行 alembic 迁移建库……")
            run_upgrade_silent(engine.database_url)

        sql_dump.restore_dump(
            engine.database_url,
            workspace / tail.id / "database" / sql_dump.DUMP_FILE_NAME,
            schema,
            ensure_schema_ready=ensure_schema_ready,
        )

        # 3. 存储恢复：清空六目录，链序叠加（各成员物理持有 = 自身清单 − 父清单）
        print("重建存储目录……")
        for directory in BACKUP_SUBDIRECTORIES:
            shutil.rmtree(engine.storage_root / directory, ignore_errors=True)
            (engine.storage_root / directory).mkdir(parents=True, exist_ok=True)
        restored = 0
        parent_files: set[str] = set()
        for member in chain:
            member_storage = workspace / member.id / "storage"
            for relative in member.files:
                if relative in parent_files:
                    continue
                destination = engine.storage_root / relative
                destination.parent.mkdir(parents=True, exist_ok=True)
                shutil.copyfile(member_storage / relative, destination)
                restored += 1
            parent_files.update(member.files)
        print(f"存储恢复完成：{restored} 个文件。")

        # 4. .env 副本落地为 .env.restored（不覆盖当前 .env，防止恢复机数据库地址不同）
        env_backup = workspace / tail.id / "env" / ".env"
        if env_backup.is_file():
            shutil.copyfile(
                env_backup, Path.cwd() / ".env.restored"
            )
            print("备份时的 .env 已存至 .env.restored（未覆盖当前 .env）。")
    except Exception as exc:  # noqa: BLE001 - CLI 汇总报错
        print(f"错误：恢复失败：{exc}")
        return 2
    finally:
        shutil.rmtree(workspace, ignore_errors=True)

    print("恢复完成。建议重启 DMS 服务后核验数据。")
    return 0


def _build_standalone_engine() -> BackupEngine:
    """离线 CLI 场景构造引擎（不依赖 Flask 上下文）。"""
    from flask import Flask

    from dms import init_dms  # 延后导入避免循环依赖

    app = Flask(__name__)
    init_dms(app)
    engine = app.extensions["dms_backup_engine"]
    assert isinstance(engine, BackupEngine)
    # init_dms 会启动定时线程，CLI 场景立即停止
    engine.stop_scheduler()
    return engine