d61526698d
将 README 改写为产品文档；移除微信/EasyTier/工作台等遗留模块；新增 youLIVE 目标端监控、摄像头推流、质量守卫与转播历史能力。 Co-authored-by: Cursor <cursoragent@cursor.com>
118 lines
3.9 KiB
Python
118 lines
3.9 KiB
Python
"""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
|