-
-
Notifications
You must be signed in to change notification settings - Fork 14
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #49 from MeshAddicts/20240716-discord-bot
20240716 Discord Bot
- Loading branch information
Showing
4 changed files
with
152 additions
and
113 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,36 +1,93 @@ | ||
| import discord | ||
| import datetime | ||
| from zoneinfo import ZoneInfo | ||
| from discord.ext import commands | ||
| import discord | ||
|
|
||
| import utils | ||
|
|
||
| class LookupFlags(commands.FlagConverter): | ||
| node_id: str = commands.flag(description='Node ID') | ||
|
|
||
| class MainCommands(commands.Cog): | ||
| def __init__(self, bot): | ||
| def __init__(self, bot, config, data): | ||
| self.bot = bot | ||
| self.config = config | ||
| self.data = data | ||
|
|
||
| @commands.Cog.listener() | ||
| async def on_ready(self): | ||
| print(f'Logged in as {self.user} (ID: {self.user.id})') | ||
| await self.tree.sync(guild=self.guilds[0]) | ||
| print('Synced Slash Commands') | ||
| print('Discord: Logged in') | ||
|
|
||
| @commands.Cog.listener() | ||
| async def on_message(message): | ||
| print(f'Discord: {message.channel.id}: {message.author}: {message.content}') | ||
| if message.content.startswith('!test'): | ||
| await message.channel.send('Test successful!') | ||
| @commands.hybrid_command(name="lookup", description="Look up a node by ID") | ||
| async def lookup_node(self, ctx, *, flags: LookupFlags): | ||
| print(f"Discord: /lookup: Looking up {flags.node_id}") | ||
| try: | ||
| id_int = int(flags.node_id, 10) | ||
| id_hex = utils.convert_node_id_from_int_to_hex(id_int) | ||
| except ValueError: | ||
| id_hex = flags.node_id | ||
|
|
||
| if id_hex not in self.data.nodes: | ||
| for node_id, node in self.data.nodes.items(): | ||
| if node['shortname'] == flags.node_id: | ||
| id_hex = node_id | ||
| break | ||
|
|
||
| @commands.command() | ||
| if id_hex not in self.data.nodes: | ||
| print(f"Discord: /lookup: Node {id_hex} not found.") | ||
| await ctx.send(f"Node {id_hex} not found.") | ||
| return | ||
|
|
||
| id_int = utils.convert_node_id_from_hex_to_int(id_hex) | ||
| node = self.data.nodes[id_hex] | ||
| print(f"Discord: /lookup: Found {node['id']}") | ||
|
|
||
| embed = discord.Embed( | ||
| title=f"Node {node['shortname']}: {node['longname']}", | ||
| url=f"https://svm1.meshinfo.network/node_{node['id']}.html", | ||
| color=discord.Color.blue()) | ||
| embed.set_thumbnail(url=f"https://api.dicebear.com/9.x/bottts-neutral/svg?seed={node['id']}") | ||
| embed.add_field(name="ID (hex)", value=id_hex, inline=True) | ||
| embed.add_field(name="ID (int)", value=id_int, inline=True) | ||
| embed.add_field(name="Shortname", value=node['shortname'], inline=False) | ||
| embed.add_field(name="Hardware", value=node['hardware'], inline=False) | ||
| embed.add_field(name="Last Seen", value=node['last_seen'], inline=False) | ||
| embed.add_field(name="Status", value=("Online" if node['active'] else "Offline"), inline=False) | ||
| await ctx.send(embed=embed) | ||
|
|
||
| @commands.hybrid_command(name="mesh", description="Information about the mesh") | ||
| async def mesh_info(self, ctx): | ||
| print(f"Discord: /mesh: Mesh info requested by {ctx.author}") | ||
| embed = discord.Embed( | ||
| title=f"Information about {self.config['mesh']['name']}", | ||
| url="https://svm1.meshinfo.network", | ||
| color=discord.Color.blue()) | ||
| embed.add_field(name="Name", value=self.config['mesh']['name'], inline=False) | ||
| embed.add_field(name="Shortname", value=self.config['mesh']['shortname'], inline=False) | ||
| embed.add_field(name="Description", value=self.config['mesh']['description'], inline=False) | ||
| embed.add_field(name="Official Website", value=self.config['mesh']['url'], inline=False) | ||
| location = f"{self.config['mesh']['metro']}, {self.config['mesh']['region']}, {self.config['mesh']['country']}" | ||
| embed.add_field(name="Location", value=location, inline=False) | ||
| embed.add_field(name="Timezone", value=self.config['server']['timezone'], inline=False) | ||
| embed.add_field(name="Known Nodes", value=len(self.data.nodes), inline=True) | ||
| embed.add_field(name="Online Nodes", value=len([n for n in self.data.nodes.values() if n['active']]), inline=True) | ||
| uptime = datetime.datetime.now().astimezone(ZoneInfo(self.config['server']['timezone'])) - self.config['server']['start_time'] | ||
| embed.add_field(name="Server Uptime", value=f"{uptime.days}d {uptime.seconds // 3600}h {uptime.seconds // 60}m {uptime.seconds % 60}s", inline=False) | ||
| embed.add_field(name="Messages Since Start", value=len(self.data.messages), inline=True) | ||
| await ctx.send(embed=embed) | ||
|
|
||
| @commands.hybrid_command(name="ping", description="Ping the bot") | ||
| async def ping(self, ctx): | ||
| print(f"Discord: /ping: Pinged by {ctx.author}") | ||
| await ctx.send(f'Pong! {round(self.bot.latency * 1000)}ms') | ||
|
|
||
| @commands.command( | ||
| name="lookup", | ||
| description="Look up a node by ID", | ||
| guild=discord.Object(id=1234910729480441947) | ||
| ) | ||
| async def lookup_node(ctx: discord.Interaction, node_id: str): | ||
| global nodes | ||
| node = nodes[node_id] | ||
| if node is None: | ||
| await ctx.response.send_message(f"Node {node_id} not found.") | ||
| return | ||
| await ctx.response.send_message(f"Node {node['id']}: Short Name = {node['shortname']}, Long Name = {node['longname']}, Hardware = {node['hardware']}, Position = {node['position']}, Last Seen = {node['last_seen']}, Active = {node['active']}") | ||
| @commands.hybrid_command(name="uptime", description="Uptime of MeshInfo instance") | ||
| async def uptime(self, ctx): | ||
| print(f"Discord: /uptime: Uptime requested by {ctx.author}") | ||
| now = datetime.datetime.now().astimezone(ZoneInfo(self.config['server']['timezone'])) | ||
| print(now) | ||
| print(self.config['server']['start_time']) | ||
| uptime = now - self.config['server']['start_time'] | ||
| print(uptime) | ||
| print(f"{uptime.days}d {uptime.seconds // 3600}h {uptime.seconds // 60}m {uptime.seconds % 60}s") | ||
| await ctx.send(f'MeshInfo uptime: {uptime.days}d {uptime.seconds // 3600}h {uptime.seconds // 60}m {uptime.seconds % 60}s') |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,55 +1,68 @@ | ||
| #!/usr/bin/env python3 | ||
|
|
||
| import asyncio | ||
| import os | ||
| from typing import List | ||
|
|
||
| import discord | ||
| from discord.ext import commands | ||
| import discord | ||
| from dotenv import load_dotenv | ||
|
|
||
| from bot.cogs.main_commands import MainCommands | ||
| from memory_data_store import MemoryDataStore | ||
|
|
||
|
|
||
| class DiscordBot(commands.Bot): | ||
| guilds = [] | ||
|
|
||
| def __init__( | ||
| self, | ||
| *args, | ||
| initial_guilds: List[int], | ||
| config: dict, | ||
| data: MemoryDataStore, | ||
| **kwargs, | ||
| ): | ||
| super().__init__(*args, **kwargs) | ||
| self.guilds = initial_guilds | ||
| self.config = config | ||
| self.data = data | ||
| self.synced = False | ||
|
|
||
| async def on_ready(self): | ||
| print('Discord: Ready!') | ||
| await self.wait_until_ready() | ||
| if not self.synced: | ||
| print("Discord: Syncing commands") | ||
| guild = discord.Object(id=self.config['integrations']['discord']['guild']) | ||
| self.tree.copy_global_to(guild=guild) | ||
| await self.tree.sync(guild = discord.Object(id=self.config['integrations']['discord']['guild'])) | ||
| self.synced = True | ||
|
|
||
| async def main(): | ||
| load_dotenv() | ||
|
|
||
| if os.environ.get("DISCORD_TOKEN") is not None: | ||
| token = os.environ["DISCORD_TOKEN"] | ||
| channel_id = os.environ["DISCORD_CHANNEL_ID"] | ||
| bot = DiscordBot( | ||
| command_prefix="!", | ||
| intents=discord.Intents.all(), | ||
| initial_guilds=[1234910729480441947], | ||
| ) | ||
| print("Adding cog MainCommands") | ||
| await bot.add_cog(MainCommands(bot)) | ||
| print("Starting bot") | ||
| await bot.start(token) | ||
| print("Bot started") | ||
| await bot.get_channel(channel_id).send("Hello.") | ||
| else: | ||
| print("Not running bot because DISCORD_TOKEN not set") | ||
|
|
||
|
|
||
| async def start_bot(): | ||
| print("Starting Discord Bot") | ||
| await main() | ||
| print("Discord Bot Done!") | ||
| async def on_message(self, message): | ||
| print(f'Discord: {message.channel.id}: {message.author}: {message.content}') | ||
| if message.content.startswith('!test'): | ||
| await message.channel.send('Test successful!') | ||
| await self.process_commands(message) | ||
|
|
||
| async def start_server(self): | ||
| print("Starting Discord Bot") | ||
| await self.add_cog(MainCommands(self, self.config, self.data)) | ||
| await self.start(self.config['integrations']['discord']['token']) | ||
| print("Discord Bot Done!") | ||
|
|
||
| async def main(): | ||
| load_dotenv() | ||
| # if os.environ.get("DISCORD_TOKEN") is not None: | ||
| # token = os.environ["DISCORD_TOKEN"] | ||
| # channel_id = os.environ["DISCORD_CHANNEL_ID"] | ||
| # bot = DiscordBot( | ||
| # command_prefix="!", | ||
| # intents=discord.Intents.all(), | ||
| # initial_guilds=[1234910729480441947], | ||
| # ) | ||
| # print("Adding cog MainCommands") | ||
| # await bot.add_cog(MainCommands(bot)) | ||
| # print("Starting bot") | ||
| # await bot.start(token) | ||
| # print("Bot started") | ||
| # await bot.get_channel(channel_id).send("Hello.") | ||
| # else: | ||
| # print("Not running bot because DISCORD_TOKEN not set") | ||
|
|
||
| if __name__ == "__main__": | ||
| asyncio.run(main(), debug=True) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters