win_monitor/backend/app/settings_service.py

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",
}


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),
    }


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)