Files
PTSRaiAgent/app/config.py
T
ElebobaandClaude Opus 5.5 531bef1e73
CI / ruff · pytest (pull_request) Successful in 2m33s
feat: crisis message after the test and non-blocking; weapon answer triggers nothing; ЕМИАС demo = Никита
- CRISIS_BLOCKING (default off for the demo): a crisis no longer locks the dialog; the message and the
  doctor's acute alert still happen. The clinical protocol (blocking, immediate stop) stays behind the flag and
  keeps its tests.
- Crisis signals inside a questionnaire (intake red flags, high C-SSRS, PHQ-9 item 9) no longer stop it: the
  test is finished and the crisis message comes at the end; the doctor's alert is raised at once/at the end.
- Weapon access: asked and stored, triggers nothing (rule disabled, no red flag, no safety block).
- ЕМИАС demo persona is «Никита В.»; the one-click demo stays «Андрей С.».

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-04 23:30:05 +03:00

96 lines
4.8 KiB
Python

"""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
# Telegram is used only for the prototype on synthetic data; prod channel is MAX / Gosuslugi.
tg_bot_token: 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_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()