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.
51 lines
1.3 KiB
Python
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
|