data_studio/config.py

from __future__ import annotations

import os
from dataclasses import dataclass
from pathlib import Path

from dotenv import load_dotenv


DEFAULT_DB_NAME = "rachana_data_studio"


class ConfigError(RuntimeError):
    """Raised when required Data Studio configuration is missing."""


@dataclass(frozen=True)
class StudioSettings:
    mongodb_uri: str
    mongodb_db_name: str
    hf_token: str
    hf_raw_dataset_repo: str
    hf_clean_dataset_repo: str
    hf_audit_dataset_repo: str
    hf_live_review_dataset_repo: str
    custom_pure_telugu_hf_dataset: str
    custom_pure_telugu_source_name: str
    active_tokenizer_version: str
    active_tokenizer_model_path: str
    active_ner_model_id: str
    target_clean_tokens: int
    phase_a_token_target: int
    phase_b_token_target: int
    phase_c_token_target: int
    studio_env: str
    deployment_version: str
    code_commit: str
    allow_legacy_import_export: bool
    single_user_mode: bool
    single_user_username: str

    @classmethod
    def from_env(cls, env_path: Path | None = None) -> "StudioSettings":
        env_file = env_path or Path(".env")
        if env_file.exists():
            load_dotenv(env_file)

        settings = cls(
            mongodb_uri=os.getenv("MONGODB_URI", "").strip(),
            mongodb_db_name=os.getenv("MONGODB_DB_NAME", DEFAULT_DB_NAME).strip() or DEFAULT_DB_NAME,
            hf_token=os.getenv("HF_Token", "").strip(),
            hf_raw_dataset_repo=os.getenv("HF_RAW_DATASET_REPO", "").strip(),
            hf_clean_dataset_repo=os.getenv("HF_CLEAN_DATASET_REPO", "").strip(),
            hf_audit_dataset_repo=os.getenv("HF_AUDIT_DATASET_REPO", "").strip(),
            hf_live_review_dataset_repo=os.getenv("HF_LIVE_REVIEW_DATASET_REPO", "").strip(),
            custom_pure_telugu_hf_dataset=os.getenv("CUSTOM_PURE_TELUGU_HF_DATASET", "").strip(),
            custom_pure_telugu_source_name=os.getenv(
                "CUSTOM_PURE_TELUGU_SOURCE_NAME",
                "Rachana Combined Telugu Content",
            ).strip()
            or "Rachana Combined Telugu Content",
            active_tokenizer_version=os.getenv("ACTIVE_TOKENIZER_VERSION", "rachana_bpe32k_v1").strip()
            or "rachana_bpe32k_v1",
            active_tokenizer_model_path=os.getenv(
                "ACTIVE_TOKENIZER_MODEL_PATH",
                str(Path("tokenizer") / "rachana_bpe32k.model"),
            ).strip()
            or str(Path("tokenizer") / "rachana_bpe32k.model"),
            active_ner_model_id=os.getenv("ACTIVE_NER_MODEL_ID", "").strip(),
            target_clean_tokens=int(os.getenv("TARGET_CLEAN_TOKENS", "0").strip() or "0"),
            phase_a_token_target=int(os.getenv("PHASE_A_TOKEN_TARGET", "25000000").strip() or "25000000"),
            phase_b_token_target=int(os.getenv("PHASE_B_TOKEN_TARGET", "100000000").strip() or "100000000"),
            phase_c_token_target=int(os.getenv("PHASE_C_TOKEN_TARGET", "250000000").strip() or "250000000"),
            studio_env=os.getenv("STUDIO_ENV", "development").strip().lower(),
            deployment_version=os.getenv("DEPLOYMENT_VERSION", "local").strip() or "local",
            code_commit=os.getenv("CODE_COMMIT", "unknown").strip() or "unknown",
            allow_legacy_import_export=os.getenv("ALLOW_LEGACY_IMPORT_EXPORT", "false").strip().lower()
            in {"1", "true", "yes"},
            single_user_mode=os.getenv("SINGLE_USER_MODE", "false").strip().lower() in {"1", "true", "yes"},
            single_user_username=os.getenv("SINGLE_USER_USERNAME", "").strip().lower(),
        )
        settings.validate()
        return settings

    def validate(self) -> None:
        missing: list[str] = []
        if not self.mongodb_uri:
            missing.append("MONGODB_URI")
        if not self.hf_token:
            missing.append("HF_Token")
        if missing:
            joined = ", ".join(missing)
            raise ConfigError(f"Missing required Data Studio environment variables: {joined}")
        if self.studio_env not in {"development", "staging", "production"}:
            raise ConfigError("STUDIO_ENV must be development, staging, or production.")
        if self.single_user_mode and not self.single_user_username:
            raise ConfigError("SINGLE_USER_USERNAME is required when SINGLE_USER_MODE is enabled.")
        tokenizer_path = Path(self.active_tokenizer_model_path)
        if not tokenizer_path.exists():
            raise ConfigError(f"ACTIVE_TOKENIZER_MODEL_PATH does not exist: {tokenizer_path}")

    def require_legacy_import_export_enabled(self) -> None:
        if not self.allow_legacy_import_export:
            raise ConfigError(
                "Legacy import/export is frozen until Phase 0 governance contracts are enforced. "
                "Set ALLOW_LEGACY_IMPORT_EXPORT=true only for explicitly approved test or recovery work."
            )

    def deployment_metadata(self) -> dict[str, str | bool]:
        return {
            "environment": self.studio_env,
            "deployment_version": self.deployment_version,
            "code_commit": self.code_commit,
            "legacy_import_export_enabled": self.allow_legacy_import_export,
            "custom_pure_telugu_hf_dataset": self.custom_pure_telugu_hf_dataset or None,
            "active_tokenizer_version": self.active_tokenizer_version,
            "active_ner_model_id": self.active_ner_model_id,
            "target_clean_tokens": self.target_clean_tokens,
            "phase_a_token_target": self.phase_a_token_target,
            "phase_b_token_target": self.phase_b_token_target,
            "phase_c_token_target": self.phase_c_token_target,
            "single_user_mode": self.single_user_mode,
            "single_user_username": self.single_user_username or None,
        }