Add ChatGPT (#65)

.env.example

@@ -1,6 +1,6 @@
 # .env
 
-# SERVER 
+# SERVER
 LOGGER_LEVEL={your-logging-level-here}
 LOG_FILE_PATH={your-log-file-path-here}
 
@@ -19,6 +19,8 @@ FORTNITE_DISCORD_ROLE_USERS_DICT={your-discord-fortnite-role-users}
 TWITCH_CLIENT_ID={your-token-here}
 TWITCH_CLIENT_SECRET={your-token-here}
 
+OPENAI_API_KEY={your-token-here}
+
 DATABASE_HOST={your-host-here}
 DATABASE_PORT={your-port-here}
 DATABASE_USERNAME={your-username-here}

clients/openai.py

@@ -0,0 +1,37 @@
+import os
+
+import openai
+
+OPENAI_MODEL = "gpt-3.5-turbo"
+
+
+def initialize():
+    """Initialize OpenAI client."""
+    openai.api_key = os.getenv("OPENAI_API_KEY")
+
+
+async def ask_chatgpt(prompt, logger):
+    """Ask OpenAI ChatGPT a question."""
+    completion = openai.ChatCompletion.create(
+        model=OPENAI_MODEL,
+        messages=[
+            {
+                "role": "user",
+                "content": prompt
+            }
+        ]
+    )
+
+    logger.info({
+        "name": "Ask ChatGPT interaction",
+        "prompt": "prompt",
+        "response": completion.to_dict_recursive()
+    })
+
+    try:
+        resp = completion["choices"][0]["message"]["content"]
+    except (IndexError, KeyError) as exc:
+        resp = f"ChatGPT response format changed: {repr(exc)}"
+        logger.error(resp)
+
+    return resp

commands.py

@@ -9,14 +9,14 @@
         "command": "hunted",
         "aliases": ["h", "player", "findnoob", "wreckedby"],
         "description": "Display player stats",
-        "examples": "`!h LigmaBalls12`, `!hunted LigmaBalls12`"    
+        "examples": "`!h LigmaBalls12`, `!hunted LigmaBalls12`"
     },
     "Track Squad": {
         "command": "track",
         "aliases": ["squad"],
         "description": ("Display current stats for the squad. If a username is provided, "
                         "display only stats for that player (ex: `!track LigmaBalls12`)."),
-        "examples": "`!track`, `!squad`"       
+        "examples": "`!track`, `!squad`"
     },
     "Stats": {
         "command": "stats",
@@ -25,7 +25,7 @@
                         "LigmaBalls12`, `!stats played`)."),
         "diff_commands": ["today", "diff"],
         "opponent_commands": ["played", "opp", "opponents", "noobs", "enemy"],
-        "examples": "`!stats diff`, `!stats today`, `!stats diff stoobish`, `!stats opponents`"       
+        "examples": "`!stats diff`, `!stats today`, `!stats diff stoobish`, `!stats opponents`"
     },
     "Rate Difficulty": {
         "command": "rate",
@@ -59,6 +59,12 @@
         "log_commands": ["log", "silent"],
         "examples": "`!replays`, `!replays LigmaBalls12`, `!replays kills`, `!replays kills LigmaBalls12`"
     },
+    "Ask": {
+        "command": "ask",
+        "aliases": ["a", "chatgpt"],
+        "description": "Ask OpenAI ChatGPT a question",
+        "examples": "``!ask`, `!chatgpt`"
+    }
 }
 
 # Help
@@ -114,4 +120,10 @@
 REPLAYS_DESCRIPTION = COMMANDS["Replays"]["description"]
 REPLAYS_ELIMINATED_COMMANDS = COMMANDS["Replays"]["eliminated_commands"]
 REPLAYS_LOG_COMMANDS = COMMANDS["Replays"]["log_commands"]
-REPLAYS_EXAMPLES = COMMANDS["Replays"]["examples"]
+REPLAYS_EXAMPLES = COMMANDS["Replays"]["examples"]
+
+# Ask OpenAI ChatGPT
+ASK_COMMAND = COMMANDS["Ask"]["command"]
+ASK_DESCRIPTION = COMMANDS["Ask"]["description"]
+ASK_ALIASES = COMMANDS["Ask"]["aliases"]
+ASK_EXAMPLES = COMMANDS["Ask"]["examples"]

fortnite_bot.py

@@ -16,6 +16,7 @@
 import clients.fortnite_api as fortnite_api
 import clients.fortnite_tracker as fortnite_tracker
 import clients.interactions as interactions
+import clients.openai as openai
 import clients.stats as stats
 import commands
 from auth import validate
@@ -38,6 +39,8 @@
 initialize_error_handlers(app)
 initialize_request_logger(app)
 
+openai.initialize()
+
 eliminated_by_me_dict = None
 eliminated_me_dict = None
 
@@ -60,10 +63,10 @@ def post():
     """
     try:
         elim_data = request.get_json()
-    except BadRequest as e:
+    except BadRequest as exc:
         return jsonify({
             "message": "Invalid JSON payload",
-            "error": e
+            "error": exc
         }), 400
 
     global eliminated_by_me_dict
@@ -98,19 +101,24 @@ async def on_guild_join(guild):
 
 @bot.event
 async def on_voice_state_update(member, before, after):
+    logger = get_logger_with_context(identifier="Main")
+
     """ Event handler to track squad stats on voice channel join """
-    if interactions.should_add_player_to_squad_player_session_list(member, before, after):
-        if member.display_name in FORTNITE_DISCORD_ROLE_USERS_DICT:
-            if FORTNITE_DISCORD_ROLE_USERS_DICT[member.display_name] not in SQUAD_PLAYERS_LIST:
-                SQUAD_PLAYERS_LIST.append(FORTNITE_DISCORD_ROLE_USERS_DICT[member.display_name])
+    try:
+        if interactions.should_add_player_to_squad_player_session_list(member, before, after):
+            if member.display_name in FORTNITE_DISCORD_ROLE_USERS_DICT:
+                if FORTNITE_DISCORD_ROLE_USERS_DICT[member.display_name] not in SQUAD_PLAYERS_LIST:
+                    SQUAD_PLAYERS_LIST.append(FORTNITE_DISCORD_ROLE_USERS_DICT[member.display_name])
 
-    if interactions.should_remove_player_from_squad_player_session_list(member, before, after):
-        if member.display_name in FORTNITE_DISCORD_ROLE_USERS_DICT:
-            if FORTNITE_DISCORD_ROLE_USERS_DICT[member.display_name] in SQUAD_PLAYERS_LIST:
-                SQUAD_PLAYERS_LIST.pop(FORTNITE_DISCORD_ROLE_USERS_DICT[member.display_name])
+        if interactions.should_remove_player_from_squad_player_session_list(member, before, after):
+            if member.display_name in FORTNITE_DISCORD_ROLE_USERS_DICT:
+                if FORTNITE_DISCORD_ROLE_USERS_DICT[member.display_name] in SQUAD_PLAYERS_LIST:
+                    SQUAD_PLAYERS_LIST.pop(FORTNITE_DISCORD_ROLE_USERS_DICT[member.display_name])
 
-    if not interactions.send_track_question(member, before, after):
-        return
+        if not interactions.send_track_question(member, before, after):
+            return
+    except Exception as e:
+        logger.warning("Failed to run on_voice_state_update: %s", repr(e), exc_info=True)
 
     ctx, silent = await interactions.send_track_question_and_wait(
         bot,
@@ -294,15 +302,15 @@ async def replays_operations(ctx, *params):
 
     if command in commands.REPLAYS_ELIMINATED_COMMANDS:
         logger.info("Outputting players that got eliminated by us")
-        await output_replay_eliminated_by_me_stats_message(ctx, eliminated_by_me_dict, username, silent=False)
+        await _output_replay_eliminated_by_me_stats_message(ctx, eliminated_by_me_dict, username, silent=False)
     elif command in commands.REPLAYS_LOG_COMMANDS:
         logger.info("Silent logging players that got eliminated by us and eliminated us")
-        await output_replay_eliminated_me_stats_message(ctx, eliminated_me_dict, username, silent=True)
-        await output_replay_eliminated_by_me_stats_message(ctx, eliminated_by_me_dict, username, silent=True)
+        await _output_replay_eliminated_me_stats_message(ctx, eliminated_me_dict, username, silent=True)
+        await _output_replay_eliminated_by_me_stats_message(ctx, eliminated_by_me_dict, username, silent=True)
     else:
         if not command:
             logger.info("Outputting players that eliminated us")
-            await output_replay_eliminated_me_stats_message(ctx, eliminated_me_dict, username, silent=False)
+            await _output_replay_eliminated_me_stats_message(ctx, eliminated_me_dict, username, silent=False)
         elif command != commands.REPLAYS_COMMAND and \
                 command not in commands.REPLAYS_ELIMINATED_COMMANDS and \
                 command not in commands.REPLAYS_LOG_COMMANDS:
@@ -311,7 +319,7 @@ async def replays_operations(ctx, *params):
             await ctx.send(f"{command} left the channel")
 
 
-async def output_replay_eliminated_me_stats_message(ctx, eliminated_me_dict, username, silent):
+async def _output_replay_eliminated_me_stats_message(ctx, eliminated_me_dict, username, silent):
     """ Create Discord Message for the stats of the opponents that eliminated us"""
     for player_guid in eliminated_me_dict:
         squad_players_eliminated_by_player = ""
@@ -330,7 +338,7 @@ async def output_replay_eliminated_me_stats_message(ctx, eliminated_me_dict, use
             await player_search(ctx, player_guid, guid=True, silent=silent)
 
 
-async def output_replay_eliminated_by_me_stats_message(ctx, eliminated_by_me_dict, username, silent):
+async def _output_replay_eliminated_by_me_stats_message(ctx, eliminated_by_me_dict, username, silent):
     """ Create Discord Message for the stats of the opponents that got eliminated by us"""
     if username:
         eliminated_by_me_dict = {username: eliminated_by_me_dict[username]}
@@ -341,6 +349,24 @@ async def output_replay_eliminated_by_me_stats_message(ctx, eliminated_by_me_dic
             await player_search(ctx, player_guid, guid=True, silent=silent)
 
 
+@bot.command(name=commands.ASK_COMMAND,
+             help=commands.ASK_DESCRIPTION,
+             aliases=commands.ASK_ALIASES)
+@log_command
+async def ask_chatgpt(ctx, *params):
+    """ Ask OpenAI ChatGPT a question """
+    # TODO: Allow each file to invoke it's own logger with the correct context
+    logger = get_logger_with_context(ctx)
+
+    prompt = " ".join(params)
+
+    try:
+        resp = await openai.ask_chatgpt(prompt, logger)
+    except Exception as exc:
+        logger.warning(exc, exc_info=_should_log_traceback(exc))
+
+    await ctx.send(resp)
+
 
 def _should_log_traceback(e):
     """ Returns True if a traceback should be logged,

requirements.txt

@@ -5,6 +5,7 @@ cloudscraper~=1.2.69
 discord.py~=1.7.1
 flask~=2.2.3
 fortnite-replay-reader~=0.3.0
+openai~=0.27.0
 pycryptodome~=3.17
 python-dotenv~=1.0.0
 requests~=2.28.2