tiku/tiku-server/app/services/backup.py

"""备份 + WebDAV 推送模块"""

import shutil
import time
from datetime import datetime
from pathlib import Path

from app.database import get_connection


def full_backup(config) -> Path:
    """全量备份：bank.db + 全部 docx 文件"""
    import shutil
    ts = datetime.now().strftime("%Y%m%d_%H%M%S")
    backup_dir = config.backups_dir / ts
    backup_dir.mkdir(parents=True, exist_ok=True)

    # bank.db
    if config.db_path.exists():
        shutil.copy2(config.db_path, backup_dir / "bank.db")

    # 全部 docx
    if config.problems_dir.exists():
        dst_dir = backup_dir / "problems"
        dst_dir.mkdir(parents=True, exist_ok=True)
        for docx in config.problems_dir.glob("*.docx"):
            shutil.copy2(docx, dst_dir / docx.name)

    # 推送到 WebDAV
    if config.webdav_enabled:
        try:
            _push_to_webdav(backup_dir, config)
        except Exception as e:
            print(f"⚠ WebDAV 推送失败: {e}")

    _cleanup_old_backups(config)
    print(f"✓ 全量备份: {backup_dir}")
    return backup_dir


def manual_backup(config, remote: bool = False):
    """手动备份"""
    backup_dir = _create_backup(config)
    print(f"✓ 本地备份: {backup_dir}")

    if remote and config.webdav_enabled:
        _push_to_webdav(backup_dir, config)
        print("✓ WebDAV 推送完成")


def auto_backup(config, changed_files: list[str] | None = None):
    """自动备份（操作前调用）"""
    backup_dir = _create_backup(config, changed_files)

    # 清理旧备份
    _cleanup_old_backups(config)

    # 如果配置了 WebDAV，尝试推送
    if config.webdav_enabled:
        try:
            _push_to_webdav(backup_dir, config)
        except Exception as e:
            print(f"⚠ WebDAV 推送失败: {e}")


def _create_backup(config, changed_files: list[str] | None = None) -> Path:
    """创建备份快照"""
    ts = datetime.now().strftime("%Y%m%d_%H%M%S")
    backup_dir = config.backups_dir / ts
    backup_dir.mkdir(parents=True, exist_ok=True)

    # 备份数据库
    if config.db_path.exists():
        shutil.copy2(config.db_path, backup_dir / "bank.db")

    # 备份被修改的 docx
    if changed_files:
        for f in changed_files:
            src = Path(f) if Path(f).is_absolute() else config.data_dir / f
            if src.exists():
                dst = backup_dir / src.parent.name / src.name
                dst.parent.mkdir(parents=True, exist_ok=True)
                shutil.copy2(src, dst)

    return backup_dir


def _cleanup_old_backups(config):
    """保留最近 N 个备份"""
    backups = sorted(config.backups_dir.glob("*"), key=lambda p: p.stat().st_mtime)
    while len(backups) > config.max_backups:
        oldest = backups.pop(0)
        if oldest.is_dir():
            shutil.rmtree(oldest)
        else:
            oldest.unlink()


def _push_to_webdav(backup_dir: Path, config):
    """推送备份到 WebDAV 服务器"""
    # 使用标准库实现简单的 WebDAV PUT 上传
    import urllib.request
    import base64

    remote_base = config.webdav_url.rstrip('/')
    remote_path = config.webdav_remote_path.rstrip('/')
    auth = base64.b64encode(f"{config.webdav_username}:{config.webdav_password}".encode()).decode()

    for file_path in backup_dir.rglob("*"):
        if file_path.is_file():
            rel_path = file_path.relative_to(backup_dir)
            url = f"{remote_base}{remote_path}/{backup_dir.name}/{rel_path}"

            # 创建目录（MKCOL）
            parent_url = '/'.join(url.split('/')[:-1])
            _webdav_mkcol(parent_url, auth)

            # 上传文件
            with open(file_path, 'rb') as f:
                data = f.read()

            req = urllib.request.Request(url, data=data, method='PUT')
            req.add_header('Authorization', f'Basic {auth}')
            req.add_header('Content-Type', 'application/octet-stream')

            try:
                urllib.request.urlopen(req, timeout=30)
            except Exception as e:
                print(f"⚠ 上传失败 {rel_path}: {e}")


def _webdav_mkcol(url: str, auth: str):
    """递归创建 WebDAV 目录"""
    import urllib.request
    parts = url.split('/')
    # 从根开始逐级创建
    for i in range(3, len(parts) + 1):
        partial = '/'.join(parts[:i])
        req = urllib.request.Request(partial, method='MKCOL')
        req.add_header('Authorization', f'Basic {auth}')
        try:
            urllib.request.urlopen(req, timeout=10)
        except Exception:
            pass  # 目录可能已存在


def get_backup_history(config) -> list[dict]:
    """获取备份历史"""
    backups = []
    for d in sorted(config.backups_dir.glob("*"), key=lambda p: p.stat().st_mtime, reverse=True):
        if d.is_dir():
            files = list(d.rglob("*"))
            backups.append({
                "name": d.name,
                "time": datetime.fromtimestamp(d.stat().st_mtime).isoformat(),
                "files": len(files),
            })
    return backups