# Copyright (c) 2025-2026 马建军. All rights reserved. # 专有软件 — 未经授权禁止复制、传播、转售。 # 严禁用于：带单/代客理财、向他人推荐期货品种或买卖建议、融资配资等业务。 # 详见 LICENSE.zh-CN.txt 与 docs/软件购买与使用协议.md """数据库备份：SQLite futures.db 或 PostgreSQL pg_dump，含 uploads 与一键恢复脚本。""" from __future__ import annotations import json import logging import os import re import shutil import sqlite3 import subprocess import sys import tarfile import tempfile import threading import time from datetime import datetime from pathlib import Path from typing import Any, Callable, IO, Optional from zoneinfo import ZoneInfo from modules.core.db_conn import DB_PATH, db_backend logger = logging.getLogger(__name__) TZ = ZoneInfo("Asia/Shanghai") BACKUP_FILENAME_RE = re.compile(r"^qihuo_backup_\d{8}_\d{6}\.tar\.gz$") BACKUP_LAST_KEY = "backup_last_at" BACKUP_KEEP_KEY = "backup_keep_count" BACKUP_AUTO_KEY = "backup_auto_enabled" BACKUP_HOUR_KEY = "backup_auto_hour" DEFAULT_KEEP_COUNT = 30 DEFAULT_AUTO_HOUR = 3 CHECK_INTERVAL_SEC = 3600 _backup_lock = threading.Lock() RESTORE_STATUS_FILE = "restore_status.json" RESTORE_CONFIRM_TOKEN = "RESTORE" RESTORE_MD = """# qihuo 备份恢复说明 本压缩包由 qihuo 系统自动生成，可在另一台 Linux 服务器上恢复交易数据。 ## 包内文件 | 文件/目录 | 说明 | |-----------|------| | `futures.db` | SQLite 主库（仅 SQLite 模式备份） | | `postgres_dump.sql` | PostgreSQL 逻辑备份（仅 PostgreSQL 模式） | | `uploads/` | 复盘截图与 K 线图（若备份时存在） | | `.env` | 环境配置（`config/.env` 或根目录 `.env`） | | `manifest.json` | 备份元数据（含 `backend` 字段） | | `restore.sh` | 一键恢复脚本 | ## 快速恢复（推荐） 1. 将本压缩包上传到目标服务器（例如 `/root/`） 2. 解压并执行恢复脚本： ```bash cd /root tar -xzf qihuo_backup_YYYYMMDD_HHMMSS.tar.gz cd qihuo_backup_YYYYMMDD_HHMMSS chmod +x restore.sh ./restore.sh ``` 默认恢复到 **`/root/qihuo`**（SQLite）或导入到 `.env` 中的 PostgreSQL（见 manifest）。 指定应用目录： ```bash RESTORE_DIR=/opt/qihuo ./restore.sh ``` 3. 在新服务器部署 qihuo 代码与 Python 环境（见 `docs/POSTGRES.md` / `docs/DEPLOY.md`） 4. 若包内含 `.env`，`restore.sh` 会自动恢复到应用目录（无需手工复制） 5. 重启服务：`pm2 restart qihuo` ## PostgreSQL 恢复 若 `manifest.json` 中 `"backend": "postgres"`： 1. 确保目标机已安装 PostgreSQL，且 `.env` 中 `DATABASE_URL` 指向空库或待覆盖库 2. 执行 `./restore.sh`（会调用 `psql` 导入 `postgres_dump.sql`） 手工导入： ```bash export DATABASE_URL=postgresql://qihuo:密码@127.0.0.1:5432/qihuo psql "$DATABASE_URL" -f postgres_dump.sql ``` ## SQLite 手工恢复 ```bash mkdir -p /opt/qihuo/uploads cp futures.db /opt/qihuo/futures.db cp -a uploads/. /opt/qihuo/uploads/ ``` ## 注意 - 恢复前请停止 qihuo 进程 - `.env` 含敏感信息，请妥善保管备份包 - 详见 `docs/POSTGRES.md` 与 `docs/BACKUP.md` """ def _app_root() -> Path: from modules.core.paths import ROOT return ROOT def default_backup_dir() -> str: env = (os.getenv("QIHUO_BACKUP_DIR") or "").strip() if env: return env if os.name == "nt": return str(_app_root() / "qihuo_backup") return "/root/qihuo_backup" def default_restore_dir() -> str: env = (os.getenv("QIHUO_RESTORE_DIR") or "").strip() if env: return env if os.name == "nt": return str(_app_root()) return "/root/qihuo" def restore_target_dir() -> Path: """Web/API 恢复目标目录，默认当前应用根目录。""" env = (os.getenv("QIHUO_RESTORE_DIR") or "").strip() if env: return Path(env) return _app_root() def backup_dir() -> Path: path = Path(default_backup_dir()) path.mkdir(parents=True, exist_ok=True) return path def backup_in_progress() -> bool: return _backup_lock.locked() def _restore_status_path() -> Path: from modules.core.paths import DATA_DIR DATA_DIR.mkdir(parents=True, exist_ok=True) return DATA_DIR / RESTORE_STATUS_FILE def _write_restore_status(state: str, message: str = "", **extra: Any) -> None: payload = { "state": state, "message": message, "updated_at": datetime.now(TZ).isoformat(timespec="seconds"), } payload.update(extra) _restore_status_path().write_text( json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8", ) def get_restore_status() -> dict: path = _restore_status_path() if not path.is_file(): return {"state": "idle", "message": "", "updated_at": ""} try: data = json.loads(path.read_text(encoding="utf-8")) if isinstance(data, dict): data.setdefault("state", "idle") data.setdefault("message", "") return data except Exception: pass return {"state": "idle", "message": "", "updated_at": ""} def restore_in_progress() -> bool: return get_restore_status().get("state") in ("pending", "running") def _manifest_root_prefix(tar: tarfile.TarFile) -> str: for member in tar.getmembers(): name = member.name.rstrip("/") if name.endswith("/manifest.json") or name == "manifest.json": if name == "manifest.json": return "" return name[: -len("/manifest.json")] raise ValueError("备份包缺少 manifest.json") def _read_manifest_from_tar(tar: tarfile.TarFile) -> dict: root = _manifest_root_prefix(tar) manifest_name = f"{root}/manifest.json" if root else "manifest.json" try: member = tar.getmember(manifest_name) except KeyError as exc: raise ValueError("备份包缺少 manifest.json") from exc extracted = tar.extractfile(member) if not extracted: raise ValueError("无法读取 manifest.json") data = json.loads(extracted.read().decode("utf-8")) if not isinstance(data, dict): raise ValueError("manifest.json 格式无效") return data def _validate_manifest(manifest: dict, *, current_backend: str | None = None) -> str: if manifest.get("app") != "qihuo": raise ValueError("不是有效的 qihuo 备份包") backend = (manifest.get("backend") or "").strip() if backend not in ("sqlite", "postgres"): raise ValueError("manifest 缺少或无效的数据库类型") if current_backend and backend != current_backend: label = "PostgreSQL" if backend == "postgres" else "SQLite" cur = "PostgreSQL" if current_backend == "postgres" else "SQLite" raise ValueError(f"备份为 {label}，当前服务为 {cur}，无法恢复") return backend def _member_exists(tar: tarfile.TarFile, root: str, name: str) -> bool: candidates = [name] if root: candidates.append(f"{root}/{name}") return any(_tar_has_member(tar, path) for path in candidates) def _tar_has_member(tar: tarfile.TarFile, path: str) -> bool: try: tar.getmember(path) return True except KeyError: return False def _validate_archive_contents(tar: tarfile.TarFile, manifest: dict, root: str) -> None: backend = manifest["backend"] if backend == "sqlite": if not _member_exists(tar, root, "futures.db"): raise ValueError("SQLite 备份缺少 futures.db") elif not _member_exists(tar, root, "postgres_dump.sql"): raise ValueError("PostgreSQL 备份缺少 postgres_dump.sql") def _manifest_preview(manifest: dict, path: Path) -> dict: backend = manifest.get("backend", "") stat = path.stat() created_at = (manifest.get("created_at") or "").strip() return { "name": path.name, "backend": backend, "backend_label": "PostgreSQL" if backend == "postgres" else "SQLite", "created_at": created_at, "includes_uploads": bool(manifest.get("includes_uploads")), "includes_env": bool(manifest.get("includes_env")), "env_restore_path": (manifest.get("env_restore_path") or "").strip(), "size": stat.st_size, "size_mb": round(stat.st_size / (1024 * 1024), 2), "mtime": datetime.fromtimestamp(stat.st_mtime, TZ).isoformat(timespec="seconds"), } def peek_manifest(path: Path) -> dict: with tarfile.open(path, "r:gz") as tar: return _read_manifest_from_tar(tar) def inspect_backup_archive(path: Path, *, check_backend: bool = True) -> dict: with tarfile.open(path, "r:gz") as tar: manifest = _read_manifest_from_tar(tar) current = db_backend() if check_backend else None _validate_manifest(manifest, current_backend=current) root = _manifest_root_prefix(tar) _validate_archive_contents(tar, manifest, root) return _manifest_preview(manifest, path) def _allocate_backup_filename(manifest: dict, preferred: str = "") -> str: preferred = (preferred or "").strip() if preferred and BACKUP_FILENAME_RE.match(preferred): candidate = backup_dir() / preferred if not candidate.exists(): return preferred created = (manifest.get("created_at") or "").strip() stamp = "" if created: try: stamp = datetime.fromisoformat(created).strftime("%Y%m%d_%H%M%S") except ValueError: stamp = "" if not stamp: stamp = datetime.now(TZ).strftime("%Y%m%d_%H%M%S") name = f"qihuo_backup_{stamp}.tar.gz" if not (backup_dir() / name).exists(): return name stamp = datetime.now(TZ).strftime("%Y%m%d_%H%M%S") return f"qihuo_backup_{stamp}.tar.gz" def save_uploaded_backup(stream: IO[bytes], original_filename: str = "") -> tuple[str, dict]: with tempfile.NamedTemporaryFile(delete=False, suffix=".tar.gz") as tmp: shutil.copyfileobj(stream, tmp) tmp_path = Path(tmp.name) try: info = inspect_backup_archive(tmp_path, check_backend=True) manifest = peek_manifest(tmp_path) filename = _allocate_backup_filename(manifest, original_filename) dest = backup_dir() / filename shutil.move(str(tmp_path), str(dest)) info["name"] = filename return filename, info except Exception: tmp_path.unlink(missing_ok=True) raise def _pm2_available() -> bool: return shutil.which("pm2") is not None def _pm2_stop() -> None: if not _pm2_available(): logger.warning("pm2 not found, skip stop") return proc = subprocess.run( ["pm2", "stop", "qihuo"], capture_output=True, text=True, check=False, ) if proc.returncode != 0: logger.warning("pm2 stop qihuo: %s", proc.stderr.strip() or proc.stdout.strip()) def _pm2_restart() -> None: if not _pm2_available(): logger.warning("pm2 not found, skip restart") return proc = subprocess.run( ["pm2", "restart", "qihuo"], capture_output=True, text=True, check=False, ) if proc.returncode != 0: proc = subprocess.run( ["pm2", "start", "qihuo"], capture_output=True, text=True, check=False, ) if proc.returncode != 0: raise RuntimeError(proc.stderr.strip() or proc.stdout.strip() or "pm2 restart 失败") def _extract_member_to_path(tar: tarfile.TarFile, member_name: str, dest: Path) -> None: try: member = tar.getmember(member_name) except KeyError: return extracted = tar.extractfile(member) if not extracted: return dest.parent.mkdir(parents=True, exist_ok=True) with open(dest, "wb") as out: shutil.copyfileobj(extracted, out) def _restore_uploads_dir(tar: tarfile.TarFile, root: str, restore_dir: Path) -> None: prefix = f"{root}/uploads" if root else "uploads" uploads_dest = restore_dir / "uploads" uploads_dest.mkdir(parents=True, exist_ok=True) found = False for member in tar.getmembers(): if member.name == prefix or member.name.startswith(prefix + "/"): found = True rel = member.name[len(prefix) :].lstrip("/") if not rel: continue target = uploads_dest / rel if member.isdir(): target.mkdir(parents=True, exist_ok=True) else: target.parent.mkdir(parents=True, exist_ok=True) extracted = tar.extractfile(member) if extracted: with open(target, "wb") as out: shutil.copyfileobj(extracted, out) if not found: logger.info("backup has no uploads/") def _reload_env_file(env_path: Path) -> None: if not env_path.is_file(): return try: from dotenv import load_dotenv load_dotenv(str(env_path), override=True) except Exception as exc: logger.warning("reload .env failed: %s", exc) def _restore_postgres_dump(dump_path: Path) -> None: url = (os.getenv("DATABASE_URL") or "").strip() if not url: raise RuntimeError("PostgreSQL 恢复需要 DATABASE_URL（请先恢复 .env 或检查环境变量）") if not shutil.which("psql"): raise RuntimeError("未找到 psql，请先安装 PostgreSQL 客户端") proc = subprocess.run( ["psql", url, "-f", str(dump_path)], capture_output=True, text=True, check=False, ) if proc.returncode != 0: raise RuntimeError(proc.stderr.strip() or proc.stdout.strip() or "psql 导入失败") def _perform_restore(archive_path: Path, restore_dir: Path) -> dict: restore_dir.mkdir(parents=True, exist_ok=True) with tempfile.TemporaryDirectory(prefix="qihuo_restore_") as tmp: work = Path(tmp) with tarfile.open(archive_path, "r:gz") as tar: manifest = _read_manifest_from_tar(tar) backend = _validate_manifest(manifest, current_backend=db_backend()) root = _manifest_root_prefix(tar) _validate_archive_contents(tar, manifest, root) env_member = f"{root}/.env" if root else ".env" env_restore_path = (manifest.get("env_restore_path") or "config/.env").strip() if manifest.get("includes_env") and _tar_has_member(tar, env_member): env_dest = restore_dir / env_restore_path _extract_member_to_path(tar, env_member, env_dest) _reload_env_file(env_dest) if backend == "sqlite": db_member = f"{root}/futures.db" if root else "futures.db" db_dest = Path(DB_PATH) if not db_dest.is_absolute(): db_dest = restore_dir / db_dest.name _extract_member_to_path(tar, db_member, db_dest) else: dump_member = f"{root}/postgres_dump.sql" if root else "postgres_dump.sql" dump_path = work / "postgres_dump.sql" _extract_member_to_path(tar, dump_member, dump_path) _restore_postgres_dump(dump_path) _restore_uploads_dir(tar, root, restore_dir) return { "backend": backend, "restore_dir": str(restore_dir), "includes_env": bool(manifest.get("includes_env")), "includes_uploads": bool(manifest.get("includes_uploads")), } def run_restore_job(archive_path: Path) -> None: filename = archive_path.name restore_dir = restore_target_dir() try: _write_restore_status( "running", "正在停止服务…", filename=filename, step="stop", restore_dir=str(restore_dir), ) _pm2_stop() _write_restore_status( "running", "正在恢复数据…", filename=filename, step="restore", restore_dir=str(restore_dir), ) summary = _perform_restore(archive_path.resolve(), restore_dir) _write_restore_status( "running", "正在重启服务…", filename=filename, step="restart", restore_dir=str(restore_dir), ) _pm2_restart() _write_restore_status( "done", "恢复完成，服务已重启", filename=filename, restore_dir=str(restore_dir), summary=summary, ) except Exception as exc: logger.exception("restore failed: %s", exc) _write_restore_status( "error", str(exc), filename=filename, restore_dir=str(restore_dir), ) try: _pm2_restart() except Exception as restart_exc: logger.warning("restart after restore error: %s", restart_exc) def schedule_restore(filename: str) -> tuple[bool, str]: if _backup_lock.locked(): return False, "备份进行中，请稍后再试" if restore_in_progress(): return False, "恢复进行中，请稍后再试" try: path = resolve_backup_file(filename) inspect_backup_archive(path, check_backend=True) except (ValueError, FileNotFoundError) as exc: return False, str(exc) _write_restore_status( "pending", "恢复任务已提交…", filename=filename, restore_dir=str(restore_target_dir()), ) script = Path(__file__).resolve().parent / "restore_job.py" subprocess.Popen( [sys.executable, str(script), str(path.resolve())], start_new_session=True, cwd=str(_app_root()), stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) return True, "恢复已开始，服务将短暂中断后自动重启" def get_backup_last_at(get_setting: Callable[[str, str], str]) -> str: return (get_setting(BACKUP_LAST_KEY, "") or "").strip() def _backup_sqlite(src_path: str, dst_path: str) -> None: src = sqlite3.connect(src_path, timeout=30) try: try: src.execute("PRAGMA wal_checkpoint(TRUNCATE)") except sqlite3.OperationalError: pass dst = sqlite3.connect(dst_path) try: src.backup(dst) dst.commit() finally: dst.close() finally: src.close() def _backup_postgres(dst_path: str) -> None: url = (os.getenv("DATABASE_URL") or "").strip() if not url: raise RuntimeError("PostgreSQL 备份需要 DATABASE_URL") env = os.environ.copy() proc = subprocess.run( ["pg_dump", "--no-owner", "--no-acl", "-f", dst_path, url], capture_output=True, text=True, env=env, check=False, ) if proc.returncode != 0: raise RuntimeError(f"pg_dump 失败: {proc.stderr.strip() or proc.stdout.strip()}") def _env_backup_info() -> tuple[Optional[Path], str]: """返回 (源 .env 路径, 恢复到应用目录的相对路径)。""" from modules.core.paths import CONFIG_DIR, LEGACY_ENV_FILE, ROOT, resolve_env_file src = Path(resolve_env_file()) if not src.is_file(): return None, "config/.env" try: if src.resolve().parent == CONFIG_DIR.resolve(): return src, "config/.env" if src.resolve().parent == ROOT.resolve() and src.name == ".env": if LEGACY_ENV_FILE.is_file() and not (CONFIG_DIR / ".env").is_file(): return src, ".env" except Exception: pass return src, "config/.env" def _write_restore_script(dest: Path, *, backend: str, env_restore_path: str = "") -> None: pg_block = "" if backend == "postgres": pg_block = """ if [ -f "$SCRIPT_DIR/postgres_dump.sql" ]; then if [ -z "${DATABASE_URL:-}" ]; then if [ -f "$RESTORE_DIR/.env" ]; then set -a # shellcheck disable=SC1090 source "$RESTORE_DIR/.env" set +a fi fi if [ -z "${DATABASE_URL:-}" ]; then echo "错误: PostgreSQL 恢复需要 DATABASE_URL（环境变量或 $RESTORE_DIR/.env）" exit 1 fi if ! command -v psql >/dev/null; then echo "错误: 未找到 psql，请先安装 PostgreSQL 客户端" exit 1 fi echo "导入 PostgreSQL: postgres_dump.sql" psql "$DATABASE_URL" -f "$SCRIPT_DIR/postgres_dump.sql" echo "PostgreSQL 导入完成" fi """ env_block = "" if env_restore_path: env_block = f""" if [ -f "$SCRIPT_DIR/.env" ]; then ENV_DEST="$RESTORE_DIR/{env_restore_path}" mkdir -p "$(dirname "$ENV_DEST")" cp -f "$SCRIPT_DIR/.env" "$ENV_DEST" echo "已复制 .env -> $ENV_DEST" fi """ script = f"""#!/bin/bash set -euo pipefail RESTORE_DIR="${{RESTORE_DIR:-{default_restore_dir()}}}" SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" mkdir -p "$RESTORE_DIR/uploads" {pg_block}{env_block} if [ -f "$SCRIPT_DIR/futures.db" ]; then cp -f "$SCRIPT_DIR/futures.db" "$RESTORE_DIR/futures.db" echo "已复制 futures.db -> $RESTORE_DIR/futures.db" fi if [ -d "$SCRIPT_DIR/uploads" ]; then cp -a "$SCRIPT_DIR/uploads/." "$RESTORE_DIR/uploads/" echo "已复制 uploads -> $RESTORE_DIR/uploads/" fi echo "" echo "恢复完成。目标目录: $RESTORE_DIR" echo "下一步: pm2 restart qihuo" echo "详见 RESTORE.md 与 docs/POSTGRES.md" """ dest.write_text(script, encoding="utf-8") def create_backup(*, include_uploads: bool = True) -> tuple[str, str]: """创建 tar.gz 备份，返回 (文件名, 说明)。""" backend = db_backend() if backend == "sqlite" and not os.path.isfile(DB_PATH): raise FileNotFoundError(f"数据库不存在: {DB_PATH}") if backend == "postgres" and not (os.getenv("DATABASE_URL") or "").strip(): raise RuntimeError("PostgreSQL 模式需要 DATABASE_URL") with _backup_lock: stamp = datetime.now(TZ).strftime("%Y%m%d_%H%M%S") folder_name = f"qihuo_backup_{stamp}" filename = f"{folder_name}.tar.gz" out_path = backup_dir() / filename app_root = _app_root() upload_src = app_root / "uploads" with tempfile.TemporaryDirectory(prefix="qihuo_bak_") as tmp: work = Path(tmp) / folder_name work.mkdir() if backend == "postgres": _backup_postgres(str(work / "postgres_dump.sql")) else: _backup_sqlite(DB_PATH, str(work / "futures.db")) if include_uploads and upload_src.is_dir(): shutil.copytree(upload_src, work / "uploads", dirs_exist_ok=True) env_src, env_restore_path = _env_backup_info() includes_env = False if env_src and env_src.is_file(): shutil.copy2(env_src, work / ".env") includes_env = True manifest = { "app": "qihuo", "backend": backend, "created_at": datetime.now(TZ).isoformat(timespec="seconds"), "db_path": DB_PATH if backend == "sqlite" else (os.getenv("DATABASE_URL") or ""), "includes_uploads": include_uploads and upload_src.is_dir(), "includes_env": includes_env, "env_restore_path": env_restore_path if includes_env else "", "default_restore_dir": default_restore_dir(), "files": sorted(p.name for p in work.iterdir()), } (work / "manifest.json").write_text( json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8", ) (work / "RESTORE.md").write_text(RESTORE_MD, encoding="utf-8") _write_restore_script( work / "restore.sh", backend=backend, env_restore_path=env_restore_path if includes_env else "", ) with tarfile.open(out_path, "w:gz") as tar: tar.add(work, arcname=folder_name) size_mb = out_path.stat().st_size / (1024 * 1024) label = "PostgreSQL" if backend == "postgres" else "SQLite" return filename, f"备份已生成 {filename}（{label}，{size_mb:.2f} MB）" def list_backups(*, with_manifest: bool = True) -> list[dict]: items: list[dict] = [] for path in sorted(backup_dir().glob("qihuo_backup_*.tar.gz"), reverse=True): if not BACKUP_FILENAME_RE.match(path.name): continue stat = path.stat() item = { "name": path.name, "size": stat.st_size, "size_mb": round(stat.st_size / (1024 * 1024), 2), "mtime": datetime.fromtimestamp(stat.st_mtime, TZ).isoformat(timespec="seconds"), "backend": "", "backend_label": "", "created_at": "", "includes_env": False, "includes_uploads": False, } if with_manifest: try: manifest = peek_manifest(path) item["backend"] = manifest.get("backend", "") item["backend_label"] = ( "PostgreSQL" if manifest.get("backend") == "postgres" else "SQLite" ) item["created_at"] = (manifest.get("created_at") or "").strip() item["includes_env"] = bool(manifest.get("includes_env")) item["includes_uploads"] = bool(manifest.get("includes_uploads")) except Exception as exc: logger.debug("read manifest %s: %s", path.name, exc) items.append(item) return items def resolve_backup_file(filename: str) -> Path: name = (filename or "").strip() if not BACKUP_FILENAME_RE.match(name): raise ValueError("无效的备份文件名") path = (backup_dir() / name).resolve() root = backup_dir().resolve() if not str(path).startswith(str(root) + os.sep) and path != root: raise ValueError("无效的备份路径") if not path.is_file(): raise FileNotFoundError("备份文件不存在") return path def prune_old_backups(keep: int) -> int: keep_n = max(1, int(keep or DEFAULT_KEEP_COUNT)) files = list_backups() removed = 0 for item in files[keep_n:]: try: resolve_backup_file(item["name"]).unlink() removed += 1 except Exception as exc: logger.warning("prune backup %s: %s", item["name"], exc) return removed def run_backup_job( *, get_setting: Callable[[str, str], str], set_setting: Callable[[str, str], None], include_uploads: bool = True, ) -> tuple[str, str]: keep = DEFAULT_KEEP_COUNT try: keep = max(5, min(200, int(get_setting(BACKUP_KEEP_KEY, str(DEFAULT_KEEP_COUNT)) or DEFAULT_KEEP_COUNT))) except ValueError: pass filename, msg = create_backup(include_uploads=include_uploads) set_setting(BACKUP_LAST_KEY, datetime.now(TZ).isoformat(timespec="seconds")) removed = prune_old_backups(keep) if removed: msg = f"{msg}，已清理 {removed} 个旧备份" return filename, msg def schedule_backup( *, get_setting: Callable[[str, str], str], set_setting: Callable[[str, str], None], include_uploads: bool = True, ) -> tuple[bool, str]: if _backup_lock.locked(): return False, "备份进行中，请稍后再试" if restore_in_progress(): return False, "恢复进行中，请稍后再试" def _run() -> None: try: run_backup_job( get_setting=get_setting, set_setting=set_setting, include_uploads=include_uploads, ) except Exception as exc: logger.exception("backup failed: %s", exc) threading.Thread(target=_run, daemon=True, name="qihuo-backup-run").start() return True, "已在后台开始备份，请稍后刷新本页查看" def _should_auto_backup(get_setting: Callable[[str, str], str]) -> bool: if (get_setting(BACKUP_AUTO_KEY, "1") or "1").strip() not in ("1", "true", "yes"): return False try: hour = int(get_setting(BACKUP_HOUR_KEY, str(DEFAULT_AUTO_HOUR)) or DEFAULT_AUTO_HOUR) except ValueError: hour = DEFAULT_AUTO_HOUR hour = max(0, min(23, hour)) now = datetime.now(TZ) if now.hour != hour: return False last = get_backup_last_at(get_setting) if last and last[:10] == now.date().isoformat(): return False return True def start_backup_worker( *, get_setting_fn: Callable[[str, str], str], set_setting_fn: Callable[[str, str], None], interval: int = CHECK_INTERVAL_SEC, ) -> None: """后台线程：按设定小时每日自动备份。""" def _loop() -> None: time.sleep(30) while True: try: if _should_auto_backup(get_setting_fn): filename, msg = run_backup_job( get_setting=get_setting_fn, set_setting=set_setting_fn, include_uploads=True, ) logger.info("auto backup: %s — %s", filename, msg) except Exception as exc: logger.warning("backup worker: %s", exc) time.sleep(max(300, interval)) threading.Thread(target=_loop, daemon=True, name="qihuo-backup-worker").start()