Codex
133 lines
4.2 KiB
Python
133 lines
4.2 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import shutil
|
|
import tempfile
|
|
from typing import Iterable
|
|
|
|
|
|
BACKUP_DIR_RE = re.compile(r"^\d{8}-\d{6}-\d{6}-")
|
|
BACKUP_KEEP_COUNT = 50
|
|
|
|
|
|
def atomic_write_text(path: Path, text: str) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
original_stat = path.stat() if path.exists() else None
|
|
temp_path: Path | None = None
|
|
try:
|
|
with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=path.parent, delete=False) as handle:
|
|
temp_path = Path(handle.name)
|
|
handle.write(text)
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
if original_stat is not None:
|
|
os.chmod(temp_path, original_stat.st_mode & 0o7777)
|
|
try:
|
|
os.chown(temp_path, original_stat.st_uid, original_stat.st_gid)
|
|
except OSError:
|
|
pass
|
|
os.replace(temp_path, path)
|
|
temp_path = None
|
|
try:
|
|
dir_fd = os.open(path.parent, os.O_DIRECTORY)
|
|
try:
|
|
os.fsync(dir_fd)
|
|
finally:
|
|
os.close(dir_fd)
|
|
except OSError:
|
|
pass
|
|
finally:
|
|
if temp_path is not None and temp_path.exists():
|
|
temp_path.unlink()
|
|
|
|
|
|
def apply_source_permissions(target: Path, source: Path) -> None:
|
|
if not source.exists():
|
|
return
|
|
source_stat = source.stat()
|
|
os.chmod(target, source_stat.st_mode & 0o7777)
|
|
try:
|
|
os.chown(target, source_stat.st_uid, source_stat.st_gid)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def data_backup_root(paths: Iterable[Path]) -> Path:
|
|
path_list = list(paths)
|
|
if not path_list:
|
|
raise ValueError("备份文件不能为空")
|
|
return path_list[0].parent / "backups"
|
|
|
|
|
|
def create_backup_directory(backup_root: Path, operation: str) -> Path:
|
|
backup_root.mkdir(parents=True, exist_ok=True)
|
|
timestamp = datetime.now().strftime("%Y%m%d-%H%M%S-%f")
|
|
base_name = f"{timestamp}-{operation}"
|
|
for index in range(1000):
|
|
suffix = "" if index == 0 else f"-{index}"
|
|
backup_dir = backup_root / f"{base_name}{suffix}"
|
|
try:
|
|
backup_dir.mkdir()
|
|
return backup_dir
|
|
except FileExistsError:
|
|
continue
|
|
raise RuntimeError("无法创建唯一备份目录")
|
|
|
|
|
|
def create_data_backup(
|
|
operation: str,
|
|
file_contents: dict[Path, str],
|
|
submitted_lines: list[str],
|
|
) -> Path:
|
|
backup_root = data_backup_root(file_contents.keys())
|
|
backup_dir = create_backup_directory(backup_root, operation)
|
|
try:
|
|
submitted_text = "\n".join(submitted_lines)
|
|
metadata = {
|
|
"backup_id": backup_dir.name,
|
|
"created_at": datetime.now().isoformat(timespec="microseconds"),
|
|
"operation": operation,
|
|
"submitted_lines_count": len(submitted_lines),
|
|
"submitted_lines_sha256": hashlib.sha256(submitted_text.encode("utf-8")).hexdigest(),
|
|
"files": [],
|
|
}
|
|
for source_path, content in file_contents.items():
|
|
target_path = backup_dir / source_path.name
|
|
atomic_write_text(target_path, content)
|
|
apply_source_permissions(target_path, source_path)
|
|
source_stat = source_path.stat()
|
|
metadata["files"].append(
|
|
{
|
|
"name": source_path.name,
|
|
"source_path": str(source_path),
|
|
"size": len(content.encode("utf-8")),
|
|
"mtime": source_stat.st_mtime,
|
|
}
|
|
)
|
|
metadata_path = backup_dir / "metadata.json"
|
|
atomic_write_text(
|
|
metadata_path,
|
|
json.dumps(metadata, ensure_ascii=False, indent=2) + "\n",
|
|
)
|
|
return backup_dir
|
|
except Exception:
|
|
shutil.rmtree(backup_dir, ignore_errors=True)
|
|
raise
|
|
|
|
|
|
def prune_data_backups(backup_root: Path, keep: int = BACKUP_KEEP_COUNT) -> None:
|
|
if keep < 1 or not backup_root.exists():
|
|
return
|
|
backup_dirs = sorted(
|
|
path
|
|
for path in backup_root.iterdir()
|
|
if path.is_dir() and BACKUP_DIR_RE.match(path.name)
|
|
)
|
|
for backup_dir in backup_dirs[:-keep]:
|
|
shutil.rmtree(backup_dir)
|