PTSRaiAgent/app/config.py

"""Typed application settings. The only place that reads environment variables."""

import re
from functools import lru_cache
from pathlib import Path
from typing import Literal
from urllib.parse import urlsplit

from pydantic import Field, SecretStr, model_validator
from pydantic_settings import BaseSettings, SettingsConfigDict

ROOT_DIR = Path(__file__).resolve().parent.parent


class Settings(BaseSettings):
    # Empty values in .env (e.g. `TG_BOT_TOKEN=`) fall back to defaults instead of failing validation.
    model_config = SettingsConfigDict(env_file=".env", extra="ignore", env_ignore_empty=True, hide_input_in_errors=True)

    # Two separate databases (ADR-003): identity holds direct identifiers, clinical holds pseudonymised data.
    identity_db_url: str = "postgresql+asyncpg://identity_app:identity_app@localhost:5432/identity"
    clinical_db_url: str = "postgresql+asyncpg://clinical_app:clinical_app@localhost:5432/clinical"

    # HMAC key linking identity and clinical records (hex, 32+ bytes).
    pseudo_key: str = Field(default="dev-only-insecure-key-change-me-0000000000000000", min_length=32)

    # LLM: GigaChat only (152-FZ: no foreign APIs). `mock` is the offline default (ADR-004).
    llm_provider: Literal["mock", "gigachat"] = "mock"
    gigachat_credentials: str = ""
    gigachat_scope: str = "GIGACHAT_API_PERS"
    gigachat_model: str = "GigaChat"
    gigachat_verify_ssl: bool = False

    # Chat agent (tool-using LLM over RAG + patient record). `auto` = OpenRouter if a key is set, else off
    # (extractive RAG / instruction answers stay). OpenRouter is a TEMPORARY prototype provider on synthetic
    # data only — prod must use GigaChat (152-FZ, foreign services are forbidden for real PII).
    chat_agent: Literal["auto", "off", "mock", "openrouter", "gigachat"] = "auto"
    openrouter_api_key: str = ""
    openrouter_model: str = "google/gemini-2.5-flash-lite"
    openrouter_base_url: str = "https://openrouter.ai/api/v1"
    openrouter_proxy_url: str = ""  # OpenRouter blocks RF IPs: the server goes through an HTTP proxy (as vobot does)
    gigachat_agent_model: str = "GigaChat-2-Pro"  # function calling quality matters more than price here
    chat_agent_timeout_s: float = 25.0
    chat_agent_max_steps: int = 5

    # Voice input (#116, #121): speech-to-text for the patient's messages. `auto` = SaluteSpeech if credentials are
    # set; otherwise, in the demo only, OpenRouter if a key is set, else mock. OpenRouter STT is synthetic data only
    # (ADR-014): the audio leaves the RF and cannot be scrubbed — forbidden with DEMO_CLOCK=false (validator below).
    stt_provider: Literal["auto", "off", "mock", "openrouter", "salute"] = "auto"
    stt_model: str = "google/gemini-2.5-flash"  # needs audio input
    stt_timeout_s: float = 30.0
    stt_max_seconds: int = Field(default=60, ge=5, le=300)
    # SaluteSpeech (Sber, RF contour) — the pilot provider (#121, ADR-016). Credentials = the base64 authorization key.
    salute_speech_credentials: SecretStr = SecretStr("")
    salute_speech_scope: str = "SALUTE_SPEECH_PERS"
    # Russian Trusted Root CA (Минцифры) is not in certifi: point to the .crt/.pem to verify Sber's TLS chain.
    salute_speech_ca: Path | None = None

    # Telegram is used only for the prototype on synthetic data; prod channel is MAX / Gosuslugi.
    tg_bot_token: str = ""
    # The RF server reaches api.telegram.org over IPv6 only; docker has IPv4 only → go through a proxy (http/socks).
    tg_proxy_url: str = ""
    doctor_notify_chat_id: int | None = None

    # MAX is opt-in. Its adapter is delivered in the related backend/channel changes.
    max_enabled: bool = False
    max_bot_token: SecretStr = SecretStr("")
    max_bot_username: str = ""
    max_webhook_secret: SecretStr = SecretStr("")
    max_webhook_url: str = ""
    max_site_url: str = ""
    max_ca_bundle: Path | None = None
    max_link_ttl_seconds: int = Field(default=600, ge=60, le=1800)

    @model_validator(mode="after")
    def validate_stt(self) -> "Settings":
        if self.stt_provider == "openrouter" and not self.demo_clock:
            raise ValueError(
                "STT_PROVIDER=openrouter is for synthetic data only (ADR-014) and is forbidden with DEMO_CLOCK=false: "
                "use STT_PROVIDER=salute"
            )
        if self.salute_speech_ca is not None and not self.salute_speech_ca.is_file():
            raise ValueError("SALUTE_SPEECH_CA must point to an existing CA certificate file")
        return self

    @model_validator(mode="after")
    def validate_max(self) -> "Settings":
        if not self.max_enabled:
            return self
        if not self.max_bot_token.get_secret_value().strip():
            raise ValueError("MAX_BOT_TOKEN is required when MAX_ENABLED=true")
        if not re.fullmatch(r"[A-Za-z0-9_]{1,128}", self.max_bot_username):
            raise ValueError("MAX_BOT_USERNAME must be the bot nickname without @ or a URL")
        if not re.fullmatch(r"[A-Za-z0-9_-]{32,256}", self.max_webhook_secret.get_secret_value()):
            raise ValueError("MAX_WEBHOOK_SECRET must contain 32..256 URL-safe characters")
        for name, value in (("MAX_SITE_URL", self.max_site_url), ("MAX_WEBHOOK_URL", self.max_webhook_url)):
            url = urlsplit(value)
            if url.scheme != "https" or not url.hostname or url.username or url.password or url.port not in (None, 443):
                raise ValueError(f"{name} must use HTTPS on port 443 without credentials")
            if url.query or (name == "MAX_WEBHOOK_URL" and url.fragment):
                raise ValueError(f"{name} must not contain a query string (or a fragment for webhook)")
        if self.max_ca_bundle is not None and not self.max_ca_bundle.is_file():
            raise ValueError("MAX_CA_BUNDLE must point to an existing trusted CA bundle")
        return self

    # Virtual clock for the demo (ADR-005).
    demo_clock: bool = True
    # Initial screening (onboarding questionnaires) on/off — the default; the demo console can flip it at runtime.
    screening_enabled: bool = True
    # Crisis blocks the dialog until the doctor's ack (clinical protocol) — off for the demo: the crisis message
    # and the doctor's alert still happen, the patient keeps using everything.
    crisis_blocking: bool = False
    tz: str = "Europe/Moscow"

    api_base_url: str = "http://localhost:8000"
    config_dir: Path = ROOT_DIR / "config"
    data_dir: Path = ROOT_DIR / "data"


@lru_cache
def get_settings() -> Settings:
    return Settings()