Spaces:
Running
Running
File size: 5,309 Bytes
58f00c9 a670808 58f00c9 a670808 58f00c9 529b973 a670808 58f00c9 a670808 58f00c9 a670808 58f00c9 529b973 a670808 58f00c9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 | """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"}
@dataclass(frozen=True)
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
@property
def max_file_bytes(self) -> int:
return self.max_file_mb * 1024 * 1024
@property
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,
)
|