Business-Management/backend/app/core/app_config.py

"""
Per-service runtime config helpers.

Config files live on the shared `app_config` Docker volume at /config/.
Each service has its own JSON file.

Atomic write pattern: write to .tmp in same dir, then os.replace() so
services never read a partial file.
"""
import copy
import json
import os
import re
from pathlib import Path

from pydantic import BaseModel

_CONFIG_DIR = Path(os.environ.get("APP_CONFIG_DIR", "/config"))

# ── AI service config schemas ──────────────────────────────────────────────────


class AnthropicConfig(BaseModel):
    api_key: str = ""
    model: str = "claude-haiku-4-5-20251001"


class OllamaConfig(BaseModel):
    base_url: str = "http://host.docker.internal:11434/v1"
    model: str = "llama3.2"
    api_key: str = "ollama"


class LMStudioConfig(BaseModel):
    base_url: str = "http://host.docker.internal:1234/v1"
    model: str = "local-model"
    api_key: str = "lm-studio"


class AIServiceConfig(BaseModel):
    provider: str = "lmstudio"
    timeout_seconds: int = 60
    max_retries: int = 2
    anthropic: AnthropicConfig = AnthropicConfig()
    ollama: OllamaConfig = OllamaConfig()
    lmstudio: LMStudioConfig = LMStudioConfig()


# ── Doc service config schemas ─────────────────────────────────────────────────

_DOC_SYSTEM_PROMPT_DEFAULT = (
    "You are a financial document analysis assistant. "
    "Given the text extracted from a PDF document, return ONLY a JSON object "
    "with no markdown, no code fences, and no explanation."
)

_DOC_USER_TEMPLATE_DEFAULT = (
    'Analyze the following document text and return a JSON object with exactly these keys:\n'
    'title (a short, descriptive human-readable title for this document, e.g. "ACME Corp Invoice April 2026", "Office Supplies Receipt", "Q1 Flower Delivery Order"),\n'
    'document_type (one of: invoice, bill, receipt, order, expense, revenue, unknown),\n'
    'total_amount (string or null),\n'
    'currency (string or null),\n'
    'vendor_name (string or null),\n'
    'customer_name (string or null),\n'
    'billing_address (string or null),\n'
    'customer_address (string or null),\n'
    'invoice_number (string or null),\n'
    'invoice_date (string or null),\n'
    'due_date (string or null),\n'
    'tags (array of short keyword strings describing the document),\n'
    'line_items (array of objects, each with keys: description, amount),\n'
    'suggested_categories (array of 2 to 5 short category name strings a user might want to file this document under, e.g. "Utilities", "Travel", "Software Subscriptions", "Client Invoices").\n'
    '\n'
    'Document text:\n'
    '{text}'
)


class DocumentsConfig(BaseModel):
    max_pdf_bytes: int = 20 * 1024 * 1024


class DocServiceSystemPrompts(BaseModel):
    system: str = _DOC_SYSTEM_PROMPT_DEFAULT
    user_template: str = _DOC_USER_TEMPLATE_DEFAULT


class DocServiceConfig(BaseModel):
    documents: DocumentsConfig = DocumentsConfig()
    system_prompts: DocServiceSystemPrompts = DocServiceSystemPrompts()


# ── Masking ────────────────────────────────────────────────────────────────────

def _mask_key(key: str) -> str:
    if not key or len(key) <= 8:
        return "••••"
    return key[:7] + "••••"


def _mask_ai_config(data: dict) -> dict:
    masked = copy.deepcopy(data)
    for provider in ("anthropic", "ollama", "lmstudio"):
        if provider in masked and "api_key" in masked[provider]:
            masked[provider]["api_key"] = _mask_key(masked[provider]["api_key"])
    return masked


# ── Load / Save ────────────────────────────────────────────────────────────────

def _config_path(service: str) -> Path:
    return _CONFIG_DIR / f"{service}_config.json"


def load_service_config(service: str) -> dict:
    path = _config_path(service)
    if not path.exists():
        if service == "ai_service":
            return AIServiceConfig().model_dump()
        if service == "doc_service":
            return DocServiceConfig().model_dump()
        return {}
    with path.open() as f:
        return json.load(f)


def save_service_config(service: str, data: dict) -> None:
    path = _config_path(service)
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_suffix(".tmp")
    tmp.write_text(json.dumps(data, indent=2))
    os.replace(tmp, path)


# AI service helpers

def load_ai_service_config() -> AIServiceConfig:
    raw = load_service_config("ai_service")
    return AIServiceConfig.model_validate(raw)


def save_ai_service_config(config: AIServiceConfig) -> None:
    save_service_config("ai_service", config.model_dump())


def load_ai_service_config_masked() -> dict:
    raw = load_service_config("ai_service")
    return _mask_ai_config(raw)


# Doc service helpers

def load_doc_service_config() -> DocServiceConfig:
    raw = load_service_config("doc_service")
    return DocServiceConfig.model_validate(raw)


def save_doc_service_config(config: DocServiceConfig) -> None:
    save_service_config("doc_service", config.model_dump())


def load_doc_service_config_masked() -> dict:
    return load_service_config("doc_service")


def _merge_api_key(new_key: str, existing_key: str) -> str:
    """If new_key is empty or a masked value, keep the existing key."""
    if not new_key or "••••" in new_key:
        return existing_key
    return new_key


# ── System prompts helpers ─────────────────────────────────────────────────────

# Registry of all services that have editable system prompts.
# key = service identifier, value = human-readable label
SYSTEM_PROMPT_SERVICES: dict[str, str] = {
    "doc_service": "Document Service",
}


def load_all_system_prompts() -> dict:
    """Return {service_id: {label, system, user_template, default_system, default_user_template}}."""
    result: dict = {}
    for service_id, label in SYSTEM_PROMPT_SERVICES.items():
        config = load_service_config(service_id)
        prompts = config.get("system_prompts", {})
        defaults = _get_service_prompt_defaults(service_id)
        result[service_id] = {
            "label": label,
            "system": prompts.get("system", defaults["system"]),
            "user_template": prompts.get("user_template", defaults["user_template"]),
            "default_system": defaults["system"],
            "default_user_template": defaults["user_template"],
        }
    return result


def save_service_system_prompts(service_id: str, system: str, user_template: str) -> None:
    """Persist updated system prompts into the service's config file."""
    if service_id not in SYSTEM_PROMPT_SERVICES:
        raise ValueError(f"Unknown service: {service_id!r}")
    config = load_service_config(service_id)
    config.setdefault("system_prompts", {})
    config["system_prompts"]["system"] = system
    config["system_prompts"]["user_template"] = user_template
    save_service_config(service_id, config)


def _get_service_prompt_defaults(service_id: str) -> dict:
    if service_id == "doc_service":
        d = DocServiceSystemPrompts()
        return {"system": d.system, "user_template": d.user_template}
    return {"system": "", "user_template": ""}


# ── Appearance config ──────────────────────────────────────────────────────────

class AppearanceConfig(BaseModel):
    theme: str = "default"
    default_mode: str = "system"


def load_appearance_config() -> AppearanceConfig:
    path = _CONFIG_DIR / "appearance_config.json"
    if not path.exists():
        return AppearanceConfig()
    with path.open() as f:
        return AppearanceConfig.model_validate(json.load(f))


def save_appearance_config(config: AppearanceConfig) -> None:
    path = _CONFIG_DIR / "appearance_config.json"
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_suffix(".tmp")
    tmp.write_text(json.dumps(config.model_dump(), indent=2))
    os.replace(tmp, path)


# ── Theme file management ──────────────────────────────────────────────────────

_THEMES_DIR = _CONFIG_DIR / "themes"

# 9 required colour tokens per mode
_REQUIRED_TOKENS = frozenset({
    "primary", "primary_hover", "accent", "accent_hover",
    "background", "surface", "border", "text_primary", "text_muted",
})

_RGB_RE = re.compile(r"^\d{1,3} \d{1,3} \d{1,3}$")

_BUILTIN_THEMES: list[dict] = [
    {
        "id": "default",
        "label": "Default",
        "builtin": True,
        "light": {
            "primary":       "37 99 235",
            "primary_hover": "29 78 216",
            "accent":        "234 179 8",
            "accent_hover":  "202 138 4",
            "background":    "248 250 252",
            "surface":       "255 255 255",
            "border":        "226 232 240",
            "text_primary":  "15 23 42",
            "text_muted":    "100 116 139",
        },
        "dark": {
            "primary":       "59 130 246",
            "primary_hover": "37 99 235",
            "accent":        "250 204 21",
            "accent_hover":  "234 179 8",
            "background":    "15 23 42",
            "surface":       "30 41 59",
            "border":        "51 65 85",
            "text_primary":  "203 213 225",
            "text_muted":    "148 163 184",
        },
    },
    {
        "id": "pastel",
        "label": "Pastel",
        "builtin": True,
        "light": {
            "primary":       "124 58 237",
            "primary_hover": "109 40 217",
            "accent":        "236 72 153",
            "accent_hover":  "219 39 119",
            "background":    "253 244 255",
            "surface":       "250 245 255",
            "border":        "233 213 255",
            "text_primary":  "30 27 75",
            "text_muted":    "107 114 128",
        },
        "dark": {
            "primary":       "167 139 250",
            "primary_hover": "196 181 253",
            "accent":        "244 114 182",
            "accent_hover":  "251 164 200",
            "background":    "30 20 51",
            "surface":       "45 27 78",
            "border":        "76 53 117",
            "text_primary":  "233 213 255",
            "text_muted":    "196 181 253",
        },
    },
    {
        "id": "high-contrast",
        "label": "High Contrast",
        "builtin": True,
        "light": {
            "primary":       "30 58 138",
            "primary_hover": "30 64 175",
            "accent":        "220 38 38",
            "accent_hover":  "185 28 28",
            "background":    "255 255 255",
            "surface":       "255 255 255",
            "border":        "156 163 175",
            "text_primary":  "0 0 0",
            "text_muted":    "75 85 99",
        },
        "dark": {
            "primary":       "96 165 250",
            "primary_hover": "147 197 253",
            "accent":        "248 113 113",
            "accent_hover":  "252 165 165",
            "background":    "0 0 0",
            "surface":       "10 10 10",
            "border":        "55 65 81",
            "text_primary":  "255 255 255",
            "text_muted":    "156 163 175",
        },
    },
    {
        "id": "ocean",
        "label": "Ocean Blue",
        "builtin": True,
        "light": {
            "primary":       "29 78 216",
            "primary_hover": "30 58 138",
            "accent":        "8 145 178",
            "accent_hover":  "14 116 144",
            "background":    "239 246 255",
            "surface":       "219 234 254",
            "border":        "147 197 253",
            "text_primary":  "30 58 138",
            "text_muted":    "59 130 246",
        },
        "dark": {
            "primary":       "96 165 250",
            "primary_hover": "147 197 253",
            "accent":        "34 211 238",
            "accent_hover":  "103 232 249",
            "background":    "10 22 40",
            "surface":       "15 36 68",
            "border":        "29 78 216",
            "text_primary":  "219 234 254",
            "text_muted":    "147 197 253",
        },
    },
]


def seed_builtin_themes() -> None:
    """Create /config/themes/ and write built-in theme files if missing."""
    _THEMES_DIR.mkdir(parents=True, exist_ok=True)
    for theme in _BUILTIN_THEMES:
        path = _THEMES_DIR / f"{theme['id']}.json"
        if not path.exists():
            path.write_text(json.dumps(theme, indent=2))


def load_all_themes() -> list[dict]:
    """Return all themes from /config/themes/*.json, built-ins first."""
    if not _THEMES_DIR.exists():
        seed_builtin_themes()
    themes = []
    for f in sorted(_THEMES_DIR.glob("*.json")):
        try:
            themes.append(json.loads(f.read_text()))
        except (json.JSONDecodeError, OSError):
            pass
    # Sort: built-ins first (preserving their original order), then custom by label
    builtin_ids = [t["id"] for t in _BUILTIN_THEMES]
    def sort_key(t: dict) -> tuple:
        tid = t.get("id", "")
        try:
            return (0, builtin_ids.index(tid))
        except ValueError:
            return (1, t.get("label", tid).lower())
    return sorted(themes, key=sort_key)


def validate_theme_tokens(colors: dict) -> list[str]:
    """Return a list of validation error messages, empty if valid."""
    errors = []
    missing = _REQUIRED_TOKENS - set(colors.keys())
    if missing:
        errors.append(f"Missing tokens: {', '.join(sorted(missing))}")
    for key, val in colors.items():
        if key in _REQUIRED_TOKENS and not _RGB_RE.match(str(val)):
            errors.append(f"Token '{key}' must be an RGB triplet like '37 99 235', got: {val!r}")
    return errors


def save_theme(theme: dict) -> None:
    """Write a theme file atomically."""
    _THEMES_DIR.mkdir(parents=True, exist_ok=True)
    path = _THEMES_DIR / f"{theme['id']}.json"
    tmp = path.with_suffix(".tmp")
    tmp.write_text(json.dumps(theme, indent=2))
    os.replace(tmp, path)


def delete_theme(theme_id: str) -> None:
    """Delete a custom theme file. Raises ValueError for built-ins, FileNotFoundError if missing."""
    path = _THEMES_DIR / f"{theme_id}.json"
    if not path.exists():
        raise FileNotFoundError(theme_id)
    data = json.loads(path.read_text())
    if data.get("builtin"):
        raise ValueError("Cannot delete a built-in theme")
    path.unlink()