335 lines
16 KiB
Python
335 lines
16 KiB
Python
# bot_discord.py
|
|
import discord
|
|
from discord import app_commands
|
|
from discord.ext import commands
|
|
import importlib
|
|
import cmd_discord
|
|
|
|
import modules
|
|
import modules.utility
|
|
from modules.db import log_message, lookup_user, log_bot_event
|
|
|
|
class DiscordBot(commands.Bot):
|
|
def __init__(self, config, log_func):
|
|
super().__init__(command_prefix="!", intents=discord.Intents.all())
|
|
self.remove_command("help") # Remove built-in help function
|
|
self.config = config
|
|
self.log = log_func # Use the logging function from bots.py
|
|
self.db_conn = None # We'll set this later
|
|
self.help_data = None # We'll set this later
|
|
self.load_commands()
|
|
|
|
self.log("Discord bot initiated")
|
|
|
|
# async def sync_slash_commands(self):
|
|
# """Syncs slash commands for the bot."""
|
|
# await self.wait_until_ready()
|
|
# try:
|
|
# await self.tree.sync()
|
|
# primary_guild = discord.Object(id=int(self.config["discord_guilds"][0]))
|
|
# await self.tree.sync(guild=primary_guild)
|
|
# self.log("Discord slash commands synced.")
|
|
# except Exception as e:
|
|
# self.log(f"Unable to sync Discord slash commands: {e}", "ERROR")
|
|
|
|
|
|
def set_db_connection(self, db_conn):
|
|
"""
|
|
Store the DB connection in the bot so commands can use it.
|
|
"""
|
|
self.db_conn = db_conn
|
|
|
|
def load_commands(self):
|
|
"""
|
|
Load all commands from cmd_discord.py
|
|
"""
|
|
try:
|
|
importlib.reload(cmd_discord) # Reload the commands file
|
|
cmd_discord.setup(self) # Ensure commands are registered
|
|
self.log("Discord commands loaded successfully.")
|
|
|
|
# Load help info
|
|
help_json_path = "dictionary/help_discord.json"
|
|
modules.utility.initialize_help_data(
|
|
bot=self,
|
|
help_json_path=help_json_path,
|
|
is_discord=True,
|
|
log_func=self.log
|
|
)
|
|
except Exception as e:
|
|
self.log(f"Error loading Discord commands: {e}", "ERROR", True)
|
|
|
|
|
|
async def on_message(self, message):
|
|
self.log(f"Message detected, attempting UUI lookup on {message.author.name} ...", "DEBUG")
|
|
try:
|
|
# If it's a bot message, ignore or pass user_is_bot=True
|
|
is_bot = message.author.bot
|
|
user_id = str(message.author.id)
|
|
user_name = message.author.name # no discriminator
|
|
display_name = message.author.display_name
|
|
|
|
modules.utility.track_user_activity(
|
|
db_conn=self.db_conn,
|
|
log_func=self.log,
|
|
platform="discord",
|
|
user_id=user_id,
|
|
username=user_name,
|
|
display_name=display_name,
|
|
user_is_bot=is_bot
|
|
)
|
|
|
|
self.log(f"... UUI lookup complete", "DEBUG")
|
|
|
|
user_data = lookup_user(db_conn=self.db_conn, log_func=self.log, identifier=user_id, identifier_type="discord_user_id")
|
|
user_uuid = user_data["UUID"] if user_data else "UNKNOWN"
|
|
if user_uuid:
|
|
# The "platform" can be e.g. "discord" or you can store the server name
|
|
platform_str = f"discord-{message.guild.name}" if message.guild else "discord-DM"
|
|
# The channel name can be message.channel.name or "DM" if it's a private channel
|
|
channel_str = message.channel.name if hasattr(message.channel, "name") else "DM"
|
|
|
|
# If you have attachments, you could gather them as links.
|
|
try:
|
|
attachments = ", ".join(a.url for a in message.attachments) if message.attachments else ""
|
|
except Exception:
|
|
attachments = ""
|
|
|
|
log_message(
|
|
db_conn=self.db_conn,
|
|
log_func=self.log,
|
|
user_uuid=user_uuid,
|
|
message_content=message.content or "",
|
|
platform=platform_str,
|
|
channel=channel_str,
|
|
attachments=attachments
|
|
)
|
|
|
|
# PLACEHOLDER FOR FUTURE MESSAGE PROCESSING
|
|
except Exception as e:
|
|
self.log(f"... UUI lookup failed: {e}", "WARNING")
|
|
pass
|
|
|
|
try:
|
|
# Pass message contents to commands processing
|
|
await self.process_commands(message)
|
|
self.log(f"Command processing complete", "DEBUG")
|
|
except Exception as e:
|
|
self.log(f"Command processing failed: {e}", "ERROR")
|
|
|
|
|
|
async def on_command(self, ctx):
|
|
"""Logs every command execution at DEBUG level."""
|
|
_cmd_args = str(ctx.message.content).split(" ")[1:]
|
|
channel_name = "Direct Message" if "Direct Message with" in str(ctx.channel) else ctx.channel
|
|
self.log(f"Command '{ctx.command}' (Discord) initiated by {ctx.author} in #{channel_name}", "DEBUG")
|
|
if len(_cmd_args) > 1: self.log(f"!{ctx.command} arguments: {_cmd_args}", "DEBUG")
|
|
|
|
async def on_ready(self):
|
|
"""Runs when the bot successfully logs in."""
|
|
# Sync Slash Commands
|
|
try:
|
|
# Sync slash commands globally
|
|
#await self.tree.sync()
|
|
#self.log("Discord slash commands synced.")
|
|
primary_guild_int = int(self.config["discord_guilds"][0])
|
|
primary_guild = discord.Object(id=primary_guild_int)
|
|
await self.tree.sync(guild=primary_guild)
|
|
self.log(f"Discord slash commands force synced to guild: {primary_guild_int}")
|
|
except Exception as e:
|
|
self.log(f"Unable to sync Discord slash commands: {e}")
|
|
|
|
# Log successful bot startup
|
|
self.log(f"Discord bot is online as {self.user}")
|
|
log_bot_event(self.db_conn, self.log, "DISCORD_RECONNECTED", "Discord bot logged in.")
|
|
|
|
async def on_disconnect(self):
|
|
self.log("Discord bot has lost connection!", "WARNING")
|
|
log_bot_event(self.db_conn, self.log, "DISCORD_DISCONNECTED", "Discord bot lost connection.")
|
|
|
|
async def on_voice_state_update(self, member, before, after):
|
|
"""
|
|
Tracks user joins, leaves, mutes, deafens, streams, and voice channel moves.
|
|
"""
|
|
guild_id = str(member.guild.id)
|
|
discord_user_id = str(member.id)
|
|
voice_channel = after.channel.name if after.channel else before.channel.name if before.channel else None
|
|
|
|
# Ensure user exists in the UUI system
|
|
user_uuid = modules.db.lookup_user(self.db_conn, self.log, identifier=discord_user_id, identifier_type="discord_user_id", target_identifier="UUID")
|
|
|
|
if not user_uuid:
|
|
self.log(f"User {member.name} ({discord_user_id}) not found in 'users'. Attempting to add...", "WARNING")
|
|
modules.utility.track_user_activity(
|
|
db_conn=self.db_conn,
|
|
log_func=self.log,
|
|
platform="discord",
|
|
user_id=discord_user_id,
|
|
username=member.name,
|
|
display_name=member.display_name,
|
|
user_is_bot=member.bot
|
|
)
|
|
user_uuid= modules.db.lookup_user(self.db_conn, self.log, identifier=discord_user_id, identifier_type="discord_user_id", target_identifier="UUID")
|
|
if not user_uuid:
|
|
self.log(f"ERROR: Failed to associate {member.name} ({discord_user_id}) with a UUID. Skipping activity log.", "ERROR")
|
|
return # Prevent logging with invalid UUID
|
|
|
|
# Detect join and leave events
|
|
if before.channel is None and after.channel is not None:
|
|
modules.db.log_discord_activity(self.db_conn, self.log, guild_id, user_uuid, "JOIN", after.channel.name)
|
|
elif before.channel is not None and after.channel is None:
|
|
modules.db.log_discord_activity(self.db_conn, self.log, guild_id, user_uuid, "LEAVE", before.channel.name)
|
|
|
|
# Detect VC moves (self/moved)
|
|
if before.channel and after.channel and before.channel != after.channel:
|
|
move_detail = f"{before.channel.name} -> {after.channel.name}"
|
|
modules.db.log_discord_activity(self.db_conn, self.log, guild_id, user_uuid, "VC_MOVE", after.channel.name, move_detail)
|
|
|
|
# Detect mute/unmute
|
|
if before.self_mute != after.self_mute:
|
|
mute_action = "MUTE" if after.self_mute else "UNMUTE"
|
|
modules.db.log_discord_activity(self.db_conn, self.log, guild_id, user_uuid, mute_action, voice_channel)
|
|
|
|
# Detect deafen/undeafen
|
|
if before.self_deaf != after.self_deaf:
|
|
deaf_action = "DEAFEN" if after.self_deaf else "UNDEAFEN"
|
|
modules.db.log_discord_activity(self.db_conn, self.log, guild_id, user_uuid, deaf_action, voice_channel)
|
|
|
|
# Detect streaming
|
|
if before.self_stream != after.self_stream:
|
|
stream_action = "STREAM_START" if after.self_stream else "STREAM_STOP"
|
|
modules.db.log_discord_activity(self.db_conn, self.log, guild_id, user_uuid, stream_action, voice_channel)
|
|
|
|
# Detect camera usage
|
|
if before.self_video != after.self_video:
|
|
camera_action = "CAMERA_ON" if after.self_video else "CAMERA_OFF"
|
|
modules.db.log_discord_activity(self.db_conn, self.log, guild_id, user_uuid, camera_action, voice_channel)
|
|
|
|
|
|
async def on_presence_update(self, before, after):
|
|
"""
|
|
Detects when a user starts or stops a game, Spotify, or Discord activity.
|
|
Ensures the activity is logged using the correct UUID from the UUI system.
|
|
"""
|
|
if not after.guild: # Ensure it's in a guild (server)
|
|
return
|
|
|
|
guild_id = str(after.guild.id)
|
|
discord_user_id = str(after.id)
|
|
|
|
# Ensure user exists in the UUI system
|
|
user_uuid = modules.db.lookup_user(
|
|
self.db_conn,
|
|
self.log,
|
|
identifier=discord_user_id,
|
|
identifier_type="discord_user_id",
|
|
target_identifier="UUID"
|
|
)
|
|
|
|
if not user_uuid:
|
|
self.log(f"User {after.name} ({discord_user_id}) not found in 'users'. Attempting to add...", "WARNING")
|
|
modules.utility.track_user_activity(
|
|
db_conn=self.db_conn,
|
|
log_func=self.log,
|
|
platform="discord",
|
|
user_id=discord_user_id,
|
|
username=after.name,
|
|
display_name=after.display_name,
|
|
user_is_bot=after.bot
|
|
)
|
|
user_uuid = modules.db.lookup_user(self.db_conn, self.log, identifier=discord_user_id, identifier_type="discord_user_id", target_identifier="UUID")
|
|
if not user_uuid:
|
|
self.log(f"ERROR: Failed to associate {after.name} ({discord_user_id}) with a UUID. Skipping activity log.", "ERROR")
|
|
return
|
|
|
|
# Check all activities
|
|
new_activity = None
|
|
for n_activity in after.activities:
|
|
if isinstance(n_activity, discord.Game):
|
|
new_activity = ("GAME_START", n_activity.name)
|
|
elif isinstance(n_activity, discord.Spotify):
|
|
# Get artist name(s) and format as "{artist_name} - {song_title}"
|
|
artist_name = ", ".join(n_activity.artists)
|
|
song_name = n_activity.title
|
|
spotify_detail = f"{artist_name} - {song_name}"
|
|
new_activity = ("LISTENING_SPOTIFY", spotify_detail)
|
|
elif isinstance(n_activity, discord.Streaming):
|
|
new_activity = ("STREAM_START", n_activity.game or "Sharing screen")
|
|
|
|
# Check all activities
|
|
old_activity = None
|
|
for o_activity in before.activities:
|
|
if isinstance(o_activity, discord.Game):
|
|
old_activity = ("GAME_STOP", o_activity.name)
|
|
# IGNORE OLD SPOTIFY EVENTS
|
|
elif isinstance(o_activity, discord.Streaming):
|
|
old_activity = ("STREAM_STOP", o_activity.game or "Sharing screen")
|
|
|
|
if new_activity:
|
|
modules.db.log_discord_activity(self.db_conn, self.log, guild_id, user_uuid, new_activity[0], None, new_activity[1])
|
|
if old_activity:
|
|
modules.db.log_discord_activity(self.db_conn, self.log, guild_id, user_uuid, old_activity[0], None, old_activity[1])
|
|
|
|
# async def start_account_linking(self, interaction: discord.Interaction):
|
|
# """Starts the linking process by generating a link code and displaying instructions."""
|
|
# user_id = str(interaction.user.id)
|
|
|
|
# # Check if the user already has a linked account
|
|
# user_data = modules.db.lookup_user(self.db_conn, self.log, identifier=user_id, identifier_type="discord_user_id")
|
|
# if user_data and user_data["twitch_user_id"]:
|
|
# link_date = user_data["datetime_linked"]
|
|
# await interaction.response.send_message(
|
|
# f"Your Discord account is already linked to Twitch user **{user_data['twitch_user_display_name']}** "
|
|
# f"(linked on {link_date}). You must remove the link before linking another account.", ephemeral=True)
|
|
# return
|
|
|
|
# # Generate a unique link code
|
|
# link_code = modules.utility.generate_link_code()
|
|
# modules.db.run_db_operation(
|
|
# self.db_conn, "write",
|
|
# "INSERT INTO link_codes (DISCORD_USER_ID, LINK_CODE) VALUES (?, ?)",
|
|
# (user_id, link_code), self.log
|
|
# )
|
|
|
|
# # Show the user the link modal
|
|
# await interaction.response.send_message(
|
|
# f"To link your Twitch account, post the following message in Twitch chat:\n"
|
|
# f"`!acc_link {link_code}`\n\n"
|
|
# f"Then, return here and click 'Done'.", ephemeral=True
|
|
# )
|
|
|
|
# async def finalize_account_linking(self, interaction: discord.Interaction):
|
|
# """Finalizes the linking process by merging duplicate UUIDs."""
|
|
# from modules import db
|
|
# user_id = str(interaction.user.id)
|
|
|
|
# # Fetch the updated user info
|
|
# user_data = modules.db.lookup_user(self.db_conn, self.log, identifier=user_id, identifier_type="discord_user_id")
|
|
# if not user_data or not user_data["twitch_user_id"]:
|
|
# await interaction.response.send_message(
|
|
# "No linked Twitch account found. Please complete the linking process first.", ephemeral=True)
|
|
# return
|
|
|
|
# discord_uuid = user_data["UUID"]
|
|
# twitch_uuid = modules.db.lookup_user(self.db_conn, self.log, identifier=user_data["twitch_user_id"], identifier_type="twitch_user_id")["UUID"]
|
|
|
|
# if discord_uuid == twitch_uuid:
|
|
# await interaction.response.send_message("Your accounts are already fully linked.", ephemeral=True)
|
|
# return
|
|
|
|
# # Merge all records from `twitch_uuid` into `discord_uuid`
|
|
# db.merge_uuid_data(self.db_conn, self.log, old_uuid=twitch_uuid, new_uuid=discord_uuid)
|
|
|
|
# # Delete the old Twitch UUID entry
|
|
# db.run_db_operation(self.db_conn, "write", "DELETE FROM users WHERE UUID = ?", (twitch_uuid,), self.log)
|
|
|
|
# # Confirm the final linking
|
|
# await interaction.response.send_message("Your Twitch and Discord accounts are now fully linked.", ephemeral=True)
|
|
|
|
|
|
async def run(self, token):
|
|
try:
|
|
await super().start(token)
|
|
except Exception as e:
|
|
self.log(f"Discord bot error: {e}", "CRITICAL")
|