core/settings_manager.py

"""
Settings Manager — runtime-configurable values persisted in a "Settings" worksheet.

Each setting is a (key, value) row. Complex values (e.g. visit_levels) are JSON-encoded.
Reads are cached for a short TTL to avoid hammering the Google Sheets API. Defaults
fall back to the constants in `config.py`.
"""
import json
import time
import threading
from typing import Any

import config
from core import data_manager


_SETTINGS_WORKSHEET = "Settings"
_CACHE_TTL_SECONDS = 30

_lock = threading.RLock()
_cache: dict = {"timestamp": 0.0, "values": {}}


# Schema:
#   key -> (default_value, json_encoded?, type_hint)
# type_hint is used to coerce the string-cell value back to the right Python type.
_SCHEMA: dict[str, tuple] = {
    "min_lead_time_hours":  (config.MIN_LEAD_TIME_HOURS, False, int),
    "buffer_minutes":       (config.BUFFER_MINUTES,      False, int),
    "visit_levels":         (config.VISIT_LEVELS,        True,  "visit_levels"),
    "on_call_start_hour":   ("07:00",                    False, str),
    "region_name":          (config.REGION_NAME,         False, str),
    "timezone":             (config.TIMEZONE,            False, str),
    "service_area":         (config.SERVICE_AREA,        False, str),
    "openai_model":         (config.OPENAI_MODEL,        False, str),
}


def _ensure_settings_worksheet() -> None:
    """Create the Settings worksheet on first use if it doesn't exist."""
    try:
        data_manager._get_worksheet(_SETTINGS_WORKSHEET)
        data_manager._ensure_columns(_SETTINGS_WORKSHEET, ["key", "value"])
    except Exception:
        # Worksheet doesn't exist — create it.
        ss = data_manager._get_spreadsheet()
        ws = ss.add_worksheet(title=_SETTINGS_WORKSHEET, rows=100, cols=2)
        ws.update(values=[["key", "value"]], range_name="A1:B1", value_input_option="RAW")
        # Bust caches so subsequent reads see the new worksheet.
        with data_manager._lock:
            data_manager._state["worksheets"].pop(_SETTINGS_WORKSHEET, None)
            data_manager._state["headers"].pop(_SETTINGS_WORKSHEET, None)
            data_manager._state["read_cache"].pop(_SETTINGS_WORKSHEET, None)


def _read_all_settings() -> dict[str, str]:
    """Return {key: raw_string_value} for everything in the Settings sheet.
    Returns {} if the sheet doesn't exist yet (callers fall back to defaults).
    """
    try:
        rows = data_manager._read_all(_SETTINGS_WORKSHEET)
        return {str(r.get("key", "")): str(r.get("value", "")) for r in rows if r.get("key")}
    except Exception:
        return {}


def _refresh_cache_if_stale() -> None:
    with _lock:
        if (time.time() - _cache["timestamp"]) < _CACHE_TTL_SECONDS:
            return
        _cache["values"] = _read_all_settings()
        _cache["timestamp"] = time.time()


def _coerce(raw: str, schema_entry: tuple) -> Any:
    default, is_json, type_hint = schema_entry
    if raw is None or raw == "":
        return default
    if is_json:
        try:
            parsed = json.loads(raw)
        except Exception:
            return default
        if type_hint == "visit_levels" and isinstance(parsed, dict):
            # JSON keys are always strings; rebuild int keys for visit levels.
            return {
                (int(k) if isinstance(k, str) and k.isdigit() else k): v
                for k, v in parsed.items()
            }
        return parsed
    if type_hint is int:
        try:
            return int(raw)
        except Exception:
            return default
    if type_hint is float:
        try:
            return float(raw)
        except Exception:
            return default
    return raw


def get_setting(key: str, default: Any = None) -> Any:
    """Return the current value for `key`. Falls back to schema default or `default`."""
    schema = _SCHEMA.get(key)
    if schema is None:
        return default
    _refresh_cache_if_stale()
    raw = _cache["values"].get(key)
    if raw is None or raw == "":
        return schema[0] if default is None else default
    return _coerce(raw, schema)


def get_all_settings() -> dict[str, Any]:
    """Return every known setting with its current effective value."""
    return {k: get_setting(k) for k in _SCHEMA.keys()}


def set_setting(key: str, value: Any) -> Any:
    """Persist `value` for `key`. Returns the parsed-back value (after round-trip)."""
    if key not in _SCHEMA:
        raise KeyError(f"Unknown setting: {key}")
    schema = _SCHEMA[key]
    _, is_json, _ = schema

    _ensure_settings_worksheet()

    if is_json:
        # Normalize: for visit_levels, ensure int durations stay int.
        serialized = json.dumps(value)
    elif value is None:
        serialized = ""
    elif isinstance(value, bool):
        serialized = str(value)
    else:
        serialized = str(value)

    row_idx = data_manager._find_row_index(_SETTINGS_WORKSHEET, "key", key)
    payload = {"key": key, "value": serialized}
    if row_idx:
        data_manager._update_row(_SETTINGS_WORKSHEET, row_idx, payload)
    else:
        data_manager._append_row(_SETTINGS_WORKSHEET, payload)

    # Invalidate cache so the next read sees the new value.
    with _lock:
        _cache["timestamp"] = 0.0

    return get_setting(key)


def get_schema_defaults() -> dict[str, Any]:
    """Return the original config.py defaults for every setting (for "reset to default" UIs)."""
    return {k: v[0] for k, v in _SCHEMA.items()}


# --- Convenience accessors ---
# These let consumers write `settings.get_min_lead_time_hours()` instead of magic strings.

def get_min_lead_time_hours() -> int:    return get_setting("min_lead_time_hours")
def get_buffer_minutes() -> int:         return get_setting("buffer_minutes")
def get_visit_levels() -> dict:          return get_setting("visit_levels")
def get_on_call_start_hour() -> str:     return get_setting("on_call_start_hour")
def get_region_name() -> str:            return get_setting("region_name")
def get_timezone() -> str:               return get_setting("timezone")
def get_service_area() -> str:           return get_setting("service_area")
def get_openai_model() -> str:           return get_setting("openai_model")