Spaces:
Running
Running
| """Environment-driven configuration.""" | |
| from __future__ import annotations | |
| import os | |
| from dataclasses import dataclass | |
| from typing import Mapping, Optional, Tuple | |
| class ConfigError(RuntimeError): | |
| pass | |
| DEFAULT_POST_BOT_ID = 8950016484 # the channel post bot: the only account whose answers in the communication chat count | |
| def _parse_ids(value: Optional[str]) -> Tuple[int, ...]: | |
| ids = [] | |
| for chunk in (value or "").replace(";", ",").replace(" ", ",").split(","): | |
| chunk = chunk.strip() | |
| if not chunk: | |
| continue | |
| try: | |
| ids.append(int(chunk)) | |
| except ValueError as exc: | |
| raise ConfigError(f"Invalid id {chunk!r} in ADMIN_IDS") from exc | |
| return tuple(ids) | |
| def _get_int(env: Mapping[str, str], name: str, default: int) -> int: | |
| raw = env.get(name, "").strip() | |
| if not raw: | |
| return default | |
| try: | |
| return int(raw) | |
| except ValueError as exc: | |
| raise ConfigError(f"{name} must be an integer, got {raw!r}") from exc | |
| def _env_any(env: Mapping[str, str], name: str) -> str: | |
| """A variable by name whatever its capital letters (``DB_Channel``, ``DB_CHANNEL`` and ``db_channel`` are one setting).""" | |
| raw = env.get(name) | |
| if raw is None: | |
| wanted = name.lower() | |
| raw = next((v for k, v in env.items() if k.lower() == wanted), "") | |
| return (raw or "").strip() | |
| def _get_chat(env: Mapping[str, str], name: str) -> Optional[int]: | |
| """A chat id such as ``-1001234567890`` (unset or empty: ``None``; a bare channel id gets its ``-100`` in front).""" | |
| raw = _env_any(env, name) | |
| if not raw or raw.lower() in {"0", "off", "none", "false", "no"}: | |
| return None | |
| try: | |
| value = int(raw) | |
| except ValueError as exc: | |
| raise ConfigError(f"{name} must be a numeric chat id like -1001234567890, got {raw!r}") from exc | |
| return -1_000_000_000_000 - value if value > 0 else value | |
| def _get_bool(env: Mapping[str, str], name: str, default: bool) -> bool: | |
| raw = env.get(name, "").strip().lower() | |
| if not raw: | |
| return default | |
| return raw in {"1", "true", "yes", "on"} | |
| class Config: | |
| api_id: int | |
| api_hash: str | |
| bot_token: str | |
| owner_id: int | |
| admin_ids: Tuple[int, ...] | |
| database_url: str | |
| session_string: str | |
| log_channel: Optional[int] | |
| port: int | |
| temp_dir: str | |
| max_jobs: int | |
| connections: int | |
| max_file_mb: int | |
| fast_transfer: bool | |
| progress_interval: float | |
| stream_transfer: bool = True # big files go straight from the download into the upload instead of through the disk | |
| db_channel: Optional[int] = None # the Database Channel a manually renamed episode can be uploaded to | |
| comm_chat: Optional[int] = None # the chat where this bot and the channel post bot talk to each other | |
| post_bot_id: int = DEFAULT_POST_BOT_ID | |
| def max_file_bytes(self) -> int: | |
| return self.max_file_mb * 1024 * 1024 | |
| def upload_enabled(self) -> bool: | |
| """Uploading to the Database Channel (and having the post published) needs both chats.""" | |
| return bool(self.db_channel and self.comm_chat) | |
| def load_config(env: Optional[Mapping[str, str]] = None) -> Config: | |
| env = os.environ if env is None else env | |
| missing = [k for k in ("API_ID", "API_HASH", "BOT_TOKEN", "OWNER_ID") if not env.get(k, "").strip()] | |
| if missing: | |
| raise ConfigError("Missing required environment variables: " + ", ".join(missing)) | |
| log_channel_raw = env.get("LOG_CHANNEL", "").strip() | |
| log_channel = None | |
| if log_channel_raw: | |
| try: | |
| log_channel = int(log_channel_raw) | |
| except ValueError as exc: | |
| raise ConfigError("LOG_CHANNEL must be a numeric chat id like -1001234567890") from exc | |
| db_channel = _get_chat(env, "DB_CHANNEL") | |
| comm_chat = _get_chat(env, "COMMUNICATION_CHAT") | |
| post_bot_raw = _env_any(env, "CHANNEL_POST_BOT_ID") | |
| try: | |
| post_bot_id = int(post_bot_raw) if post_bot_raw else DEFAULT_POST_BOT_ID | |
| except ValueError as exc: | |
| raise ConfigError(f"CHANNEL_POST_BOT_ID must be a numeric user id, got {post_bot_raw!r}") from exc | |
| connections = max(1, min(_get_int(env, "TRANSFER_CONNECTIONS", 8), 16)) | |
| max_jobs = max(1, min(_get_int(env, "MAX_CONCURRENT_JOBS", 1), 4)) | |
| return Config( | |
| api_id=_get_int(env, "API_ID", 0), | |
| api_hash=env["API_HASH"].strip(), | |
| bot_token=env["BOT_TOKEN"].strip(), | |
| owner_id=_get_int(env, "OWNER_ID", 0), | |
| admin_ids=_parse_ids(env.get("ADMIN_IDS")), | |
| database_url=env.get("DATABASE_URL", "").strip() or "sqlite+aiosqlite:///bot.db", | |
| session_string=env.get("SESSION_STRING", "").strip(), | |
| log_channel=log_channel, | |
| port=_get_int(env, "PORT", 7860), | |
| temp_dir=env.get("TEMP_DIR", "").strip() or "/tmp/rename", | |
| max_jobs=max_jobs, | |
| connections=connections, | |
| max_file_mb=max(1, _get_int(env, "MAX_FILE_MB", 2000)), | |
| fast_transfer=_get_bool(env, "FAST_TRANSFER", True), | |
| progress_interval=max(2.0, float(env.get("PROGRESS_INTERVAL", "4") or 4)), | |
| stream_transfer=_get_bool(env, "STREAM_RENAME", True), | |
| db_channel=db_channel, | |
| comm_chat=comm_chat, | |
| post_bot_id=post_bot_id, | |
| ) | |