arm-scan-gateway/api/src/core/config.py

"""Настройки сервиса (pydantic-settings).

Вложенные модели — тоже BaseSettings (как в шаблоне): имена полей в ENV регистронезависимы только для них.
Секреты платформы приходят вложенными переменными `OPENBAO__MINIO__*`, `OPENBAO__NATS__*`, `OPENBAO__SMB__*`;
прикладные параметры — обычными именами верхнего уровня (`SCAN_HOTFOLDER_MODE`, `SCAN_POLL_SECONDS`, ...), а там, где
обычные переменные окружения задать нельзя (контур платформы), — как `OPENBAO__APP__<ИМЯ>`
(`src/core/settings_sources.py`;
приоритет: обычный env -> `OPENBAO__APP__*` -> `.env` -> умолчание; неизвестное имя — ошибка при старте).
Файл окружения выбирается переменной `ENV_FILE` (по умолчанию `.env`, путь считается от текущего каталога — `api/`);
переменные окружения приоритетнее файла. Настройки читаются при первом вызове `get_settings()`, не при импорте.
"""

import os
from functools import lru_cache
from typing import Any, Literal

from pydantic import NonNegativeFloat, PositiveFloat, PositiveInt, model_validator
from pydantic_settings import BaseSettings, PydanticBaseSettingsSource, SettingsConfigDict

from src.core.settings_sources import OpenBaoAppSettingsSource


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 не требуется

    @model_validator(mode="before")
    @classmethod
    def _drop_app(cls, data: Any) -> Any:
        """`OPENBAO__APP__<ИМЯ>` — настройки верхнего уровня `Settings` (читает `OpenBaoAppSettingsSource`), не группа
        секретов. Вложенный разбор `__` кладёт их сюда ключом `APP`, а `extra=forbid` отверг бы его; убирается только
        этот ключ — опечатка в имени группы (`OPENBAO__MINO__...`) по-прежнему ошибка."""
        if isinstance(data, dict):
            return {key: value for key, value in data.items() if str(key).upper() != "APP"}
        return data


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 после первой попытки (всего до 4)
    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_TMP_DIR: str = "Tmp"  # служебный каталог временных файлов записи (рядом с Input, на той же ФС/шаре)
    SCAN_POLL_SECONDS: PositiveFloat = 5  # период опроса «Исходящих»
    # набор *.json ответа готов, когда имена и размеры не менялись столько секунд; 0 — готов сразу (тесты)
    SCAN_SETTLE_SECONDS: NonNegativeFloat = 2
    SCAN_TIMEOUT_SECONDS: PositiveFloat = 300  # потолок ожидания ответа ИС (дополнительно к reply_deadline запроса)
    SCAN_MAX_INFLIGHT: PositiveInt = 4  # одновременно обрабатываемых заданий
    # Очистка горячей папки после прочтения набора ответа (D-2): любое удаление, включая .part в Tmp, только при true.
    # На SMB-шаре контура права на удаление не подтверждены — по умолчанию выключено
    SCAN_CLEANUP: bool = False
    SCAN_RULES_DIR: str = "rules"  # справочники нормализатора (yaml), путь от текущего каталога (api/)

    model_config = SettingsConfigDict(
        extra="ignore",
        env_file_encoding="utf-8",
        env_nested_delimiter="__",
        nested_model_default_partial_update=False,
    )

    @classmethod
    def settings_customise_sources(
        cls,
        settings_cls: type[BaseSettings],
        init_settings: PydanticBaseSettingsSource,
        env_settings: PydanticBaseSettingsSource,
        dotenv_settings: PydanticBaseSettingsSource,
        file_secret_settings: PydanticBaseSettingsSource,
    ) -> tuple[PydanticBaseSettingsSource, ...]:
        """Обычный env -> `OPENBAO__APP__*` -> `.env` -> значение по умолчанию."""
        return (
            init_settings,
            env_settings,
            OpenBaoAppSettingsSource(settings_cls),
            dotenv_settings,
            file_secret_settings,
        )

    @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)}"
                )
        if self.SCAN_SETTLE_SECONDS >= self.SCAN_TIMEOUT_SECONDS:
            raise ValueError(
                f"SCAN_SETTLE_SECONDS ({self.SCAN_SETTLE_SECONDS}) должен быть меньше SCAN_TIMEOUT_SECONDS "
                f"({self.SCAN_TIMEOUT_SECONDS}): иначе набор ответа не успеет настояться и каждое задание завершится "
                "TIMEOUT"
            )
        return self


@lru_cache
def get_settings() -> Settings:
    return Settings(_env_file=os.environ.get("ENV_FILE", ".env"))