phase-raid-bot/main.py at main · Phase-Project/phase-raid-bot · GitHub

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
import asyncio
import logging
import os
import discord
import sys as _sys
from discord.ext import commands

from core.config import TOKEN, WEBSITE
from core.utils.checks import global_interaction_check
from core.context import Context
from core.utils.db import init_db
from core.utils.leaderboard import load_leaderboard, track_command
from core.utils.logger import setup_logger

intents = discord.Intents.all()
bot = commands.Bot(command_prefix="p", intents=intents)
bot.context_cls = Context
logger = setup_logger()
bot.tree.interaction_check = global_interaction_check

bot.color = 0xff0000
bot.prefix = "p"


def get_global_total_commands() -> int:
    _, total_commands = load_leaderboard()
    return total_commands


async def update_bot_status():
    activity = discord.Activity(
        name=f"{WEBSITE} | {get_global_total_commands()} raids...",
        type=discord.ActivityType.streaming,
        url="https://twitch.tv/diddy"
    )
    await bot.change_presence(activity=activity)


@bot.event
async def on_interaction(interaction: discord.Interaction):
    if interaction.type != discord.InteractionType.application_command:
        return

    if not interaction.command:
        return

    await track_command(
        str(interaction.user.id),
        interaction.command.qualified_name,
        display_name=interaction.user.display_name,
        avatar_url=interaction.user.display_avatar.url,
    )


@bot.event
async def on_message(message):
    if message.author.bot:
        return
    await bot.process_commands(message)

@bot.event
async def on_command_error(ctx, error):
    import traceback
    traceback.print_exception(type(error), error, error.__traceback__)


def discover_cogs(commands_dir: str = "cogs") -> list[str]:
    cogs = []
    for filename in os.listdir(commands_dir):
        full_path = os.path.join(commands_dir, filename)
        if filename.endswith(".py") and not filename.startswith("__"):
            cogs.append(f"cogs.{filename[:-3]}")
        elif os.path.isdir(full_path) and filename != "__pycache__":
            if os.path.exists(os.path.join(full_path, "__init__.py")):
                cogs.append(f"cogs.{filename}")
    cogs.sort()
    return cogs


async def load_cog(cog: str):
    try:
        module = __import__(cog, fromlist=[""])
    except Exception as e:
        logger.error(f"Failed to import cog {cog}: {e}", exc_info=True)
        return

    if hasattr(module, "cog_setup"):
        try:
            module.cog_setup(bot)
            logger.info(f"new cog loaded: {cog}")
        except Exception as e:
            logger.error(f"Failed to load cog {cog}: {e}", exc_info=True)
        return

    try:
        await bot.load_extension(cog)
        logger.info(f"new cog loaded: {cog}")
    except Exception as e:
        logger.error(f"Failed to load cog {cog}: {e}", exc_info=True)


@bot.event
async def on_ready():
    await init_db()
    logger.info(f"i am {bot.user}")

    for cog in discover_cogs():
        await load_cog(cog)

    try:
        await bot.tree.sync()
    except Exception as e:
        logger.error(f"Error syncing tree: {e}")

    await update_bot_status()

    invite_url = f"https://discord.com/api/oauth2/authorize?client_id={bot.user.id}&permissions=8&scope=bot"
    logger.info(f"Bot invite: {invite_url}")


async def main():
    logger.info("Starting bot...")
    await bot.start(TOKEN)
    logger.info("Bot disconnected.")


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        exit