X-Financial/server/src/app/core/config.py

from __future__ import annotations

from os import environ
from pathlib import Path

from dotenv import dotenv_values
from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict

SERVER_DIR = Path(__file__).resolve().parents[3]
ROOT_DIR = SERVER_DIR.parent
DEFAULT_ENV_FILES = (ROOT_DIR / ".env", SERVER_DIR / ".env")
ONLYOFFICE_FIELD_NAMES = {
    "ONLYOFFICE_ENABLED": "onlyoffice_enabled",
    "ONLYOFFICE_PUBLIC_URL": "onlyoffice_public_url",
    "ONLYOFFICE_BACKEND_URL": "onlyoffice_backend_url",
    "ONLYOFFICE_JWT_SECRET": "onlyoffice_jwt_secret",
}

_settings_cache: Settings | None = None
_settings_cache_signature: tuple[tuple[str, bool, int | None, int | None], ...] | None = None


class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=DEFAULT_ENV_FILES,
        env_file_encoding="utf-8",
        extra="ignore",
    )

    app_name: str = Field(default="X-Financial Server", alias="APP_NAME")
    app_env: str = Field(default="local", alias="APP_ENV")
    app_debug: bool = Field(default=True, alias="APP_DEBUG")
    setup_completed: bool = Field(default=False, alias="SETUP_COMPLETED")

    company_name: str = Field(default="", alias="COMPANY_NAME")
    company_code: str = Field(default="", alias="COMPANY_CODE")
    admin_email: str = Field(default="", alias="ADMIN_EMAIL")

    web_host: str = Field(default="0.0.0.0", alias="WEB_HOST")
    web_port: int = Field(default=5273, alias="WEB_PORT")
    app_host: str = Field(default="0.0.0.0", alias="SERVER_HOST")
    app_port: int = Field(default=8000, alias="SERVER_PORT")
    server_workers: int = Field(default=1, alias="SERVER_WORKERS")
    web_concurrency: int | None = Field(default=None, alias="WEB_CONCURRENCY")
    startup_bootstrap_enabled: bool = Field(default=True, alias="STARTUP_BOOTSTRAP_ENABLED")
    startup_cache_warmup_enabled: bool = Field(default=False, alias="STARTUP_CACHE_WARMUP_ENABLED")
    background_schedulers_enabled: bool = Field(
        default=True,
        alias="BACKGROUND_SCHEDULERS_ENABLED",
    )
    api_v1_prefix: str = Field(default="/api/v1", alias="API_V1_PREFIX")

    postgres_host: str = Field(default="127.0.0.1", alias="POSTGRES_HOST")
    postgres_port: int = Field(default=5432, alias="POSTGRES_PORT")
    postgres_db: str = Field(default="x_financial", alias="POSTGRES_DB")
    postgres_user: str = Field(default="postgres", alias="POSTGRES_USER")
    postgres_password: str = Field(default="postgres", alias="POSTGRES_PASSWORD")

    database_url: str | None = Field(default=None, alias="DATABASE_URL")
    sqlalchemy_echo: bool = Field(default=False, alias="SQLALCHEMY_ECHO")
    sqlalchemy_pool_size: int = Field(default=10, alias="SQLALCHEMY_POOL_SIZE")
    sqlalchemy_max_overflow: int = Field(default=20, alias="SQLALCHEMY_MAX_OVERFLOW")
    sqlalchemy_pool_timeout: int = Field(default=30, alias="SQLALCHEMY_POOL_TIMEOUT")

    redis_url: str | None = Field(default=None, alias="REDIS_URL")
    cors_origins: list[str] = Field(default_factory=list, alias="CORS_ORIGINS")
    vite_api_base_url: str = Field(
        default="http://127.0.0.1:8000/api/v1", alias="VITE_API_BASE_URL"
    )
    onlyoffice_enabled: bool = Field(default=False, alias="ONLYOFFICE_ENABLED")
    onlyoffice_public_url: str = Field(default="", alias="ONLYOFFICE_PUBLIC_URL")
    onlyoffice_backend_url: str = Field(default="", alias="ONLYOFFICE_BACKEND_URL")
    onlyoffice_jwt_secret: str = Field(default="", alias="ONLYOFFICE_JWT_SECRET")
    hermes_agent_shared_token: str = Field(default="", alias="HERMES_AGENT_SHARED_TOKEN")
    steward_agent_runtime: str = Field(default="langgraph", alias="STEWARD_AGENT_RUNTIME")

    log_level: str = Field(default="INFO", alias="LOG_LEVEL")
    log_dir: str = Field(default="logs", alias="LOG_DIR")
    log_file_enabled: bool = Field(default=True, alias="LOG_FILE_ENABLED")
    storage_root_dir: str = Field(default="storage", alias="STORAGE_ROOT_DIR")
    ocr_python_bin: str = Field(default="", alias="OCR_PYTHON_BIN")
    ocr_device: str = Field(default="", alias="OCR_DEVICE")
    ocr_timeout_seconds: int = Field(default=180, alias="OCR_TIMEOUT_SECONDS")
    ocr_max_file_size_mb: int = Field(default=20, alias="OCR_MAX_FILE_SIZE_MB")
    ocr_max_concurrent_workers: int = Field(default=1, alias="OCR_MAX_CONCURRENT_WORKERS")
    ocr_language: str = Field(default="ch", alias="OCR_LANGUAGE")
    seed_demo_financial_records: bool = Field(
        default=False,
        alias="SEED_DEMO_FINANCIAL_RECORDS",
    )
    ocr_text_detection_model: str = Field(
        default="PP-OCRv5_mobile_det",
        alias="OCR_TEXT_DETECTION_MODEL",
    )
    ocr_text_recognition_model: str = Field(
        default="PP-OCRv5_mobile_rec",
        alias="OCR_TEXT_RECOGNITION_MODEL",
    )

    @property
    def resolved_database_url(self) -> str:
        if self.database_url:
            return self.database_url

        return (
            f"postgresql+psycopg://{self.postgres_user}:{self.postgres_password}"
            f"@{self.postgres_host}:{self.postgres_port}/{self.postgres_db}"
        )

    @property
    def resolved_storage_root_dir(self) -> Path:
        path = Path(self.storage_root_dir)
        if not path.is_absolute():
            path = SERVER_DIR / path
        return path.resolve()

    @property
    def resolved_ocr_temp_dir(self) -> Path:
        return (self.resolved_storage_root_dir / "ocr_temp").resolve()

def _resolve_env_files() -> tuple[Path, ...]:
    env_files = Settings.model_config.get("env_file") or ()
    return tuple(Path(item) for item in env_files)


def _build_settings_signature() -> tuple[tuple[str, bool, int | None, int | None], ...]:
    signature: list[tuple[str, bool, int | None, int | None]] = []

    for env_file in _resolve_env_files():
        if not env_file.exists():
            signature.append((str(env_file), False, None, None))
            continue

        stat = env_file.stat()
        signature.append((str(env_file), True, stat.st_mtime_ns, stat.st_size))

    return tuple(signature)


def _parse_onlyoffice_enabled(value: object) -> bool:
    return str(value).strip().lower() in {"1", "true", "yes", "on"}


def _load_onlyoffice_env_file_overrides() -> dict[str, object]:
    overrides: dict[str, object] = {}

    for env_file in _resolve_env_files():
        if not env_file.exists():
            continue

        values = dotenv_values(env_file)
        for alias, field_name in ONLYOFFICE_FIELD_NAMES.items():
            if alias not in values:
                continue

            value = values[alias]
            if field_name == "onlyoffice_enabled":
                overrides[field_name] = _parse_onlyoffice_enabled(value)
            else:
                overrides[field_name] = "" if value is None else str(value)

    return overrides


def _clear_settings_cache() -> None:
    global _settings_cache, _settings_cache_signature

    _settings_cache = None
    _settings_cache_signature = None


def clear_runtime_settings_cache() -> int:
    cleared_count = int(_settings_cache is not None)
    _clear_settings_cache()
    return cleared_count


def get_settings() -> Settings:
    global _settings_cache, _settings_cache_signature

    signature = _build_settings_signature()
    if _settings_cache is None or _settings_cache_signature != signature:
        settings = Settings()
        onlyoffice_overrides = _load_onlyoffice_env_file_overrides()
        if onlyoffice_overrides:
            settings = settings.model_copy(update=onlyoffice_overrides)

        _settings_cache = settings
        _settings_cache_signature = signature

    return _settings_cache


get_settings.cache_clear = _clear_settings_cache  # type: ignore[attr-defined]


def refresh_settings(updated_values: dict[str, str]) -> Settings:
    for key, value in updated_values.items():
        environ[key] = value

    get_settings.cache_clear()
    return get_settings()