LegacyHUB/app/config.py

"""Centralized typed configuration loaded from environment variables.

All other modules import :data:`settings` and never touch ``os.environ`` directly.
"""

from __future__ import annotations

from functools import lru_cache
from typing import Literal

from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict

_DEV_LIKE_ENVS = {"dev", "development", "test", "testing", "local"}


class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        case_sensitive=False,
        extra="ignore",
    )

    # ---------------- App ----------------
    app_env: str = Field("dev", alias="APP_ENV")
    app_log_level: str = Field("INFO", alias="APP_LOG_LEVEL")
    app_host: str = Field("0.0.0.0", alias="APP_HOST")
    app_port: int = Field(8000, alias="APP_PORT")
    app_input_dir: str = Field("/data/input", alias="APP_INPUT_DIR")
    app_work_dir: str = Field("/data/work", alias="APP_WORK_DIR")
    app_api_prefix: str = Field("/api/v1", alias="APP_API_PREFIX")
    cors_allowed_origins: str = Field(
        "http://localhost:5273,http://localhost:4173",
        alias="CORS_ALLOWED_ORIGINS",
    )
    api_key: str = Field("", alias="API_KEY")

    # ---------------- Trusted-header identity (gateway/SSO, D2) ----------------
    # When true the API rejects requests under app_api_prefix that lack a valid
    # X-TeamHub-Actor header. Enabled behind the platform gateway; left false for
    # direct local development. See app/integrations/identity.py.
    auth_require_identity: bool = Field(False, alias="AUTH_REQUIRE_IDENTITY")
    auth_app_code: str = Field("legacyhub", alias="AUTH_APP_CODE")
    # Service-to-service key required on participant ingest endpoints. Falls back
    # to API_KEY when empty. Enforced whenever either is set.
    ingest_api_key: str = Field("", alias="INGEST_API_KEY")
    # Deprecated local folder ingest is off by default; use /knowledge-ingest.
    enable_folder_ingest: bool = Field(False, alias="ENABLE_FOLDER_INGEST")

    # ---------------- Dispatch (inter-module event bus, D4) ----------------
    dispatch_enabled: bool = Field(False, alias="DISPATCH_ENABLED")
    dispatch_api_url: str = Field("", alias="DISPATCH_API_URL")
    dispatch_api_key: str = Field("", alias="DISPATCH_API_KEY")
    dispatch_participant_code: str = Field("legacyhub", alias="DISPATCH_PARTICIPANT_CODE")
    dispatch_timeout_seconds: int = Field(5, alias="DISPATCH_TIMEOUT_SECONDS")

    @property
    def cors_origins(self) -> list[str]:
        return [o.strip() for o in self.cors_allowed_origins.split(",") if o.strip()]

    # ---------------- Postgres ----------------
    postgres_host: str = Field("postgres", alias="POSTGRES_HOST")
    postgres_port: int = Field(5432, alias="POSTGRES_PORT")
    postgres_db: str = Field("legacyhub", alias="POSTGRES_DB")
    postgres_user: str = Field("legacyhub", alias="POSTGRES_USER")
    # No working default: dev values come from .env / compose; non-dev must set it.
    postgres_password: str = Field("", alias="POSTGRES_PASSWORD")

    @property
    def database_url(self) -> str:
        return (
            f"postgresql+psycopg://{self.postgres_user}:{self.postgres_password}"
            f"@{self.postgres_host}:{self.postgres_port}/{self.postgres_db}"
        )

    # ---------------- MinIO ----------------
    minio_endpoint: str = Field("minio:9000", alias="MINIO_ENDPOINT")
    minio_access_key: str = Field("", alias="MINIO_ACCESS_KEY")
    minio_secret_key: str = Field("", alias="MINIO_SECRET_KEY")
    # Bucket names follow the platform asset standard: teamhub-<module>-<role>.
    minio_bucket_originals: str = Field("teamhub-legacyhub-originals", alias="MINIO_BUCKET_ORIGINALS")
    minio_bucket_derived: str = Field("teamhub-legacyhub-derived", alias="MINIO_BUCKET_DERIVED")
    minio_bucket_quarantine: str = Field(
        "teamhub-legacyhub-quarantine", alias="MINIO_BUCKET_QUARANTINE"
    )
    minio_bucket_tmp: str = Field("teamhub-legacyhub-tmp", alias="MINIO_BUCKET_TMP")
    minio_bucket_exports: str = Field("teamhub-legacyhub-exports", alias="MINIO_BUCKET_EXPORTS")
    minio_secure: bool = Field(False, alias="MINIO_SECURE")
    minio_region: str = Field("us-east-1", alias="MINIO_REGION")

    # ---------------- OpenSearch ----------------
    opensearch_host: str = Field("opensearch", alias="OPENSEARCH_HOST")
    opensearch_port: int = Field(9200, alias="OPENSEARCH_PORT")
    opensearch_use_ssl: bool = Field(False, alias="OPENSEARCH_USE_SSL")
    opensearch_verify_certs: bool = Field(False, alias="OPENSEARCH_VERIFY_CERTS")
    opensearch_user: str = Field("", alias="OPENSEARCH_USER")
    opensearch_password: str = Field("", alias="OPENSEARCH_PASSWORD")
    opensearch_index_chunks: str = Field("legacy_chunks", alias="OPENSEARCH_INDEX_CHUNKS")

    # ---------------- Qdrant ----------------
    qdrant_host: str = Field("qdrant", alias="QDRANT_HOST")
    qdrant_port: int = Field(6333, alias="QDRANT_PORT")
    qdrant_api_key: str = Field("", alias="QDRANT_API_KEY")
    qdrant_collection_chunks: str = Field("legacy_chunks", alias="QDRANT_COLLECTION_CHUNKS")

    # ---------------- Redis ----------------
    redis_url: str = Field("redis://redis:6379/0", alias="REDIS_URL")

    # ---------------- OCR ----------------
    ocr_languages: str = Field("rus+eng", alias="OCR_LANGUAGES")
    ocr_enabled: bool = Field(True, alias="OCR_ENABLED")
    docling_ocr_enabled: bool = Field(False, alias="DOCLING_OCR_ENABLED")
    max_document_timeout_seconds: int = Field(180, alias="MAX_DOCUMENT_TIMEOUT_SECONDS")
    ocr_deskew: bool = Field(True, alias="OCR_DESKEW")
    ocr_clean: bool = Field(True, alias="OCR_CLEAN")
    ocr_optimize: int = Field(1, alias="OCR_OPTIMIZE")

    # ---------------- Embeddings / Reranker ----------------
    embedding_model: str = Field("BAAI/bge-m3", alias="EMBEDDING_MODEL")
    embedding_dim: int = Field(1024, alias="EMBEDDING_DIM")
    embedding_device: Literal["cpu", "cuda", "mps"] = Field("cpu", alias="EMBEDDING_DEVICE")
    embedding_batch_size: int = Field(8, alias="EMBEDDING_BATCH_SIZE")
    embedding_normalize: bool = Field(True, alias="EMBEDDING_NORMALIZE")

    reranker_model: str = Field("BAAI/bge-reranker-v2-m3", alias="RERANKER_MODEL")
    reranker_device: Literal["cpu", "cuda", "mps"] = Field("cpu", alias="RERANKER_DEVICE")
    reranker_enabled: bool = Field(True, alias="RERANKER_ENABLED")
    reranker_batch_size: int = Field(8, alias="RERANKER_BATCH_SIZE")

    # ---------------- Chunking ----------------
    chunk_target_tokens: int = Field(700, alias="CHUNK_TARGET_TOKENS")
    chunk_min_tokens: int = Field(120, alias="CHUNK_MIN_TOKENS")
    chunk_max_tokens: int = Field(900, alias="CHUNK_MAX_TOKENS")
    chunk_overlap_tokens: int = Field(100, alias="CHUNK_OVERLAP_TOKENS")

    # ---------------- Hybrid search ----------------
    hybrid_opensearch_top_k: int = Field(50, alias="HYBRID_OPENSEARCH_TOP_K")
    hybrid_qdrant_top_k: int = Field(50, alias="HYBRID_QDRANT_TOP_K")
    hybrid_rrf_k: int = Field(60, alias="HYBRID_RRF_K")
    rerank_candidates: int = Field(40, alias="RERANK_CANDIDATES")


def _enforce_secret_policy(settings: Settings) -> Settings:
    """Fail loud if a required secret is empty outside dev.

    Raises a plain ``RuntimeError`` naming only the missing keys. We deliberately
    do *not* use a pydantic validator here: pydantic would render the whole input
    (including secret values) into the ``ValidationError`` string.
    """
    if settings.app_env.lower() in _DEV_LIKE_ENVS:
        return settings
    missing = [
        name
        for name, value in (
            ("POSTGRES_PASSWORD", settings.postgres_password),
            ("MINIO_ACCESS_KEY", settings.minio_access_key),
            ("MINIO_SECRET_KEY", settings.minio_secret_key),
        )
        if not (value and value.strip())
    ]
    if missing:
        raise RuntimeError(
            f"APP_ENV={settings.app_env!r} requires secrets to be set: "
            f"{', '.join(missing)} (values are never logged)."
        )
    return settings


@lru_cache(maxsize=1)
def get_settings() -> Settings:
    return _enforce_secret_policy(Settings())  # type: ignore[call-arg]


settings = get_settings()