imageforge/backend/app/storage/settings.py

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")