Spaces:
Sleeping
Sleeping
File size: 2,025 Bytes
32c5da4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 | from __future__ import annotations
import json
from datetime import datetime, timezone
from threading import Lock
from ..core.config import (
ADMIN_AUDIT_FILE,
CONTENT_PROFILE,
OUTPUT_RETENTION_DAYS,
RATE_LIMIT_PER_MINUTE,
SETTINGS_FILE,
)
DEFAULT_SETTINGS = {
"content_profile": CONTENT_PROFILE,
"rate_limit_per_minute": RATE_LIMIT_PER_MINUTE,
"output_retention_days": OUTPUT_RETENTION_DAYS,
"adult_enabled": False,
}
class SettingsStore:
def __init__(self) -> None:
self._lock = Lock()
def get(self) -> dict:
with self._lock:
return self._merged()
def update(self, patch: dict, actor: str) -> dict:
with self._lock:
current = self._merged()
current.update({k: v for k, v in patch.items() if v is not None})
self._write(current)
self._audit(actor=actor, patch=patch, current=current)
return current
def _merged(self) -> dict:
current = DEFAULT_SETTINGS.copy()
current.update(self._read())
return current
def _read(self) -> dict:
if not SETTINGS_FILE.exists():
return {}
try:
parsed = json.loads(SETTINGS_FILE.read_text(encoding="utf-8"))
return parsed if isinstance(parsed, dict) else {}
except json.JSONDecodeError:
return {}
def _write(self, data: dict) -> None:
SETTINGS_FILE.parent.mkdir(parents=True, exist_ok=True)
SETTINGS_FILE.write_text(json.dumps(data, indent=2), encoding="utf-8")
def _audit(self, actor: str, patch: dict, current: dict) -> None:
entry = {
"timestamp": datetime.now(timezone.utc).isoformat(),
"actor": actor,
"patch": patch,
"current": current,
}
ADMIN_AUDIT_FILE.parent.mkdir(parents=True, exist_ok=True)
with ADMIN_AUDIT_FILE.open("a", encoding="utf-8") as fp:
fp.write(json.dumps(entry, ensure_ascii=True) + "\n")
|