4f57cccea5
Enhance the bot's architecture by introducing sharding capabilities, allowing for better scalability and performance. Update environment files to include sharding options, and modify the bot's core logic to handle shard connections and latencies. Implement shard-specific logging and API synchronization, ensuring that only the primary shard performs certain tasks. Update API schemas and routes to reflect shard information, improving the overall status reporting and monitoring of the bot's performance across multiple shards.
130 lines
5.5 KiB
Python
130 lines
5.5 KiB
Python
# ╔══════════════════════════════════════════════════════════════════╗
|
|
# ║ ║
|
|
# ║ +-+-+-+-+-+-+-+-+ ║
|
|
# ║ |H|e|x|a|H|o|s|t| ║
|
|
# ║ +-+-+-+-+-+-+-+-+ ║
|
|
# ║ ║
|
|
# ║ © 2026 HexaHost — All Rights Reserved ║
|
|
# ║ ║
|
|
# ║ discord ── https://discord.gg/hexahost ║
|
|
# ║ github ── https://github.com/theoneandonlymace ║
|
|
# ║ ║
|
|
# ╚══════════════════════════════════════════════════════════════════╝
|
|
|
|
import os
|
|
from dotenv import load_dotenv
|
|
|
|
load_dotenv()
|
|
|
|
TOKEN = os.environ.get("TOKEN")
|
|
BRAND_NAME = os.environ.get("brand_name", "Axiom")
|
|
NAME = BRAND_NAME
|
|
BotName = BRAND_NAME
|
|
|
|
server = "https://discord.gg/hexahost"
|
|
serverLink = "https://discord.gg/hexahost"
|
|
|
|
# ── Security / feature toggles ────────────────────────────────────────────────
|
|
|
|
def _env_bool(key: str, default: bool = False) -> bool:
|
|
raw = os.getenv(key, str(default)).strip().lower()
|
|
return raw in ("1", "true", "yes", "on")
|
|
|
|
|
|
JISHAKU_ENABLED = _env_bool("JISHAKU_ENABLED", False)
|
|
API_ENABLED = _env_bool("API_ENABLED", False)
|
|
TUNNEL_ENABLED = _env_bool("TUNNEL_ENABLED", False)
|
|
TUNNEL_ALLOW_BINARY_DOWNLOAD = _env_bool("TUNNEL_ALLOW_BINARY_DOWNLOAD", False)
|
|
|
|
API_HOST = os.getenv("API_HOST", "127.0.0.1").strip() or "127.0.0.1"
|
|
API_PORT = int(os.getenv("API_PORT", "8000"))
|
|
|
|
# ── Webhooks (only active when explicitly configured) ─────────────────────────
|
|
|
|
def _valid_webhook_url(url: str | None) -> str | None:
|
|
if not url:
|
|
return None
|
|
url = url.strip()
|
|
if not url or url == "https://discord.com/api/webhooks/":
|
|
return None
|
|
if "discord.com/api/webhooks/" not in url and "discordapp.com/api/webhooks/" not in url:
|
|
return None
|
|
return url
|
|
|
|
CMD_WEBHOOK_URL = _valid_webhook_url(os.getenv("CMD_WEBHOOK_URL"))
|
|
|
|
# ── Optional Discord log / stats channels (unset = disabled) ──────────────────
|
|
|
|
def _parse_optional_id(env_key: str) -> int | None:
|
|
raw = os.getenv(env_key, "").strip()
|
|
if not raw or not raw.isdigit():
|
|
return None
|
|
return int(raw)
|
|
|
|
LOG_CHANNEL_ID = _parse_optional_id("LOG_CHANNEL_ID")
|
|
SERVER_COUNT_CHANNEL_ID = _parse_optional_id("SERVER_COUNT_CHANNEL_ID")
|
|
USER_COUNT_CHANNEL_ID = _parse_optional_id("USER_COUNT_CHANNEL_ID")
|
|
|
|
# ── Owner / Staff IDs ─────────────────────────────────────────────────────────
|
|
# REQUIRED: set OWNER_IDS in .env — no hardcoded fallback IDs.
|
|
|
|
def _parse_ids(env_key: str) -> list[int]:
|
|
raw = os.getenv(env_key, "").strip()
|
|
if not raw:
|
|
return []
|
|
return [int(p.strip()) for p in raw.split(",") if p.strip().isdigit()]
|
|
|
|
OWNER_IDS: list[int] = _parse_ids("OWNER_IDS")
|
|
OWNER_IDS_STR: list[str] = [str(i) for i in OWNER_IDS]
|
|
|
|
# Discord application / OAuth client id (same as dashboard DISCORD_CLIENT_ID)
|
|
DISCORD_CLIENT_ID = os.getenv("DISCORD_CLIENT_ID", "").strip()
|
|
|
|
# ── Discord sharding ──────────────────────────────────────────────────────────
|
|
# SHARD_COUNT empty = AutoShardedBot asks Discord for recommended count (in-process).
|
|
# SHARD_IDS (e.g. "0,1") + SHARD_COUNT = this process only runs those shards (cluster).
|
|
# Primary process runs API / slash sync / emoji sync / stats (shard 0 or override).
|
|
|
|
def _parse_optional_int(env_key: str) -> int | None:
|
|
raw = os.getenv(env_key, "").strip()
|
|
if not raw or not raw.isdigit():
|
|
return None
|
|
return int(raw)
|
|
|
|
|
|
SHARD_COUNT: int | None = _parse_optional_int("SHARD_COUNT")
|
|
SHARD_IDS: list[int] = _parse_ids("SHARD_IDS")
|
|
|
|
|
|
def get_shard_kwargs() -> dict:
|
|
"""Kwargs for AutoShardedBot.__init__ (omit keys for full auto-sharding)."""
|
|
kwargs: dict = {}
|
|
if SHARD_COUNT is not None:
|
|
kwargs["shard_count"] = SHARD_COUNT
|
|
if SHARD_IDS:
|
|
if SHARD_COUNT is None:
|
|
raise ValueError("SHARD_IDS requires SHARD_COUNT to be set")
|
|
kwargs["shard_ids"] = SHARD_IDS
|
|
kwargs["shard_count"] = SHARD_COUNT
|
|
return kwargs
|
|
|
|
|
|
def _resolve_is_shard_primary() -> bool:
|
|
raw = os.getenv("SHARD_PRIMARY", "").strip().lower()
|
|
if raw in ("1", "true", "yes", "on"):
|
|
return True
|
|
if raw in ("0", "false", "no", "off"):
|
|
return False
|
|
if not SHARD_IDS:
|
|
return True
|
|
return 0 in SHARD_IDS
|
|
|
|
|
|
IS_SHARD_PRIMARY: bool = _resolve_is_shard_primary()
|
|
|
|
# Aliases kept for backwards compatibility with files that import these names
|
|
BOT_OWNER_IDS = OWNER_IDS
|
|
BOT_OWNER_IDS_STR = OWNER_IDS_STR
|
|
STAFF_IDS = OWNER_IDS
|
|
STAFF_IDS_STR = OWNER_IDS_STR
|