from __future__ import annotations import json import os import tomllib from dataclasses import dataclass, field from pathlib import Path from typing import Any try: from dotenv import load_dotenv except ModuleNotFoundError: # pragma: no cover - optional dependency fallback def load_dotenv(*args: Any, **kwargs: Any) -> bool: # type: ignore[override] return False @dataclass(slots=True) class AccountConfig: name: str session: str api_id: int | None = None api_hash: str | None = None session_string: str | None = None phone_number: str | None = None password: str | None = None phone_code: str | None = None bot_token: str | None = None proxy: dict[str, Any] | str | None = None workdir: Path | None = None in_memory: bool = False no_updates: bool = False takeout: bool = False hide_password: bool = False use_qr: bool = False @dataclass(slots=True) class ElevenLabsConfig: api_key: str | None = None model_id: str = "scribe_v2" language_code: str | None = None request_timeout_seconds: float = 300.0 @dataclass(slots=True) class AiProviderConfig: api_url: str api_key: str | None = None model: str = "" @dataclass(slots=True) class AiConfig: glm: AiProviderConfig = field( default_factory=lambda: AiProviderConfig( api_url="https://api.z.ai/api/paas/v4", model="glm-5.3", ) ) deepseek: AiProviderConfig = field( default_factory=lambda: AiProviderConfig( api_url="https://api.deepseek.com", model="deepseek-v4-flash", ) ) claude: AiProviderConfig = field( default_factory=lambda: AiProviderConfig( api_url="https://api.anthropic.com", model="claude-haiku-5-5", ) ) openrouter: AiProviderConfig = field( default_factory=lambda: AiProviderConfig( api_url="https://openrouter.ai/api/v1", model="anthropic/claude-haiku-5.5", ) ) prompt_dir: Path = Path("prompts") base_dir: Path = Path(".") max_context_messages: int = 100 max_media_bytes: int = 5 * 1024 * 1024 max_tool_rounds: int = 16 brave_api_key: str | None = None brave_timeout_seconds: float = 10.0 web_page_timeout_seconds: float = 45.0 web_page_max_chars: int = 60_000 @dataclass(slots=True) class AppConfig: api_id: int | None api_hash: str | None session_dir: Path accounts: list[AccountConfig] elevenlabs: ElevenLabsConfig = field(default_factory=ElevenLabsConfig) ai: AiConfig = field(default_factory=AiConfig) def _read_toml_file(path: Path) -> dict[str, Any]: with path.open("rb") as handle: return tomllib.load(handle) def _parse_int(value: Any, *, field_name: str) -> int: try: return int(value) except (TypeError, ValueError) as exc: raise ValueError(f"{field_name} must be an integer") from exc def _parse_bool(value: Any, *, field_name: str) -> bool: if isinstance(value, bool): return value if isinstance(value, str): normalized = value.strip().lower() if normalized in {"1", "true", "yes", "on"}: return True if normalized in {"0", "false", "no", "off"}: return False if isinstance(value, int): return bool(value) raise ValueError(f"{field_name} must be a boolean") def _parse_float(value: Any, *, field_name: str) -> float: try: return float(value) except (TypeError, ValueError) as exc: raise ValueError(f"{field_name} must be a number") from exc def _parse_mapping(value: Any, *, field_name: str) -> dict[str, Any]: if value is None: return {} if not isinstance(value, dict): raise ValueError(f"{field_name} must be an object") return value def _parse_elevenlabs_config(raw: Any) -> ElevenLabsConfig: raw_config = _parse_mapping(raw, field_name="elevenlabs") raw_api_key = raw_config.get("api_key") api_key = str(raw_api_key).strip() if raw_api_key is not None else None api_key = api_key or None model_id = str(raw_config.get("model_id", "scribe_v2") or "scribe_v2").strip() raw_language = raw_config.get("language_code") language_code = str(raw_language).strip() if raw_language is not None else None language_code = language_code or None request_timeout_seconds = _parse_float( raw_config.get("request_timeout_seconds", 300), field_name="elevenlabs.request_timeout_seconds", ) if os.environ.get("ELEVENLABS_API_KEY") is not None: api_key = os.environ["ELEVENLABS_API_KEY"].strip() or None if os.environ.get("ELEVENLABS_MODEL_ID") is not None: model_id = os.environ["ELEVENLABS_MODEL_ID"].strip() if os.environ.get("ELEVENLABS_LANGUAGE_CODE") is not None: language_code = os.environ["ELEVENLABS_LANGUAGE_CODE"].strip() or None if os.environ.get("ELEVENLABS_REQUEST_TIMEOUT_SECONDS") is not None: request_timeout_seconds = _parse_float( os.environ["ELEVENLABS_REQUEST_TIMEOUT_SECONDS"], field_name="ELEVENLABS_REQUEST_TIMEOUT_SECONDS", ) if not model_id: raise ValueError("elevenlabs.model_id is required") if request_timeout_seconds <= 0: raise ValueError("elevenlabs.request_timeout_seconds must be greater than zero") return ElevenLabsConfig( api_key=api_key, model_id=model_id, language_code=language_code, request_timeout_seconds=request_timeout_seconds, ) def _parse_ai_config(raw: Any) -> AiConfig: raw_config = _parse_mapping(raw, field_name="ai") raw_prompt_dir = raw_config.get("prompt_dir", "prompts") prompt_dir_value = str(raw_prompt_dir).strip() if os.environ.get("AI_PROMPT_DIR") is not None: prompt_dir_value = os.environ["AI_PROMPT_DIR"].strip() if not prompt_dir_value: raise ValueError("ai.prompt_dir is required") prompt_dir = Path(prompt_dir_value).expanduser() def parse_provider( name: str, *, default_url: str, default_model: str, env_prefix: str, ) -> AiProviderConfig: provider = _parse_mapping(raw_config.get(name), field_name=f"ai.{name}") api_url = str(provider.get("api_url", default_url)).strip() raw_api_key = provider.get("api_key") api_key = str(raw_api_key).strip() if raw_api_key is not None else None api_key = api_key or None model = str(provider.get("model", default_model)).strip() if os.environ.get(f"{env_prefix}_API_URL") is not None: api_url = os.environ[f"{env_prefix}_API_URL"].strip() if os.environ.get(f"{env_prefix}_API_KEY") is not None: api_key = os.environ[f"{env_prefix}_API_KEY"].strip() or None if os.environ.get(f"{env_prefix}_MODEL") is not None: model = os.environ[f"{env_prefix}_MODEL"].strip() if not api_url: raise ValueError(f"ai.{name}.api_url is required") if not model: raise ValueError(f"ai.{name}.model is required") return AiProviderConfig(api_url=api_url, api_key=api_key, model=model) glm = parse_provider( "glm", default_url="https://api.z.ai/api/paas/v4", default_model="glm-5.3", env_prefix="GLM", ) deepseek = parse_provider( "deepseek", default_url="https://api.deepseek.com", default_model="deepseek-v4-flash", env_prefix="DEEPSEEK", ) claude = parse_provider( "claude", default_url="https://api.anthropic.com", default_model="claude-haiku-5-5", env_prefix="CLAUDE", ) openrouter = parse_provider( "openrouter", default_url="https://openrouter.ai/api/v1", default_model="anthropic/claude-haiku-5.5", env_prefix="OPENROUTER", ) max_context_messages = _parse_int( raw_config.get("max_context_messages", 100), field_name="ai.max_context_messages", ) max_media_bytes = _parse_int( raw_config.get("max_media_bytes", 5 * 1024 * 1024), field_name="ai.max_media_bytes", ) max_tool_rounds = _parse_int( raw_config.get("max_tool_rounds", 16), field_name="ai.max_tool_rounds", ) brave_timeout_seconds = _parse_float( raw_config.get("brave_timeout_seconds", 10), field_name="ai.brave_timeout_seconds", ) web_page_timeout_seconds = _parse_float( raw_config.get("web_page_timeout_seconds", 45), field_name="ai.web_page_timeout_seconds", ) web_page_max_chars = _parse_int( raw_config.get("web_page_max_chars", 60_000), field_name="ai.web_page_max_chars", ) raw_brave_key = raw_config.get("brave_api_key") brave_api_key = str(raw_brave_key).strip() if raw_brave_key is not None else None brave_api_key = brave_api_key or None if os.environ.get("BRAVE_SEARCH_API_KEY") is not None: brave_api_key = os.environ["BRAVE_SEARCH_API_KEY"].strip() or None if os.environ.get("AI_MAX_CONTEXT_MESSAGES") is not None: max_context_messages = _parse_int( os.environ["AI_MAX_CONTEXT_MESSAGES"], field_name="AI_MAX_CONTEXT_MESSAGES", ) if os.environ.get("AI_MAX_MEDIA_BYTES") is not None: max_media_bytes = _parse_int( os.environ["AI_MAX_MEDIA_BYTES"], field_name="AI_MAX_MEDIA_BYTES", ) if os.environ.get("AI_MAX_TOOL_ROUNDS") is not None: max_tool_rounds = _parse_int( os.environ["AI_MAX_TOOL_ROUNDS"], field_name="AI_MAX_TOOL_ROUNDS", ) if os.environ.get("BRAVE_SEARCH_TIMEOUT_SECONDS") is not None: brave_timeout_seconds = _parse_float( os.environ["BRAVE_SEARCH_TIMEOUT_SECONDS"], field_name="BRAVE_SEARCH_TIMEOUT_SECONDS", ) if os.environ.get("AI_WEB_PAGE_TIMEOUT_SECONDS") is not None: web_page_timeout_seconds = _parse_float( os.environ["AI_WEB_PAGE_TIMEOUT_SECONDS"], field_name="AI_WEB_PAGE_TIMEOUT_SECONDS", ) if os.environ.get("AI_WEB_PAGE_MAX_CHARS") is not None: web_page_max_chars = _parse_int( os.environ["AI_WEB_PAGE_MAX_CHARS"], field_name="AI_WEB_PAGE_MAX_CHARS", ) if max_context_messages < 0: raise ValueError("ai.max_context_messages must not be negative") if max_media_bytes < 1: raise ValueError("ai.max_media_bytes must be greater than zero") if max_tool_rounds < 1: raise ValueError("ai.max_tool_rounds must be greater than zero") if brave_timeout_seconds <= 0: raise ValueError("ai.brave_timeout_seconds must be greater than zero") if web_page_timeout_seconds <= 0: raise ValueError("ai.web_page_timeout_seconds must be greater than zero") if web_page_max_chars < 1: raise ValueError("ai.web_page_max_chars must be greater than zero") return AiConfig( glm=glm, deepseek=deepseek, claude=claude, openrouter=openrouter, prompt_dir=prompt_dir, max_context_messages=max_context_messages, max_media_bytes=max_media_bytes, max_tool_rounds=max_tool_rounds, brave_api_key=brave_api_key, brave_timeout_seconds=brave_timeout_seconds, web_page_timeout_seconds=web_page_timeout_seconds, web_page_max_chars=web_page_max_chars, ) def _parse_accounts(raw_accounts: Any, *, default_api_id: int | None, default_api_hash: str | None, default_session_dir: Path) -> list[AccountConfig]: if raw_accounts is None: return [] if not isinstance(raw_accounts, list): raise ValueError("accounts must be a list") accounts: list[AccountConfig] = [] for index, raw_account in enumerate(raw_accounts, start=1): if isinstance(raw_account, str): raw_account = {"name": raw_account, "session": raw_account} if not isinstance(raw_account, dict): raise ValueError(f"accounts[{index}] must be a string or object") name = str( raw_account.get("name") or raw_account.get("label") or raw_account.get("session") or raw_account.get("session_name") or "" ).strip() session = str(raw_account.get("session") or raw_account.get("session_name") or name).strip() if not name: raise ValueError(f"accounts[{index}] is missing name") if not session: raise ValueError(f"accounts[{index}] is missing session") account_api_id = raw_account.get("api_id", default_api_id) account_api_hash = raw_account.get("api_hash", default_api_hash) if account_api_id is not None: account_api_id = _parse_int(account_api_id, field_name=f"accounts[{index}].api_id") if account_api_hash is not None: account_api_hash = str(account_api_hash) account_workdir = raw_account.get("workdir") if account_workdir is not None: account_workdir = Path(str(account_workdir)).expanduser() if not account_workdir.is_absolute(): account_workdir = (default_session_dir.parent / account_workdir).resolve() accounts.append( AccountConfig( name=name, session=session, api_id=account_api_id, api_hash=account_api_hash, session_string=raw_account.get("session_string"), phone_number=raw_account.get("phone_number"), password=raw_account.get("password"), phone_code=raw_account.get("phone_code"), bot_token=raw_account.get("bot_token"), proxy=raw_account.get("proxy"), workdir=account_workdir, in_memory=bool(raw_account.get("in_memory", False) or raw_account.get("session_string")), no_updates=bool(raw_account.get("no_updates", False)), takeout=bool(raw_account.get("takeout", False)), hide_password=bool(raw_account.get("hide_password", False)), use_qr=_parse_bool( raw_account.get("use_qr", False), field_name=f"accounts[{index}].use_qr", ), ) ) return accounts def _accounts_from_env() -> list[dict[str, Any]] | None: raw_accounts = os.environ.get("ACCOUNTS_JSON") if raw_accounts: parsed = json.loads(raw_accounts) if not isinstance(parsed, list): raise ValueError("ACCOUNTS_JSON must contain a JSON list") return parsed raw_sessions = os.environ.get("ACCOUNT_SESSIONS") if raw_sessions: sessions = [item.strip() for item in raw_sessions.split(",") if item.strip()] return [{"name": session, "session": session} for session in sessions] return None def load_app_config(root: Path) -> AppConfig: dotenv_path = os.environ.get("BOT_DOTENV_PATH") if dotenv_path: candidate_dotenv = Path(dotenv_path).expanduser() if not candidate_dotenv.is_absolute(): candidate_dotenv = (root / candidate_dotenv).resolve() else: candidate_dotenv = root / ".env" load_dotenv(candidate_dotenv, override=False) config_path = os.environ.get("BOT_CONFIG") if config_path: candidate = Path(config_path).expanduser() if not candidate.is_absolute(): candidate = (root / candidate).resolve() else: candidate = root / "config.toml" raw: dict[str, Any] = {} if candidate.exists(): raw = _read_toml_file(candidate) env_api_id = os.environ.get("TG_API_ID") env_api_hash = os.environ.get("TG_API_HASH") env_session_dir = os.environ.get("SESSION_DIR") if env_api_id is not None: raw["api_id"] = env_api_id if env_api_hash is not None: raw["api_hash"] = env_api_hash if env_session_dir is not None: raw["session_dir"] = env_session_dir env_accounts = _accounts_from_env() if env_accounts is not None: raw["accounts"] = env_accounts if "accounts" not in raw: raise ValueError( "Missing accounts configuration. Provide accounts in config.toml or set ACCOUNTS_JSON / ACCOUNT_SESSIONS." ) api_id = _parse_int(raw["api_id"], field_name="api_id") if "api_id" in raw else None api_hash = str(raw["api_hash"]) if "api_hash" in raw else None session_dir = Path(str(raw.get("session_dir", "sessions"))).expanduser() if not session_dir.is_absolute(): session_dir = (root / session_dir).resolve() accounts = _parse_accounts( raw["accounts"], default_api_id=api_id, default_api_hash=api_hash, default_session_dir=session_dir, ) elevenlabs = _parse_elevenlabs_config(raw.get("elevenlabs")) ai = _parse_ai_config(raw.get("ai")) if not ai.prompt_dir.is_absolute(): ai.prompt_dir = root / ai.prompt_dir ai.prompt_dir = ai.prompt_dir.resolve() ai.base_dir = root if not accounts: raise ValueError("At least one account must be configured") seen_names: set[str] = set() seen_sessions: set[str] = set() for account in accounts: if account.name in seen_names: raise ValueError(f"Duplicate account name: {account.name}") if account.session in seen_sessions: raise ValueError(f"Duplicate session name: {account.session}") seen_names.add(account.name) seen_sessions.add(account.session) return AppConfig( api_id=api_id, api_hash=api_hash, session_dir=session_dir, accounts=accounts, elevenlabs=elevenlabs, ai=ai, )