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