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
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}} for all registered services."""
    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"]),
        }
    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": ""}