""" Per-service runtime config helpers. All config files are stored in the 'config' bucket of the storage-service. Every function is async — callers must await them. Key layout in the config bucket: ai_service_config.json doc_service_config.json appearance_config.json themes/{id}.json """ import copy import logging import re from copy import deepcopy from pydantic import BaseModel from app.core import config_storage logger = logging.getLogger(__name__) # ── 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 ──────────────────────────────────────────────────────────────── async def load_service_config(service: str) -> dict: data = await config_storage.read_json(f"{service}_config.json") if data is None: if service == "ai_service": return AIServiceConfig().model_dump() if service == "doc_service": return DocServiceConfig().model_dump() return {} return data async def save_service_config(service: str, data: dict) -> None: await config_storage.write_json(f"{service}_config.json", data) # AI service helpers async def load_ai_service_config() -> AIServiceConfig: raw = await load_service_config("ai_service") return AIServiceConfig.model_validate(raw) async def save_ai_service_config(config: AIServiceConfig) -> None: await save_service_config("ai_service", config.model_dump()) async def load_ai_service_config_masked() -> dict: raw = await load_service_config("ai_service") return _mask_ai_config(raw) # Doc service helpers async def load_doc_service_config() -> DocServiceConfig: raw = await load_service_config("doc_service") return DocServiceConfig.model_validate(raw) async def save_doc_service_config(config: DocServiceConfig) -> None: await save_service_config("doc_service", config.model_dump()) async def load_doc_service_config_masked() -> dict: return await 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 ───────────────────────────────────────────────────── SYSTEM_PROMPT_SERVICES: dict[str, str] = { "doc_service": "Document Service", } async 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 = await 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 async def save_service_system_prompts(service_id: str, system: str, user_template: str) -> None: if service_id not in SYSTEM_PROMPT_SERVICES: raise ValueError(f"Unknown service: {service_id!r}") config = await load_service_config(service_id) config.setdefault("system_prompts", {}) config["system_prompts"]["system"] = system config["system_prompts"]["user_template"] = user_template await 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" async def load_appearance_config() -> AppearanceConfig: data = await config_storage.read_json("appearance_config.json") if data is None: return AppearanceConfig() return AppearanceConfig.model_validate(data) async def save_appearance_config(config: AppearanceConfig) -> None: await config_storage.write_json("appearance_config.json", config.model_dump()) # ── Theme file management ────────────────────────────────────────────────────── # 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", }, }, ] async def seed_builtin_themes() -> None: """Write built-in theme files to storage-service if they are not already there.""" existing_keys = await config_storage.list_keys(prefix="themes/") existing_ids = {k.removeprefix("themes/").removesuffix(".json") for k in existing_keys} for theme in _BUILTIN_THEMES: if theme["id"] not in existing_ids: await config_storage.write_json(f"themes/{theme['id']}.json", theme) logger.info("Built-in themes seeded (%d themes)", len(_BUILTIN_THEMES)) async def load_all_themes() -> list[dict]: """Return all themes from storage-service, built-ins first then custom by label.""" keys = await config_storage.list_keys(prefix="themes/") themes: list[dict] = [] for key in keys: data = await config_storage.read_json(key) if data: themes.append(data) 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) async def load_theme_by_id(theme_id: str) -> dict | None: """Return a single theme dict, or None if not found.""" return await config_storage.read_json(f"themes/{theme_id}.json") async def save_theme(theme: dict) -> None: """Write a theme to storage-service.""" await config_storage.write_json(f"themes/{theme['id']}.json", theme) async def delete_theme(theme_id: str) -> None: """Delete a custom theme. Raises ValueError for built-ins, KeyError if not found.""" data = await config_storage.read_json(f"themes/{theme_id}.json") if data is None: raise FileNotFoundError(theme_id) if data.get("builtin"): raise ValueError("Cannot delete a built-in theme") await config_storage.delete_key(f"themes/{theme_id}.json") 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