import discord
from discord.ext import commands
from dotenv import load_dotenv
from configparser import ConfigParser
from data_manager import DataManager
from modules.common.boot_notice import post_boot_notice
import pathlib
import os, signal, asyncio, xml.etree.ElementTree as ET
import aiohttp
# Version consists of the following:
# Major version
# Enhancement version
# Minor version
# Patch version
# Test/Dev version -> Does not trigger automatic update
VERSION="0.3.9.1.a2"
# ---------- Env & config loading ----------
load_dotenv()
TOKEN = os.getenv('DISCORD_TOKEN', '').strip()
CONFIG_PATH = os.getenv('SHAI_CONFIG', '/config/settings.conf')
config = ConfigParser()
read_files = config.read(CONFIG_PATH)
if not read_files:
print(f"[Config] INFO: no config at {CONFIG_PATH} (or unreadable). Will rely on env + defaults.")
# Ensure DEFAULT section exists
if 'DEFAULT' not in config:
config['DEFAULT'] = {}
def _overlay_env_into_config(cfg: ConfigParser):
"""
Overlay all SHAI_* environment variables into cfg['DEFAULT'] so env wins.
Also accept SHAI_DATA_FILE or SHAI_DATA for data_file.
"""
d = cfg['DEFAULT']
# Map SHAI_* -> lower-case keys (e.g. SHAI_MOD_CHANNEL_ID -> 'mod_channel_id')
for k, v in os.environ.items():
if not k.startswith('SHAI_'):
continue
key = k[5:].lower() # drop 'SHAI_' prefix
if key == 'data':
key = 'data_file'
d[key] = str(v)
if not d.get('data_file', '').strip():
d['data_file'] = '/data/data.json'
# Apply overlay so env takes precedence everywhere
_overlay_env_into_config(config)
# ---------- 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 ----------
data_file = config['DEFAULT']['data_file'] # guaranteed present by overlay
if not TOKEN:
print("[Config] WARNING: DISCORD_TOKEN not set (env). Bot will fail to log in.")
bot = commands.Bot(command_prefix='!', intents=intents)
bot.config = config
bot.data_manager = DataManager(data_file)
# ---------- Self-check helpers ----------
async def _guild_selfcheck(g: discord.Guild, cfg):
problems = []
def _need_channel(id_key, *perms):
raw = cfg.get(id_key)
if not raw:
problems.append(f"Missing config key: {id_key}")
return
try:
cid = int(raw)
except Exception:
problems.append(f"Bad channel id for {id_key}: {raw}")
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)
async def _fetch_latest_from_rss(url: str):
try:
timeout = aiohttp.ClientTimeout(total=8)
async with aiohttp.ClientSession(timeout=timeout) as sess:
async with sess.get(url) as resp:
if resp.status != 200:
return None, None
text = await resp.text()
# Gitea RSS structure: - …
root = ET.fromstring(text)
item = root.find('./channel/item')
if item is None:
return None, None
title = (item.findtext('title') or '').strip()
link = (item.findtext('link') or '').strip()
# Try to extract short sha from link tail if it's a commit URL
sha = None
if '/commit/' in link:
sha = link.rsplit('/commit/', 1)[-1][:7]
# Many Gitea feeds put the commit subject in
subject = title if title else None
return subject, sha
except Exception:
return None, None
# ---------- boot notice ----------
async def _maybe_post_boot_notice(bot):
status = os.getenv("SHAI_BOOT_STATUS", "")
if not status:
return
desc = os.getenv("SHAI_BOOT_DESC", "")
old_v = os.getenv("SHAI_BOOT_OLD", "")
new_v = os.getenv("SHAI_BOOT_NEW", "")
if status == "fetched_new":
line = f"Successfully fetched, cached, and booted new version: v{old_v or '0.0.0.0'} -> v{new_v}"
elif status == "cached_no_update":
line = f"Successfully booted from cached version: v{new_v}. No new update found"
else:
line = f"Successfully booted from cached version: v{new_v}. Program repository not accessible!"
ch_id = int(bot.config['DEFAULT'].get('modlog_channel_id', "0") or 0)
ch = None
for g in bot.guilds:
ch = g.get_channel(ch_id)
if ch: break
if ch:
try:
msg = line if not desc else f"{line}\n_{desc}_"
await ch.send(msg, allowed_mentions=discord.AllowedMentions.none())
except Exception:
pass
async def _post_boot_notice():
msg = f"Self-update and reboot successful! (v.{VERSION})"
ch_id_raw = bot.config['DEFAULT'].get('modlog_channel_id', '')
try:
ch_id = int(ch_id_raw) if ch_id_raw else 0
except Exception:
ch_id = 0
if not ch_id:
return
for g in bot.guilds:
ch = g.get_channel(ch_id)
if ch:
try:
await ch.send(msg)
except Exception:
pass
break
# ---------- events ----------
@bot.event
async def on_ready():
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)
await asyncio.gather(*[_guild_selfcheck(g, bot.config['DEFAULT']) for g in bot.guilds])
# Slash command sync
try:
dev_gid = bot.config['DEFAULT'].get('dev_guild_id')
if dev_gid:
guild = bot.get_guild(int(dev_gid))
if guild:
synced = await bot.tree.sync(guild=guild)
print(f"[Slash] Synced {len(synced)} commands to {guild.name}")
else:
synced = await bot.tree.sync()
print(f"[Slash] Synced {len(synced)} commands globally (dev_guild_id not in cache)")
else:
synced = await bot.tree.sync()
print(f"[Slash] Synced {len(synced)} commands globally")
except Exception as e:
print("[Slash] Sync failed:", repr(e))
# Post BSM if present
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 folder.is_dir():
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):
def _graceful(*_):
# ask discord.py to close cleanly
loop.create_task(bot.close())
for s in (signal.SIGTERM, signal.SIGINT):
try:
loop.add_signal_handler(s, _graceful)
except NotImplementedError:
pass # Windows
async def main():
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())