Initializing repo

Author: sternd

.gitignore

@@ -0,0 +1,5 @@
+.idea
+*.zip
+*.log
+.env
+*.pyc

DiscordRequests.py

@@ -0,0 +1,108 @@
+import requests
+from requests.exceptions import HTTPError, Timeout
+import discord
+
+
+# Helper class to send Discord API requests
+class DiscordRequests:
+    DISCORD_BASE_URL = "https://discordapp.com/api/v7"
+    DISCORD_MESSAGE_HISTORY = "/channels/:channel_id/messages"
+    DISCORD_CREATE_MESSAGE = "/channels/:channel_id/messages"
+    DISCORD_EDIT_MESSAGE = "/channels/:channel_id/messages/:message_id"
+
+    TOKEN = None
+    BOT_CLIENT_ID = None
+
+    def __init__(self, config):
+        if 'auth_token' not in config or config['token'] is None:
+            raise Exception('Missing Discord Token')
+        elif 'bot_client_id' not in config or config['bot_client_id'] is None:
+            raise Exception('Missing Bot Client ID')
+
+        self.TOKEN = config['token']
+        self.BOT_CLIENT_ID = config['bot_client_id']
+
+    # Send a Discord API request
+    def sendDiscordRequest(self, action, url, params=None):
+        # Do request here
+
+        auth_token = f'Bot {self.TOKEN}'
+
+        try:
+            if action == 'GET':
+                response = requests.get(url, timeout=5, headers={"Authorization": auth_token})
+            elif action == 'POST':
+                response = requests.post(url, timeout=5, headers={"Authorization": auth_token}, json=params)
+            elif action == 'PATCH':
+                response = requests.patch(url, timeout=5, headers={"Authorization": auth_token}, json=params)
+            else:
+                raise Exception(f'Attempting to send request with unknown action: {action}')
+
+            response.raise_for_status()
+        except HTTPError as http_err:
+            print(f'HTTP error occurred for {id}: {http_err}')
+        except Timeout as timeout:
+            print(f'HTTP timeout occurred for {id}: {timeout}')
+        except Exception as err:
+            print(f'Other error occurred for {id}: {err}')
+        else:
+            return response.json()
+
+        return None
+
+    # Get message ID of most recent message sent by bot
+    def getLastMessageByBot(self, channel_id):
+        # Make request for message history
+        # Parse message ID
+        url = self.DISCORD_BASE_URL + self.DISCORD_MESSAGE_HISTORY.replace(':channel_id', channel_id, 1)
+
+        response = self.sendDiscordRequest('GET', url)
+
+        if response is None:
+            return None
+
+        for message in response:
+            if message["author"]["id"] != self.BOT_CLIENT_ID:
+                continue
+
+            return message
+
+        return None
+
+    # Edit an existing Rich Embed message
+    def editMessage(self, channel_id, message_id, data):
+        # Make request to edit message
+        url = self.DISCORD_BASE_URL + self.DISCORD_EDIT_MESSAGE.replace(':channel_id', channel_id, 1).replace(
+            ':message_id', message_id, 1)
+
+        response = self.sendDiscordRequest('PATCH', url, data)
+
+        if response is None:
+            return None
+
+        return response
+
+    # Create a new Rich Embed message
+    def createMessage(self, channel_id, data):
+        # Make request to edit message
+        url = self.DISCORD_BASE_URL + self.DISCORD_CREATE_MESSAGE.replace(':channel_id', channel_id, 1)
+
+        response = self.sendDiscordRequest('POST', url, data)
+
+        if response is None:
+            return None
+
+        return response
+
+    # Create an initial connection to Discord to allow future messages
+    # Only needs to be called once per Discord server
+    def intialConnection(self):
+        client = discord.Client()
+
+        @client.event
+        async def on_ready():
+            print('Closing connection')
+            await client.close()
+            return None
+
+        client.run(self.TOKEN)

DynamoDBHelper.py

@@ -0,0 +1,32 @@
+from boto3 import resource
+
+
+class DynamoDBHelper:
+    DB = None
+
+    def __init__(self, db_type, db_region):
+        self.DB = resource(db_type, region_name=db_region)
+
+    def getDocument(self, table_name, key, val):
+        table = self.DB.Table(table_name)
+        item = table.get_item(Key={key: val})
+
+        if not item:
+            return None
+
+        if 'Item' not in item:
+            return None
+
+        return item['Item']
+
+    def createDocument(self, table_name, doc):
+        table = self.DB.Table(table_name)
+        response = table.put_item(Item=doc)
+
+        return response
+
+    def updateDocument(self, table_name, key_dict, update_expression, expression_attribute_values):
+        table = self.DB.Table(table_name)
+        response = table.update_item(Key=key_dict, UpdateExpression=update_expression, ExpressionAttributeValues=expression_attribute_values, ReturnValues="UPDATED_NEW")
+
+        return response

NitradeoRequests.py

@@ -0,0 +1,52 @@
+import requests
+from requests.exceptions import HTTPError, Timeout
+
+
+class NitradoRequests:
+    NITRAPI_BASE_URL = "https://api.nitrado.net"
+    NITRAPI_GAMESERVER_BOOST_HISTORY = "/services/:id/gameservers/boost/history"
+    NITRAPI_GAMESERVER_DETAILS = "/services/:id/gameservers"
+
+    # Send a Discord API request
+    def sendNitrapiRequest(self, action, token, url, params=None):
+        # Do request here
+
+        auth_token = f'Bearer {token}'
+
+        try:
+            if action == 'GET':
+                response = requests.get(url, timeout=5, headers={"Authorization": auth_token})
+            elif action == 'POST':
+                response = requests.post(url, timeout=5, headers={"Authorization": auth_token}, json=params)
+            elif action == 'PATCH':
+                response = requests.patch(url, timeout=5, headers={"Authorization": auth_token}, json=params)
+            else:
+                raise Exception(f'Attempting to send request with unknown action: {action}')
+
+            response.raise_for_status()
+        except HTTPError as http_err:
+            print(f'HTTP error occurred for {id}: {http_err}')
+        except Timeout as timeout:
+            print(f'HTTP timeout occurred for {id}: {timeout}')
+        except Exception as err:
+            print(f'Other error occurred for {id}: {err}')
+        else:
+            return response.json()
+
+        return None
+
+    # Make GET request to NitrAPI for boost history
+    # Return: Dict | None
+    # See for more info: https://doc.nitrado.net/
+    def getBoostHistory(self, token, gameserver_id):
+        url = self.NITRAPI_BASE_URL + self.NITRAPI_GAMESERVER_BOOST_HISTORY.replace(':id', gameserver_id, 1)
+
+        return self.sendNitrapiRequest("GET", token, url)
+
+    # Make GET request to NitrAPI for gameserver details
+    # Return: Dict | None
+    # See for more info: https://doc.nitrado.net/#api-Gameserver-Details
+    def getGameserverDetails(self, token, gameserver_id):
+        url = self.NITRAPI_BASE_URL + self.NITRAPI_GAMESERVER_DETAILS.replace(':id', gameserver_id, 1)
+
+        return self.sendNitrapiRequest("GET", token, url)

README.md

@@ -0,0 +1 @@
+# Python modules for use in other projects

requirements.txt

@@ -0,0 +1,4 @@
+discord.py
+requests
+python-dotenv
+boto3