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, }