From 3d111dce7d2f410c6460479837c980966146384e Mon Sep 17 00:00:00 2001 From: cyclop-bot <178948048+cyclop-bot@users.noreply.github.com> Date: Tue, 3 Jun 2025 15:43:05 -0500 Subject: [PATCH] feat: Flesh out api_helper.py with command handling --- api_helper.py | 136 +++++++++++++++++++++++++++++++++++++++++--------- 1 file changed, 113 insertions(+), 23 deletions(-) diff --git a/api_helper.py b/api_helper.py index 501bc4b..e59b871 100644 --- a/api_helper.py +++ b/api_helper.py @@ -1,8 +1,32 @@ - import http.server import socketserver import os import logging +import asyncio +# Assuming InferenceBot is available in the same environment or can be imported +# For demonstration, we'll use a placeholder if not explicitly provided. +try: + from inference_bot import InferenceBot +except ImportError: + logging.warning("InferenceBot not found. Using a placeholder for APIHelper.") + class InferenceBot: + def __init__(self): + self.history = {} + self.status_message = "Bot is operational." + self.processing_status = {} + async def start(self): return "Placeholder Bot started." + def clear_conversation_history(self, user_id): self.history[user_id] = [] + def get_bot_status(self): return self.status_message + async def switch_model(self): return "Placeholder model switched." + async def handle_message(self, user_id, message): + self.history.setdefault(user_id, []).append(f"User: {message}") + response = f"Placeholder Bot received: {message}" + self.history[user_id].append(f"Bot: {response}") + return response + async def abort_processing(self, user_id): return "Placeholder processing aborted." + def set_processing_status(self, user_id, message_id): self.processing_status[user_id] = message_id + def clear_processing_status(self, user_id): self.processing_status.pop(user_id, None) + # Configure logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') @@ -11,40 +35,102 @@ COPILOT_HOST = os.getenv("COPILOT_HOST", "0.0.0.0") COPILOT_PORT = int(os.getenv("COPILOT_PORT", 8000)) COPILOT_PATH = "/copilot" +# A single user ID for API interactions, as there's no multi-user concept here +API_USER_ID = 1 + +class APIHelper: + def __init__(self, bot: InferenceBot): + self.bot = bot + + async def _start_logic(self) -> str: + return await self.bot.start() + + async def _clear_logic(self, user_id: int) -> str: + self.bot.clear_conversation_history(user_id) + return "Conversation history cleared. Let's start fresh!" + + def _status_logic(self) -> str: + return self.bot.get_bot_status() + + async def _switch_logic(self) -> str: + if hasattr(self.bot, 'switch_model'): + return await self.bot.switch_model() + else: + return "Model switching is not supported for this bot." + + async def _handle_message_logic(self, user_id: int, user_message: str) -> str: + try: + response = await self.bot.handle_message(user_id, user_message) + return response + except Exception as e: + logging.error(f"Error in _handle_message_logic for user {user_id}: {str(e)}") + return f"Error processing message: {str(e)}" + + class CopilotRequestHandler(http.server.BaseHTTPRequestHandler): + # This will be set by the server when it's created + api_helper_instance: APIHelper = None + + def _send_response(self, status_code: int, content: str): + self.send_response(status_code) + self.send_header('Content-type', 'text/plain; charset=utf-8') + self.end_headers() + self.wfile.write(content.encode('utf-8')) + def do_POST(self): if self.path == COPILOT_PATH: content_length = int(self.headers['Content-Length']) post_data_bytes = self.rfile.read(content_length) - post_data_str = post_data_bytes.decode('utf-8') + user_message = post_data_bytes.decode('utf-8').strip() - logging.info(f"Received data from {self.client_address[0]}: {post_data_str}") + logging.info(f"Received POST from {self.client_address[0]}: {user_message}") - # In a real scenario, you would pass post_data_str to your AI model - # and get a response. For now, we just echo it back. - response_text = f"Copilot received: {post_data_str}" - - self.send_response(200) - self.send_header('Content-type', 'text/plain; charset=utf-8') - self.end_headers() - self.wfile.write(response_text.encode('utf-8')) + response_text = "" + # Use a fixed user ID for the API interaction + user_id = API_USER_ID + + if self.api_helper_instance is None: + logging.error("APIHelper instance not set on request handler.") + self._send_response(500, "Internal Server Error: API Helper not initialized.") + return + + # Simulate command handling based on message content + if user_message.startswith('/'): + command_parts = user_message.split(' ', 1) + command = command_parts[0] + + if command == '/start': + response_text = asyncio.run(self.api_helper_instance._start_logic()) + elif command == '/clear': + response_text = asyncio.run(self.api_helper_instance._clear_logic(user_id)) + elif command == '/status': + response_text = self.api_helper_instance._status_logic() + elif command == '/switch': + response_text = asyncio.run(self.api_helper_instance._switch_logic()) + else: + # For unknown commands, treat as a regular message or an error + response_text = asyncio.run(self.api_helper_instance._handle_message_logic(user_id, user_message)) + + else: + # Treat as a regular message + response_text = asyncio.run(self.api_helper_instance._handle_message_logic(user_id, user_message)) + + self._send_response(200, response_text) else: - self.send_response(404) - self.end_headers() - self.wfile.write(b"Not Found") + self._send_response(404, "Not Found") def do_GET(self): if self.path == "/health": - self.send_response(200) - self.send_header('Content-type', 'text/plain; charset=utf-8') - self.end_headers() - self.wfile.write(b"API Helper is running") + self._send_response(200, "API Helper is running") else: - self.send_response(404) - self.end_headers() - self.wfile.write(b"Not Found") + self._send_response(404, "Not Found") + +def run_server(bot_instance: InferenceBot, server_class=http.server.HTTPServer, handler_class=CopilotRequestHandler, host=COPILOT_HOST, port=COPILOT_PORT): + # Create an instance of APIHelper + api_helper = APIHelper(bot_instance) + # Attach the APIHelper instance to the handler class + handler_class.api_helper_instance = api_helper -def run_server(server_class=http.server.HTTPServer, handler_class=CopilotRequestHandler, host=COPILOT_HOST, port=COPILOT_PORT): server_address = (host, port) httpd = server_class(server_address, handler_class) logging.info(f"Starting Copilot API helper on http://{host}:{port}{COPILOT_PATH}") @@ -56,4 +142,8 @@ def run_server(server_class=http.server.HTTPServer, handler_class=CopilotRequest httpd.server_close() if __name__ == '__main__': - run_server() + # In a real deployment, you would pass a properly configured InferenceBot instance here. + # For standalone execution, we instantiate the placeholder InferenceBot. + logging.warning("Running api_helper.py in standalone mode with a placeholder InferenceBot.") + logging.warning("Ensure a proper InferenceBot instance is passed when integrating into a larger system.") + run_server(bot_instance=InferenceBot())