diff --git a/py/api.py b/py/api.py index 70c848e..6ba23c8 100644 --- a/py/api.py +++ b/py/api.py @@ -1,5 +1,4 @@ from time import sleep - from flask import Flask, request, jsonify from flask_cors import CORS import secrets @@ -13,53 +12,58 @@ from tts import TTS class API: def __init__(self): - self.crypt_size = 64 - self.app = Flask(__name__) - self.ai_response = {} - self.ai = AI() - self.db = DB() - self.weather = Weather() - self.voice = VoiceRecognition() - self.tts = TTS() - self.db.load_database() - self.ai_response_lock = threading.Lock() - CORS(self.app) + # Initialize the API class with necessary components and configurations + self.crypt_size = 64 # Size for generating secure tokens + self.app = Flask(__name__) # Create a Flask app instance + self.ai_response = {} # Dictionary to store AI responses keyed by access token + self.ai = AI() # AI processing instance + self.db = DB() # Database instance + self.weather = Weather() # Weather data retrieval instance + self.voice = VoiceRecognition() # Voice recognition instance + self.tts = TTS() # Text-to-Speech instance + self.db.load_database() # Load the database on startup + self.ai_response_lock = threading.Lock() # Lock for thread-safe access to AI responses + CORS(self.app) # Enable Cross-Origin Resource Sharing def run(self): + # Route to create a new AI session @self.app.route("/interstellar_ai/api/ai_create", methods=["GET"]) def create_ai(): - access_token = secrets.token_urlsafe(self.crypt_size) + access_token = secrets.token_urlsafe(self.crypt_size) # Generate a secure access token + # Ensure the access token is unique while access_token in self.ai_response: access_token = secrets.token_urlsafe(self.crypt_size) - self.ai_response[access_token] = "" + self.ai_response[access_token] = "" # Initialize the response for the new session return jsonify({"status": 200, "access_token": access_token}) + # Route to send messages to the AI @self.app.route("/interstellar_ai/api/ai_send", methods=["POST"]) def send_ai(): - data = request.get_json() - messages = data.get("messages") - model_type = data.get("model_type") - ai_model = data.get("ai_model") - access_token = data.get("access_token") + data = request.get_json() # Get JSON data from the request + messages = data.get("messages") # Extract messages + model_type = data.get("model_type") # Extract model type + ai_model = data.get("ai_model") # Extract AI model name + access_token = data.get("access_token") # Extract access token - print(model_type) + print(model_type) # Debugging output if access_token not in self.ai_response: - return jsonify({"status": 401, "error": "Invalid access token"}) + return jsonify({"status": 401, "error": "Invalid access token"}) # Token validation + # Start a new thread to process AI requests based on the model type if model_type == "local": thread = threading.Thread( target=self.ai.process_local, args=(ai_model, messages, self, access_token), ) thread.start() - thread.join() - sleep(0.5) + thread.join() # Wait for the thread to complete + sleep(0.5) # Sleep for a short duration return jsonify({"status": 200}) elif model_type == "mistral": - print(model_type) - api_key = data.get("api_key") + print(model_type) # Debugging output + api_key = data.get("api_key") # Get API key thread = threading.Thread( target=self.ai.process_mistralai, args=(ai_model, messages, self, access_token, api_key), @@ -69,7 +73,7 @@ class API: sleep(0.5) return jsonify({"status": 200}) elif model_type == "openai": - api_key = data.get("api_key") + api_key = data.get("api_key") # Get API key thread = threading.Thread( target=self.ai.process_openai, args=(ai_model, messages, self, access_token, api_key), @@ -79,7 +83,7 @@ class API: sleep(0.5) return jsonify({"status": 200}) elif model_type == "anthropic": - api_key = data.get("api_key") + api_key = data.get("api_key") # Get API key thread = threading.Thread( target=self.ai.process_anthropic, args=(ai_model, messages, self, access_token, api_key), @@ -89,7 +93,7 @@ class API: sleep(0.5) return jsonify({"status": 200}) elif model_type == "google": - api_key = data.get("api_key") + api_key = data.get("api_key") # Get API key thread = threading.Thread( target=self.ai.process_google, args=(ai_model, messages, self, access_token, api_key), @@ -99,79 +103,85 @@ class API: sleep(0.5) return jsonify({"status": 200}) - return jsonify({"status": 401, "error": "Invalid AI model type"}) + return jsonify({"status": 401, "error": "Invalid AI model type"}) # Model type validation + # Route to retrieve AI response based on access token @self.app.route("/interstellar_ai/api/ai_get", methods=["GET"]) def get_ai(): - data = request.args.get("access_token") + data = request.args.get("access_token") # Get access token from query parameters if data not in self.ai_response: - return jsonify({"status": 401, "error": "Invalid access token"}) - return jsonify({"status": 200, "response": self.ai_response[data]}) + return jsonify({"status": 401, "error": "Invalid access token"}) # Token validation + return jsonify({"status": 200, "response": self.ai_response[data]}) # Return AI response + # Route for database operations @self.app.route("/interstellar_ai/db", methods=["POST"]) def db_manipulate(): - sent_data = request.get_json() - action = sent_data.get("action") + sent_data = request.get_json() # Get JSON data from the request + action = sent_data.get("action") # Extract action type if action == "create_account": - return jsonify({"status": 200, "response": self.db.add_user(sent_data)}) + return jsonify({"status": 200, "response": self.db.add_user(sent_data)}) # Add user elif action == "change_password": return jsonify( - {"status": 200, "response": self.db.update_password(sent_data)} + {"status": 200, "response": self.db.update_password(sent_data)} # Update password ) elif action == "get_settings": return jsonify( - {"status": 200, "response": self.db.get_settings(sent_data)} + {"status": 200, "response": self.db.get_settings(sent_data)} # Get user settings ) elif action == "change_settings": return jsonify( - {"status": 200, "response": self.db.change_settings(sent_data)} + {"status": 200, "response": self.db.change_settings(sent_data)} # Change user settings ) elif action == "get_history": return jsonify( - {"status": 200, "response": self.db.get_history(sent_data)} + {"status": 200, "response": self.db.get_history(sent_data)} # Get user history ) elif action == "change_history": return jsonify( - {"status": 200, "response": self.db.change_history(sent_data)} + {"status": 200, "response": self.db.change_history(sent_data)} # Change user history ) elif action == "check_credentials": return jsonify( - {"status": 200, "response": self.db.check_credentials(sent_data)} + {"status": 200, "response": self.db.check_credentials(sent_data)} # Check user credentials ) elif action == "delete_account": return jsonify( - {"status": 200, "response": self.db.delete_user(sent_data)} + {"status": 200, "response": self.db.delete_user(sent_data)} # Delete user account ) elif action == "get_email": return jsonify( - {"status": 200, "response": self.db.get_email(sent_data)} + {"status": 200, "response": self.db.get_email(sent_data)} # Get user email ) elif action == "get_name": - return jsonify({"status": 200, "response": self.db.get_name(sent_data)}) + return jsonify({"status": 200, "response": self.db.get_name(sent_data)}) # Get user name - return jsonify({"status": 401, "response": "Invalid action"}) + return jsonify({"status": 401, "response": "Invalid action"}) # Action validation + # Route for voice recognition @self.app.route("/interstellar_ai/api/voice_recognition", methods=["POST"]) def voice_recognition(): - audio = request.files.get("audio") - text = self.voice.recognition(audio) - return jsonify({"status": 200, "response": text}) + audio = request.files.get("audio") # Get audio file from request + text = self.voice.recognition(audio) # Perform voice recognition + return jsonify({"status": 200, "response": text}) # Return recognized text + # Route for weather information retrieval @self.app.route("/interstellar_ai/api/weather", methods=["POST"]) def get_weather(): - sent_data = request.get_json() - unit_type = sent_data.get("unit_type") - city = sent_data.get("city") - weather_data = self.weather.getweather(unit_type, city) - return jsonify({"status": 200, "response": weather_data}) + sent_data = request.get_json() # Get JSON data from the request + unit_type = sent_data.get("unit_type") # Extract unit type (metric, imperial) + city = sent_data.get("city") # Extract city name + weather_data = self.weather.getweather(unit_type, city) # Get weather data + return jsonify({"status": 200, "response": weather_data}) # Return weather data - self.app.run(debug=True, host="0.0.0.0", port=5000) + self.app.run(debug=True, host="0.0.0.0", port=5000) # Start the Flask app + # Route for Text-to-Speech conversion @self.app.route("/interstellar_ai/api/tts", methods=["POST"]) def tts(): - text = request.args.get("text") - return jsonify({"status": 200, "response": self.tts.gen_tts(text)}) + text = request.args.get("text") # Get text from query parameters + return jsonify({"status": 200, "response": self.tts.gen_tts(text)}) # Generate TTS and return response +# Initialize the API class and run the application api = API() api.run()