"""Настройки сервиса (pydantic-settings). Вложенные модели — тоже BaseSettings (как в шаблоне): имена полей в ENV регистронезависимы только для них. Секреты платформы приходят вложенными переменными `OPENBAO__MINIO__*`, `OPENBAO__NATS__*`, `OPENBAO__SMB__*`; прикладные параметры — обычными именами верхнего уровня (`SCAN_HOTFOLDER_MODE`, `SCAN_POLL_SECONDS`, ...). Файл окружения выбирается переменной `ENV_FILE` (по умолчанию `.env`, путь считается от текущего каталога — `api/`); переменные окружения приоритетнее файла. Настройки читаются при первом вызове `get_settings()`, не при импорте. """ import os from functools import lru_cache from typing import Literal from pydantic import PositiveFloat, PositiveInt, model_validator from pydantic_settings import BaseSettings, SettingsConfigDict class MinioSettings(BaseSettings): MINIO_URL: str MINIO_ADMIN_LOGIN: str MINIO_ADMIN_PASSWORD: str MINIO_BUCKET: str MINIO_SECURE: bool = True # https (платформа); false — http (MinIO из docker compose) class NATSSettings(BaseSettings): NATS_URL: str NATS_SEED: str = "" # пусто — подключение к NATS без nkeys/JWT (локальная разработка, тесты) NATS_JWT: str = "" class SMBSettings(BaseSettings): """SMB-шара горячих папок ИС Сканирование. Нужна только при `SCAN_HOTFOLDER_MODE=smb` (там все поля обязательны, проверяет `Settings`); в режиме local весь блок можно не задавать.""" SMB_HOST: str = "" SMB_PATH: str = "" SMB_LOGIN: str = "" SMB_PASSWORD: str = "" class OpenBAOSettings(BaseSettings): MINIO: MinioSettings NATS: NATSSettings SMB: SMBSettings = SMBSettings() # весь блок необязателен: в режиме local не требуется class ServerSettings(BaseSettings): HOST: str = "0.0.0.0" PORT: int = 8000 class Settings(BaseSettings): OPENBAO: OpenBAOSettings DEBUG: bool = False server: ServerSettings = ServerSettings() ROOT_PATH: str = "/" VERSION: str = "1.0.0" SERVICE_NAME: str = "scan-gateway" # service.name в конверте recognition.result # NATS: подписка на recognition.image.request и публикация. Выключено — воркер не запускается (dev без NATS) NATS_ENABLED: bool = True PUBLISH_RETRIES: PositiveInt = 3 # попытки публикации результата/ack DONE_CACHE_SIZE: PositiveInt = 1000 # размер LRU обработанных job_id (дедупликация повторов) # Горячие папки ИС Сканирование: local — обычный каталог (dev, тесты), smb — SMB-шара платформы SCAN_HOTFOLDER_MODE: Literal["local", "smb"] = "local" SCAN_HOTFOLDER_PATH: str = "" # корень горячей папки в режиме local (каталог должен существовать) SCAN_INPUT_DIR: str = "Input" SCAN_OUTPUT_DIR: str = "Output" SCAN_POLL_SECONDS: PositiveFloat = 5 # период опроса «Исходящих» SCAN_TIMEOUT_SECONDS: PositiveFloat = 300 # потолок ожидания ответа ИС (дополнительно к reply_deadline запроса) SCAN_MAX_INFLIGHT: PositiveInt = 4 # одновременно обрабатываемых заданий model_config = SettingsConfigDict( extra="ignore", env_file_encoding="utf-8", env_nested_delimiter="__", nested_model_default_partial_update=False, ) @model_validator(mode="after") def _check_hotfolder(self) -> "Settings": if self.SCAN_HOTFOLDER_MODE == "local" and not self.SCAN_HOTFOLDER_PATH.strip(): raise ValueError("SCAN_HOTFOLDER_MODE=local требует SCAN_HOTFOLDER_PATH (корень горячей папки)") if self.SCAN_HOTFOLDER_MODE == "smb": missing = [name for name, value in self.OPENBAO.SMB.model_dump().items() if not value.strip()] if missing: raise ValueError( "SCAN_HOTFOLDER_MODE=smb требует OPENBAO__SMB__SMB_HOST/SMB_PATH/SMB_LOGIN/SMB_PASSWORD; " f"не заданы: {', '.join(missing)}" ) return self @lru_cache def get_settings() -> Settings: return Settings(_env_file=os.environ.get("ENV_FILE", ".env"))