"""Config file backup/restore before writes (simplified from sh2 config_governance.py).""" from __future__ import annotations import errno import shutil from datetime import datetime from pathlib import Path from typing import Any MAX_BACKUPS_PER_FILE = 24 def _now_stamp() -> str: return datetime.now().strftime("%Y-%m-%d_%H-%M-%S") def _backup_root(base_dir: Path) -> Path: return base_dir / "backup_config" / "__web__" def _history_dir(base_dir: Path, rel_name: str) -> Path: rel = Path(rel_name.replace("\\", "/").lstrip("/")) return _backup_root(base_dir) / rel.parent / f"{rel.name}.history" def backup_config_file(config_dir: Path, rel_name: str, *, reason: str = "save") -> dict[str, Any] | None: """Copy current config to history before overwrite. rel_name e.g. youtube.ini or relay_pro/foo.ini""" config_dir = Path(config_dir) target = config_dir / rel_name.replace("\\", "/").lstrip("/") if not target.is_file(): return None hist_dir = _history_dir(config_dir, rel_name) try: hist_dir.mkdir(parents=True, exist_ok=True) except OSError: return None suffix = target.suffix or ".bak" backup_name = f"{_now_stamp()}__{reason}{suffix}" backup_path = hist_dir / backup_name try: shutil.copy2(target, backup_path) except OSError: return None files = sorted( (p for p in hist_dir.iterdir() if p.is_file()), key=lambda p: p.stat().st_mtime, reverse=True, ) for old in files[MAX_BACKUPS_PER_FILE:]: try: old.unlink(missing_ok=True) except OSError: pass stat = backup_path.stat() return { "id": backup_name, "reason": reason, "label": backup_name, "created_at": datetime.fromtimestamp(stat.st_mtime).isoformat(timespec="seconds"), "size": stat.st_size, } def list_config_backups(config_dir: Path, rel_name: str) -> list[dict[str, Any]]: hist_dir = _history_dir(config_dir, rel_name) if not hist_dir.is_dir(): return [] out: list[dict[str, Any]] = [] for item in sorted(hist_dir.iterdir(), key=lambda p: p.stat().st_mtime, reverse=True): if not item.is_file(): continue stat = item.stat() stem = item.stem reason = "snapshot" if "__" in stem: _, _, reason = stem.partition("__") out.append( { "id": item.name, "reason": reason or "snapshot", "label": item.name, "created_at": datetime.fromtimestamp(stat.st_mtime).isoformat(timespec="seconds"), "size": stat.st_size, } ) return out def restore_config_backup(config_dir: Path, rel_name: str, backup_id: str) -> tuple[bool, str]: config_dir = Path(config_dir) rel = rel_name.replace("\\", "/").lstrip("/") target = config_dir / rel hist_dir = _history_dir(config_dir, rel_name) backup_path = hist_dir / Path(backup_id).name if not backup_path.is_file() or backup_path.parent != hist_dir: return False, "备份不存在" try: target.parent.mkdir(parents=True, exist_ok=True) if target.is_file(): backup_config_file(config_dir, rel_name, reason="pre_restore") shutil.copy2(backup_path, target) return True, "已恢复配置" except OSError as exc: if exc.errno in {errno.EACCES, errno.EPERM}: return False, f"无写入权限: {target}" return False, str(exc) def write_config_text(config_dir: Path, rel_name: str, content: str, *, reason: str = "save") -> Path: config_dir = Path(config_dir) rel = rel_name.replace("\\", "/").lstrip("/") target = config_dir / rel backup_config_file(config_dir, rel_name, reason=reason) target.parent.mkdir(parents=True, exist_ok=True) target.write_text(content, encoding="utf-8") return target