Files
kairee 5c2ab7b187 feat(p1): add daily snapshots and fan tracking
Implement P1: seven-metric view snapshots, relation/stat fan snapshots, failure placeholders, resume by daily completion markers, Cookie login warnings, daily SQLite backups, Shanghai daily scheduling, and CLI snapshot/auth-check commands. Real read-only run captured 795/795 videos successfully over 1124s with a verified daily backup.
2026-10-31 19:19:24 +08:00

51 lines
1.3 KiB
Python

# -*- coding: utf-8 -*-
"""备份与保留期清理。"""
from __future__ import annotations
import time
from pathlib import Path
from ..database import Database
def daily_backup_path(backup_dir: str | Path, stat_date: str) -> Path:
compact = stat_date.replace("-", "")
return Path(backup_dir) / f"app-{compact}.db"
def backup_daily(
db: Database,
backup_dir: str | Path,
stat_date: str,
*,
logger=None,
) -> Path:
target = daily_backup_path(backup_dir, stat_date)
db.backup_to(target)
if logger is not None:
logger.info("backup.daily", path=str(target), stat_date=stat_date)
return target
def prune_daily_backups(
backup_dir: str | Path,
*,
keep_days: int,
now: float | None = None,
) -> list[Path]:
"""只清理 ``app-YYYYMMDD.db`` 日备份,不动迁移备份。"""
target_dir = Path(backup_dir)
if not target_dir.is_dir():
return []
cutoff = (time.time() if now is None else now) - keep_days * 86400
removed: list[Path] = []
for path in target_dir.glob("app-[0-9][0-9][0-9][0-9][0-9][0-9][0-9][0-9].db"):
try:
if path.stat().st_mtime < cutoff:
path.unlink()
removed.append(path)
except OSError:
continue
return removed