""" config.py ───────── Centralised configuration loader. All settings are read from environment variables (populated from the .env file via python-dotenv). Sensible defaults are provided where possible so the module never raises at import time — validation errors surface only when you call `get_config()`. """ from __future__ import annotations import logging import os from dataclasses import dataclass, field from pathlib import Path from typing import List from dotenv import load_dotenv # Load .env from the project root (the directory that contains this file). load_dotenv(dotenv_path=Path(__file__).parent / ".env", override=False) # ───────────────────────────────────────────────────────────────────────────── # Data-class that holds the entire runtime configuration # ───────────────────────────────────────────────────────────────────────────── @dataclass class Config: # ── Telegram API ────────────────────────────────────────────────────────── api_id: int api_hash: str phone_number: str # ── Channels ────────────────────────────────────────────────────────────── source_channels: List[str] # usernames or numeric IDs as strings destination_channel: str # username or numeric ID # ── Session ─────────────────────────────────────────────────────────────── session_name: str = "monitor_bot" # ── Paths ───────────────────────────────────────────────────────────────── temp_dir: Path = field(default_factory=lambda: Path("./temp_media")) db_path: Path = field(default_factory=lambda: Path("./processed_messages.db")) # ── Groq AI ─────────────────────────────────────────────────────────────── groq_api_key: str = "" # required when ai_enabled=True ai_enabled: bool = True # set False to bypass all AI processing similarity_threshold: float = 0.80 # duplicate-detection threshold (0–1) recent_news_window: int = 200 # sliding window size (# of recent items) # ── Hugging Face persistent store (optional, replaces SQLite on Render) ── hf_token: str = "" # HF write token (Settings → Access Tokens) hf_dataset_repo: str = "" # e.g. "yourname/pepbielsa-processed" hf_push_every: int = 20 # push to HF every N marks # ── Session String (for Render/cloud deployments) ───────────────────────── session_string: str = "" # export once with session_export.py # ── Logging ─────────────────────────────────────────────────────────────── log_level: str = "INFO" def __post_init__(self) -> None: # Ensure directories exist at startup. self.temp_dir.mkdir(parents=True, exist_ok=True) # ───────────────────────────────────────────────────────────────────────────── # Helper — parse comma-separated channel list # ───────────────────────────────────────────────────────────────────────────── def _parse_channels(raw: str) -> List[str]: """Split a comma-separated string into a clean list of channel identifiers. Each element is stripped of whitespace. Numeric IDs are returned as strings so callers can decide whether to cast them. """ return [ch.strip() for ch in raw.split(",") if ch.strip()] # ───────────────────────────────────────────────────────────────────────────── # Public factory # ───────────────────────────────────────────────────────────────────────────── def get_config() -> Config: """Build and return a validated :class:`Config` instance. Raises ------ ValueError If any *required* environment variable is missing or invalid. """ errors: List[str] = [] # ── Required fields ─────────────────────────────────────────────────────── raw_api_id = os.getenv("API_ID", "") try: api_id = int(raw_api_id) except ValueError: errors.append(f"API_ID must be an integer, got: {raw_api_id!r}") api_id = 0 # placeholder so we can collect more errors api_hash = os.getenv("API_HASH", "") if not api_hash: errors.append("API_HASH is required.") phone_number = os.getenv("PHONE_NUMBER", "") if not phone_number: errors.append("PHONE_NUMBER is required.") destination_channel = os.getenv("DESTINATION_CHANNEL", "") if not destination_channel: errors.append("DESTINATION_CHANNEL is required.") raw_sources = os.getenv("SOURCE_CHANNELS", "") source_channels = _parse_channels(raw_sources) if not source_channels: errors.append( "SOURCE_CHANNELS is required. Provide a comma-separated list of " "channel usernames or numeric IDs." ) if errors: raise ValueError( "Configuration errors found:\n • " + "\n • ".join(errors) ) # ── Optional fields (with defaults) ─────────────────────────────────────── session_name = os.getenv("SESSION_NAME", "monitor_bot") log_level = os.getenv("LOG_LEVEL", "INFO").upper() temp_dir = Path(os.getenv("TEMP_DIR", "./temp_media")) db_path = Path(os.getenv("DB_PATH", "./processed_messages.db")) # ── Groq AI (optional, but required when AI_ENABLED=true) ──────────────── ai_enabled_str = os.getenv("AI_ENABLED", "true").lower() ai_enabled = ai_enabled_str not in ("false", "0", "no", "off") groq_api_key = os.getenv("GROQ_API_KEY", "") if ai_enabled and not groq_api_key: errors.append( "GROQ_API_KEY is required when AI_ENABLED=true. " "Get a free key (no credit card) at https://console.groq.com " "or set AI_ENABLED=false to disable AI features." ) try: similarity_threshold = float(os.getenv("SIMILARITY_THRESHOLD", "0.80")) if not 0.0 <= similarity_threshold <= 1.0: raise ValueError() except ValueError: errors.append("SIMILARITY_THRESHOLD must be a float between 0.0 and 1.0.") similarity_threshold = 0.80 try: recent_news_window = int(os.getenv("RECENT_NEWS_WINDOW", "200")) if recent_news_window < 1: raise ValueError() except ValueError: errors.append("RECENT_NEWS_WINDOW must be a positive integer.") recent_news_window = 200 if errors: raise ValueError( "Configuration errors found:\n • " + "\n • ".join(errors) ) # ── Hugging Face (optional) ─────────────────────────────────────────────── hf_token = os.getenv("HF_TOKEN", "") hf_dataset_repo = os.getenv("HF_DATASET_REPO", "") try: hf_push_every = int(os.getenv("HF_PUSH_EVERY", "20")) except ValueError: hf_push_every = 20 # ── Session String (cloud deployments) ──────────────────────────────────── session_string = os.getenv("SESSION_STRING", "") if errors: raise ValueError( "Configuration errors found:\n • " + "\n • ".join(errors) ) return Config( api_id=api_id, api_hash=api_hash, phone_number=phone_number, source_channels=source_channels, destination_channel=destination_channel, session_name=session_name, log_level=log_level, temp_dir=temp_dir, db_path=db_path, groq_api_key=groq_api_key, ai_enabled=ai_enabled, similarity_threshold=similarity_threshold, recent_news_window=recent_news_window, hf_token=hf_token, hf_dataset_repo=hf_dataset_repo, hf_push_every=hf_push_every, session_string=session_string, ) # ───────────────────────────────────────────────────────────────────────────── # Logging setup (called once from main.py) # ───────────────────────────────────────────────────────────────────────────── def configure_logging(level: str = "INFO") -> logging.Logger: """Configure root logger and return the application logger.""" numeric = getattr(logging, level, logging.INFO) fmt = ( "%(asctime)s | %(levelname)-8s | %(name)-25s | %(message)s" ) date_fmt = "%Y-%m-%d %H:%M:%S" logging.basicConfig( level=numeric, format=fmt, datefmt=date_fmt, handlers=[ logging.StreamHandler(), ], ) # Silence noisy third-party loggers logging.getLogger("telethon").setLevel(logging.WARNING) return logging.getLogger("telegram_monitor")