dekun
88 lines
2.3 KiB
Python
88 lines
2.3 KiB
Python
"""SQLite 在线备份与恢复。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
import sqlite3
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from zoneinfo import ZoneInfo
|
|
|
|
_BACKUP_NAME_RE = re.compile(r"^market_intel_\d{8}_\d{6}\.db$")
|
|
|
|
|
|
def _now_stamp(tz: str = "Asia/Shanghai") -> str:
|
|
return datetime.now(ZoneInfo(tz)).strftime("%Y%m%d_%H%M%S")
|
|
|
|
|
|
def ensure_backup_dir(path: Path) -> Path:
|
|
path.mkdir(parents=True, exist_ok=True)
|
|
return path
|
|
|
|
|
|
def is_valid_backup_name(name: str) -> bool:
|
|
return bool(_BACKUP_NAME_RE.fullmatch(name))
|
|
|
|
|
|
def resolve_backup_file(backup_dir: Path, name: str) -> Path:
|
|
if not is_valid_backup_name(name):
|
|
raise ValueError("invalid backup filename")
|
|
path = (backup_dir / name).resolve()
|
|
root = backup_dir.resolve()
|
|
if not str(path).startswith(str(root)):
|
|
raise ValueError("invalid backup path")
|
|
if not path.is_file():
|
|
raise FileNotFoundError(name)
|
|
return path
|
|
|
|
|
|
def backup_db(db_path: Path, backup_dir: Path, tz: str = "Asia/Shanghai") -> Path:
|
|
ensure_backup_dir(backup_dir)
|
|
if not db_path.is_file():
|
|
raise FileNotFoundError(f"database not found: {db_path}")
|
|
|
|
dest = backup_dir / f"market_intel_{_now_stamp(tz)}.db"
|
|
src = sqlite3.connect(str(db_path))
|
|
try:
|
|
dst = sqlite3.connect(str(dest))
|
|
try:
|
|
src.backup(dst)
|
|
finally:
|
|
dst.close()
|
|
finally:
|
|
src.close()
|
|
return dest
|
|
|
|
|
|
def restore_db(backup_path: Path, db_path: Path) -> None:
|
|
if not backup_path.is_file():
|
|
raise FileNotFoundError(f"backup not found: {backup_path}")
|
|
|
|
src = sqlite3.connect(str(backup_path))
|
|
try:
|
|
dst = sqlite3.connect(str(db_path))
|
|
try:
|
|
src.backup(dst)
|
|
finally:
|
|
dst.close()
|
|
finally:
|
|
src.close()
|
|
|
|
|
|
def list_backups(backup_dir: Path) -> list[dict]:
|
|
if not backup_dir.is_dir():
|
|
return []
|
|
items: list[dict] = []
|
|
for p in sorted(backup_dir.glob("market_intel_*.db"), reverse=True):
|
|
if not p.is_file() or not is_valid_backup_name(p.name):
|
|
continue
|
|
st = p.stat()
|
|
items.append(
|
|
{
|
|
"name": p.name,
|
|
"size_bytes": st.st_size,
|
|
"modified_ms": int(st.st_mtime * 1000),
|
|
}
|
|
)
|
|
return items
|