0349c36880
- Changed interaction check on command executions in an attempt to prevent hybrid commands from counting twice on the docsite
255 lines
8.3 KiB
Python
255 lines
8.3 KiB
Python
import os, signal, asyncio, pathlib
|
||
import discord
|
||
from discord.ext import commands
|
||
from dotenv import load_dotenv
|
||
|
||
from data_manager import DataManager
|
||
from modules.common.settings import cfg as cfg_helper
|
||
from modules.common.boot_notice import post_boot_notice
|
||
|
||
# Version consists of:
|
||
# Major.Enhancement.Minor.Patch.Test (Test is alphanumeric; doesn’t trigger auto update)
|
||
VERSION = "0.4.1.0.a7"
|
||
|
||
# ---------- Env loading ----------
|
||
|
||
load_dotenv()
|
||
|
||
def _get_env(name: str, default: str = "") -> str:
|
||
v = os.getenv(name, "")
|
||
return (v or "").strip().strip('"').strip("'") or default
|
||
|
||
TOKEN = _get_env("DISCORD_TOKEN")
|
||
DATA_FILE = _get_env("SHAI_DATA") or _get_env("SHAI_DATA_FILE") or "/data/data.json"
|
||
|
||
print("[Config] DISCORD_TOKEN set:", bool(TOKEN))
|
||
print("[Config] DATA_FILE:", DATA_FILE)
|
||
|
||
# ---------- Discord intents ----------
|
||
|
||
intents = discord.Intents.default()
|
||
intents.guilds = True
|
||
intents.members = True
|
||
intents.message_content = True
|
||
intents.reactions = True
|
||
intents.emojis_and_stickers = True
|
||
intents.voice_states = True
|
||
|
||
# ---------- Bot + DataManager ----------
|
||
|
||
if not TOKEN:
|
||
print("[Config] WARNING: DISCORD_TOKEN is empty. The bot will fail to log in.")
|
||
|
||
bot = commands.Bot(command_prefix="!", intents=intents)
|
||
|
||
# Ensure data path exists and is seeded
|
||
os.makedirs(os.path.dirname(DATA_FILE) or ".", exist_ok=True)
|
||
if not os.path.exists(DATA_FILE):
|
||
with open(DATA_FILE, "w", encoding="utf-8") as f:
|
||
f.write("{}")
|
||
|
||
bot.data_manager = DataManager(DATA_FILE)
|
||
|
||
# ---------- Self-check: resolve from ENV first, then cfg_helper ----------
|
||
def _resolve_channel_id(c, key: str) -> int:
|
||
# 1) ENV always wins
|
||
env_key = f"SHAI_{key.upper()}"
|
||
raw = os.getenv(env_key, "").strip().strip('"').strip("'")
|
||
if raw.isdigit():
|
||
return int(raw)
|
||
|
||
# 2) Try cfg_helper (if it happens to know)
|
||
try:
|
||
v = int(c.int(key, 0))
|
||
if v:
|
||
return v
|
||
except Exception:
|
||
pass
|
||
|
||
# 3) Last resort: legacy bot.config shapes
|
||
try:
|
||
# bot.config like dict
|
||
v = int(getattr(c, "get", lambda *_: 0)(key, 0))
|
||
if v:
|
||
return v
|
||
except Exception:
|
||
pass
|
||
|
||
return 0
|
||
|
||
|
||
async def _guild_selfcheck(g: discord.Guild, c):
|
||
problems = []
|
||
|
||
def _need_channel(id_key, *perms):
|
||
cid = _resolve_channel_id(c, id_key)
|
||
if not cid:
|
||
problems.append(f"Missing config key: {id_key}")
|
||
return
|
||
|
||
ch = g.get_channel(cid)
|
||
if not ch:
|
||
problems.append(f"Channel not found: {id_key}={cid}")
|
||
return
|
||
|
||
me = g.me
|
||
p = ch.permissions_for(me)
|
||
for perm in perms:
|
||
if not getattr(p, perm, False):
|
||
problems.append(f"Missing permission on #{ch.name}: {perm}")
|
||
|
||
_need_channel('mod_channel_id', 'read_messages', 'send_messages', 'add_reactions', 'read_message_history')
|
||
_need_channel('modlog_channel_id', 'read_messages', 'send_messages')
|
||
_need_channel('pirates_list_channel_id', 'read_messages', 'send_messages')
|
||
|
||
if problems:
|
||
print(f"[SelfCheck:{g.name}]")
|
||
for p in problems:
|
||
print(" -", p)
|
||
|
||
# ---------- events ----------
|
||
|
||
@bot.event
|
||
async def on_ready():
|
||
import asyncio
|
||
|
||
print(f"Logged in as {bot.user} (ID: {bot.user.id})")
|
||
print("[Intents] members:", bot.intents.members,
|
||
"/ message_content:", bot.intents.message_content,
|
||
"/ voice_states:", bot.intents.voice_states)
|
||
|
||
env_cfg = cfg_helper(bot)
|
||
|
||
# Per-guild permission sanity checks (env-aware)
|
||
try:
|
||
await asyncio.gather(*[_guild_selfcheck(g, env_cfg) for g in bot.guilds])
|
||
except Exception as e:
|
||
print("[SelfCheck] failed:", repr(e))
|
||
|
||
# ---------- Slash command scope & sync ----------
|
||
#
|
||
# Toggle here (or set SHAI_SLASH_GUILD_ONLY=true/false):
|
||
guild_only = env_cfg.bool("slash_guild_only", True)
|
||
|
||
# Choose target guilds for "instant" registration
|
||
target_gids = set()
|
||
for key in ("home_guild_id", "dev_guild_id"):
|
||
val = env_cfg.get(key)
|
||
if val:
|
||
try:
|
||
target_gids.add(int(val))
|
||
except Exception:
|
||
pass
|
||
|
||
try:
|
||
if guild_only and target_gids:
|
||
print(f"[Slash] Mode: GUILD-ONLY to {sorted(target_gids)}")
|
||
|
||
# Copy all currently-loaded global commands to each target guild
|
||
for gid in sorted(target_gids):
|
||
g = bot.get_guild(gid)
|
||
if not g:
|
||
print(f"[Slash] Guild {gid}: not in cache; skipping copy/sync.")
|
||
continue
|
||
bot.tree.copy_global_to(guild=g)
|
||
g_cmds = await bot.tree.sync(guild=g)
|
||
names = ", ".join(f"/{c.name}" for c in g_cmds) if g_cmds else "(none)"
|
||
print(f"[Slash] Synced {len(g_cmds)} commands to {g.name} ({g.id}): {names}")
|
||
|
||
# Now remove global commands so only guild-scoped remain
|
||
bot.tree.clear_commands(guild=None)
|
||
cleared = await bot.tree.sync() # push empty global set (purges old global copies)
|
||
print(f"[Slash] Cleared global commands (now {len(cleared)}).")
|
||
|
||
else:
|
||
print("[Slash] Mode: GLOBAL")
|
||
# Purge any old per-guild copies in target guilds (to avoid dupes),
|
||
# then sync globally once.
|
||
for gid in sorted(target_gids):
|
||
g = bot.get_guild(gid)
|
||
if not g:
|
||
print(f"[Slash] Guild {gid}: not in cache; skip purge.")
|
||
continue
|
||
bot.tree.clear_commands(guild=g)
|
||
await bot.tree.sync(guild=g)
|
||
print(f"[Slash] Purged guild-specific commands in {g.name} ({g.id}).")
|
||
|
||
global_cmds = await bot.tree.sync()
|
||
names = ", ".join(f"/{c.name}" for c in global_cmds) if global_cmds else "(none)"
|
||
print(f"[Slash] Synced {len(global_cmds)} commands globally: {names}")
|
||
|
||
# --- Always print what actually exists after sync ---
|
||
def _fmt_cmds(cmds):
|
||
try:
|
||
return ", ".join(f"/{c.name}" for c in cmds) if cmds else "(none)"
|
||
except Exception:
|
||
return "(unreadable)"
|
||
|
||
# Global list
|
||
try:
|
||
global_cmds = await bot.tree.fetch_commands()
|
||
print(f"[Slash] Global commands ({len(global_cmds)}): {_fmt_cmds(global_cmds)}")
|
||
except Exception as e:
|
||
print("[Slash] Failed to fetch global commands:", repr(e))
|
||
|
||
# Guild lists
|
||
for gid in sorted(target_gids):
|
||
g = bot.get_guild(gid)
|
||
if not g:
|
||
print(f"[Slash] Guild {gid}: not in cache; cannot fetch commands.")
|
||
continue
|
||
try:
|
||
g_cmds = await bot.tree.fetch_commands(guild=g)
|
||
print(f"[Slash] {g.name} ({g.id}) guild commands ({len(g_cmds)}): {_fmt_cmds(g_cmds)}")
|
||
except Exception as e:
|
||
print(f"[Slash] Failed to fetch commands for guild {gid}:", repr(e))
|
||
|
||
except Exception as e:
|
||
print("[Slash] Sync failed:", repr(e))
|
||
|
||
# Post boot status message
|
||
try:
|
||
await post_boot_notice(bot)
|
||
except Exception as e:
|
||
print("[BootNotice] failed:", repr(e))
|
||
|
||
# ---------- Auto-discover extensions ----------
|
||
|
||
modules_path = pathlib.Path(__file__).parent / "modules"
|
||
extensions = []
|
||
for folder in modules_path.iterdir():
|
||
if not folder.is_dir():
|
||
continue
|
||
# skip non-cog helpers under modules/common
|
||
if folder.name == "common":
|
||
continue
|
||
for file in folder.glob("*.py"):
|
||
if file.name == "__init__.py":
|
||
continue
|
||
extensions.append(f"modules.{folder.name}.{file.stem}")
|
||
|
||
def _install_signal_handlers(loop, bot_obj):
|
||
def _graceful(*_):
|
||
loop.create_task(bot_obj.close())
|
||
for s in (signal.SIGTERM, signal.SIGINT):
|
||
try:
|
||
loop.add_signal_handler(s, _graceful)
|
||
except NotImplementedError:
|
||
pass # Windows
|
||
|
||
async def main():
|
||
print(f"[STARTUP] ShaiWatcher booting v{VERSION}")
|
||
async with bot:
|
||
for ext in extensions:
|
||
try:
|
||
await bot.load_extension(ext)
|
||
print(f"[Modules] Loaded: {ext}")
|
||
except Exception as e:
|
||
print(f"[Modules] Failed to load {ext}:", repr(e))
|
||
loop = asyncio.get_running_loop()
|
||
_install_signal_handlers(loop, bot)
|
||
await bot.start(TOKEN)
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|