from __future__ import annotations from pathlib import Path from typing import Any from fastapi import HTTPException from .database import DATA_DIR, get_db from .scanner import now_iso from .schemas import SystemSettingsUpdate SETTING_KEYS = { "default_ai_provider_id", "default_ai_model_id", "default_ai_temperature", "automation_file_root", "automation_screen_path", "automation_error_path", "automation_runtime_path", "automation_auto_screenshot_enabled", "automation_auto_screenshot_interval", "automation_remote_token", } def list_settings() -> dict[str, Any]: with get_db() as conn: rows = conn.execute("SELECT * FROM app_settings ORDER BY key ASC").fetchall() values = {row["key"]: row["value"] for row in rows} return {"settings": normalize_settings(values), "items": rows} def update_settings(payload: SystemSettingsUpdate) -> dict[str, Any]: values = payload.model_dump(exclude_unset=True) now = now_iso() with get_db() as conn: for key, value in values.items(): if key not in SETTING_KEYS: continue if key.endswith("_path") or key == "automation_file_root": value = normalize_relative_path(value) conn.execute( """ INSERT INTO app_settings (key, value, description, updated_at) VALUES (?, ?, COALESCE((SELECT description FROM app_settings WHERE key = ?), ''), ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at """, (key, serialize_value(value), key, now), ) return list_settings() def normalize_settings(values: dict[str, str | None]) -> dict[str, Any]: return { "default_ai_provider_id": optional_int(values.get("default_ai_provider_id")), "default_ai_model_id": optional_int(values.get("default_ai_model_id")), "default_ai_temperature": optional_float(values.get("default_ai_temperature"), 0.1), "automation_file_root": values.get("automation_file_root") or "automation", "automation_screen_path": values.get("automation_screen_path") or "automation/screens", "automation_error_path": values.get("automation_error_path") or "automation/errors", "automation_runtime_path": values.get("automation_runtime_path") or "automation/runtime", "automation_auto_screenshot_enabled": parse_bool(values.get("automation_auto_screenshot_enabled")), "automation_auto_screenshot_interval": optional_int(values.get("automation_auto_screenshot_interval"), 30), "automation_remote_token": values.get("automation_remote_token") or "", } def automation_remote_token() -> str: return str(get_settings_dict().get("automation_remote_token") or "").strip() def get_settings_dict() -> dict[str, Any]: return list_settings()["settings"] def default_ai_params() -> dict[str, Any]: settings = get_settings_dict() return { "provider_id": settings.get("default_ai_provider_id"), "model_id": settings.get("default_ai_model_id"), "temperature": settings.get("default_ai_temperature", 0.1), } def resolve_data_path(setting_key: str, fallback: str) -> Path: settings = get_settings_dict() relative = settings.get(setting_key) or fallback normalized = normalize_relative_path(relative) path = (DATA_DIR / normalized).resolve() data_root = DATA_DIR.resolve() if data_root != path and data_root not in path.parents: raise HTTPException(status_code=400, detail="Configured path escapes data directory") path.mkdir(parents=True, exist_ok=True) return path def normalize_relative_path(value: str | None) -> str: raw = (value or "").strip().replace("\\", "/") if not raw: return "" path = Path(raw) if path.is_absolute() or ".." in path.parts: raise HTTPException(status_code=400, detail="Path must be relative and must not contain ..") return "/".join(part for part in path.parts if part not in {"", "."}) def optional_int(value: Any, default: int | None = None) -> int | None: if value in (None, ""): return default try: return int(value) except (TypeError, ValueError): return default def optional_float(value: Any, default: float | None = None) -> float | None: if value in (None, ""): return default try: return float(value) except (TypeError, ValueError): return default def parse_bool(value: Any) -> bool: return str(value).lower() in {"1", "true", "yes", "on"} def serialize_value(value: Any) -> str: if value is None: return "" if isinstance(value, bool): return "1" if value else "0" return str(value)