60 lines
1.7 KiB
Python
60 lines
1.7 KiB
Python
"""Shared on-disk control and progress state for admin audit retention."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
def status_path(runtime_root: Path) -> Path:
|
|
return runtime_root / ".admin_audit_retention.status.json"
|
|
|
|
|
|
def pause_path(runtime_root: Path) -> Path:
|
|
return runtime_root / ".admin_audit_retention.paused"
|
|
|
|
|
|
def is_paused(runtime_root: Path) -> bool:
|
|
return pause_path(runtime_root).exists()
|
|
|
|
|
|
def set_paused(runtime_root: Path, paused: bool) -> None:
|
|
runtime_root.mkdir(parents=True, exist_ok=True)
|
|
marker = pause_path(runtime_root)
|
|
if paused:
|
|
marker.touch(exist_ok=True)
|
|
else:
|
|
marker.unlink(missing_ok=True)
|
|
|
|
|
|
def read_status(runtime_root: Path) -> dict[str, Any]:
|
|
path = status_path(runtime_root)
|
|
try:
|
|
value = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, ValueError, TypeError):
|
|
return {
|
|
"status": "scheduled",
|
|
"current_source": None,
|
|
"round": 0,
|
|
"max_rounds": 4,
|
|
"deleted": {},
|
|
"last_sweep": None,
|
|
"started_at": None,
|
|
"next_run_at": None,
|
|
"retention_days": 90,
|
|
}
|
|
return value if isinstance(value, dict) else {}
|
|
|
|
|
|
def write_status(runtime_root: Path, value: dict[str, Any]) -> None:
|
|
runtime_root.mkdir(parents=True, exist_ok=True)
|
|
path = status_path(runtime_root)
|
|
temporary = path.with_name(f"{path.name}.{os.getpid()}.tmp")
|
|
temporary.write_text(json.dumps(value, ensure_ascii=False), encoding="utf-8")
|
|
os.replace(temporary, path)
|
|
|
|
|
|
__all__ = ["is_paused", "read_status", "set_paused", "write_status"]
|