interstellar_ai/py/api.py

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):
        self.crypt_size = 4096
        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)

    def run(self):
        @self.app.route('/interstellar_ai/api/ai_create', methods=['GET'])
        def create_ai():
            access_token = secrets.token_urlsafe(self.crypt_size)

            while access_token in self.ai_response:
                access_token = secrets.token_urlsafe(self.crypt_size)

            self.ai_response[access_token] = ""
            return jsonify({'status': 200, 'access_token': access_token})

        @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')
            if access_token not in self.ai_response:
                return jsonify({'status': 401, 'error': 'Invalid access token'})

            if model_type == "local":
                thread = threading.Thread(target=self.ai.process_local, args=(ai_model, messages, self, access_token))
                thread.start()
                thread.join()
                return jsonify({'status': 200})
            elif model_type == "mistral":
                api_key = data.get('api_key')
                thread = threading.Thread(target=self.ai.process_mistralai,
                                          args=(ai_model, messages, self, access_token, api_key))
                thread.start()
                thread.join()
                return jsonify({'status': 200})
            elif model_type == "openai":
                api_key = data.get('api_key')
                thread = threading.Thread(target=self.ai.process_openai,
                                          args=(ai_model, messages, self, access_token, api_key))
                thread.start()
                thread.join()
                return jsonify({'status': 200})
            elif model_type == "anthropic":
                api_key = data.get('api_key')
                thread = threading.Thread(target=self.ai.process_anthropic,
                                          args=(ai_model, messages, self, access_token, api_key))
                thread.start()
                thread.join()
                return jsonify({'status': 200})
            elif model_type == "google":
                api_key = data.get('api_key')
                thread = threading.Thread(target=self.ai.process_google,
                                          args=(ai_model, messages, self, access_token, api_key))
                thread.start()
                thread.join()
                return jsonify({'status': 200})

            return jsonify({'status': 401, 'error': 'Invalid AI model type'})

        @self.app.route('/interstellar_ai/api/ai_get', methods=['GET'])
        def get_ai():
            data = request.args.get('access_token')
            if data not in self.ai_response:
                return jsonify({'status': 401, 'error': 'Invalid access token'})
            return jsonify({'status': 200, 'response': self.ai_response[data]})

        @self.app.route('/interstellar_ai/db', methods=['POST'])
        def db_manipulate():
            sent_data = request.get_json()
            action = sent_data.get('action')
            if action == "create_account":
                return jsonify({'status': 200, 'response': self.db.add_user(sent_data)})
            elif action == "change_password":
                return jsonify({'status': 200, 'response': self.db.update_password(sent_data)})
            elif action == "get_data":
                return jsonify({'status': 200, 'response': self.db.get_data(sent_data)})
            elif action == "change_data":
                return jsonify({'status': 200, 'response': self.db.change_data(sent_data)})
            elif action == "check_credentials":
                return jsonify({'status': 200, 'response': self.db.check_credentials(sent_data)})
            elif action == "delete_account":
                return jsonify({'status': 200, 'response': self.db.delete_user(sent_data)})

            return jsonify({'status': 401, 'response': "Invalid action"})

        @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})

        @self.app.route('/interstellar_ai/api/weather', methods=['POST'])
        def get_weather():
            unit_type = request.args.get('unit_type')
            city = request.args.get('city')
            return jsonify({'status': 200, 'response': self.weather.getweather(unit_type, city)})

        self.app.run(debug=True, host='0.0.0.0', port=5000)

        @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)})


api = API()
api.run()