mirror of
https://github.com/rojo-rbx/rojo.git
synced 2026-10-23 05:51:39 +00:00
9934094dbc
- Updated the `clap` dependency in `Cargo.toml` to include the "env" feature for improved command-line argument handling. - Introduced a new `roblox_api` module to encapsulate Roblox API interactions, including authentication and asset downloading. - Refactored the `CloneCommand` and `StudioCommand` to utilize the new `roblox_api` functions for fetching experience names and universe IDs, enhancing code clarity and reducing duplication. - Removed obsolete functions related to Roblox API calls from the CLI commands, streamlining the codebase. - Added error handling for authentication failures, ensuring better user feedback when required credentials are missing. These changes aim to improve the overall structure and functionality of the CLI while enhancing the integration with the Roblox API.
307 lines
13 KiB
Python
307 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import TYPE_CHECKING
|
|
|
|
import anthropic
|
|
import discord
|
|
from discord import app_commands
|
|
|
|
from . import _bot_ref, daemon, security
|
|
|
|
if TYPE_CHECKING:
|
|
from .config import Config
|
|
from .db import Database
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
class AtlasDaemonBot(discord.Client):
|
|
def __init__(
|
|
self,
|
|
config: Config,
|
|
db: Database,
|
|
anthropic_client: anthropic.AsyncAnthropic,
|
|
) -> None:
|
|
intents = discord.Intents.default()
|
|
intents.guilds = True
|
|
intents.guild_messages = True
|
|
super().__init__(intents=intents)
|
|
|
|
self.config = config
|
|
self.db = db
|
|
self.claude = anthropic_client
|
|
self.tree = app_commands.CommandTree(self)
|
|
|
|
_bot_ref.db = db
|
|
|
|
self._register_commands()
|
|
|
|
def _register_commands(self) -> None:
|
|
tree = self.tree
|
|
|
|
@tree.command(name="approve-server", description="Whitelist this server (or a remote one by ID) to use the bot")
|
|
@app_commands.describe(server_id="Optional: remote server ID to approve")
|
|
async def approve_server(interaction: discord.Interaction, server_id: str | None = None) -> None:
|
|
if not self.db.is_admin(str(interaction.user.id)):
|
|
await interaction.response.send_message("Only bot admins can approve servers.", ephemeral=True)
|
|
return
|
|
|
|
target_id = server_id or str(interaction.guild_id)
|
|
server_name = None
|
|
if not server_id and interaction.guild:
|
|
server_name = interaction.guild.name
|
|
|
|
self.db.approve_server(target_id, server_name, str(interaction.user.id))
|
|
await interaction.response.send_message(f"Server `{target_id}` approved.", ephemeral=True)
|
|
|
|
@tree.command(name="revoke-server", description="Remove a server from the whitelist")
|
|
@app_commands.describe(server_id="Server ID to revoke")
|
|
async def revoke_server(interaction: discord.Interaction, server_id: str) -> None:
|
|
if not self.db.is_admin(str(interaction.user.id)):
|
|
await interaction.response.send_message("Only bot admins can revoke servers.", ephemeral=True)
|
|
return
|
|
|
|
daemon.stop_all_for_server(self.db, server_id)
|
|
self.db.revoke_server(server_id)
|
|
await interaction.response.send_message(f"Server `{server_id}` revoked.", ephemeral=True)
|
|
|
|
@tree.command(name="set-api-key", description="Set your OpenCloud API key (requires legacy-asset:manage scope)")
|
|
@app_commands.describe(key="Your OpenCloud API key")
|
|
async def set_api_key(interaction: discord.Interaction, key: str) -> None:
|
|
if not self._check_server(interaction):
|
|
return
|
|
|
|
encrypted = security.encrypt_api_key(key)
|
|
for game in self.db.get_games_for_server(str(interaction.guild_id)):
|
|
if game.added_by == str(interaction.user.id):
|
|
self.db._conn.execute(
|
|
"UPDATE games SET api_key_encrypted = ? WHERE id = ?",
|
|
(encrypted, game.id),
|
|
)
|
|
self.db._conn.commit()
|
|
await interaction.response.send_message(
|
|
"API key updated for all your games in this server.", ephemeral=True
|
|
)
|
|
|
|
@tree.command(name="add-game", description="Start monitoring a Roblox game")
|
|
@app_commands.describe(
|
|
place_id="The Roblox Place ID to monitor",
|
|
channel="Channel to post reviews in (defaults to current)",
|
|
)
|
|
async def add_game(
|
|
interaction: discord.Interaction,
|
|
place_id: int,
|
|
channel: discord.TextChannel | None = None,
|
|
) -> None:
|
|
if not await self._check_server_async(interaction):
|
|
return
|
|
|
|
server_id = str(interaction.guild_id)
|
|
|
|
if self.db.count_games_for_server(server_id) >= self.config.max_games_per_server:
|
|
await interaction.response.send_message(
|
|
f"This server has reached the maximum of {self.config.max_games_per_server} monitored games.",
|
|
ephemeral=True,
|
|
)
|
|
return
|
|
|
|
if self.db.get_game(server_id, place_id):
|
|
await interaction.response.send_message(
|
|
f"PlaceId {place_id} is already being monitored in this server.",
|
|
ephemeral=True,
|
|
)
|
|
return
|
|
|
|
await interaction.response.defer(ephemeral=True)
|
|
|
|
target_channel = channel or interaction.channel
|
|
if not isinstance(target_channel, discord.TextChannel):
|
|
await interaction.followup.send("Please specify a text channel.", ephemeral=True)
|
|
return
|
|
|
|
games = self.db.get_games_for_server(server_id)
|
|
user_games = [g for g in games if g.added_by == str(interaction.user.id)]
|
|
api_key_encrypted: bytes | None = None
|
|
if user_games:
|
|
api_key_encrypted = user_games[0].api_key_encrypted
|
|
|
|
if api_key_encrypted is None:
|
|
await interaction.followup.send(
|
|
"No API key found. Please run `/set-api-key` first, then try again.",
|
|
ephemeral=True,
|
|
)
|
|
return
|
|
|
|
try:
|
|
opencloud_key = security.decrypt_api_key(api_key_encrypted)
|
|
game_dir = await daemon.init_game(
|
|
self.config, place_id, server_id, opencloud_key
|
|
)
|
|
except Exception as exc:
|
|
await interaction.followup.send(
|
|
f"Failed to initialize game: ```{str(exc)[:1500]}```", ephemeral=True
|
|
)
|
|
return
|
|
|
|
game_id = self.db.add_game(
|
|
server_id=server_id,
|
|
place_id=place_id,
|
|
channel_id=str(target_channel.id),
|
|
api_key_encrypted=api_key_encrypted,
|
|
added_by=str(interaction.user.id),
|
|
working_dir=str(game_dir),
|
|
)
|
|
|
|
game = self.db.get_game(server_id, place_id)
|
|
if game:
|
|
daemon.start_sync_loop(game, self.config, self.db, self, self.claude)
|
|
|
|
await interaction.followup.send(
|
|
f"Now monitoring PlaceId {place_id} in {target_channel.mention}. "
|
|
f"Syncing every {self.config.sync_interval}s.",
|
|
ephemeral=True,
|
|
)
|
|
|
|
@tree.command(name="remove-game", description="Stop monitoring a game")
|
|
@app_commands.describe(place_id="The Place ID to stop monitoring")
|
|
async def remove_game(interaction: discord.Interaction, place_id: int) -> None:
|
|
if not await self._check_server_async(interaction):
|
|
return
|
|
|
|
server_id = str(interaction.guild_id)
|
|
game = self.db.get_game(server_id, place_id)
|
|
if not game:
|
|
await interaction.response.send_message(
|
|
f"PlaceId {place_id} is not being monitored.", ephemeral=True
|
|
)
|
|
return
|
|
|
|
if game.added_by != str(interaction.user.id) and not self.db.is_admin(str(interaction.user.id)):
|
|
await interaction.response.send_message("You can only remove games you added.", ephemeral=True)
|
|
return
|
|
|
|
daemon.stop_sync_loop(game.id)
|
|
self.db.remove_game(server_id, place_id)
|
|
await interaction.response.send_message(f"Stopped monitoring PlaceId {place_id}.", ephemeral=True)
|
|
|
|
@tree.command(name="reset-game", description="Delete local clone and re-sync from scratch")
|
|
@app_commands.describe(place_id="The Place ID to reset")
|
|
async def reset_game(interaction: discord.Interaction, place_id: int) -> None:
|
|
if not await self._check_server_async(interaction):
|
|
return
|
|
|
|
server_id = str(interaction.guild_id)
|
|
game = self.db.get_game(server_id, place_id)
|
|
if not game:
|
|
await interaction.response.send_message(
|
|
f"PlaceId {place_id} is not being monitored.", ephemeral=True
|
|
)
|
|
return
|
|
|
|
if game.added_by != str(interaction.user.id) and not self.db.is_admin(str(interaction.user.id)):
|
|
await interaction.response.send_message("You can only reset games you added.", ephemeral=True)
|
|
return
|
|
|
|
await interaction.response.defer(ephemeral=True)
|
|
|
|
daemon.stop_sync_loop(game.id)
|
|
|
|
import shutil
|
|
from pathlib import Path
|
|
|
|
working_dir = Path(game.working_dir)
|
|
if working_dir.exists():
|
|
shutil.rmtree(working_dir, ignore_errors=True)
|
|
|
|
try:
|
|
opencloud_key = security.decrypt_api_key(game.api_key_encrypted)
|
|
await daemon.init_game(self.config, place_id, server_id, opencloud_key)
|
|
except Exception as exc:
|
|
await interaction.followup.send(
|
|
f"Failed to re-initialize: ```{str(exc)[:1500]}```", ephemeral=True
|
|
)
|
|
return
|
|
|
|
self.db._conn.execute(
|
|
"DELETE FROM issues WHERE game_id = ?", (game.id,)
|
|
)
|
|
self.db._conn.execute(
|
|
"UPDATE games SET last_sync_at = NULL, last_error = NULL, error_count = 0 WHERE id = ?",
|
|
(game.id,),
|
|
)
|
|
self.db._conn.commit()
|
|
|
|
game = self.db.get_game(server_id, place_id)
|
|
if game:
|
|
daemon.start_sync_loop(game, self.config, self.db, self, self.claude)
|
|
|
|
await interaction.followup.send(f"PlaceId {place_id} has been reset and re-synced.", ephemeral=True)
|
|
|
|
@tree.command(name="status", description="Show monitored games and sync status")
|
|
async def status(interaction: discord.Interaction) -> None:
|
|
if not await self._check_server_async(interaction):
|
|
return
|
|
|
|
games = self.db.get_games_for_server(str(interaction.guild_id))
|
|
if not games:
|
|
await interaction.response.send_message("No games are being monitored in this server.", ephemeral=True)
|
|
return
|
|
|
|
lines = []
|
|
for g in games:
|
|
status_icon = "🔴" if g.last_error else "🟢"
|
|
unresolved = len(self.db.get_unresolved_issues(g.id))
|
|
lines.append(
|
|
f"{status_icon} **PlaceId {g.place_id}** -- "
|
|
f"Last sync: {g.last_sync_at or 'Never'} -- "
|
|
f"Unresolved: {unresolved}"
|
|
)
|
|
if g.last_error:
|
|
error_preview = g.last_error[:100] + ("..." if len(g.last_error) > 100 else "")
|
|
lines.append(f" └ Error: {error_preview}")
|
|
|
|
await interaction.response.send_message("\n".join(lines), ephemeral=True)
|
|
|
|
@tree.command(name="list-games", description="List all monitored games in this server")
|
|
async def list_games(interaction: discord.Interaction) -> None:
|
|
if not await self._check_server_async(interaction):
|
|
return
|
|
|
|
games = self.db.get_games_for_server(str(interaction.guild_id))
|
|
if not games:
|
|
await interaction.response.send_message("No games monitored.", ephemeral=True)
|
|
return
|
|
|
|
lines = [
|
|
f"• PlaceId **{g.place_id}** → <#{g.channel_id}> (added by <@{g.added_by}>)"
|
|
for g in games
|
|
]
|
|
await interaction.response.send_message("\n".join(lines), ephemeral=True)
|
|
|
|
def _check_server(self, interaction: discord.Interaction) -> bool:
|
|
if not interaction.guild_id:
|
|
return False
|
|
return self.db.is_server_allowed(str(interaction.guild_id))
|
|
|
|
async def _check_server_async(self, interaction: discord.Interaction) -> bool:
|
|
if not interaction.guild_id:
|
|
await interaction.response.send_message("This command can only be used in a server.", ephemeral=True)
|
|
return False
|
|
if not self.db.is_server_allowed(str(interaction.guild_id)):
|
|
await interaction.response.send_message(
|
|
"This server is not authorized. Ask a bot admin to run `/approve-server`.",
|
|
ephemeral=True,
|
|
)
|
|
return False
|
|
return True
|
|
|
|
async def on_ready(self) -> None:
|
|
log.info("Bot ready as %s", self.user)
|
|
await self.tree.sync()
|
|
log.info("Slash commands synced")
|
|
|
|
for game in self.db.get_all_games():
|
|
if self.db.is_server_allowed(game.discord_server_id):
|
|
daemon.start_sync_loop(game, self.config, self.db, self, self.claude)
|
|
log.info("Started sync loop for PlaceId %d", game.place_id)
|