From ef1e44780e649adeeb9ba119c43f6197931afe20 Mon Sep 17 00:00:00 2001 From: idk-maybe-none Date: Fri, 31 Oct 2025 17:08:58 +0200 Subject: [PATCH 1/3] Separate into files & make counting more smart Separate app.py into many files, and make that counting now accepts mathematical expressions --- .gitignore | 1 + README.md | 22 +- app.py | 1264 ---------------------------------------- bot.py | 28 + cogs/admin.py | 205 +++++++ cogs/counting.py | 72 +++ cogs/devices.py | 133 +++++ cogs/events.py | 99 ++++ cogs/maintenance.py | 75 +++ cogs/moderation.py | 368 ++++++++++++ cogs/utility.py | 79 +++ config.py | 59 ++ config_real.py | 59 ++ main.py | 11 + requirements.txt | 4 + utils/__init__.py | 0 utils/file_handlers.py | 21 + utils/helpers.py | 44 ++ utils/logger.py | 32 + 19 files changed, 1304 insertions(+), 1272 deletions(-) delete mode 100644 app.py create mode 100644 bot.py create mode 100644 cogs/admin.py create mode 100644 cogs/counting.py create mode 100644 cogs/devices.py create mode 100644 cogs/events.py create mode 100644 cogs/maintenance.py create mode 100644 cogs/moderation.py create mode 100644 cogs/utility.py create mode 100644 config.py create mode 100644 config_real.py create mode 100644 main.py create mode 100644 requirements.txt create mode 100644 utils/__init__.py create mode 100644 utils/file_handlers.py create mode 100644 utils/helpers.py create mode 100644 utils/logger.py diff --git a/.gitignore b/.gitignore index 9fd3707..909d721 100644 --- a/.gitignore +++ b/.gitignore @@ -1 +1,2 @@ /token +__pycache__/ diff --git a/README.md b/README.md index e2127ca..77b12d8 100644 --- a/README.md +++ b/README.md @@ -1,39 +1,45 @@ # FreeXR-Bot FreeXR Discord Bot -v2 changelog: ## Changelog -### General +### v2 changelog: + +#### General - **All commands are now hybrid commands** (`@bot.hybrid_command()`), supporting both prefix and slash usage. Thanks @Anonymous941 ! - You can now run the bot with `-t TOKEN` to specify the Discord bot token. - **Improved error handling**: - Unauthorized command usage now pings the user and says they are not authorized. - Missing required arguments are now reported to the user. - **Fixed`.replies_cmd` command** (replaced with a single `replies` hybrid command) +- **Counting now support arithmetics**, (and SPOILER: it's not just eval()) + - Bot now uses arithmeval, python library that gives bot ability to do basic math ( and not only ) + - Now you can use Addition (+), True division (/), Floor division (//), Modulus (%), Multiplication (*), Exponentiation (**), Subtraction (-), Logical AND (and), Logical OR (or), Logical NOT (not), and even pi (Ļ€), e, tau (Ļ„), the golden ratio (Ļ†), Euler-Mascheroni's gamma (gamma), the speed of light (c), Newton's gravitational constant (G), and Planck's constant (h). (IDK why you would use them) -### Device Management +#### Device Management - **New persistent device system**: - `devices` lists your devices or another user's devices. - `deviceadd` adds a device using a Discord modal form (slash command only). - `deviceremove ` removes a device by its per-user index. - `deviceinfo ` shows info for a specific device. - - Devices are stored in devices.json + - Devices are stored in data/devices.json Don't know what this would be used for... :) -### Regex Management +#### Regex Management - **Regex block command now uses a modal** (slash command only) for adding new regex patterns. - **Regex list, toggle, and unblock commands now use 1-based indices** for user-facing commands (internally adjusted to 0-based). This was one of the earliest bugs present when we first implemented the regex system. - -### Replies System +#### Replies System - **Removed explicit `replies` handling from `on_message`**, now handled by the hybrid command. -### Miscellaneous +#### Miscellaneous - **Update and hotupdate commands**: - `update` pulls the whole repo and restarts the bot. - `hotupdate` pulls the repo and reloads replies without restarting, replacing the old `updatereplies` command. +- **Separate files**: + - One-file bot code was split into multiple files. + - Adding new features and reviewing them is now easier. - **General code cleanup and refactoring** for clarity and maintainability. Thanks to all of FreeXR supporting, you 667 members all rock <3 diff --git a/app.py b/app.py deleted file mode 100644 index 82f9671..0000000 --- a/app.py +++ /dev/null @@ -1,1264 +0,0 @@ -# FreeXR Bot -# Made with love by ilovecats4606 <3 -BOTVERSION = "2.1.5" -DISABLED_IN_BETA = {"slowmode", "q", "uq"} -import discord -from discord.ext import commands -import asyncio -import json -from pathlib import Path -import re -import datetime -import os -import requests -import time -import platform -import sys -import tasklib -from discord.ext import tasks, commands -from discord.ext import commands -from datetime import datetime, timedelta -import git -from git import Repo -import argparse -parser = argparse.ArgumentParser(description="FreeXR Bot") -parser.add_argument("-t", "--token", type=str, help="Discord bot token") -args = parser.parse_args() - -intents = discord.Intents.default() -intents.messages = True -intents.message_content = True -intents.guilds = True -intents.dm_messages = True -intents.members = True -start_time = time.time() - -class DiscordConsoleLogger: - def __init__(self, bot, channel_id): - self.bot = bot - self.channel_id = channel_id - self.buffer = "" - self.lock = asyncio.Lock() - - def write(self, message): - self.buffer += message - if '\n' in self.buffer: - lines = self.buffer.split('\n') - self.buffer = lines[-1] - for line in lines[:-1]: - if line.strip(): - asyncio.ensure_future(self.send_to_discord(line.strip())) - - def flush(self): - if self.buffer.strip(): - asyncio.ensure_future(self.send_to_discord(self.buffer.strip())) - self.buffer = "" - - async def send_to_discord(self, message): - async with self.lock: - channel = self.bot.get_channel(self.channel_id) - if channel: - try: - await channel.send(f"```\n{message[:1900].replace('```', '')}\n```") - except Exception: - pass - -def get_uptime(): - seconds = int(time.time() - start_time) - days, seconds = divmod(seconds, 86400) - hours, seconds = divmod(seconds, 3600) - minutes, seconds = divmod(seconds, 60) - return f"{days}d {hours}h {minutes}m {seconds}s" - - -bot = commands.Bot(command_prefix=".", intents=intents) - -REPORT_LOG_CHANNEL_ID = 1361285583195869265 -ADMIN_ROLE_ID = 1376159693021646900 -QUARANTINE_ROLE_ID = 1373608273306976276 - -if args.token: - TOKEN = args.token -else: - with open("token", "r") as file: - TOKEN = file.read().strip() - -# In-memory report buffer per user -active_reports = {} -regex_filters = [] - -COUNT_FILE = "count_data.json" - -IS_BETA = "b" in BOTVERSION.lower() - -def load_count_data(): - if not os.path.exists(COUNT_FILE): - return {"current_count": 0, "last_counter_id": None} - with open(COUNT_FILE, "r") as f: - return json.load(f) - - -def save_count_data(data): - with open(COUNT_FILE, "w") as f: - json.dump(data, f) - - -# JSON file path -REPORTS_FILE = Path("reports.json") - - -# Load/save report mapping -def load_reports(): - if REPORTS_FILE.exists(): - with open(REPORTS_FILE, "r") as f: - return json.load(f) - return {} - - -def save_reports(data): - with open(REPORTS_FILE, "w") as f: - json.dump(data, f, indent=4) - - -report_log_map = load_reports() - -FILTER_FILE = "filters.json" - -REPO_URL = "https://github.com/FreeXR/FreeXR-Bot.git" -REPO_DIR = "FreeXR-Bot" -REPLIES_DIR = os.path.join(REPO_DIR, "quick_replies") - -DEVICES_FILE = "devices.json" - - -def load_replies(): - replies = {} - if not os.path.exists(REPLIES_DIR): - return replies - for filename in os.listdir(REPLIES_DIR): - if filename.endswith(".md"): - filepath = os.path.join(REPLIES_DIR, filename) - with open(filepath, "r", encoding="utf-8") as f: - content = f.read().split("---") - if len(content) >= 2: - summary_line = content[0].strip().splitlines()[0] - reply_text = content[1].strip() - command_name = filename[:-3] # remove .md - replies[command_name] = (summary_line, reply_text) - return replies - - -# Initial load -replies = load_replies() - - -# Load regex filters from file -def load_filters(): - if os.path.exists(FILTER_FILE): - with open(FILTER_FILE, "r") as f: - return json.load(f) - return [] - - -# Save regex filters to file -def save_filters(): - with open(FILTER_FILE, "w") as f: - json.dump(regex_filters, f, indent=2) - - -regex_filters = load_filters() - -def load_devices(): - if os.path.exists(DEVICES_FILE): - with open(DEVICES_FILE, "r", encoding="utf-8") as f: - return json.load(f) - return {} - -def save_devices(devices): - with open(DEVICES_FILE, "w", encoding="utf-8") as f: - json.dump(devices, f, indent=2) - -devices_data = load_devices() - -BACKUP_FILE = "message_backups.json" -if os.path.exists(BACKUP_FILE): - with open(BACKUP_FILE, "r") as f: - message_backups = json.load(f) -else: - message_backups = {} - - -def save_backups(): - with open(BACKUP_FILE, "w") as f: - json.dump(message_backups, f, indent=2) - - -@bot.event -async def on_ready(): - await bot.tree.sync() - sys.stdout = DiscordConsoleLogger(bot, 1376528272204103721) - sys.stderr = sys.stdout - print(f"Logged in as {bot.user}") - os_info = platform.system() - release = platform.release() - architecture = platform.machine() - python_version = platform.python_version() - uptime = get_uptime() - - env_message = ( - f"āœ… Bot is running in **{os_info} {release} ({architecture})** environment " - f"with **Python {python_version}**\n" - f"šŸ›  Version: **{BOTVERSION}**\n" - f"ā± Load time: **{uptime}**" - ) - if "b" in BOTVERSION.lower(): - env_message += "\nāš ļø Beta version detected ā€“ may be unstable! Potentially destructive commands have been disabled." - channel = bot.get_channel(1376528272204103721) - await channel.send(env_message) - - print(env_message) - load_quarantine_data() - # On startup, verify all quarantined users still have role, otherwise cleanup - guild = bot.guilds[0] - quarantine_role = guild.get_role(QUARANTINE_ROLE_ID) - to_remove = [] - for user_id_str, unq_time_str in active_quarantines.items(): - user_id = int(user_id_str) - member = guild.get_member(user_id) - unq_time = datetime.fromisoformat(unq_time_str) - if member is None: - # Member not found in guild, remove from active - to_remove.append(user_id_str) - continue - if quarantine_role not in member.roles: - # Role missing, remove from active (maybe manually removed) - to_remove.append(user_id_str) - for user_id_str in to_remove: - active_quarantines.pop(user_id_str) - save_quarantine_data() - - check_quarantine_expiry.start() - -@bot.event -async def on_member_join(member): - welcome_channel_id = 1348562119469305958 - - try: - await member.add_roles(member.guild.get_role(1406195112714829899)) - except discord.Forbidden: - print(f"Couldn't add role to {member} (permissions issue)") - - try: - await member.send( - "# šŸ‘‹ Welcome to the server!\n Hello and welcome to FreeXR. We hack headsets to root them and unlock their bootloaders, and we appreciate you for joining. To get started, please read the https://discord.com/channels/1344235945238593547/1364918149404688454.\nWe hope you have a great stay here, and thank you for joining." - ) - except discord.Forbidden: - channel = member.guild.get_channel(welcome_channel_id) - if channel: - msg = await channel.send( - f"{member.mention} šŸ‘‹ Welcome to the server! Please read the https://discord.com/channels/1344235945238593547/1364918149404688454, and we hope you have a great stay here!.\n-# Psst! Your DMs are closed, so I couldn't send you a DM." - ) - - -RAW_URL = "https://raw.githubusercontent.com/FreeXR/FreeXR-Bot/refs/heads/main/app.py" -LOCAL_PATH = "/home/freexr/app.py" - - -@bot.hybrid_command() -@commands.has_role(ADMIN_ROLE_ID) -async def update(ctx): - """ - Updates the bot by pulling the latest code from the repository and restarting. - """ - await ctx.send("šŸ“„ Downloading latest version of `app.py`...") - - try: - await ctx.send("šŸ“„ Pulling from repository...") - if os.path.exists(REPO_DIR): - repo = Repo(REPO_DIR) - repo.remotes.origin.pull() - - else: - Repo.clone_from(REPO_URL, REPO_DIR) - - await ctx.send("āœ… Update complete. Restarting bot...") - python = sys.executable - os.execv(python, [python] + sys.argv) - - except Exception as e: - await ctx.send(f"āŒ Update failed:\n```{e}```") - - -@bot.hybrid_command() -async def role(ctx, role_id: int, user_id: int): - """ - Toggles a role for a user. - Only the user with ID 981463678698266664 is authorized to use this command. - """ - allowed_user_id = 981463678698266664 - - if ctx.author.id != allowed_user_id: - await ctx.send("āŒ You are not authorized to use this command.") - return - - guild = ctx.guild - member = guild.get_member(user_id) - role = guild.get_role(role_id) - - if not member: - await ctx.send("āŒ User not found.") - return - if not role: - await ctx.send("āŒ Role not found.") - return - - try: - if role in member.roles: - await member.remove_roles(role) - await ctx.send(f"āœ… Removed role **{role.name}** from {member.mention}.") - else: - await member.add_roles(role) - await ctx.send(f"āœ… Added role **{role.name}** to {member.mention}.") - except discord.Forbidden: - await ctx.send("āŒ I don't have permission to manage that role.") - except discord.HTTPException as e: - await ctx.send(f"āŒ Failed to modify role: {e}") - - -@bot.hybrid_command() -async def pin(ctx): - """ - Pins a message that you reply to. - Anyone can use this command. - """ - if not ctx.message.reference: - await ctx.send("Please reply to the message you want to pin.") - return - - try: - msg = await ctx.channel.fetch_message(ctx.message.reference.message_id) - await msg.pin() - await ctx.send("šŸ“Œ Message pinned.") - except discord.Forbidden: - await ctx.send("I don't have permission to pin messages in this channel.") - except discord.HTTPException as e: - await ctx.send(f"Failed to pin message: {e}") - - -@bot.hybrid_command() -async def unpin(ctx): - """ - Unpins a message that you reply to. - Anyone can use this command. - """ - if not ctx.message.reference: - await ctx.send("Please reply to the message you want to unpin.") - return - - try: - msg = await ctx.channel.fetch_message(ctx.message.reference.message_id) - await msg.unpin() - await ctx.send("šŸ“ Message unpinned.") - except discord.Forbidden: - await ctx.send("I don't have permission to unpin messages in this channel.") - except discord.HTTPException as e: - await ctx.send(f"Failed to unpin message: {e}") - - -@bot.hybrid_command() -async def report(ctx): - """ - Starts a report in DMs. - Only works in DMs. - """ - if not isinstance(ctx.channel, discord.DMChannel): - await ctx.send("Please DM me this command.") - return - - await ctx.send( - "You're reporting to the server admins. All messages from this point will be recorded.\n" - "Please state your issue. Upload images as links (attachments won't work).\n" - "When you're finished, type `.iamdone`. Messages will stop being recorded." - ) - active_reports[ctx.author.id] = [] - - -@bot.tree.context_menu(name="Add to report") -async def add_to_report(interaction: discord.Interaction, message: discord.Message): - user_id = interaction.user.id - - # Not in report - if user_id not in active_reports: - await interaction.response.send_message( - "āŒ You don't have an active report. Please start one by DMing me `.report`.", - ephemeral=True, - ) - return - - # Prepare the message info - content = f"**Message from {message.author} in <#{message.channel.id}>:**\n{message.content}" - active_reports[user_id].append(content) - - # Backup - backup_entry = { - "author": str(message.author), - "channel_id": message.channel.id, - "message_id": message.id, - "content": message.content, - "timestamp": str(message.created_at), - "jump_url": message.jump_url, - } - if str(user_id) not in message_backups: - message_backups[str(user_id)] = [] - message_backups[str(user_id)].append(backup_entry) - save_backups() - - await interaction.response.send_message( - "āœ… Message added to your report.", ephemeral=True - ) - - -@bot.hybrid_command() -async def iamdone(ctx): - """ - Ends the report and sends it to the admins. - Only works in DMs. - """ - if not isinstance(ctx.channel, discord.DMChannel): - return - - user_id = ctx.author.id - if user_id not in active_reports or not active_reports[user_id]: - await ctx.send( - "No messages recorded or you haven't started a report with `.report`." - ) - return - - channel = bot.get_channel(REPORT_LOG_CHANNEL_ID) - report_content = "\n".join(active_reports[user_id]) - extra = "" - if str(user_id) in message_backups: - extra += "\n\n**Attached messages:**\n" - for entry in message_backups[str(user_id)]: - extra += ( - f"[Jump to message]({entry['jump_url']}) | " - f"**{entry['author']}** ({entry['timestamp']}):\n" - f"{entry['content']}\n\n" - ) - - full_content = report_content + extra - - embed = discord.Embed( - title="New Report", - description=full_content[:4000], - color=discord.Color.orange(), - ) - embed.set_author(name=f"{ctx.author}", icon_url=ctx.author.display_avatar.url) - - report_message = await channel.send(embed=embed) - - report_log_map[str(report_message.id)] = user_id - save_reports(report_log_map) - - await ctx.send("Thank you! Your report has been sent.") - - # Clear user's report data - active_reports[user_id] = [] - if str(user_id) in message_backups: - del message_backups[str(user_id)] - save_backups() - - -@bot.hybrid_command() -@commands.has_role(ADMIN_ROLE_ID) -async def resolve(ctx, msg_id: int = None): - """ - Marks a report as resolved. - Only admins can use this command. - Reply to a report message or provide its ID. - """ - # Try to get message ID from reply if not given - if not msg_id and ctx.message.reference: - msg_id = ctx.message.reference.message_id - - if not msg_id: - await ctx.send("Please reply to a report or provide a message ID.") - return - - report_id = str(msg_id) - - if report_id in report_log_map: - del report_log_map[report_id] - save_reports(report_log_map) - - try: - msg = await ctx.channel.fetch_message(msg_id) - await msg.reply("āœ… Marked as resolved. Further interaction closed.") - except discord.NotFound: - await ctx.send( - "Marked as resolved, but couldn't find the original message." - ) - - else: - await ctx.send("That message isn't tracked as an active report.") - - -@bot.hybrid_command() -@commands.has_role(ADMIN_ROLE_ID) -async def createchannel(ctx, msg_id: int = None): - """ - Creates a private channel for a report. - Only admins can use this command. - Reply to a report message or provide its ID. - """ - # Fallback to reply if no ID provided - if not msg_id and ctx.message.reference: - msg_id = ctx.message.reference.message_id - - embed = None - if msg_id: - try: - log_channel = bot.get_channel(REPORT_LOG_CHANNEL_ID) - report_msg = await log_channel.fetch_message(msg_id) - if report_msg.embeds: - embed = report_msg.embeds[0] - except discord.NotFound: - await ctx.send("Couldn't find the report message.") - return - except Exception as e: - await ctx.send(f"Error fetching report: {e}") - return - - await ctx.send("What should the channel be called?") - - def check(m): - return m.author == ctx.author and m.channel == ctx.channel - - try: - name_msg = await bot.wait_for("message", check=check, timeout=60) - guild = ctx.guild - overwrites = { - guild.default_role: discord.PermissionOverwrite(read_messages=False), - guild.get_role(ADMIN_ROLE_ID): discord.PermissionOverwrite( - read_messages=True - ), - } - channel = await guild.create_text_channel( - name=name_msg.content, overwrites=overwrites - ) - await ctx.send(f"Created channel: {channel.mention}") - - if embed: - await channel.send("Report linked to this channel:", embed=embed) - - except asyncio.TimeoutError: - await ctx.send("Timed out.") - except Exception as e: - await ctx.send(f"Failed to create channel: {e}") - - -@bot.hybrid_command() -@commands.has_role(ADMIN_ROLE_ID) -async def createchannelp(ctx, msg_id: int = None): - """ - Creates a private channel for a report with the original reporter. - Only admins can use this command. - Reply to a report message or provide its ID. - """ - if not msg_id and ctx.message.reference: - msg_id = ctx.message.reference.message_id - - if not msg_id: - await ctx.send("Please reply to a report or provide a message ID.") - return - - report_id = str(msg_id) - - if report_id not in report_log_map: - await ctx.send("Couldn't find the original reporter for that ID.") - return - - user_id = report_log_map[report_id] - guild = ctx.guild - member = guild.get_member(user_id) - - if not member: - await ctx.send("The original reporter is not in this server.") - return - - await ctx.send("What should the channel be called?") - - def check(m): - return m.author == ctx.author and m.channel == ctx.channel - - try: - name_msg = await bot.wait_for("message", check=check, timeout=60) - overwrites = { - guild.default_role: discord.PermissionOverwrite(read_messages=False), - guild.get_role(ADMIN_ROLE_ID): discord.PermissionOverwrite( - read_messages=True - ), - member: discord.PermissionOverwrite(read_messages=True), - } - channel = await guild.create_text_channel( - name=name_msg.content, overwrites=overwrites - ) - await ctx.send( - f"Created private channel: {channel.mention} with access to {member.mention} given." - ) - except asyncio.TimeoutError: - await ctx.send("Timed out.") - except Exception as e: - await ctx.send(f"Failed to create channel: {e}") - - -@bot.hybrid_command() -@commands.has_role(ADMIN_ROLE_ID) -async def listreport(ctx): - """ - Lists all active reports. - Only admins can use this command. - """ - if not report_log_map: - await ctx.send("No reports found.") - return - - log_channel = bot.get_channel(REPORT_LOG_CHANNEL_ID) - report_lines = [] - - for i, (msg_id, user_id) in enumerate(report_log_map.items()): - try: - msg = await log_channel.fetch_message(int(msg_id)) - title = msg.embeds[0].title if msg.embeds else "No Title" - report_lines.append(f"{i+1}. {title} (ID: {msg.id})") - except discord.NotFound: - report_lines.append(f"{i+1}. [Message not found] (ID: {msg_id})") - - if report_lines: - await ctx.send("**Active Reports:**\n" + "\n".join(report_lines)) - else: - await ctx.send("No valid report messages found.") - - -def is_admin(member): - return any(role.id == ADMIN_ROLE_ID for role in member.roles) - - -@bot.hybrid_command() -@commands.has_role(ADMIN_ROLE_ID) -async def block(ctx): - """ - Blocks a regex pattern from being sent in the server. - Only admins can use this command. - """ - # Only allow modal for slash commands - if ctx.interaction: - class RegexModal(discord.ui.Modal, title="Block Regex Pattern"): - pattern = discord.ui.TextInput(label="Regex Pattern", required=True) - - async def on_submit(self, interaction: discord.Interaction): - try: - re.compile(self.pattern.value) # Validate regex - regex_filters.append({"pattern": self.pattern.value, "enabled": True}) - save_filters() - await interaction.response.send_message(f"Blocked regex added: `{self.pattern.value}`", ephemeral=True) - except re.error: - await interaction.response.send_message("Invalid regex pattern.", ephemeral=True) - - await ctx.interaction.response.send_modal(RegexModal()) - else: - await ctx.send("Please use the `/block` slash command to add a regex pattern.") - - - -@bot.hybrid_command() -@commands.has_role(ADMIN_ROLE_ID) -async def listregex(ctx): - """ - Lists all blocked regex patterns. - Only admins can use this command. - """ - if not regex_filters: - return await ctx.send("No regex patterns are currently blocked.") - - message = "Blocked Regex Patterns:\n" - for i, entry in enumerate(regex_filters): - message += f"{i}. `{entry['pattern']}` - {'āœ… Enabled' if entry['enabled'] else 'āŒ Disabled'}\n" - message += "\nUse `.toggle ` to enable/disable a regex." - await ctx.send(message) - - -def get_user_id(ctx, user_str): - if user_str.isdigit(): - return int(user_str) - if user_str.startswith("<@") and user_str.endswith(">"): - return int(user_str.strip("<@!>")) - member = ctx.guild.get_member_named(user_str) - if member: - return member.id - return None - -@bot.hybrid_command(name="devices") -async def devices_cmd(ctx, user: discord.User = None): - """ - Lists your devices or another user's devices. - Usage: .devices [@user] - """ - user_id = user.id if user else ctx.author.id - user_devices = devices_data.get(str(user_id), []) - if not user_devices: - await ctx.send("No devices found for this user.") - return - msg = f"Devices for <@{user_id}>:\n" - for idx, device in enumerate(user_devices, 1): - msg += f"**{idx}.** {device['Name']} (`{device['Codename']}`)\n" - await ctx.send(msg) - -@bot.hybrid_command(name="deviceinfo") -async def devicelist_cmd(ctx, user: discord.User, device_id: int): - """ - Shows info for a specific device. - Usage: .devicelist @user - """ - user_id = user.id - user_devices = devices_data.get(str(user_id), []) - if 1 <= device_id <= len(user_devices): - device = user_devices[device_id - 1] - cocaine_trade_status = "N/A" - if device.get("Model", "").lower() == "eureka": - try: - resp = requests.get("https://cocaine.trade/Quest_3_firmware", timeout=10) - if resp.ok: - if device.get("Build Version", "") in resp.text: - cocaine_trade_status = "True" - else: - cocaine_trade_status = "False" - else: - cocaine_trade_status = "Unknown (site error)" - except Exception: - cocaine_trade_status = "Unknown (request failed)" - msg = ( - f"**Device {device_id} for <@{user_id}>:**\n" - f"**Name:** {device.get('Name','')}\n" - f"**Model:** {device.get('Model','')}\n" - f"**Security Patch:** {device.get('Security Patch','')}\n" - f"**Build Version:** {device.get('Build Version','')}\n" - f"**Version on cocaine.trade:** {cocaine_trade_status}\n" - f"**Vulnerable to:** None" - ) - await ctx.send(msg) - else: - await ctx.send("Device not found for this user.") - -@bot.hybrid_command(name="deviceadd") -async def deviceadd_cmd(ctx): - """ - Add a device using a Discord modal. - Don't know why I made this command. - """ -# Sanitize input to remove @ symbols and escape markdown - def sanitize_input(value: str) -> str: - cleaned = value.strip().replace("@", "") # Prevent mentions - return discord.utils.escape_markdown(cleaned) - - class DeviceModal(discord.ui.Modal, title="Add Device"): - name = discord.ui.TextInput(label="Name", required=True) - model = discord.ui.TextInput(label="Model", required=True) - patch = discord.ui.TextInput(label="Security Patch", required=True) - build = discord.ui.TextInput(label="Build Version", required=True) - - async def on_submit(self, interaction: discord.Interaction): - errors = [] - - # Check for @ - if "@" in self.name.value: - errors.append("**Name** cannot contain `@` symbols.") - if "@" in self.model.value: - errors.append("**Model** cannot contain `@` symbols.") - if "@" in self.patch.value: - errors.append("**Security Patch** cannot contain `@` symbols.") - if "@" in self.build.value: - errors.append("**Build Version** cannot contain `@` symbols.") - - # Validate Build Version: must be a 17 digit number - build_value = self.build.value.strip() - if not (build_value.isdigit() and len(build_value) == 17): - errors.append("**Build Version** must be a 17-digit number (from `adb shell getprop ro.build.version.incremental`).") - - # Validate Security Patch: must be YYYY-MM-DD - patch_value = self.patch.value.strip() - if not re.match(r"^\d{4}-\d{2}-\d{2}$", patch_value): - errors.append("**Security Patch** must be a date in YYYY-MM-DD format (from `adb shell getprop ro.build.version.security_patch`).") - - if errors: - await interaction.response.send_message( - "āŒ There were errors with your input:\n" + "\n".join(errors), - ephemeral=True - ) - return - - user_id = str(interaction.user.id) - device = { - "Name": self.name.value.strip(), - "Model": self.model.value.strip(), - "Security Patch": patch_value, - "Build Version": build_value, - } - devices_data.setdefault(user_id, []).append(device) - save_devices(devices_data) - - await interaction.response.send_message("āœ… Device added!", ephemeral=True) - - # Use the interaction to send the modal - if ctx.interaction: - await ctx.interaction.response.send_modal(DeviceModal()) - else: - await ctx.send("This command must be used as a slash command.") - - -@bot.hybrid_command(name="deviceremove") -async def deviceremove_cmd(ctx, device_id: int): - """ - Remove one of your devices by its ID. - Usage: .deviceremove - """ - user_id = str(ctx.author.id) - user_devices = devices_data.get(user_id, []) - if 1 <= device_id <= len(user_devices): - removed = user_devices.pop(device_id - 1) - save_devices(devices_data) - await ctx.send(f"Removed device: {removed['Name']} (`{removed['Codename']}`)") - else: - await ctx.send("Device not found.") - -@bot.hybrid_command() -@commands.has_role(ADMIN_ROLE_ID) -async def toggle(ctx, index: int): - """ - Toggles a regex pattern. - Only admins can use this command. - """ - real_index = index - 1 # User sees 1-based, code uses 0-based - if 0 <= real_index < len(regex_filters): - regex_filters[real_index]["enabled"] = not regex_filters[real_index]["enabled"] - save_filters() - await ctx.send( - f"Toggled regex `{regex_filters[real_index]['pattern']}` to {'enabled' if regex_filters[real_index]['enabled'] else 'disabled'}." - ) - else: - await ctx.send("Invalid index. Use the number shown in `.listregex`.") - - -@bot.hybrid_command() -@commands.has_role(ADMIN_ROLE_ID) -async def unblock(ctx): - """ - Unblocks a regex pattern. - Only admins can use this command. - """ - if not regex_filters: - return await ctx.send("No regex patterns to remove.") - - await ctx.send("Please enter the index of the regex to remove (as shown in `.listregex`):") - - def check(msg): - return msg.author == ctx.author and msg.channel == ctx.channel - - try: - msg = await bot.wait_for("message", check=check, timeout=60) - index = int(msg.content) - real_index = index - 1 # User sees 1-based, code uses 0-based - removed = regex_filters.pop(real_index) - save_filters() - await ctx.send(f"Removed regex `{removed['pattern']}`") - except (ValueError, IndexError): - await ctx.send("Invalid index.") - except TimeoutError: - await ctx.send("Timeout. Please try again.") - - -@bot.hybrid_command() -async def status(ctx): - """ - Displays the bot's status and environment information. - """ - os_info = platform.system() - release = platform.release() - architecture = platform.machine() - python_version = platform.python_version() - uptime = get_uptime() - - env_message = ( - f"āœ… Bot is running in **{os_info} {release} ({architecture})** environment " - f"with **Python {python_version}**\n" - f"šŸ›  Version: **{BOTVERSION}**\n" - f"ā± Uptime: **{uptime}**" - ) - if "b" in BOTVERSION.lower(): - env_message += "\nāš ļø Beta version detected ā€“ may be unstable! Potentially destructive commands have been disabled." - await ctx.send(env_message) - - -@bot.hybrid_command() -@commands.has_role(ADMIN_ROLE_ID) -async def slowmode(ctx, seconds: int): - """ - Sets the slowmode for the current channel. - Only admins can use this command. - """ - await ctx.channel.edit(slowmode_delay=seconds) - await ctx.send(f"This channel now has a slowmode of {seconds} seconds!") - - -# File to store quarantine data persistently -QUARANTINE_DATA_FILE = "quarantine_data.json" -LOG_FILE = "quarantine_log.txt" - -# Active quarantines dictionary: user_id -> unquarantine timestamp ISO string -active_quarantines = {} - - -def log_to_file(entry: str): - with open(LOG_FILE, "a", encoding="utf-8") as f: - f.write(f"{datetime.utcnow().isoformat()} - {entry}\n") - - -def save_quarantine_data(): - with open(QUARANTINE_DATA_FILE, "w", encoding="utf-8") as f: - json.dump(active_quarantines, f) - - -def load_quarantine_data(): - global active_quarantines - if os.path.exists(QUARANTINE_DATA_FILE): - with open(QUARANTINE_DATA_FILE, "r", encoding="utf-8") as f: - active_quarantines = json.load(f) - else: - active_quarantines = {} - - -def is_admin_quarantine(): - def predicate(ctx): - return any(role.id == ADMIN_ROLE_ID for role in ctx.author.roles) - - return commands.check(predicate) - - -@bot.hybrid_command() -@is_admin_quarantine() -async def q( - ctx, member: discord.Member, duration: str, *, reason: str = "No reason provided" -): - """ - Quarantine a member for a duration (e.g. 10m, 1h, 1d). - """ - quarantine_role = ctx.guild.get_role(QUARANTINE_ROLE_ID) - if quarantine_role in member.roles: - await ctx.send(f"{member.display_name} is already quarantined.") - return - - # Parse duration - try: - amount = int(duration[:-1]) - unit = duration[-1].lower() - if unit == "m": - delta = timedelta(minutes=amount) - elif unit == "h": - delta = timedelta(hours=amount) - elif unit == "d": - delta = timedelta(days=amount) - else: - await ctx.send( - "Invalid duration format. Use m (minutes), h (hours), or d (days)." - ) - return - except Exception as e: - await ctx.send( - f"Invalid duration format. Use m (minutes), h (hours), or d (days). Example: 10m, 1h, 2d {e}" - ) - return - - await member.add_roles( - quarantine_role, reason=f"Quarantine by {ctx.author} for {reason}" - ) - unquarantine_time = datetime.utcnow() + delta - active_quarantines[str(member.id)] = unquarantine_time.isoformat() - save_quarantine_data() - - await ctx.send( - f"{member.display_name} has been quarantined for {duration}. Reason: {reason}" - ) - - log_entry = f"{ctx.author} quarantined {member} for {duration}. Reason: {reason}" - log_to_file(log_entry) - - log_channel = ctx.guild.get_channel(REPORT_LOG_CHANNEL_ID) - if log_channel: - embed = discord.Embed( - title="User Quarantined", - color=discord.Color.orange(), - timestamp=datetime.utcnow(), - ) - embed.add_field(name="User", value=member.mention, inline=True) - embed.add_field(name="By", value=ctx.author.mention, inline=True) - embed.add_field(name="Duration", value=duration, inline=True) - embed.add_field(name="Reason", value=reason, inline=False) - await log_channel.send(embed=embed) - - -@bot.hybrid_command() -@is_admin_quarantine() -async def uq(ctx, member: discord.Member, *, reason: str = "No reason provided"): - """ - Unquarantine a member immediately. - """ - quarantine_role = ctx.guild.get_role(QUARANTINE_ROLE_ID) - if quarantine_role not in member.roles: - await ctx.send(f"{member.display_name} is not quarantined.") - return - - await member.remove_roles( - quarantine_role, reason=f"Unquarantined by {ctx.author} for {reason}" - ) - active_quarantines.pop(str(member.id), None) - save_quarantine_data() - - await ctx.send(f"{member.display_name} has been unquarantined. Reason: {reason}") - - log_entry = f"{ctx.author} unquarantined {member}. Reason: {reason}" - log_to_file(log_entry) - - log_channel = ctx.guild.get_channel(REPORT_LOG_CHANNEL_ID) - if log_channel: - embed = discord.Embed( - title="User Unquarantined", - color=discord.Color.green(), - timestamp=datetime.utcnow(), - ) - embed.add_field(name="User", value=member.mention, inline=True) - embed.add_field(name="By", value=ctx.author.mention, inline=True) - embed.add_field(name="Reason", value=reason, inline=False) - await log_channel.send(embed=embed) - - -@tasks.loop(seconds=60) -async def check_quarantine_expiry(): - now = datetime.utcnow() - guild = bot.guilds[0] - quarantine_role = guild.get_role(QUARANTINE_ROLE_ID) - to_remove = [] - - for user_id_str, unq_time_str in active_quarantines.items(): - user_id = int(user_id_str) - unq_time = datetime.fromisoformat(unq_time_str) - if now >= unq_time: - member = guild.get_member(user_id) - if member and quarantine_role in member.roles: - try: - await member.remove_roles( - quarantine_role, reason="Automatic quarantine expiry" - ) - except Exception as e: - print(f"Error removing quarantine role from {member}: {e}") - - log_entry = f"Automatic unquarantine for {member} (quarantine expired)." - log_to_file(log_entry) - - log_channel = guild.get_channel(REPORT_LOG_CHANNEL_ID) - if log_channel: - embed = discord.Embed( - title="Quarantine Expired", - color=discord.Color.blue(), - timestamp=datetime.utcnow(), - ) - embed.add_field(name="User", value=member.mention) - embed.add_field(name="Reason", value="Quarantine time expired") - await log_channel.send(embed=embed) - - to_remove.append(user_id_str) - - for user_id_str in to_remove: - active_quarantines.pop(user_id_str) - if to_remove: - save_quarantine_data() - - -@bot.hybrid_command() -@commands.has_role(ADMIN_ROLE_ID) -async def hotupdate(ctx): - """ - Pulls everything from the repository, but does not restart the bot. - """ - global replies - try: - await ctx.send("šŸ“„ Pulling everything from the repository...") - if os.path.exists(REPO_DIR): - repo = Repo(REPO_DIR) - repo.remotes.origin.pull() - else: - Repo.clone_from(REPO_URL, REPO_DIR) - replies = load_replies() - await ctx.send("āœ… Hot update complete.") - except Exception as e: - await ctx.send(f"āŒ Hot update failed:\n```{e}```") - - -@bot.hybrid_command() -async def ratelimitcheck(ctx): - try: - await ctx.send("If you see this message, the bot is not rate limited.") - except discord.HTTPException as e: - if e.status == 429: - print("Rate limited: Try again after", e.retry_after) - else: - await ctx.send( - "An error occurred while trying to send the message. Check console!!" - ) - print(f"Unexpected HTTP error: {e}") - - -@bot.event -async def on_message(message): - if message.author.bot: - return - - invisible_chars = ["\u200b", "\u200c", "\u200d", "\u200e", "\u200f"] - content = message.content - for ch in invisible_chars: - content = content.replace(ch, "") - content = content.strip() - message.content = content # Update message for command processing later - - for entry in regex_filters: - if entry["enabled"]: - try: - if re.search(entry["pattern"], content): - await message.delete() - try: - await message.author.send( - f"šŸš« Your message was not allowed:\n`{content}`\n(Reason: Matches blocked pattern)" - ) - except discord.Forbidden: - pass - - log_channel = bot.get_channel(REPORT_LOG_CHANNEL_ID) - if log_channel: - timestamp = datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S UTC") - await log_channel.send( - f"šŸšØ **Blocked Message**\n" - f"**User:** {message.author.mention} (`{message.author.id}`)\n" - f"**Message:** `{content}`\n" - f"**Time:** {timestamp}" - ) - return - except re.error: - continue - - if message.channel.id == 1374296035798814804: - data = load_count_data() - current_count = data["current_count"] - last_counter_id = data["last_counter_id"] - - if content.isdigit() and "\n" not in content: - number = int(content) - if number == current_count + 1 and message.author.id != last_counter_id: - # Valid message - data["current_count"] = number - data["last_counter_id"] = message.author.id - save_count_data(data) - return - else: - reason = ( - "Double message" - if message.author.id == last_counter_id - else "Incorrect number" - ) - else: - reason = "Invalid message format" - - data["current_count"] = 0 - data["last_counter_id"] = None - save_count_data(data) - - await message.delete() - countreport = bot.get_channel(1348562119469305958) - count = bot.get_channel(1374296035798814804) - if countreport: - await countreport.send( - f"āš ļø <@{message.author.id}> broke the counting streak in <#{message.channel.id}>! ({reason})" - ) - await count.send("Streak has been broken! Start from 1.") - return - - if content.startswith("."): - cmd = content[1:] - if cmd in replies: - await message.channel.send(replies[cmd][1]) - return - - - if isinstance(message.channel, discord.DMChannel): - user_id = message.author.id - if user_id in active_reports and not content.startswith("."): - active_reports[user_id].append(content) - return - - await bot.process_commands(message) - - - -@bot.hybrid_command() -@commands.has_role(ADMIN_ROLE_ID) -async def reboot(ctx): - """ - Reboots the bot. - Only admins can use this command. - """ - await ctx.send("šŸ”‚ Rebooting bot...") - python = sys.executable - os.execv(python, [python] + sys.argv) - - -@bot.hybrid_command() -async def streak(ctx): - """ - Displays the current counting streak. - """ - data = load_count_data() - await ctx.send(f"The current counting streak is **{data['current_count']}**.") - -@bot.hybrid_command(name="replies") -async def replies_command(ctx): - """Lists all available quick replies.""" - if not replies: - await ctx.send("āš ļø No replies available.") - response = "\n".join([f"* {key}: {val[0]}" for key, val in replies.items()]) - await ctx.send(f"```\n{response}\n```") - -@bot.event -async def on_command_error(ctx, error): - if isinstance(error, commands.CommandNotFound): - print(f"Ignoring exception: CommandNotFound: '{ctx.message.content}'") - elif isinstance(error, commands.MissingRequiredArgument): - await ctx.send(f"āŒ Missing required argument: {error.param.name}") - elif isinstance(error, commands.MissingRole): - await ctx.send(f"{ctx.author.mention}āŒ You are not authorized to use this command.") - else: - print(f"Unhandled command error: {error}") - -async def main(): - try: - await bot.start(TOKEN) - except discord.HTTPException as e: - if e.status == 429: - print(f"Startup failed: Rate limited. Retry after {e.retry_after} seconds.") - else: - print(f"Startup failed with HTTP error: {e}") - except Exception as e: - print(f"Startup failed with unexpected error: {e}") - -if IS_BETA: - for name in DISABLED_IN_BETA: - if name in bot.all_commands: - bot.remove_command(name) -asyncio.run(main()) diff --git a/bot.py b/bot.py new file mode 100644 index 0000000..54f1484 --- /dev/null +++ b/bot.py @@ -0,0 +1,28 @@ +from discord.ext import commands +from config import intents, BOT_VERSION, DISABLED_IN_BETA + + +class FreeXRBot(commands.Bot): + def __init__(self): + super().__init__( + command_prefix=".", + intents=intents, + help_command=None + ) + + async def setup_hook(self): + await self.load_extension("cogs.admin") + await self.load_extension("cogs.moderation") + await self.load_extension("cogs.devices") + await self.load_extension("cogs.counting") + await self.load_extension("cogs.utility") + await self.load_extension("cogs.maintenance") + await self.load_extension("cogs.events") + + if "b" in BOT_VERSION.lower(): + for name in DISABLED_IN_BETA: + if name in self.all_commands: + self.remove_command(name) + + +bot = FreeXRBot() \ No newline at end of file diff --git a/cogs/admin.py b/cogs/admin.py new file mode 100644 index 0000000..2809528 --- /dev/null +++ b/cogs/admin.py @@ -0,0 +1,205 @@ +import discord +from discord.ext import commands +from config import ADMIN_ROLE_ID +import asyncio + + +class Admin(commands.Cog): + def __init__(self, bot): + self.bot = bot + + @commands.hybrid_command() + @commands.has_role(ADMIN_ROLE_ID) + async def slowmode(self, ctx, seconds: int): + """Sets the slowmode for the current channel.""" + await ctx.channel.edit(slowmode_delay=seconds) + await ctx.send(f"This channel now has a slowmode of {seconds} seconds!") + + @commands.hybrid_command() + @commands.has_role(ADMIN_ROLE_ID) + async def resolve(self, ctx, msg_id: int = None): + """Marks a report as resolved.""" + if not msg_id and ctx.message.reference: + msg_id = ctx.message.reference.message_id + + if not msg_id: + await ctx.send("Please reply to a report or provide a message ID.") + return + + from utils.file_handlers import load_json, save_json + from config import REPORTS_FILE + + report_log_map = load_json(REPORTS_FILE, {}) + report_id = str(msg_id) + + if report_id in report_log_map: + del report_log_map[report_id] + save_json(REPORTS_FILE, report_log_map) + + try: + msg = await ctx.channel.fetch_message(msg_id) + await msg.reply("āœ… Marked as resolved. Further interaction closed.") + except discord.NotFound: + await ctx.send("Marked as resolved, but couldn't find the original message.") + else: + await ctx.send("That message isn't tracked as an active report.") + + @commands.hybrid_command() + @commands.has_role(ADMIN_ROLE_ID) + async def createchannel(self, ctx, msg_id: int = None): + """Creates a private channel for a report.""" + if not msg_id and ctx.message.reference: + msg_id = ctx.message.reference.message_id + + embed = None + if msg_id: + try: + from config import REPORT_LOG_CHANNEL_ID + log_channel = self.bot.get_channel(REPORT_LOG_CHANNEL_ID) + report_msg = await log_channel.fetch_message(msg_id) + if report_msg.embeds: + embed = report_msg.embeds[0] + except discord.NotFound: + await ctx.send("Couldn't find the report message.") + return + except Exception as e: + await ctx.send(f"Error fetching report: {e}") + return + + await ctx.send("What should the channel be called?") + + def check(m): + return m.author == ctx.author and m.channel == ctx.channel + + try: + name_msg = await self.bot.wait_for("message", check=check, timeout=60) + guild = ctx.guild + overwrites = { + guild.default_role: discord.PermissionOverwrite(read_messages=False), + guild.get_role(ADMIN_ROLE_ID): discord.PermissionOverwrite(read_messages=True), + } + channel = await guild.create_text_channel(name=name_msg.content, overwrites=overwrites) + await ctx.send(f"Created channel: {channel.mention}") + + if embed: + await channel.send("Report linked to this channel:", embed=embed) + + except asyncio.TimeoutError: + await ctx.send("Timed out.") + except Exception as e: + await ctx.send(f"Failed to create channel: {e}") + + @commands.hybrid_command() + @commands.has_role(ADMIN_ROLE_ID) + async def createchannelp(self, ctx, msg_id: int = None): + """Creates a private channel for a report with the original reporter.""" + if not msg_id and ctx.message.reference: + msg_id = ctx.message.reference.message_id + + if not msg_id: + await ctx.send("Please reply to a report or provide a message ID.") + return + + from utils.file_handlers import load_json + from config import REPORTS_FILE + + report_log_map = load_json(REPORTS_FILE, {}) + report_id = str(msg_id) + + if report_id not in report_log_map: + await ctx.send("Couldn't find the original reporter for that ID.") + return + + user_id = report_log_map[report_id] + guild = ctx.guild + member = guild.get_member(user_id) + + if not member: + await ctx.send("The original reporter is not in this server.") + return + + await ctx.send("What should the channel be called?") + + def check(m): + return m.author == ctx.author and m.channel == ctx.channel + + try: + name_msg = await self.bot.wait_for("message", check=check, timeout=60) + overwrites = { + guild.default_role: discord.PermissionOverwrite(read_messages=False), + guild.get_role(ADMIN_ROLE_ID): discord.PermissionOverwrite(read_messages=True), + member: discord.PermissionOverwrite(read_messages=True), + } + channel = await guild.create_text_channel(name=name_msg.content, overwrites=overwrites) + await ctx.send(f"Created private channel: {channel.mention} with access to {member.mention} given.") + except asyncio.TimeoutError: + await ctx.send("Timed out.") + except Exception as e: + await ctx.send(f"Failed to create channel: {e}") + + @commands.hybrid_command() + @commands.has_role(ADMIN_ROLE_ID) + async def listreport(self, ctx): + """Lists all active reports.""" + from utils.file_handlers import load_json + from config import REPORTS_FILE, REPORT_LOG_CHANNEL_ID + + report_log_map = load_json(REPORTS_FILE, {}) + if not report_log_map: + await ctx.send("No reports found.") + return + + log_channel = self.bot.get_channel(REPORT_LOG_CHANNEL_ID) + report_lines = [] + + for i, (msg_id, user_id) in enumerate(report_log_map.items()): + try: + msg = await log_channel.fetch_message(int(msg_id)) + title = msg.embeds[0].title if msg.embeds else "No Title" + report_lines.append(f"{i + 1}. {title} (ID: {msg.id})") + except discord.NotFound: + report_lines.append(f"{i + 1}. [Message not found] (ID: {msg_id})") + + if report_lines: + await ctx.send("**Active Reports:**\n" + "\n".join(report_lines)) + else: + await ctx.send("No valid report messages found.") + + @commands.hybrid_command() + async def role(self, ctx, role_id: int, user_id: int): + """ + Toggles a role for a user. + Only the user with ID 981463678698266664 is authorized to use this command. + """ + allowed_user_id = 981463678698266664 + + if ctx.author.id != allowed_user_id: + await ctx.send("āŒ You are not authorized to use this command.") + return + + guild = ctx.guild + member = guild.get_member(user_id) + role = guild.get_role(role_id) + + if not member: + await ctx.send("āŒ User not found.") + return + if not role: + await ctx.send("āŒ Role not found.") + return + + try: + if role in member.roles: + await member.remove_roles(role) + await ctx.send(f"āœ… Removed role **{role.name}** from {member.mention}.") + else: + await member.add_roles(role) + await ctx.send(f"āœ… Added role **{role.name}** to {member.mention}.") + except discord.Forbidden: + await ctx.send("āŒ I don't have permission to manage that role.") + except discord.HTTPException as e: + await ctx.send(f"āŒ Failed to modify role: {e}") + + +async def setup(bot): + await bot.add_cog(Admin(bot)) \ No newline at end of file diff --git a/cogs/counting.py b/cogs/counting.py new file mode 100644 index 0000000..463e58d --- /dev/null +++ b/cogs/counting.py @@ -0,0 +1,72 @@ +from discord.ext import commands + +from utils.file_handlers import load_json, save_json +from config import COUNTING_CHANNEL_ID, COUNTING_REPORT_CHANNEL_ID, MATH_CONSTANTS, COUNT_FILE +from arithmetic_eval import evaluate + + +class Counting(commands.Cog): + def __init__(self, bot): + self.bot = bot + + @commands.hybrid_command() + async def streak(self, ctx): + data = load_json(COUNT_FILE, {"current_count": 0, "last_counter_id": None}) + await ctx.send(f"The current counting streak is **{data['current_count']}**.") + + @commands.Cog.listener() + async def on_message(self, message): + if message.channel.id == COUNTING_CHANNEL_ID and not message.author.bot: + data = load_json(COUNT_FILE, {"current_count": 0, "last_counter_id": None}) + current_count = data["current_count"] + last_counter_id = data["last_counter_id"] + content = message.content + + try: + number = evaluate(content, MATH_CONSTANTS) + + if number == current_count + 1 and message.author.id != last_counter_id: + data["current_count"] = number + data["last_counter_id"] = message.author.id + save_json(COUNT_FILE, data) + return + else: + if message.author.id == last_counter_id: + reason = "Double counting - same user consecutively" + elif number != current_count + 1: + reason = f"Incorrect number - expected {current_count + 1}, got {number}" + else: + reason = "Unknown error" + + except (ValueError, SyntaxError, ZeroDivisionError, TypeError) as e: # TODO: Refactor entire file because I'm not sure it even works + if content.isdigit() and "\n" not in content: + number = int(content) + if number == current_count + 1 and message.author.id != last_counter_id: + data["current_count"] = number + data["last_counter_id"] = message.author.id + save_json(COUNT_FILE, data) + return + else: + if message.author.id == last_counter_id: + reason = "Double counting - same user consecutively" + else: + reason = f"Incorrect number - expected {current_count + 1}, got {number}" + else: + reason = f"Invalid mathematical expression: {str(e)}" + + data["current_count"] = 0 + data["last_counter_id"] = None + save_json(COUNT_FILE, data) + + await message.delete() + count_report = self.bot.get_channel(COUNTING_REPORT_CHANNEL_ID) + count = self.bot.get_channel(COUNTING_CHANNEL_ID) + if count_report: + await count_report.send( + f"āš ļø <@{message.author.id}> broke the counting streak in <#{message.channel.id}>! ({reason})" + ) + await count.send("Streak has been broken! Start from 1.") + + +async def setup(bot): + await bot.add_cog(Counting(bot)) \ No newline at end of file diff --git a/cogs/devices.py b/cogs/devices.py new file mode 100644 index 0000000..c73dd0a --- /dev/null +++ b/cogs/devices.py @@ -0,0 +1,133 @@ +import discord +from discord.ext import commands +import requests +import re + +from utils.file_handlers import load_json, save_json +from config import DEVICES_FILE + + +class Devices(commands.Cog): + def __init__(self, bot): + self.bot = bot + self.devices_data = load_json(DEVICES_FILE, {}) + + @commands.hybrid_command(name="devices") + async def devices_cmd(self, ctx, user: discord.User = None): + """Lists your devices or another user's devices.""" + user_id = user.id if user else ctx.author.id + user_devices = self.devices_data.get(str(user_id), []) + if not user_devices: + await ctx.send("No devices found for this user.") + return + msg = f"Devices for <@{user_id}>:\n" + for idx, device in enumerate(user_devices, 1): + msg += f"**{idx}.** {device['Name']} (`{device.get('Codename', 'N/A')}`)\n" + await ctx.send(msg) + + @commands.hybrid_command(name="deviceinfo") + async def deviceinfo_cmd(self, ctx, user: discord.User, device_id: int): + """Shows info for a specific device.""" + user_id = user.id + user_devices = self.devices_data.get(str(user_id), []) + if 1 <= device_id <= len(user_devices): + device = user_devices[device_id - 1] + cocaine_trade_status = "N/A" + if device.get("Model", "").lower() == "eureka": + try: + resp = requests.get("https://cocaine.trade/Quest_3_firmware", timeout=10) + if resp.ok: + if device.get("Build Version", "") in resp.text: + cocaine_trade_status = "True" + else: + cocaine_trade_status = "False" + else: + cocaine_trade_status = "Unknown (site error)" + except Exception: + cocaine_trade_status = "Unknown (request failed)" + msg = ( + f"**Device {device_id} for <@{user_id}>:**\n" + f"**Name:** {device.get('Name', '')}\n" + f"**Model:** {device.get('Model', '')}\n" + f"**Security Patch:** {device.get('Security Patch', '')}\n" + f"**Build Version:** {device.get('Build Version', '')}\n" + f"**Version on cocaine.trade:** {cocaine_trade_status}\n" + f"**Vulnerable to:** None" + ) + await ctx.send(msg) + else: + await ctx.send("Device not found for this user.") + + @commands.hybrid_command(name="deviceadd") + async def deviceadd_cmd(self, ctx): + """Add a device using a Discord modal.""" + + class DeviceModal(discord.ui.Modal, title="Add Device"): + name = discord.ui.TextInput(label="Name", required=True) + model = discord.ui.TextInput(label="Model", required=True) + patch = discord.ui.TextInput(label="Security Patch", required=True) + build = discord.ui.TextInput(label="Build Version", required=True) + + async def on_submit(self, interaction: discord.Interaction): + errors = [] + + if "@" in self.name.value: + errors.append("**Name** cannot contain `@` symbols.") + if "@" in self.model.value: + errors.append("**Model** cannot contain `@` symbols.") + if "@" in self.patch.value: + errors.append("**Security Patch** cannot contain `@` symbols.") + if "@" in self.build.value: + errors.append("**Build Version** cannot contain `@` symbols.") + + build_value = self.build.value.strip() + if not (build_value.isdigit() and len(build_value) == 17): + errors.append( + "**Build Version** must be a 17-digit number (from `adb shell getprop ro.build.version.incremental`).") + + patch_value = self.patch.value.strip() + if not re.match(r"^\d{4}-\d{2}-\d{2}$", patch_value): + errors.append( + "**Security Patch** must be a date in YYYY-MM-DD format (from `adb shell getprop ro.build.version.security_patch`).") + + if errors: + await interaction.response.send_message( + "āŒ There were errors with your input:\n" + "\n".join(errors), + ephemeral=True + ) + return + + user_id = str(interaction.user.id) + device = { + "Name": self.name.value.strip(), + "Model": self.model.value.strip(), + "Security Patch": patch_value, + "Build Version": build_value, + } + self.cog.devices_data.setdefault(user_id, []).append(device) + save_json(DEVICES_FILE, self.cog.devices_data) + + await interaction.response.send_message("āœ… Device added!", ephemeral=True) + + modal = DeviceModal() + modal.cog = self + if ctx.interaction: + await ctx.interaction.response.send_modal(modal) + else: + await ctx.send("This command must be used as a slash command.") + + @commands.hybrid_command(name="deviceremove") + async def deviceremove_cmd(self, ctx, device_id: int): + """Remove one of your devices by its ID.""" + user_id = str(ctx.author.id) + user_devices = self.devices_data.get(user_id, []) + if 1 <= device_id <= len(user_devices): + removed = user_devices.pop(device_id - 1) + save_json(DEVICES_FILE, self.devices_data) + await ctx.send(f"Removed device: {removed['Name']} (`{removed.get('Codename', 'N/A')}`)") + else: + await ctx.send("Device not found.") + + +async def setup(bot): + await bot.add_cog(Devices(bot)) \ No newline at end of file diff --git a/cogs/events.py b/cogs/events.py new file mode 100644 index 0000000..d229b0f --- /dev/null +++ b/cogs/events.py @@ -0,0 +1,99 @@ +import discord +from discord.ext import commands +import platform +import time +import sys + +from config import BOT_VERSION, QUARANTINE_ROLE_ID, QUARANTINE_DATA_FILE, BOT_CONSOLE_CHANNEL_ID, WELCOME_CHANNEL_ID, MEMBER_ROLE_ID +from utils.file_handlers import load_json +from utils.logger import DiscordConsoleLogger +from utils.helpers import get_uptime + +start_time = time.time() + + +class Events(commands.Cog): + def __init__(self, bot): + self.bot = bot + + @commands.Cog.listener() + async def on_ready(self): + await self.bot.tree.sync() + sys.stdout = DiscordConsoleLogger(self.bot, BOT_CONSOLE_CHANNEL_ID) + sys.stderr = sys.stdout + print(f"Logged in as {self.bot.user}") + + os_info = platform.system() + release = platform.release() + architecture = platform.machine() + python_version = platform.python_version() + uptime = get_uptime(start_time) + + env_message = ( + f"āœ… Bot is running in **{os_info} {release} ({architecture})** environment " + f"with **Python {python_version}**\n" + f"šŸ›  Version: **{BOT_VERSION}**\n" + f"ā± Load time: **{uptime}**" + ) + if "b" in BOT_VERSION.lower(): + env_message += "\nāš ļø Beta version detected ā€“ may be unstable! Potentially destructive commands have been disabled." + channel = self.bot.get_channel(BOT_CONSOLE_CHANNEL_ID) + await channel.send(env_message) + + print(env_message) + + active_quarantines = load_json(QUARANTINE_DATA_FILE, {}) + guild = self.bot.guilds[0] + quarantine_role = guild.get_role(QUARANTINE_ROLE_ID) + to_remove = [] + + for user_id_str, unq_time_str in active_quarantines.items(): + user_id = int(user_id_str) + member = guild.get_member(user_id) + from datetime import datetime + unq_time = datetime.fromisoformat(unq_time_str) + if member is None: + to_remove.append(user_id_str) + continue + if quarantine_role not in member.roles: + to_remove.append(user_id_str) + + for user_id_str in to_remove: + active_quarantines.pop(user_id_str) + + from utils.file_handlers import save_json + save_json(QUARANTINE_DATA_FILE, active_quarantines) + + @commands.Cog.listener() + async def on_member_join(self, member): + try: + await member.add_roles(member.guild.get_role(MEMBER_ROLE_ID)) + except discord.Forbidden: + print(f"Couldn't add role to {member} (permissions issue)") + + try: + await member.send( + "# šŸ‘‹ Welcome to the server!\n Hello and welcome to FreeXR. We hack headsets to root them and unlock their bootloaders, and we appreciate you for joining. To get started, please read the https://discord.com/channels/1344235945238593547/1364918149404688454.\nWe hope you have a great stay here, and thank you for joining." + ) + except discord.Forbidden: + channel = member.guild.get_channel(WELCOME_CHANNEL_ID) + if channel: + msg = await channel.send( + f"{member.mention} šŸ‘‹ Welcome to the server! Please read the https://discord.com/channels/1344235945238593547/1364918149404688454, and we hope you have a great stay here!.\n-# Psst! Your DMs are closed, so I couldn't send you a DM." + ) + + @commands.Cog.listener() + async def on_command_error(self, ctx, error): + if isinstance(error, commands.CommandNotFound): + if not self.bot.get_cog('Utility').replies[ctx.message.content]: + print(f"Ignoring exception: CommandNotFound: '{ctx.message.content}'") + elif isinstance(error, commands.MissingRequiredArgument): + await ctx.send(f"āŒ Missing required argument: {error.param.name}") + elif isinstance(error, commands.MissingRole): + await ctx.send(f"{ctx.author.mention}āŒ You are not authorized to use this command.") + else: + print(f"Unhandled command error: {error}") + + +async def setup(bot): + await bot.add_cog(Events(bot)) \ No newline at end of file diff --git a/cogs/maintenance.py b/cogs/maintenance.py new file mode 100644 index 0000000..9544754 --- /dev/null +++ b/cogs/maintenance.py @@ -0,0 +1,75 @@ +import discord +from discord.ext import commands +import sys +import os +from git import Repo + +from config import ADMIN_ROLE_ID, REPO_DIR, REPO_URL +from utils.helpers import load_replies + + +class Maintenance(commands.Cog): + def __init__(self, bot): + self.bot = bot + + @commands.hybrid_command() + @commands.has_role(ADMIN_ROLE_ID) + async def update(self, ctx): + """Updates the bot by pulling latest code and restarting""" + await ctx.send("šŸ“„ Downloading latest version...") + + try: + await ctx.send("šŸ“„ Pulling from repository...") + if os.path.exists(REPO_DIR): + repo = Repo(REPO_DIR) + repo.remotes.origin.pull() + else: + Repo.clone_from(REPO_URL, REPO_DIR) + + await ctx.send("āœ… Update complete. Restarting bot...") + python = sys.executable + os.execv(python, [python] + sys.argv) + except Exception as e: + await ctx.send(f"āŒ Update failed:\n```{e}```") + + @commands.hybrid_command() + @commands.has_role(ADMIN_ROLE_ID) + async def hotupdate(self, ctx): + """Pulls everything from repository without restarting""" + try: + await ctx.send("šŸ“„ Pulling everything from the repository...") + if os.path.exists(REPO_DIR): + repo = Repo(REPO_DIR) + repo.remotes.origin.pull() + else: + Repo.clone_from(REPO_URL, REPO_DIR) + + self.bot.get_cog('Utility').replies = load_replies() + await ctx.send("āœ… Hot update complete.") + except Exception as e: + await ctx.send(f"āŒ Hot update failed:\n```{e}```") + + @commands.hybrid_command() + @commands.has_role(ADMIN_ROLE_ID) + async def reboot(self, ctx): + """Reboots the bot""" + await ctx.send("šŸ”‚ Rebooting bot...") + python = sys.executable + os.execv(python, [python] + sys.argv) + + @commands.hybrid_command() + async def ratelimitcheck(self, ctx): + try: + await ctx.send("If you see this message, the bot is not rate limited.") + except discord.HTTPException as e: + if e.status == 429: + print("Rate limited: Try again after", e.retry_after) + else: + await ctx.send( + "An error occurred while trying to send the message. Check console!!" + ) + print(f"Unexpected HTTP error: {e}") + + +async def setup(bot): + await bot.add_cog(Maintenance(bot)) \ No newline at end of file diff --git a/cogs/moderation.py b/cogs/moderation.py new file mode 100644 index 0000000..c463347 --- /dev/null +++ b/cogs/moderation.py @@ -0,0 +1,368 @@ +import discord +from discord.ext import commands, tasks +from datetime import datetime, timedelta, timezone +import re +import asyncio + +from config import ADMIN_ROLE_ID, QUARANTINE_ROLE_ID, QUARANTINE_DATA_FILE, BACKUP_FILE, FILTER_FILE, REPORT_LOG_CHANNEL_ID, QUARANTINE_LOG_FILE +from utils.file_handlers import load_json, save_json +from utils.helpers import log_to_file, clean_message_content + + +class Moderation(commands.Cog): + def __init__(self, bot): + self.context_menu = None + self.bot = bot + self.active_reports = {} + self.regex_filters = load_json(FILTER_FILE, []) + self.active_quarantines = load_json(QUARANTINE_DATA_FILE, {}) + self.message_backups = load_json(BACKUP_FILE, {}) + + self.check_quarantine_expiry.start() + + def cog_unload(self): + self.check_quarantine_expiry.cancel() + self.bot.tree.remove_command("Add to report", type=discord.AppCommandType.message) + + async def cog_load(self): + self.context_menu = discord.app_commands.ContextMenu( + name="Add to report", + callback=self.add_to_report, + ) + self.bot.tree.add_command(self.context_menu) + + @commands.hybrid_command() + async def report(self, ctx): + """Starts a report in DMs.""" + if not isinstance(ctx.channel, discord.DMChannel): + await ctx.send("Please DM me this command.") + return + + await ctx.send( + "You're reporting to the server admins. All messages from this point will be recorded.\n" + "Please state your issue. Upload images as links (attachments won't work).\n" + "When you're finished, type `.iamdone`. Messages will stop being recorded." + ) + self.active_reports[ctx.author.id] = [] + + #@discord.app_commands.context_menu(name="Add to report") + async def add_to_report(self, interaction: discord.Interaction, message: discord.Message): + user_id = interaction.user.id + + if user_id not in self.active_reports: + await interaction.response.send_message( + "āŒ You don't have an active report. Please start one by DMing me `.report`.", + ephemeral=True, + ) + return + + content = f"**Message from {message.author} in <#{message.channel.id}>:**\n{message.content}" + self.active_reports[user_id].append(content) + + backup_entry = { + "author": str(message.author), + "channel_id": message.channel.id, + "message_id": message.id, + "content": message.content, + "timestamp": str(message.created_at), + "jump_url": message.jump_url, + } + if str(user_id) not in self.message_backups: + self.message_backups[str(user_id)] = [] + self.message_backups[str(user_id)].append(backup_entry) + save_json(BACKUP_FILE, self.message_backups) + + await interaction.response.send_message("āœ… Message added to your report.", ephemeral=True) + + @commands.hybrid_command() + async def iamdone(self, ctx): + """Ends the report and sends it to the admins.""" + if not isinstance(ctx.channel, discord.DMChannel): + return + + user_id = ctx.author.id + if user_id not in self.active_reports or not self.active_reports[user_id]: + await ctx.send("No messages recorded or you haven't started a report with `.report`.") + return + + channel = self.bot.get_channel(REPORT_LOG_CHANNEL_ID) + report_content = "\n".join(self.active_reports[user_id]) + extra = "" + if str(user_id) in self.message_backups: + extra += "\n\n**Attached messages:**\n" + for entry in self.message_backups[str(user_id)]: + extra += ( + f"[Jump to message]({entry['jump_url']}) | " + f"**{entry['author']}** ({entry['timestamp']}):\n" + f"{entry['content']}\n\n" + ) + + full_content = report_content + extra + + embed = discord.Embed( + title="New Report", + description=full_content[:4000], + color=discord.Color.orange(), + ) + embed.set_author(name=f"{ctx.author}", icon_url=ctx.author.display_avatar.url) + + report_message = await channel.send(embed=embed) + + from utils.file_handlers import load_json, save_json + report_log_map = load_json(REPORTS_FILE, {}) + report_log_map[str(report_message.id)] = user_id + save_json(REPORTS_FILE, report_log_map) + + await ctx.send("Thank you! Your report has been sent.") + + self.active_reports[user_id] = [] + if str(user_id) in self.message_backups: + del self.message_backups[str(user_id)] + save_json(BACKUP_FILE, self.message_backups) + + @commands.hybrid_command() + @commands.has_role(ADMIN_ROLE_ID) + async def block(self, ctx): + """Blocks a regex pattern from being sent in the server.""" + if ctx.interaction: + class RegexModal(discord.ui.Modal, title="Block Regex Pattern"): + pattern = discord.ui.TextInput(label="Regex Pattern", required=True) + + async def on_submit(self, interaction: discord.Interaction): + try: + re.compile(self.pattern.value) + self.cog.regex_filters.append({"pattern": self.pattern.value, "enabled": True}) + save_json(FILTER_FILE, self.cog.regex_filters) + await interaction.response.send_message(f"Blocked regex added: `{self.pattern.value}`", + ephemeral=True) + except re.error: + await interaction.response.send_message("Invalid regex pattern.", ephemeral=True) + + modal = RegexModal() + modal.cog = self + await ctx.interaction.response.send_modal(modal) + else: + await ctx.send("Please use the `/block` slash command to add a regex pattern.") + + @commands.hybrid_command() + @commands.has_role(ADMIN_ROLE_ID) + async def listregex(self, ctx): + """Lists all blocked regex patterns.""" + if not self.regex_filters: + return await ctx.send("No regex patterns are currently blocked.") + + message = "Blocked Regex Patterns:\n" + for i, entry in enumerate(self.regex_filters): + message += f"{i}. `{entry['pattern']}` - {'āœ… Enabled' if entry['enabled'] else 'āŒ Disabled'}\n" + message += "\nUse `.toggle ` to enable/disable a regex." + await ctx.send(message) + + @commands.hybrid_command() + @commands.has_role(ADMIN_ROLE_ID) + async def toggle(self, ctx, index: int): + """Toggles a regex pattern.""" + real_index = index - 1 + if 0 <= real_index < len(self.regex_filters): + self.regex_filters[real_index]["enabled"] = not self.regex_filters[real_index]["enabled"] + save_json(FILTER_FILE, self.regex_filters) + await ctx.send( + f"Toggled regex `{self.regex_filters[real_index]['pattern']}` to {'enabled' if self.regex_filters[real_index]['enabled'] else 'disabled'}.") + else: + await ctx.send("Invalid index. Use the number shown in `.listregex`.") + + @commands.hybrid_command() + @commands.has_role(ADMIN_ROLE_ID) + async def unblock(self, ctx): + """Unblocks a regex pattern.""" + if not self.regex_filters: + return await ctx.send("No regex patterns to remove.") + + await ctx.send("Please enter the index of the regex to remove (as shown in `.listregex`):") + + def check(msg): + return msg.author == ctx.author and msg.channel == ctx.channel + + try: + msg = await self.bot.wait_for("message", check=check, timeout=60) + index = int(msg.content) + real_index = index - 1 + removed = self.regex_filters.pop(real_index) + save_json(FILTER_FILE, self.regex_filters) + await ctx.send(f"Removed regex `{removed['pattern']}`") + except (ValueError, IndexError): + await ctx.send("Invalid index.") + except asyncio.TimeoutError: + await ctx.send("Timeout. Please try again.") + + @staticmethod # IDK if this is dirty fix, it just works + def is_admin_quarantine(): + def predicate(ctx): + return any(role.id == ADMIN_ROLE_ID for role in ctx.author.roles) + + return commands.check(predicate) + + @commands.hybrid_command() + @is_admin_quarantine() + async def q(self, ctx, member: discord.Member, duration: str, *, reason: str = "No reason provided"): + """Quarantine a member for a duration (e.g. 10m, 1h, 1d).""" + quarantine_role = ctx.guild.get_role(QUARANTINE_ROLE_ID) + if quarantine_role in member.roles: + await ctx.send(f"{member.display_name} is already quarantined.") + return + + try: + amount = int(duration[:-1]) + unit = duration[-1].lower() + if unit == 'm': + delta = timedelta(minutes=amount) + elif unit == 'h': + delta = timedelta(hours=amount) + elif unit == 'd': + delta = timedelta(days=amount) + else: + await ctx.send("Invalid duration format. Use m (minutes), h (hours), or d (days).") + return + except Exception as e: + await ctx.send( + f"Invalid duration format. Use m (minutes), h (hours), or d (days). Example: 10m, 1h, 2d {e}") + return + + await member.add_roles(quarantine_role, reason=f"Quarantine by {ctx.author} for {reason}") + unquarantine_time = datetime.now(timezone.utc) + delta + self.active_quarantines[str(member.id)] = unquarantine_time.isoformat() + save_json(QUARANTINE_DATA_FILE, self.active_quarantines) + + await ctx.send(f"{member.display_name} has been quarantined for {duration}. Reason: {reason}") + + log_entry = f"{ctx.author} quarantined {member} for {duration}. Reason: {reason}" + log_to_file(log_entry, QUARANTINE_LOG_FILE) + + log_channel = ctx.guild.get_channel(REPORT_LOG_CHANNEL_ID) + if log_channel: + embed = discord.Embed( + title="User Quarantined", + color=discord.Color.orange(), + timestamp=datetime.now(timezone.utc), + ) + embed.add_field(name="User", value=member.mention, inline=True) + embed.add_field(name="By", value=ctx.author.mention, inline=True) + embed.add_field(name="Duration", value=duration, inline=True) + embed.add_field(name="Reason", value=reason, inline=False) + await log_channel.send(embed=embed) + + @commands.hybrid_command() + @is_admin_quarantine() + async def uq(self, ctx, member: discord.Member, *, reason: str = "No reason provided"): + """Unquarantine a member immediately.""" + quarantine_role = ctx.guild.get_role(QUARANTINE_ROLE_ID) + if quarantine_role not in member.roles: + await ctx.send(f"{member.display_name} is not quarantined.") + return + + await member.remove_roles(quarantine_role, reason=f"Unquarantined by {ctx.author} for {reason}") + self.active_quarantines.pop(str(member.id), None) + save_json(QUARANTINE_DATA_FILE, self.active_quarantines) + + await ctx.send(f"{member.display_name} has been unquarantined. Reason: {reason}") + + log_entry = f"{ctx.author} unquarantined {member}. Reason: {reason}" + log_to_file(log_entry, QUARANTINE_LOG_FILE) + + log_channel = ctx.guild.get_channel(REPORT_LOG_CHANNEL_ID) + if log_channel: + embed = discord.Embed( + title="User Unquarantined", + color=discord.Color.green(), + timestamp=datetime.now(timezone.utc), + ) + embed.add_field(name="User", value=member.mention, inline=True) + embed.add_field(name="By", value=ctx.author.mention, inline=True) + embed.add_field(name="Reason", value=reason, inline=False) + await log_channel.send(embed=embed) + + @tasks.loop(seconds=60) + async def check_quarantine_expiry(self): + """Check for expired quarantines.""" + now = datetime.now(timezone.utc) + guild = self.bot.guilds[0] if self.bot.guilds else None + if not guild: + return + + quarantine_role = guild.get_role(QUARANTINE_ROLE_ID) + to_remove = [] + + for user_id_str, unq_time_str in self.active_quarantines.items(): + user_id = int(user_id_str) + unq_time = datetime.fromisoformat(unq_time_str) + if now >= unq_time: + member = guild.get_member(user_id) + if member and quarantine_role in member.roles: + try: + await member.remove_roles(quarantine_role, reason="Automatic quarantine expiry") + except Exception as e: + print(f"Error removing quarantine role from {member}: {e}") + + log_entry = f"Automatic unquarantine for {member} (quarantine expired)." + log_to_file(log_entry, QUARANTINE_LOG_FILE) + + log_channel = guild.get_channel(REPORT_LOG_CHANNEL_ID) + if log_channel: + embed = discord.Embed( + title="Quarantine Expired", + color=discord.Color.blue(), + timestamp=datetime.now(timezone.utc), + ) + embed.add_field(name="User", value=member.mention) + embed.add_field(name="Reason", value="Quarantine time expired") + await log_channel.send(embed=embed) + + to_remove.append(user_id_str) + + for user_id_str in to_remove: + self.active_quarantines.pop(user_id_str) + if to_remove: + save_json(QUARANTINE_DATA_FILE, self.active_quarantines) + + @commands.Cog.listener() + async def on_message(self, message): + """Handle message filtering and report collection.""" + if message.author.bot: + return + + content = clean_message_content(message.content) + message.content = content + + for entry in self.regex_filters: + if entry["enabled"]: + try: + if re.search(entry["pattern"], content): + await message.delete() + try: + await message.author.send( + f"šŸš« Your message was not allowed:\n`{content}`\n(Reason: Matches blocked pattern)" + ) + except discord.Forbidden: + pass + + log_channel = self.bot.get_channel(REPORT_LOG_CHANNEL_ID) + if log_channel: + timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC") + await log_channel.send( + f"šŸšØ **Blocked Message**\n" + f"**User:** {message.author.mention} (`{message.author.id}`)\n" + f"**Message:** `{content}`\n" + f"**Time:** {timestamp}" + ) + return + except re.error: + continue + + if isinstance(message.channel, discord.DMChannel): + user_id = message.author.id + if user_id in self.active_reports and not content.startswith("."): + self.active_reports[user_id].append(content) + return + + +async def setup(bot): + await bot.add_cog(Moderation(bot)) \ No newline at end of file diff --git a/cogs/utility.py b/cogs/utility.py new file mode 100644 index 0000000..517ff32 --- /dev/null +++ b/cogs/utility.py @@ -0,0 +1,79 @@ +import discord +from discord.ext import commands +import platform +import time + +from config import BOT_VERSION +from utils.helpers import get_uptime, load_replies, clean_message_content + +start_time = time.time() + + +class Utility(commands.Cog): + def __init__(self, bot): + self.bot = bot + self.replies = load_replies() + + @commands.hybrid_command() + async def status(self, ctx): + """Displays the bot's status and environment information""" + os_info = platform.system() + release = platform.release() + architecture = platform.machine() + python_version = platform.python_version() + uptime = get_uptime(start_time) + + env_message = ( + f"āœ… Bot is running in **{os_info} {release} ({architecture})** environment " + f"with **Python {python_version}**\n" + f"šŸ›  Version: **{BOT_VERSION}**\n" + f"ā± Uptime: **{uptime}**" + ) + if "b" in BOT_VERSION.lower(): + env_message += "\nāš ļø Beta version detected ā€“ may be unstable!" + await ctx.send(env_message) + + @commands.hybrid_command() + async def pin(self, ctx): + """Pins a message that you reply to""" + if not ctx.message.reference: + await ctx.send("Please reply to the message you want to pin.") + return + + try: + msg = await ctx.channel.fetch_message(ctx.message.reference.message_id) + await msg.pin() + await ctx.send("šŸ“Œ Message pinned.") + except discord.Forbidden: + await ctx.send("I don't have permission to pin messages in this channel.") + except discord.HTTPException as e: + await ctx.send(f"Failed to pin message: {e}") + + @commands.hybrid_command(name="replies") + async def replies_command(self, ctx): + """Lists all available quick replies""" + if not self.replies: + await ctx.send("āš ļø No replies available.") + return + + response = "\n".join([f"* {key}: {val[0]}" for key, val in self.replies.items()]) + await ctx.send(f"```\n{response}\n```") + + @commands.Cog.listener() + async def on_message(self, message): + """Handle message filtering and report collection.""" + if message.author.bot: + return + + content = clean_message_content(message.content) + message.content = content + + if content.startswith("."): + cmd = content[1:] + if cmd in self.replies: + await message.channel.send(self.replies[cmd][1]) + return + + +async def setup(bot): + await bot.add_cog(Utility(bot)) \ No newline at end of file diff --git a/config.py b/config.py new file mode 100644 index 0000000..0ef7c9c --- /dev/null +++ b/config.py @@ -0,0 +1,59 @@ +from discord import Intents +import argparse +import math + +BOT_VERSION: str = "2.1.5" +DISABLED_IN_BETA: set[str] = {"slowmode", "q", "uq"} + +MATH_CONSTANTS: dict[str, float] = { + 'pi': math.pi, + 'e': math.e, + 'tau': math.tau, + 'inf': math.inf, + 'nan': math.nan, + 'Ļ€': math.pi, + 'Ļ„': math.tau, + 'Ļ†': (1 + math.sqrt(5)) / 2, + 'gamma': 0.57721566490153286060, + 'c': 299792458, + 'G': 6.67430e-11, + 'h': 6.62607015e-34, +} + +WELCOME_CHANNEL_ID: int = 1433809691422494770 +BOT_CONSOLE_CHANNEL_ID: int = 1433808437657075903 +COUNTING_REPORT_CHANNEL_ID: int = 1208782728083013673 +COUNTING_CHANNEL_ID: int = 1433783925414563941 +REPORT_LOG_CHANNEL_ID: int = 1433783978359259258 +ADMIN_ROLE_ID: int = 1433784069954474048 +QUARANTINE_ROLE_ID: int = 1433784445164322837 +MEMBER_ROLE_ID: int = 1433810924174704701 + +intents = Intents.default() +intents.messages = True +intents.message_content = True +intents.guilds = True +intents.dm_messages = True +intents.members = True + +COUNT_FILE: str = "data/count_data.json" +REPORTS_FILE: str = "data/reports.json" +FILTER_FILE: str = "data/filters.json" +DEVICES_FILE: str = "data/devices.json" +BACKUP_FILE: str = "data/message_backups.json" +QUARANTINE_DATA_FILE: str = "data/quarantine_data.json" +QUARANTINE_LOG_FILE: str = "data/quarantine_log.txt" + +REPO_URL: str = "https://github.com/FreeXR/FreeXR-Bot.git" +REPO_DIR: str = "FreeXR-Bot" +REPLIES_DIR: str = "quick_replies" + +parser = argparse.ArgumentParser(description="FreeXR Bot") +parser.add_argument("-t", "--token", type=str, help="Discord bot token") +args = parser.parse_args() + +if args.token: + TOKEN: str = args.token +else: + with open("token", "r") as file: + TOKEN: str = file.read().strip() \ No newline at end of file diff --git a/config_real.py b/config_real.py new file mode 100644 index 0000000..1a3c256 --- /dev/null +++ b/config_real.py @@ -0,0 +1,59 @@ +from discord import Intents +import argparse +import math + +BOT_VERSION: str = "2.1.6" +DISABLED_IN_BETA: set[str] = {"slowmode", "q", "uq"} + +MATH_CONSTANTS: dict[str, float] = { + 'pi': math.pi, + 'e': math.e, + 'tau': math.tau, + 'inf': math.inf, + 'nan': math.nan, + 'Ļ€': math.pi, + 'Ļ„': math.tau, + 'Ļ†': (1 + math.sqrt(5)) / 2, + 'gamma': 0.57721566490153286060, + 'c': 299792458, + 'G': 6.67430e-11, + 'h': 6.62607015e-34, +} + +WELCOME_CHANNEL_ID: int = 1348562119469305958 +BOT_CONSOLE_CHANNEL_ID: int = 1376528272204103721 +COUNTING_REPORT_CHANNEL_ID: int = 1348562119469305958 # IDK, but maybe it'll be better if just be same as COUNTING_CHANNEL_ID, but then how to reply? +COUNTING_CHANNEL_ID: int = 1374296035798814804 +REPORT_LOG_CHANNEL_ID: int = 1361285583195869255 +ADMIN_ROLE_ID: int = 1376159693021646900 +QUARANTINE_ROLE_ID: int = 1373608273306976276 +MEMBER_ROLE_ID: int = 1406195112714829899 # IDK what role this really is + +intents = Intents.default() +intents.messages = True +intents.message_content = True +intents.guilds = True +intents.dm_messages = True +intents.members = True + +COUNT_FILE: str = "data/count_data.json" +REPORTS_FILE: str = "data/reports.json" +FILTER_FILE: str = "data/filters.json" +DEVICES_FILE: str = "data/devices.json" +BACKUP_FILE: str = "data/message_backups.json" +QUARANTINE_DATA_FILE: str = "data/quarantine_data.json" +QUARANTINE_LOG_FILE: str = "data/quarantine_log.txt" + +REPO_URL: str = "https://github.com/FreeXR/FreeXR-Bot.git" +REPO_DIR: str = "FreeXR-Bot" +REPLIES_DIR: str = "quick_replies" + +parser = argparse.ArgumentParser(description="FreeXR Bot") +parser.add_argument("-t", "--token", type=str, help="Discord bot token") +args = parser.parse_args() + +if args.token: + TOKEN: str = args.token +else: + with open("token", "r") as file: + TOKEN: str = file.read().strip() \ No newline at end of file diff --git a/main.py b/main.py new file mode 100644 index 0000000..bb1c633 --- /dev/null +++ b/main.py @@ -0,0 +1,11 @@ +import asyncio +from bot import bot +from config import TOKEN + +async def main(): + async with bot: + await bot.start(TOKEN) + + +if __name__ == "__main__": + asyncio.run(main()) \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..2ac1ed7 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,4 @@ +discord.py~=2.6.4 +GitPython~=3.1.45 +requests~=2.32.5 +arithmeval~=0.3.4 \ No newline at end of file diff --git a/utils/__init__.py b/utils/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/utils/file_handlers.py b/utils/file_handlers.py new file mode 100644 index 0000000..d0fbf38 --- /dev/null +++ b/utils/file_handlers.py @@ -0,0 +1,21 @@ +import json +from pathlib import Path + + +def load_json(file_path, default=None): + if default is None: + default = {} + + file_path = Path(file_path) + if file_path.exists(): + with open(file_path, "r", encoding="utf-8") as f: + return json.load(f) + return default + + +def save_json(file_path, data): + file_path = Path(file_path) + file_path.parent.mkdir(parents=True, exist_ok=True) + + with open(file_path, "w", encoding="utf-8") as f: + json.dump(data, f, indent=2, ensure_ascii=False) \ No newline at end of file diff --git a/utils/helpers.py b/utils/helpers.py new file mode 100644 index 0000000..c856c9f --- /dev/null +++ b/utils/helpers.py @@ -0,0 +1,44 @@ +import os +import time +from datetime import datetime, timezone +from config import REPLIES_DIR + + +def get_uptime(start_time: float) -> str: + seconds = int(time.time() - start_time) + days, seconds = divmod(seconds, 86400) + hours, seconds = divmod(seconds, 3600) + minutes, seconds = divmod(seconds, 60) + return f"{days}d {hours}h {minutes}m {seconds}s" + + +def load_replies(): + replies = {} + + if not os.path.exists(REPLIES_DIR): + return replies + + for filename in os.listdir(REPLIES_DIR): + if filename.endswith(".md"): + filepath = os.path.join(REPLIES_DIR, filename) + with open(filepath, "r", encoding="utf-8") as f: + content = f.read().split("---") + if len(content) >= 2: + summary_line = content[0].strip().splitlines()[0] + reply_text = content[1].strip() + command_name = filename[:-3] + replies[command_name] = (summary_line, reply_text) + return replies + + +def log_to_file(entry: str, log_file: str): + with open(log_file, "a", encoding="utf-8") as f: + f.write(f"{datetime.now(timezone.utc).isoformat()} - {entry}\n") + + +def clean_message_content(content: str) -> str: + invisible_chars = ["\u200b", "\u200c", "\u200d", "\u200e", "\u200f"] + cleaned_content = content + for ch in invisible_chars: + cleaned_content = cleaned_content.replace(ch, "") + return cleaned_content.strip() \ No newline at end of file diff --git a/utils/logger.py b/utils/logger.py new file mode 100644 index 0000000..93e8541 --- /dev/null +++ b/utils/logger.py @@ -0,0 +1,32 @@ +import asyncio + + +class DiscordConsoleLogger: + def __init__(self, bot, channel_id): + self.bot = bot + self.channel_id = channel_id + self.buffer = "" + self.lock = asyncio.Lock() + + def write(self, message): + self.buffer += message + if '\n' in self.buffer: + lines = self.buffer.split('\n') + self.buffer = lines[-1] + for line in lines[:-1]: + if line.strip(): + asyncio.ensure_future(self.send_to_discord(line.strip())) + + def flush(self): + if self.buffer.strip(): + asyncio.ensure_future(self.send_to_discord(self.buffer.strip())) + self.buffer = "" + + async def send_to_discord(self, message): + async with self.lock: + channel = self.bot.get_channel(self.channel_id) + if channel: + try: + await channel.send(f"```\n{message[:1900].replace('```', '')}\n```") + except Exception: # Too broad exception clause + pass \ No newline at end of file From cdd3d8cedc45cf4b5417e8f22b7bb6c49d7b3af0 Mon Sep 17 00:00:00 2001 From: idk-maybe-none Date: Fri, 31 Oct 2025 17:11:24 +0200 Subject: [PATCH 2/3] Replace config, and delete testing config Oops, not that config, but atleast its not my testing bot token... --- config.py | 18 +++++++-------- config_real.py | 59 -------------------------------------------------- 2 files changed, 9 insertions(+), 68 deletions(-) delete mode 100644 config_real.py diff --git a/config.py b/config.py index 0ef7c9c..1a3c256 100644 --- a/config.py +++ b/config.py @@ -2,7 +2,7 @@ import argparse import math -BOT_VERSION: str = "2.1.5" +BOT_VERSION: str = "2.1.6" DISABLED_IN_BETA: set[str] = {"slowmode", "q", "uq"} MATH_CONSTANTS: dict[str, float] = { @@ -20,14 +20,14 @@ 'h': 6.62607015e-34, } -WELCOME_CHANNEL_ID: int = 1433809691422494770 -BOT_CONSOLE_CHANNEL_ID: int = 1433808437657075903 -COUNTING_REPORT_CHANNEL_ID: int = 1208782728083013673 -COUNTING_CHANNEL_ID: int = 1433783925414563941 -REPORT_LOG_CHANNEL_ID: int = 1433783978359259258 -ADMIN_ROLE_ID: int = 1433784069954474048 -QUARANTINE_ROLE_ID: int = 1433784445164322837 -MEMBER_ROLE_ID: int = 1433810924174704701 +WELCOME_CHANNEL_ID: int = 1348562119469305958 +BOT_CONSOLE_CHANNEL_ID: int = 1376528272204103721 +COUNTING_REPORT_CHANNEL_ID: int = 1348562119469305958 # IDK, but maybe it'll be better if just be same as COUNTING_CHANNEL_ID, but then how to reply? +COUNTING_CHANNEL_ID: int = 1374296035798814804 +REPORT_LOG_CHANNEL_ID: int = 1361285583195869255 +ADMIN_ROLE_ID: int = 1376159693021646900 +QUARANTINE_ROLE_ID: int = 1373608273306976276 +MEMBER_ROLE_ID: int = 1406195112714829899 # IDK what role this really is intents = Intents.default() intents.messages = True diff --git a/config_real.py b/config_real.py deleted file mode 100644 index 1a3c256..0000000 --- a/config_real.py +++ /dev/null @@ -1,59 +0,0 @@ -from discord import Intents -import argparse -import math - -BOT_VERSION: str = "2.1.6" -DISABLED_IN_BETA: set[str] = {"slowmode", "q", "uq"} - -MATH_CONSTANTS: dict[str, float] = { - 'pi': math.pi, - 'e': math.e, - 'tau': math.tau, - 'inf': math.inf, - 'nan': math.nan, - 'Ļ€': math.pi, - 'Ļ„': math.tau, - 'Ļ†': (1 + math.sqrt(5)) / 2, - 'gamma': 0.57721566490153286060, - 'c': 299792458, - 'G': 6.67430e-11, - 'h': 6.62607015e-34, -} - -WELCOME_CHANNEL_ID: int = 1348562119469305958 -BOT_CONSOLE_CHANNEL_ID: int = 1376528272204103721 -COUNTING_REPORT_CHANNEL_ID: int = 1348562119469305958 # IDK, but maybe it'll be better if just be same as COUNTING_CHANNEL_ID, but then how to reply? -COUNTING_CHANNEL_ID: int = 1374296035798814804 -REPORT_LOG_CHANNEL_ID: int = 1361285583195869255 -ADMIN_ROLE_ID: int = 1376159693021646900 -QUARANTINE_ROLE_ID: int = 1373608273306976276 -MEMBER_ROLE_ID: int = 1406195112714829899 # IDK what role this really is - -intents = Intents.default() -intents.messages = True -intents.message_content = True -intents.guilds = True -intents.dm_messages = True -intents.members = True - -COUNT_FILE: str = "data/count_data.json" -REPORTS_FILE: str = "data/reports.json" -FILTER_FILE: str = "data/filters.json" -DEVICES_FILE: str = "data/devices.json" -BACKUP_FILE: str = "data/message_backups.json" -QUARANTINE_DATA_FILE: str = "data/quarantine_data.json" -QUARANTINE_LOG_FILE: str = "data/quarantine_log.txt" - -REPO_URL: str = "https://github.com/FreeXR/FreeXR-Bot.git" -REPO_DIR: str = "FreeXR-Bot" -REPLIES_DIR: str = "quick_replies" - -parser = argparse.ArgumentParser(description="FreeXR Bot") -parser.add_argument("-t", "--token", type=str, help="Discord bot token") -args = parser.parse_args() - -if args.token: - TOKEN: str = args.token -else: - with open("token", "r") as file: - TOKEN: str = file.read().strip() \ No newline at end of file From da4f737fa60883f4dd923a1bab6e70aced327b97 Mon Sep 17 00:00:00 2001 From: idk-maybe-none Date: Mon, 3 Nov 2025 20:00:00 +0200 Subject: [PATCH 3/3] Use expr.py & improve JSON loading self-explanatory --- README.md | 2 +- cogs/counting.py | 28 +++++++++++----------------- requirements.txt | 4 ++-- utils/file_handlers.py | 11 +++++++---- 4 files changed, 21 insertions(+), 24 deletions(-) diff --git a/README.md b/README.md index 77b12d8..e5ec41e 100644 --- a/README.md +++ b/README.md @@ -14,7 +14,7 @@ FreeXR Discord Bot - **Fixed`.replies_cmd` command** (replaced with a single `replies` hybrid command) - **Counting now support arithmetics**, (and SPOILER: it's not just eval()) - Bot now uses arithmeval, python library that gives bot ability to do basic math ( and not only ) - - Now you can use Addition (+), True division (/), Floor division (//), Modulus (%), Multiplication (*), Exponentiation (**), Subtraction (-), Logical AND (and), Logical OR (or), Logical NOT (not), and even pi (Ļ€), e, tau (Ļ„), the golden ratio (Ļ†), Euler-Mascheroni's gamma (gamma), the speed of light (c), Newton's gravitational constant (G), and Planck's constant (h). (IDK why you would use them) + - Now you can use Addition (+), True division (/), Floor division (//), Modulus (%), Multiplication (*), Exponentiation (^), Subtraction (-), and even pi (Ļ€), e, tau (Ļ„), the golden ratio (Ļ†), Euler-Mascheroni's gamma (gamma), the speed of light (c), Newton's gravitational constant (G), and Planck's constant (h). (IDK why you would use them) #### Device Management - **New persistent device system**: diff --git a/cogs/counting.py b/cogs/counting.py index 463e58d..98c0a9f 100644 --- a/cogs/counting.py +++ b/cogs/counting.py @@ -2,7 +2,7 @@ from utils.file_handlers import load_json, save_json from config import COUNTING_CHANNEL_ID, COUNTING_REPORT_CHANNEL_ID, MATH_CONSTANTS, COUNT_FILE -from arithmetic_eval import evaluate +from expr import evaluate, errors class Counting(commands.Cog): @@ -23,7 +23,7 @@ async def on_message(self, message): content = message.content try: - number = evaluate(content, MATH_CONSTANTS) + number = evaluate(content, variables=MATH_CONSTANTS) if number == current_count + 1 and message.author.id != last_counter_id: data["current_count"] = number @@ -38,21 +38,15 @@ async def on_message(self, message): else: reason = "Unknown error" - except (ValueError, SyntaxError, ZeroDivisionError, TypeError) as e: # TODO: Refactor entire file because I'm not sure it even works - if content.isdigit() and "\n" not in content: - number = int(content) - if number == current_count + 1 and message.author.id != last_counter_id: - data["current_count"] = number - data["last_counter_id"] = message.author.id - save_json(COUNT_FILE, data) - return - else: - if message.author.id == last_counter_id: - reason = "Double counting - same user consecutively" - else: - reason = f"Incorrect number - expected {current_count + 1}, got {number}" - else: - reason = f"Invalid mathematical expression: {str(e)}" + except errors.ParsingError as e: + """ + TODO: Ignore if error is Gibberish, InvalidSyntax, BadOperation because they all may mean that it's not math expression, or it's just: 2**2 instead of 2^2 + TODO: Maybe implement custom error handler so instead of Invalid mathematical expression: Overflow, it says: Don't try to break this, and maybe even mutes for 15 minutes because of that + (But first ask someone, it's just my ideas of improving it) + """ + reason = f"Invalid mathematical expression: {e.friendly}" + except Exception as e: + reason = f"Unknown error: {e}" data["current_count"] = 0 data["last_counter_id"] = None diff --git a/requirements.txt b/requirements.txt index 2ac1ed7..445d40b 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,4 @@ discord.py~=2.6.4 -GitPython~=3.1.45 requests~=2.32.5 -arithmeval~=0.3.4 \ No newline at end of file +GitPython~=3.1.45 +expr-py~=0.3.0 \ No newline at end of file diff --git a/utils/file_handlers.py b/utils/file_handlers.py index d0fbf38..027d093 100644 --- a/utils/file_handlers.py +++ b/utils/file_handlers.py @@ -6,10 +6,13 @@ def load_json(file_path, default=None): if default is None: default = {} - file_path = Path(file_path) - if file_path.exists(): - with open(file_path, "r", encoding="utf-8") as f: - return json.load(f) + try: + file_path = Path(file_path) + if file_path.exists(): + with open(file_path, "r", encoding="utf-8") as f: + return json.load(f) + except json.JSONDecodeError: + return default return default