diff --git a/commands/bonk.py b/commands/bonk.py index 85fe6db..6e1bdb5 100644 --- a/commands/bonk.py +++ b/commands/bonk.py @@ -19,7 +19,7 @@ class Bonk(commands.Cog): return if user.id == interaction.user.id: - await interaction.followup.send("You cannot bonk yourself", ephemeral=True) + await interaction.followup.send("You cannot bonk yourself silly", ephemeral=False) return file = discord.File(self.bonk_path, filename="bonk.png") diff --git a/commands/minesweeper.py b/commands/minesweeper.py new file mode 100644 index 0000000..c1fb114 --- /dev/null +++ b/commands/minesweeper.py @@ -0,0 +1,448 @@ +import discord +from discord.ext import commands +from discord import app_commands +import random +import asyncio +import re + +class MinesweeperGame: + def __init__(self, rows: int = 13, cols: int = 13, mines: int = 20): + self.rows = rows + self.cols = cols + self.mine_count = mines + + # Board: -1 = mine, 0-8 = number of adjacent mines + self.board = [[0 for _ in range(cols)] for _ in range(rows)] + # Revealed: False = hidden, True = revealed + self.revealed = [[False for _ in range(cols)] for _ in range(rows)] + # Flags: False = no flag, True = flagged + self.flags = [[False for _ in range(cols)] for _ in range(rows)] + + self.game_over = False + self.won = False + self.first_move = True + + # Column number emojis (11-13 use custom bot emojis) + self.col_emojis = [ + "1️⃣", "2️⃣", "3️⃣", "4️⃣", "5️⃣", "6️⃣", "7️⃣", "8️⃣", "9️⃣", "🔟", + "<:11:1470841923785719952>", + "<:12:1470841922716172299>", + "<:13:1470841927409729600>" + ] + + def setup_board(self, safe_row: int, safe_col: int): + """Generate mines avoiding the first click position""" + positions = [(r, c) for r in range(self.rows) for c in range(self.cols) + if not (abs(r - safe_row) <= 1 and abs(c - safe_col) <= 1)] + + mine_positions = random.sample(positions, self.mine_count) + + for r, c in mine_positions: + self.board[r][c] = -1 + + # Calculate numbers + for r in range(self.rows): + for c in range(self.cols): + if self.board[r][c] != -1: + count = 0 + for dr in [-1, 0, 1]: + for dc in [-1, 0, 1]: + if dr == 0 and dc == 0: + continue + nr, nc = r + dr, c + dc + if 0 <= nr < self.rows and 0 <= nc < self.cols: + if self.board[nr][nc] == -1: + count += 1 + self.board[r][c] = count + + def reveal(self, row: int, col: int) -> bool: + """Reveal a cell. Returns True if hit a mine.""" + if self.first_move: + self.setup_board(row, col) + self.first_move = False + + if self.revealed[row][col] or self.flags[row][col]: + return False + + self.revealed[row][col] = True + + if self.board[row][col] == -1: + self.game_over = True + return True + + # Flood fill: Auto-reveal adjacent cells if it's a 0 + if self.board[row][col] == 0: + self._flood_fill(row, col) + + # Check win condition + if self.check_win(): + self.game_over = True + self.won = True + + return False + + def _flood_fill(self, start_row: int, start_col: int): + """ + Flood fill algorithm to reveal all connected empty cells and their borders. + Reveals all 0-cells connected to the starting position, plus the numbered + cells that border them (the 'edge' of the empty region). + """ + from collections import deque + + queue = deque([(start_row, start_col)]) + visited = set() + visited.add((start_row, start_col)) + + while queue: + row, col = queue.popleft() + + # Check all 8 adjacent cells + for dr in [-1, 0, 1]: + for dc in [-1, 0, 1]: + if dr == 0 and dc == 0: + continue + + nr, nc = row + dr, col + dc + + # Skip if out of bounds or already visited + if not (0 <= nr < self.rows and 0 <= nc < self.cols): + continue + if (nr, nc) in visited: + continue + + # Skip if flagged + if self.flags[nr][nc]: + continue + + # Mark as visited + visited.add((nr, nc)) + + # Reveal the cell + self.revealed[nr][nc] = True + + # If it's also a 0, add it to the queue to continue flood fill + if self.board[nr][nc] == 0: + queue.append((nr, nc)) + + def toggle_flag(self, row: int, col: int): + """Toggle flag on a cell""" + if self.revealed[row][col]: + return + self.flags[row][col] = not self.flags[row][col] + + def check_win(self) -> bool: + """Check if all non-mine cells are revealed""" + for r in range(self.rows): + for c in range(self.cols): + if self.board[r][c] != -1 and not self.revealed[r][c]: + return False + return True + + def get_cell_display(self, row: int, col: int, show_all: bool = False) -> str: + """Get the display string for a cell""" + if show_all and self.board[row][col] == -1: + return "💣" + + if self.flags[row][col]: + return "🚩" + + if not self.revealed[row][col]: + return "⬛" + + if self.board[row][col] == -1: + return "💥" + + num_to_emoji = { + 0: "⬜", + 1: "1️⃣", + 2: "2️⃣", + 3: "3️⃣", + 4: "4️⃣", + 5: "5️⃣", + 6: "6️⃣", + 7: "7️⃣", + 8: "8️⃣" + } + return num_to_emoji.get(self.board[row][col], "⬜") + + def render_board(self) -> str: + """Render the board as a string""" + # Column headers using numbers + col_headers = "⬛" + "".join(self.col_emojis[:self.cols]) + + lines = [col_headers] + for r in range(self.rows): + # Row headers using letters (A-M) + row_header = chr(0x1F1E6 + r) # Regional indicator A-M + row_str = row_header + "".join( + self.get_cell_display(r, c, self.game_over) + for c in range(self.cols) + ) + lines.append(row_str) + + return "\n".join(lines) + + def get_stats(self) -> str: + """Get game statistics""" + total_flags = sum(sum(row) for row in self.flags) + flags_remaining = self.mine_count - total_flags + cells_remaining = sum( + 1 for r in range(self.rows) + for c in range(self.cols) + if not self.revealed[r][c] and self.board[r][c] != -1 + ) + + return f"🚩 Flags: {total_flags}/{self.mine_count} | 💣 Mines: {self.mine_count} | ⬛ Cells left: {cells_remaining}" + + +class MinesweeperView(discord.ui.View): + def __init__(self, player: discord.Member, game: MinesweeperGame): + super().__init__(timeout=600) + self.player = player + self.game = game + self.message: discord.Message = None + + @discord.ui.button(label="Forfeit", style=discord.ButtonStyle.danger, emoji="🏳️") + async def forfeit(self, interaction: discord.Interaction, button: discord.ui.Button): + if interaction.user.id != self.player.id: + await interaction.response.send_message("Only the player can forfeit!", ephemeral=True) + return + + self.game.game_over = True + for child in self.children: + child.disabled = True + self.stop() + + embed = self.create_embed() + embed.color = discord.Color.red() + embed.title = "💀 Game Over - Forfeited" + + await interaction.response.edit_message(embed=embed, view=self) + + @discord.ui.button(label="How to Play", style=discord.ButtonStyle.secondary, emoji="❓") + async def help_button(self, interaction: discord.Interaction, button: discord.ui.Button): + help_embed = discord.Embed( + title="🎮 How to Play Minesweeper", + description="Send messages in this channel to make moves!", + color=discord.Color.blue() + ) + help_embed.add_field( + name="📝 Move Format", + value="```[column] [row] [flag]```", + inline=False + ) + help_embed.add_field( + name="Examples", + value=( + "• `1 A` - Reveal cell at column 1, row A\n" + "• `6 B F` - Flag cell at column 6, row B\n" + "• `7D FLAG` - Flag cell at column 7, row D\n" + "• `2Hf` - Flag cell at column 2, row H" + ), + inline=False + ) + help_embed.add_field( + name="📌 Notes", + value=( + "• Column: 1-13 (numbers)\n" + "• Row: A-M (letters)\n" + "• Flag: f, flag, or leave blank\n" + "• Inputs are NOT case sensitive!\n" + "• Your messages will be auto-deleted" + ), + inline=False + ) + + await interaction.response.send_message(embed=help_embed, ephemeral=True) + + def create_embed(self) -> discord.Embed: + """Create the game embed""" + if self.game.game_over: + if self.game.won: + embed = discord.Embed( + title="🎉 You Win!", + description=f"{self.player.mention} successfully cleared the minefield!", + color=discord.Color.green() + ) + else: + embed = discord.Embed( + title="💥 Game Over!", + description=f"{self.player.mention} hit a mine!", + color=discord.Color.red() + ) + else: + embed = discord.Embed( + title="💣 Minesweeper", + description=f"{self.player.mention}'s game", + color=discord.Color.blue() + ) + + board = self.game.render_board() + # Discord has a 4096 character limit for embed descriptions + # Split board if needed + if len(board) < 1900: + embed.add_field(name="Board", value=board, inline=False) + else: + # Split into chunks + lines = board.split('\n') + chunk_size = len(lines) // 2 + chunk1 = '\n'.join(lines[:chunk_size]) + chunk2 = '\n'.join(lines[chunk_size:]) + embed.add_field(name="Board (1/2)", value=chunk1, inline=False) + embed.add_field(name="Board (2/2)", value=chunk2, inline=False) + + stats = self.game.get_stats() + embed.add_field(name="Stats", value=stats, inline=False) + + if not self.game.game_over: + embed.set_footer(text="Send your move in chat: [column number] [row letter] [flag]") + + return embed + + async def on_timeout(self): + self.game.game_over = True + for child in self.children: + child.disabled = True + + if self.message: + embed = self.create_embed() + embed.color = discord.Color.greyple() + embed.title = "⏱️ Game Timed Out" + try: + await self.message.edit(embed=embed, view=self) + except: + pass + + +class Minesweeper(commands.Cog): + def __init__(self, bot): + self.bot = bot + self.active_games = {} # channel_id: (player_id, MinesweeperView) + + @app_commands.command(name="minesweeper", description="Start a game of Minesweeper") + @app_commands.describe( + difficulty="Choose difficulty level (Easy: 13x13 with 20 mines, Medium: 13x13 with 35 mines, Hard: 13x13 with 50 mines)" + ) + @app_commands.choices(difficulty=[ + app_commands.Choice(name="Easy (20 mines)", value=1), + app_commands.Choice(name="Medium (35 mines)", value=2), + app_commands.Choice(name="Hard (50 mines)", value=3) + ]) + async def minesweeper(self, interaction: discord.Interaction, difficulty: int = 1): + # Check if there's already an active game in this channel + if interaction.channel_id in self.active_games: + player_id, _ = self.active_games[interaction.channel_id] + await interaction.response.send_message( + f"There's already an active game in this channel! Please wait for it to finish.", + ephemeral=True + ) + return + + # Set difficulty + mine_counts = {1: 20, 2: 35, 3: 50} + mines = mine_counts.get(difficulty, 20) + + await interaction.response.defer() + + game = MinesweeperGame(rows=13, cols=13, mines=mines) + view = MinesweeperView(interaction.user, game) + + embed = view.create_embed() + message = await interaction.followup.send(embed=embed, view=view) + view.message = message + + self.active_games[interaction.channel_id] = (interaction.user.id, view) + + # Wait for the game to finish + await view.wait() + + # Clean up + if interaction.channel_id in self.active_games: + del self.active_games[interaction.channel_id] + + @commands.Cog.listener() + async def on_message(self, message: discord.Message): + # Ignore bot messages + if message.author.bot: + return + + # Check if there's an active game in this channel + if message.channel.id not in self.active_games: + return + + player_id, view = self.active_games[message.channel.id] + + # Check if message is from the player + if message.author.id != player_id: + return + + # Check if game is over + if view.game.game_over: + return + + # Parse the message + move = self.parse_move(message.content) + + if move is None: + # Invalid format, don't delete the message - it's not an input attempt + return + + col, row, is_flag = move + + # Valid input - delete the player's message to reduce clutter + try: + await message.delete() + except: + pass + + # Make the move + if is_flag: + view.game.toggle_flag(row, col) + else: + view.game.reveal(row, col) + + # Update the embed + if view.game.game_over: + for child in view.children: + child.disabled = True + view.stop() + + embed = view.create_embed() + try: + await view.message.edit(embed=embed, view=view) + except: + pass + + def parse_move(self, text: str) -> tuple[int, int, bool] | None: + """ + Parse a move from text. + Returns (col, row, is_flag) or None if invalid. + Format: [column number] [row letter] [flag] + Examples: "1 A", "6 B F", "7D FLAG", "2Hf" + """ + text = text.upper().strip() + + # Try to match pattern with optional spaces and flag + # Pattern: number (1-13), letter (A-M), optional flag + pattern = r'(\d+)\s*([A-M])\s*(F|FLAG)?' + match = re.match(pattern, text) + + if not match: + return None + + col_str, row_letter, flag = match.groups() + + col = int(col_str) - 1 # Convert to 0-indexed + row = ord(row_letter) - ord('A') # Convert letter to 0-indexed number + + # Validate + if not (0 <= col < 13 and 0 <= row < 13): + return None + + is_flag = flag is not None + + return (col, row, is_flag) + + +async def setup(bot): + await bot.add_cog(Minesweeper(bot)) \ No newline at end of file diff --git a/lilac-tools/admin.py b/lilac-tools/admin.py index c581af9..205b4eb 100644 --- a/lilac-tools/admin.py +++ b/lilac-tools/admin.py @@ -6,15 +6,81 @@ import os import glob from typing import Optional import traceback +import re ADMIN_USER_ID = 252130669919076352 +# Security configuration +BLOCKED_PATHS = [ + '.env', + '.env.local', + '.env.production', + 'config.json', + 'secrets.json', + 'credentials.json', + '.git', + 'id_rsa', + 'id_ed25519', + '.pem', + '.key', + 'private_key', + 'secret_key', +] + +BLOCKED_PATTERNS = [ + r'.*\.env.*', + r'.*secret.*', + r'.*password.*', + r'.*token.*', + r'.*api[_-]?key.*', + r'.*private[_-]?key.*', + r'.*\.pem$', + r'.*\.key$', + r'.*/\.ssh/.*', + r'.*/\.aws/.*', +] + +BLOCKED_COMMANDS = [ + 'rm -rf', + 'dd if=', + 'mkfs', + 'fdisk', + ':(){ :|:& };:', # Fork bomb + 'chmod 777', + 'chmod -R 777', +] + def is_owner(): """Check if the user is the bot owner""" async def predicate(ctx): return ctx.author.id == ADMIN_USER_ID return commands.check(predicate) +def is_path_safe(path): + """Check if a path is safe to access""" + # Normalize the path + normalized = os.path.normpath(path) + + # Check against blocked paths + for blocked in BLOCKED_PATHS: + if blocked.lower() in normalized.lower(): + return False + + # Check against blocked patterns + for pattern in BLOCKED_PATTERNS: + if re.match(pattern, normalized.lower()): + return False + + return True + +def is_command_safe(command): + """Check if a command is safe to execute""" + command_lower = command.lower() + for blocked in BLOCKED_COMMANDS: + if blocked.lower() in command_lower: + return False + return True + class AdminCommands(commands.Cog): """Admin-only commands for database queries and terminal access""" @@ -36,6 +102,147 @@ class AdminCommands(commands.Cog): self.db_connections = {os.path.basename(f).replace('.db', ''): f for f in db_files} print(f"Discovered databases: {list(self.db_connections.keys())}") + @commands.command(name="admin_help", aliases=["adminhelp", "ahelp"]) + @is_owner() + async def admin_help(self, ctx, category: Optional[str] = None): + """Show all admin commands. Usage: !admin_help [category] + Categories: code, db, terminal, files, dev""" + + if category and category.lower() == "code": + embed = discord.Embed( + title="📝 Code Management Commands", + description="Commands for viewing and editing code files", + color=discord.Color.blue() + ) + + commands_list = [ + ("!code view ``", "View file with syntax highlighting and line numbers"), + ("!code search `` `[path]`", "Search for code patterns across files (regex supported)"), + ("!code find `` ``", "Find specific text in a file and show line numbers"), + ("!code edit `` `` `[end]`", "Edit line(s) with interactive popup editor"), + ("!code insert `` `` ``", "Insert new line at position"), + ("!code delete `` ``", "Delete specific line"), + ("!code replace `` `\"find\"` `\"replace\"` `[--all]`", "Find and replace text"), + ("!code backup ``", "Create timestamped backup"), + ("!code diff `` ``", "Compare two files"), + ] + + for cmd, desc in commands_list: + embed.add_field(name=cmd, value=desc, inline=False) + + embed.set_footer(text="Example: !code edit bot.py 42") + await ctx.send(embed=embed) + return + + elif category and category.lower() == "db": + embed = discord.Embed( + title="🗄️ Database Commands", + description="Commands for querying and managing SQLite databases", + color=discord.Color.green() + ) + + commands_list = [ + ("!dblist", "List all discovered .db files"), + ("!dbtables `[db_name]`", "List all tables in a database"), + ("!dbinfo `` ``", "Show table schema and column info"), + ("!dbquery `` ``", "Execute SELECT query (read-only)"), + ("!dbexec `` ``", "Execute UPDATE/DELETE/INSERT (write operations)"), + ] + + for cmd, desc in commands_list: + embed.add_field(name=cmd, value=desc, inline=False) + + embed.set_footer(text="Example: !dbquery mydb SELECT * FROM users LIMIT 10") + await ctx.send(embed=embed) + return + + elif category and category.lower() == "terminal": + embed = discord.Embed( + title="🖥️ Terminal Commands", + description="Commands for executing shell commands", + color=discord.Color.red() + ) + + commands_list = [ + ("!terminal ``", "Execute shell command (30s timeout, blocks dangerous commands)"), + ("!term ``", "Alias for !terminal"), + ("!sh ``", "Alias for !terminal"), + ] + + for cmd, desc in commands_list: + embed.add_field(name=cmd, value=desc, inline=False) + + embed.add_field(name="⚠️ Blocked Commands", value="rm -rf, dd, mkfs, fdisk, fork bombs, chmod 777", inline=False) + embed.set_footer(text="Example: !terminal git status") + await ctx.send(embed=embed) + return + + elif category and category.lower() == "files": + embed = discord.Embed( + title="📁 File System Commands", + description="Commands for navigating and viewing files", + color=discord.Color.purple() + ) + + commands_list = [ + ("!pwd", "Show current working directory"), + ("!ls `[path]`", "List files and directories"), + ] + + for cmd, desc in commands_list: + embed.add_field(name=cmd, value=desc, inline=False) + + embed.set_footer(text="Example: !ls cogs/") + await ctx.send(embed=embed) + return + + elif category and category.lower() == "dev": + embed = discord.Embed( + title="🔧 Development Commands", + description="Commands for testing and debugging", + color=discord.Color.orange() + ) + + commands_list = [ + ("!eval ``", "Evaluate Python code with bot context"), + ] + + for cmd, desc in commands_list: + embed.add_field(name=cmd, value=desc, inline=False) + + embed.add_field(name="Available in eval context", value="`bot`, `ctx`, `discord`, `commands`, `asyncio`, `os`, `aiosqlite`", inline=False) + embed.set_footer(text="Example: !eval await ctx.send('Hello!')") + await ctx.send(embed=embed) + return + + # Main help menu + embed = discord.Embed( + title="🛡️ Admin Commands Help", + description="All commands available to the bot owner", + color=discord.Color.gold() + ) + + categories = [ + ("📝 Code Management", "`!admin_help code`", "View, edit, search, and manage code files"), + ("🗄️ Database", "`!admin_help db`", "Query and manage SQLite databases"), + ("🖥️ Terminal", "`!admin_help terminal`", "Execute shell commands"), + ("📁 File System", "`!admin_help files`", "Navigate and view files"), + ("🔧 Development", "`!admin_help dev`", "Testing and debugging tools"), + ] + + for title, command, description in categories: + embed.add_field(name=f"{title}", value=f"{description}\nUse {command} for details", inline=False) + + embed.add_field( + name="🔒 Security Features", + value="• Blocks access to .env, secrets, keys\n• Prevents dangerous commands\n• File size limits (1MB)\n• Command timeouts (30s)", + inline=False + ) + + embed.set_footer(text=f"Bot Owner: {ADMIN_USER_ID} • All commands require owner permissions") + await ctx.send(embed=embed) + + async def _format_results(self, results, description): """Format SQL results into a readable string""" if not results: @@ -299,6 +506,11 @@ class AdminCommands(commands.Cog): @is_owner() async def terminal(self, ctx, *, command: str): """Execute a shell command. Usage: !terminal """ + # Security check + if not is_command_safe(command): + await ctx.send("❌ Command blocked for security reasons.") + return + try: # Execute command with timeout using async subprocess process = await asyncio.create_subprocess_shell( @@ -449,6 +661,731 @@ class AdminCommands(commands.Cog): await ctx.send(embed=embed) except Exception as e: await ctx.send(f"❌ Error: {str(e)}") + + @commands.group(name="code", invoke_without_command=True) + @is_owner() + async def code(self, ctx): + """Code management commands. Use !code help for more info.""" + await ctx.send("Use `!code help` to see all available code commands.") + + @code.command(name="help") + @is_owner() + async def code_help(self, ctx): + """Show all code commands""" + embed = discord.Embed( + title="📝 Code Commands", + description="All commands for viewing and editing code files", + color=discord.Color.blue() + ) + + commands_list = [ + ("**view** ``", "View file contents with syntax highlighting"), + ("**search** `` `[path]`", "Search for code patterns in files"), + ("**edit** `` `` `[end_line]`", "Edit line(s) interactively with popup editor"), + ("**insert** `` `` ``", "Insert a new line at position"), + ("**delete** `` ``", "Delete a specific line"), + ("**replace** `` `\"find\"` `\"replace\"` `[--all]`", "Find and replace text"), + ("**backup** ``", "Create timestamped backup"), + ("**diff** `` ``", "Compare two files"), + ("**find** `` ``", "Find specific text and show line numbers"), + ] + + for cmd, desc in commands_list: + embed.add_field(name=cmd, value=desc, inline=False) + + embed.set_footer(text="Example: !code edit bot.py 42") + await ctx.send(embed=embed) + + @code.command(name="view", aliases=["cat", "read"]) + @is_owner() + async def code_view(self, ctx, *, path: str): + """View file contents. Usage: !code view """ + # Security check + if not is_path_safe(path): + await ctx.send("❌ Access to this file is blocked for security reasons.") + return + + try: + if not os.path.isfile(path): + await ctx.send(f"❌ File not found: `{path}`") + return + + # Check file size + file_size = os.path.getsize(path) + if file_size > 1_000_000: # 1MB limit + await ctx.send(f"❌ File is too large ({file_size:,} bytes). Max: 1MB") + return + + loop = asyncio.get_event_loop() + + def _read(): + with open(path, 'r', encoding='utf-8') as f: + return f.read() + + try: + content = await loop.run_in_executor(None, _read) + except UnicodeDecodeError: + await ctx.send("❌ Cannot read file: binary or non-UTF-8 encoded") + return + + # Determine language for syntax highlighting + ext = os.path.splitext(path)[1].lower() + lang_map = { + '.py': 'python', + '.js': 'javascript', + '.ts': 'typescript', + '.json': 'json', + '.yml': 'yaml', + '.yaml': 'yaml', + '.md': 'markdown', + '.sh': 'bash', + '.sql': 'sql', + '.html': 'html', + '.css': 'css', + } + lang = lang_map.get(ext, '') + + # Add line numbers + lines = content.split('\n') + numbered_lines = [] + line_num_width = len(str(len(lines))) + for i, line in enumerate(lines, 1): + numbered_lines.append(f"{i:>{line_num_width}} | {line}") + numbered_content = '\n'.join(numbered_lines) + + # Handle output + if len(numbered_content) > 1900: + filename = f"file_content_{ctx.message.id}.txt" + await loop.run_in_executor(None, self._write_file, filename, numbered_content) + await ctx.send(f"📄 `{path}` ({file_size:,} bytes, {len(lines)} lines):", file=discord.File(filename)) + await loop.run_in_executor(None, os.remove, filename) + else: + await ctx.send(f"📄 `{path}` ({len(lines)} lines):\n```{lang}\n{numbered_content}\n```") + + except Exception as e: + await ctx.send(f"❌ Error: {str(e)}") + traceback.print_exc() + + @code.command(name="search", aliases=["grep", "find-all"]) + @is_owner() + async def code_search(self, ctx, pattern: str, path: str = "."): + """Search for code patterns in files. Usage: !code search "" [path]""" + try: + loop = asyncio.get_event_loop() + + def _search(): + import re + results = [] + + if os.path.isfile(path): + files = [path] + elif os.path.isdir(path): + files = [] + for root, dirs, filenames in os.walk(path): + # Skip hidden and backup directories + dirs[:] = [d for d in dirs if not d.startswith('.') and d not in ['backups', '__pycache__', 'node_modules', 'venv', '.venv']] + for filename in filenames: + if filename.endswith(('.py', '.js', '.json', '.txt', '.md', '.yml', '.yaml', '.html', '.css', '.sql')): + files.append(os.path.join(root, filename)) + else: + return None, "Invalid path" + + for filepath in files[:200]: # Limit to 200 files + if not is_path_safe(filepath): + continue + + try: + with open(filepath, 'r', encoding='utf-8') as f: + for i, line in enumerate(f, 1): + if re.search(pattern, line, re.IGNORECASE): + results.append({ + 'file': filepath, + 'line': i, + 'content': line.rstrip() + }) + if len(results) >= 100: # Limit results + return results, "truncated" + except: + continue + + return results, None + + results, status = await loop.run_in_executor(None, _search) + + if results is None: + await ctx.send(f"❌ {status}") + return + + if not results: + await ctx.send(f"❌ No matches found for pattern: `{pattern}`") + return + + # Group by file + files_dict = {} + for result in results: + if result['file'] not in files_dict: + files_dict[result['file']] = [] + files_dict[result['file']].append(result) + + # Build output + output_parts = [] + output_parts.append(f"🔍 Found {len(results)} match(es) in {len(files_dict)} file(s)") + if status == "truncated": + output_parts.append("(Showing first 100 matches)") + output_parts.append("") + + for filepath, file_results in files_dict.items(): + output_parts.append(f"\n📄 **{filepath}** ({len(file_results)} match(es)):") + for result in file_results[:10]: # Max 10 per file in output + output_parts.append(f" Line {result['line']}: `{result['content'][:100]}`") + if len(file_results) > 10: + output_parts.append(f" ... and {len(file_results) - 10} more") + + output = "\n".join(output_parts) + + if len(output) > 1900: + # Create detailed file + detailed_output = [f"Search results for: {pattern}\n{'=' * 50}\n"] + for filepath, file_results in files_dict.items(): + detailed_output.append(f"\n{filepath}:") + for result in file_results: + detailed_output.append(f" {result['line']}: {result['content']}") + + filename = f"search_{ctx.message.id}.txt" + await loop.run_in_executor(None, self._write_file, filename, '\n'.join(detailed_output)) + await ctx.send(f"🔍 Found {len(results)} matches (see attached file):", file=discord.File(filename)) + await loop.run_in_executor(None, os.remove, filename) + else: + await ctx.send(output) + + except Exception as e: + await ctx.send(f"❌ Error: {str(e)}") + traceback.print_exc() + + @code.command(name="find") + @is_owner() + async def code_find(self, ctx, path: str, *, text: str): + """Find specific text in a file and show line numbers. Usage: !code find """ + # Security check + if not is_path_safe(path): + await ctx.send("❌ Access to this file is blocked for security reasons.") + return + + try: + if not os.path.isfile(path): + await ctx.send(f"❌ File not found: `{path}`") + return + + loop = asyncio.get_event_loop() + + def _find(): + matches = [] + with open(path, 'r', encoding='utf-8') as f: + for i, line in enumerate(f, 1): + if text.lower() in line.lower(): + matches.append((i, line.rstrip())) + return matches + + matches = await loop.run_in_executor(None, _find) + + if not matches: + await ctx.send(f"❌ Text not found in `{path}`") + return + + embed = discord.Embed( + title=f"🔍 Found in `{os.path.basename(path)}`", + description=f"Found {len(matches)} occurrence(s) of: `{text}`", + color=discord.Color.green() + ) + + # Show matches + for line_num, line_content in matches[:15]: # Show max 15 + # Highlight the match + highlighted = line_content.replace(text, f"**{text}**") + if len(highlighted) > 100: + # Find the match position and show context + pos = line_content.lower().find(text.lower()) + start = max(0, pos - 40) + end = min(len(line_content), pos + len(text) + 40) + snippet = line_content[start:end] + if start > 0: + snippet = "..." + snippet + if end < len(line_content): + snippet = snippet + "..." + highlighted = snippet.replace(text, f"**{text}**") + + embed.add_field( + name=f"Line {line_num}", + value=f"`{highlighted}`", + inline=False + ) + + if len(matches) > 15: + embed.set_footer(text=f"Showing first 15 of {len(matches)} matches") + + await ctx.send(embed=embed) + + except Exception as e: + await ctx.send(f"❌ Error: {str(e)}") + traceback.print_exc() + + except Exception as e: + await ctx.send(f"❌ Error: {str(e)}") + traceback.print_exc() + + @code.command(name="edit") + @is_owner() + async def code_edit(self, ctx, path: str, line_start: int, line_end: Optional[int] = None): + """Edit line(s) in a file interactively. Usage: !code edit OR !code edit """ + # Security check + if not is_path_safe(path): + await ctx.send("❌ Access to this file is blocked for security reasons.") + return + + try: + if not os.path.isfile(path): + await ctx.send(f"❌ File not found: `{path}`") + return + + # If no end line, edit single line + if line_end is None: + line_end = line_start + + loop = asyncio.get_event_loop() + + def _read_lines(): + with open(path, 'r', encoding='utf-8') as f: + lines = f.readlines() + + # Validate line numbers + if line_start < 1 or line_start > len(lines): + return None, None, f"Line {line_start} is out of range (file has {len(lines)} lines)" + if line_end < line_start or line_end > len(lines): + return None, None, f"Line {line_end} is out of range" + + # Get the lines to edit + selected_lines = ''.join(lines[line_start-1:line_end]).rstrip() + + return lines, selected_lines, None + + lines, selected_lines, error = await loop.run_in_executor(None, _read_lines) + + if error: + await ctx.send(f"❌ {error}") + return + + # Create modal for editing + class EditModal(discord.ui.Modal, title=f"Edit {os.path.basename(path)}"): + def __init__(self, parent_cog, path, lines, line_start, line_end, selected_lines): + super().__init__() + self.cog = parent_cog + self.path = path + self.lines = lines + self.line_start = line_start + self.line_end = line_end + + # Add text input with current content + self.code_input = discord.ui.TextInput( + label=f"Lines {line_start}-{line_end}" if line_start != line_end else f"Line {line_start}", + style=discord.TextStyle.paragraph, + default=selected_lines, + required=True, + max_length=4000 + ) + self.add_item(self.code_input) + + async def on_submit(self, interaction: discord.Interaction): + new_content = self.code_input.value + + loop = asyncio.get_event_loop() + + def _write(): + # Replace the lines + new_lines = new_content.split('\n') + # Ensure newlines at end + new_lines = [line + '\n' if not line.endswith('\n') else line for line in new_lines] + + # Remove the old lines and insert new ones + self.lines[self.line_start-1:self.line_end] = new_lines + + # Write back + with open(self.path, 'w', encoding='utf-8') as f: + f.writelines(self.lines) + + try: + await loop.run_in_executor(None, _write) + + embed = discord.Embed( + title=f"✅ Edited `{os.path.basename(self.path)}`", + description=f"Lines {self.line_start}-{self.line_end}" if self.line_start != self.line_end else f"Line {self.line_start}", + color=discord.Color.green() + ) + + # Show a preview if not too long + if len(new_content) < 1000: + embed.add_field(name="New Content", value=f"```python\n{new_content}\n```", inline=False) + else: + embed.add_field(name="New Content", value=f"```\n{new_content[:500]}...\n(truncated)\n```", inline=False) + + await interaction.response.send_message(embed=embed) + except Exception as e: + await interaction.response.send_message(f"❌ Error saving: {str(e)}", ephemeral=True) + + # Create a view with button to open modal + class EditView(discord.ui.View): + def __init__(self, cog, path, lines, line_start, line_end, selected_lines): + super().__init__(timeout=300) + self.cog = cog + self.path = path + self.lines = lines + self.line_start = line_start + self.line_end = line_end + self.selected_lines = selected_lines + + @discord.ui.button(label="✏️ Edit Code", style=discord.ButtonStyle.primary) + async def edit_button(self, interaction: discord.Interaction, button: discord.ui.Button): + if interaction.user.id != ADMIN_USER_ID: + await interaction.response.send_message("❌ Only the bot owner can edit files.", ephemeral=True) + return + + modal = EditModal(self.cog, self.path, self.lines.copy(), self.line_start, self.line_end, self.selected_lines) + await interaction.response.send_modal(modal) + + @discord.ui.button(label="❌ Cancel", style=discord.ButtonStyle.secondary) + async def cancel_button(self, interaction: discord.Interaction, button: discord.ui.Button): + if interaction.user.id != ADMIN_USER_ID: + await interaction.response.send_message("❌ Only the bot owner can use this.", ephemeral=True) + return + await interaction.response.send_message("✅ Edit cancelled.", ephemeral=True) + self.stop() + + # Send message with current code + embed = discord.Embed( + title=f"📝 Ready to edit `{os.path.basename(path)}`", + description=f"Lines {line_start}-{line_end}" if line_start != line_end else f"Line {line_start}", + color=discord.Color.blue() + ) + + # Show current content + if len(selected_lines) < 1000: + embed.add_field(name="Current Content", value=f"```python\n{selected_lines}\n```", inline=False) + else: + embed.add_field(name="Current Content", value=f"```python\n{selected_lines[:500]}...\n(truncated, will show full in editor)\n```", inline=False) + + embed.set_footer(text="Click 'Edit Code' to open the editor") + + view = EditView(self, path, lines, line_start, line_end, selected_lines) + await ctx.send(embed=embed, view=view) + + except Exception as e: + await ctx.send(f"❌ Error: {str(e)}") + traceback.print_exc() + + @code.command(name="insert") + @is_owner() + async def code_insert(self, ctx, path: str, line_num: int, *, content: str): + """Insert a new line at a specific position. Usage: !code insert """ + # Security check + if not is_path_safe(path): + await ctx.send("❌ Access to this file is blocked for security reasons.") + return + + try: + if not os.path.isfile(path): + await ctx.send(f"❌ File not found: `{path}`") + return + + loop = asyncio.get_event_loop() + + def _insert(): + with open(path, 'r', encoding='utf-8') as f: + lines = f.readlines() + + # Validate line number + if line_num < 1 or line_num > len(lines) + 1: + return f"Line {line_num} is out of range (file has {len(lines)} lines)" + + # Insert new line + lines.insert(line_num - 1, content + '\n') + + # Write back + with open(path, 'w', encoding='utf-8') as f: + f.writelines(lines) + + return None + + error = await loop.run_in_executor(None, _insert) + + if error: + await ctx.send(f"❌ {error}") + return + + await ctx.send(f"✅ Inserted line {line_num} in `{path}`:\n```{content}```") + + except Exception as e: + await ctx.send(f"❌ Error: {str(e)}") + traceback.print_exc() + + @code.command(name="delete", aliases=["del"]) + @is_owner() + async def code_delete(self, ctx, path: str, line_num: int): + """Delete a specific line from a file. Usage: !code delete """ + # Security check + if not is_path_safe(path): + await ctx.send("❌ Access to this file is blocked for security reasons.") + return + + try: + if not os.path.isfile(path): + await ctx.send(f"❌ File not found: `{path}`") + return + + loop = asyncio.get_event_loop() + + def _delete(): + with open(path, 'r', encoding='utf-8') as f: + lines = f.readlines() + + # Validate line number + if line_num < 1 or line_num > len(lines): + return None, f"Line {line_num} is out of range (file has {len(lines)} lines)" + + # Store deleted line for confirmation + deleted_line = lines[line_num - 1].rstrip() + + # Delete line + del lines[line_num - 1] + + # Write back + with open(path, 'w', encoding='utf-8') as f: + f.writelines(lines) + + return deleted_line, None + + deleted_line, error = await loop.run_in_executor(None, _delete) + + if error: + await ctx.send(f"❌ {error}") + return + + await ctx.send(f"✅ Deleted line {line_num} from `{path}`:\n```{deleted_line}```") + + except Exception as e: + await ctx.send(f"❌ Error: {str(e)}") + traceback.print_exc() + + @code.command(name="replace") + @is_owner() + async def code_replace(self, ctx, path: str, find: str, replace: str, *, options: str = ""): + """Find and replace text in a file. Usage: !code replace "" "" [--all]""" + # Security check + if not is_path_safe(path): + await ctx.send("❌ Access to this file is blocked for security reasons.") + return + + try: + if not os.path.isfile(path): + await ctx.send(f"❌ File not found: `{path}`") + return + + replace_all = "--all" in options.lower() + + loop = asyncio.get_event_loop() + + def _replace(): + with open(path, 'r', encoding='utf-8') as f: + content = f.read() + + # Count occurrences + count = content.count(find) + if count == 0: + return 0, None + + # Replace + if replace_all: + new_content = content.replace(find, replace) + else: + new_content = content.replace(find, replace, 1) + + # Write back + with open(path, 'w', encoding='utf-8') as f: + f.write(new_content) + + return count, 1 if not replace_all else count + + total_found, replaced = await loop.run_in_executor(None, _replace) + + if total_found == 0: + await ctx.send(f"❌ Text not found in `{path}`") + return + + msg = f"✅ Replaced {replaced} occurrence(s) in `{path}`" + if not replace_all and total_found > 1: + msg += f"\n(Found {total_found} total occurrences. Use `--all` to replace all)" + + embed = discord.Embed( + title=msg, + color=discord.Color.green() + ) + embed.add_field(name="Find", value=f"```{find}```", inline=False) + embed.add_field(name="Replace", value=f"```{replace}```", inline=False) + await ctx.send(embed=embed) + + except Exception as e: + await ctx.send(f"❌ Error: {str(e)}") + traceback.print_exc() + + @code.command(name="backup") + @is_owner() + async def code_backup(self, ctx, path: str): + """Create a backup of a file. Usage: !code backup """ + # Security check + if not is_path_safe(path): + await ctx.send("❌ Access to this file is blocked for security reasons.") + return + + try: + if not os.path.isfile(path): + await ctx.send(f"❌ File not found: `{path}`") + return + + # Create backup filename + import datetime + timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") + backup_path = f"{path}.backup_{timestamp}" + + loop = asyncio.get_event_loop() + + def _backup(): + import shutil + shutil.copy2(path, backup_path) + + await loop.run_in_executor(None, _backup) + await ctx.send(f"✅ Created backup: `{backup_path}`") + + except Exception as e: + await ctx.send(f"❌ Error: {str(e)}") + traceback.print_exc() + + @code.command(name="diff") + @is_owner() + async def code_diff(self, ctx, path1: str, path2: str): + """Compare two files. Usage: !code diff """ + # Security check + if not is_path_safe(path1) or not is_path_safe(path2): + await ctx.send("❌ Access to one or more files is blocked for security reasons.") + return + + try: + if not os.path.isfile(path1): + await ctx.send(f"❌ File not found: `{path1}`") + return + if not os.path.isfile(path2): + await ctx.send(f"❌ File not found: `{path2}`") + return + + loop = asyncio.get_event_loop() + + def _diff(): + import difflib + + with open(path1, 'r', encoding='utf-8') as f: + lines1 = f.readlines() + with open(path2, 'r', encoding='utf-8') as f: + lines2 = f.readlines() + + diff = difflib.unified_diff( + lines1, lines2, + fromfile=path1, + tofile=path2, + lineterm='' + ) + + return '\n'.join(diff) + + diff_output = await loop.run_in_executor(None, _diff) + + if not diff_output: + await ctx.send("✅ Files are identical") + return + + if len(diff_output) > 1900: + filename = f"diff_{ctx.message.id}.txt" + await loop.run_in_executor(None, self._write_file, filename, diff_output) + await ctx.send("📊 Diff:", file=discord.File(filename)) + await loop.run_in_executor(None, os.remove, filename) + else: + await ctx.send(f"```diff\n{diff_output}\n```") + + except Exception as e: + await ctx.send(f"❌ Error: {str(e)}") + traceback.print_exc() + + @commands.command(name="grep") + @is_owner() + async def grep(self, ctx, pattern: str, path: str = "."): + """Search for text in files. Usage: !grep "" [path]""" + try: + loop = asyncio.get_event_loop() + + def _grep(): + import re + results = [] + + if os.path.isfile(path): + files = [path] + elif os.path.isdir(path): + files = [] + for root, dirs, filenames in os.walk(path): + # Skip hidden and backup directories + dirs[:] = [d for d in dirs if not d.startswith('.') and d not in ['backups', '__pycache__', 'node_modules']] + for filename in filenames: + if filename.endswith(('.py', '.js', '.json', '.txt', '.md', '.yml', '.yaml')): + files.append(os.path.join(root, filename)) + else: + return None, "Invalid path" + + for filepath in files[:100]: # Limit to 100 files + if not is_path_safe(filepath): + continue + + try: + with open(filepath, 'r', encoding='utf-8') as f: + for i, line in enumerate(f, 1): + if re.search(pattern, line): + results.append(f"{filepath}:{i}: {line.rstrip()}") + if len(results) >= 50: # Limit results + return results, "truncated" + except: + continue + + return results, None + + results, status = await loop.run_in_executor(None, _grep) + + if results is None: + await ctx.send(f"❌ {status}") + return + + if not results: + await ctx.send(f"❌ No matches found for pattern: `{pattern}`") + return + + output = "\n".join(results) + if status == "truncated": + output += "\n... (truncated, showing first 50 matches)" + + if len(output) > 1900: + filename = f"grep_{ctx.message.id}.txt" + await loop.run_in_executor(None, self._write_file, filename, output) + await ctx.send(f"🔍 Search results for `{pattern}`:", file=discord.File(filename)) + await loop.run_in_executor(None, os.remove, filename) + else: + await ctx.send(f"🔍 Search results for `{pattern}`:\n```\n{output}\n```") + + except Exception as e: + await ctx.send(f"❌ Error: {str(e)}") + traceback.print_exc() async def setup(bot): await bot.add_cog(AdminCommands(bot)) \ No newline at end of file