interstellar_ai/py/api.py

from time import sleep
from flask import Flask, request, jsonify
from flask_cors import CORS
import secrets
import threading
from ai import AI
from db import DB
from weather import Weather
from voice import VoiceRecognition
from tts import TTS


class API:
    def __init__(self):
        # 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)  # 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] = ""  # 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()  # 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)  # Debugging output
            if access_token not in self.ai_response:
                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()  # 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)  # 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),
                )
                thread.start()
                thread.join()
                sleep(0.5)
                return jsonify({"status": 200})
            elif model_type == "openai":
                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),
                )
                thread.start()
                thread.join()
                sleep(0.5)
                return jsonify({"status": 200})
            elif model_type == "anthropic":
                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),
                )
                thread.start()
                thread.join()
                sleep(0.5)
                return jsonify({"status": 200})
            elif model_type == "google":
                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),
                )
                thread.start()
                thread.join()
                sleep(0.5)
                return jsonify({"status": 200})

            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")  # Get access token from query parameters
            if data not in self.ai_response:
                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()  # 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)})  # Add user
            elif action == "change_password":
                return jsonify(
                    {"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)}  # Get user settings
                )
            elif action == "change_settings":
                return jsonify(
                    {"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)}  # Get user history
                )
            elif action == "change_history":
                return jsonify(
                    {"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)}  # Check user credentials
                )
            elif action == "delete_account":
                return jsonify(
                    {"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)}  # Get user email
                )
            elif action == "get_name":
                return jsonify({"status": 200, "response": self.db.get_name(sent_data)})  # Get user name

            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")  # 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()  # 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)  # 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")  # 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()