crypto_key/backup.py

"""备份与恢复：打包 .env / data.json / settings.json / uploads"""
import io
import json
import os
import shutil
import tarfile
import threading
import time
from datetime import datetime, timedelta
from pathlib import Path
from typing import Optional, Tuple

BASE_DIR = Path(__file__).resolve().parent
UPLOAD_DIR = BASE_DIR / "uploads"
BACKUP_META = BASE_DIR / "backup_meta.json"

# Linux 生产环境写到 /root；其它环境回退到项目 backups/
ROOT_BACKUP_DIR = Path("/root/crypto_key_backups")
LOCAL_BACKUP_DIR = BASE_DIR / "backups"

BACKUP_FILES = (".env", "data.json", "settings.json")

_scheduler_started = False
_scheduler_lock = threading.Lock()


def backup_dir() -> Path:
    if Path("/root").exists() and os.access("/root", os.W_OK):
        try:
            ROOT_BACKUP_DIR.mkdir(parents=True, exist_ok=True)
            return ROOT_BACKUP_DIR
        except OSError:
            pass
    LOCAL_BACKUP_DIR.mkdir(parents=True, exist_ok=True)
    return LOCAL_BACKUP_DIR


def _load_meta() -> dict:
    if not BACKUP_META.exists():
        return {}
    try:
        return json.loads(BACKUP_META.read_text(encoding="utf-8"))
    except (json.JSONDecodeError, OSError):
        return {}


def _save_meta(data: dict):
    BACKUP_META.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")


def create_backup_archive(dest: Optional[Path] = None) -> Path:
    """创建 tar.gz 备份，返回文件路径。"""
    ts = datetime.now().strftime("%Y%m%d_%H%M%S")
    out_dir = dest.parent if dest else backup_dir()
    out_dir.mkdir(parents=True, exist_ok=True)
    out_path = dest or (out_dir / f"crypto_key_backup_{ts}.tar.gz")

    with tarfile.open(out_path, "w:gz") as tar:
        for name in BACKUP_FILES:
            p = BASE_DIR / name
            if p.exists():
                tar.add(p, arcname=name)
        if UPLOAD_DIR.exists():
            for f in UPLOAD_DIR.rglob("*"):
                if f.is_file():
                    tar.add(f, arcname=str(Path("uploads") / f.relative_to(UPLOAD_DIR)))

    meta = _load_meta()
    meta["last_backup"] = datetime.now().isoformat(timespec="seconds")
    meta["last_backup_path"] = str(out_path)
    meta["last_backup_size"] = out_path.stat().st_size
    _save_meta(meta)
    return out_path


def create_backup_bytes() -> Tuple[bytes, str]:
    """生成内存中的备份包，供前端下载。"""
    buf = io.BytesIO()
    with tarfile.open(fileobj=buf, mode="w:gz") as tar:
        for name in BACKUP_FILES:
            p = BASE_DIR / name
            if p.exists():
                tar.add(p, arcname=name)
        if UPLOAD_DIR.exists():
            for f in UPLOAD_DIR.rglob("*"):
                if f.is_file():
                    tar.add(f, arcname=str(Path("uploads") / f.relative_to(UPLOAD_DIR)))
    buf.seek(0)
    filename = f"crypto_key_backup_{datetime.now().strftime('%Y%m%d_%H%M%S')}.tar.gz"
    return buf.read(), filename


def restore_from_archive(fileobj) -> None:
    """从 tar.gz 恢复到项目目录。"""
    # 先解压到临时目录再覆盖，避免半成品
    tmp = BASE_DIR / f".restore_tmp_{int(time.time())}"
    if tmp.exists():
        shutil.rmtree(tmp)
    tmp.mkdir(parents=True)

    try:
        with tarfile.open(fileobj=fileobj, mode="r:gz") as tar:
            for member in tar.getmembers():
                name = member.name.replace("\\", "/").lstrip("./")
                if not name or name.startswith("/") or ".." in name.split("/"):
                    raise ValueError(f"非法备份路径: {member.name}")
                allowed = (
                    name in BACKUP_FILES
                    or name == "uploads"
                    or name.startswith("uploads/")
                )
                if not allowed:
                    continue
                member.name = name
                tar.extract(member, path=tmp)

        for name in BACKUP_FILES:
            src = tmp / name
            if src.exists():
                shutil.copy2(src, BASE_DIR / name)

        uploads_src = tmp / "uploads"
        if uploads_src.exists():
            UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
            for f in uploads_src.rglob("*"):
                if f.is_file():
                    rel = f.relative_to(uploads_src)
                    dest = UPLOAD_DIR / rel
                    dest.parent.mkdir(parents=True, exist_ok=True)
                    shutil.copy2(f, dest)

        meta = _load_meta()
        meta["last_restore"] = datetime.now().isoformat(timespec="seconds")
        _save_meta(meta)
    finally:
        shutil.rmtree(tmp, ignore_errors=True)


def get_backup_status() -> dict:
    meta = _load_meta()
    return {
        "last_backup": meta.get("last_backup"),
        "last_backup_path": meta.get("last_backup_path"),
        "last_backup_size": meta.get("last_backup_size"),
        "last_restore": meta.get("last_restore"),
        "auto_backup_dir": str(backup_dir()),
        "auto_backup_enabled": True,
    }


def _seconds_until_hour(hour: int = 3) -> float:
    now = datetime.now()
    target = now.replace(hour=hour, minute=0, second=0, microsecond=0)
    if target <= now:
        target += timedelta(days=1)
    return (target - now).total_seconds()


def _auto_backup_loop():
    while True:
        try:
            time.sleep(max(60, _seconds_until_hour(3)))
            create_backup_archive()
            # 清理超过 30 天的自动备份
            _cleanup_old_backups(keep_days=30)
        except Exception:
            time.sleep(3600)


def _cleanup_old_backups(keep_days: int = 30):
    cutoff = time.time() - keep_days * 86400
    d = backup_dir()
    for f in d.glob("crypto_key_backup_*.tar.gz"):
        try:
            if f.stat().st_mtime < cutoff:
                f.unlink()
        except OSError:
            pass


def start_auto_backup_scheduler():
    """启动每日自动备份后台线程（仅启动一次）。"""
    global _scheduler_started
    with _scheduler_lock:
        if _scheduler_started:
            return
        _scheduler_started = True
        t = threading.Thread(target=_auto_backup_loop, name="crypto-key-backup", daemon=True)
        t.start()