158 lines
5.0 KiB
Python
158 lines
5.0 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
|
||
|
||
|
||
def _normalize_url_line_for_dedupe(line: str) -> str:
|
||
from web2_relay_pro import extract_douyin_url_from_config_line
|
||
|
||
url = extract_douyin_url_from_config_line(line)
|
||
if url:
|
||
return url.rstrip("/").lower()
|
||
t = (line or "").strip()
|
||
if not t or t.startswith("#"):
|
||
return ""
|
||
return t.rstrip("/").lower()
|
||
|
||
|
||
def dedupe_url_config_content(content: str) -> tuple[str, int]:
|
||
"""去除 URL_config 中重复的抖音直播间行（参照 d2ypp2 config_governance dedupe_urls）。"""
|
||
seen: set[str] = set()
|
||
removed = 0
|
||
out: list[str] = []
|
||
for raw in content.splitlines():
|
||
line = raw.strip()
|
||
if not line or line.startswith("#"):
|
||
out.append(raw.rstrip())
|
||
continue
|
||
key = _normalize_url_line_for_dedupe(line)
|
||
if not key:
|
||
out.append(raw.rstrip())
|
||
continue
|
||
if "douyin" not in key:
|
||
out.append(raw.rstrip())
|
||
continue
|
||
if key in seen:
|
||
removed += 1
|
||
continue
|
||
seen.add(key)
|
||
out.append(raw.rstrip())
|
||
text = "\n".join(out).rstrip()
|
||
if text:
|
||
text += "\n"
|
||
return text, removed
|