sageTheDm/interstellar_ai
1
0
Fork 0
forked from React-Group/interstellar_ai
Code Pull requests Activity

api.py comments

Browse source
This commit is contained in:
sageTheDM 2024-10-11 10:12:25 +02:00
parent 4debcc6fad
commit 2eb4311f0a
1 changed files with 66 additions and 56 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

122
py/api.py
View file

@ -1,5 +1,4 @@
from time import sleep from time import sleep
from flask import Flask, request, jsonify from flask import Flask, request, jsonify
from flask_cors import CORS from flask_cors import CORS
import secrets import secrets
@ -13,53 +12,58 @@ from tts import TTS
class API: class API:
def __init__(self): def __init__(self):
self.crypt_size = 64 # Initialize the API class with necessary components and configurations
self.app = Flask(__name__) self.crypt_size = 64 # Size for generating secure tokens
self.ai_response = {} self.app = Flask(__name__) # Create a Flask app instance
self.ai = AI() self.ai_response = {} # Dictionary to store AI responses keyed by access token
self.db = DB() self.ai = AI() # AI processing instance
self.weather = Weather() self.db = DB() # Database instance
self.voice = VoiceRecognition() self.weather = Weather() # Weather data retrieval instance
self.tts = TTS() self.voice = VoiceRecognition() # Voice recognition instance
self.db.load_database() self.tts = TTS() # Text-to-Speech instance
self.ai_response_lock = threading.Lock() self.db.load_database() # Load the database on startup
CORS(self.app) 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): def run(self):
# Route to create a new AI session
@self.app.route("/interstellar_ai/api/ai_create", methods=["GET"]) @self.app.route("/interstellar_ai/api/ai_create", methods=["GET"])
def create_ai(): 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: while access_token in self.ai_response:
access_token = secrets.token_urlsafe(self.crypt_size) 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}) 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"]) @self.app.route("/interstellar_ai/api/ai_send", methods=["POST"])
def send_ai(): def send_ai():
data = request.get_json() data = request.get_json() # Get JSON data from the request
messages = data.get("messages") messages = data.get("messages") # Extract messages
model_type = data.get("model_type") model_type = data.get("model_type") # Extract model type
ai_model = data.get("ai_model") ai_model = data.get("ai_model") # Extract AI model name
access_token = data.get("access_token") 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: 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": if model_type == "local":
thread = threading.Thread( thread = threading.Thread(
target=self.ai.process_local, target=self.ai.process_local,
args=(ai_model, messages, self, access_token), args=(ai_model, messages, self, access_token),
) )
thread.start() thread.start()
thread.join() thread.join() # Wait for the thread to complete
sleep(0.5) sleep(0.5) # Sleep for a short duration
return jsonify({"status": 200}) return jsonify({"status": 200})
elif model_type == "mistral": elif model_type == "mistral":
print(model_type) print(model_type) # Debugging output
api_key = data.get("api_key") api_key = data.get("api_key") # Get API key
thread = threading.Thread( thread = threading.Thread(
target=self.ai.process_mistralai, target=self.ai.process_mistralai,
args=(ai_model, messages, self, access_token, api_key), args=(ai_model, messages, self, access_token, api_key),
@ -69,7 +73,7 @@ class API:
sleep(0.5) sleep(0.5)
return jsonify({"status": 200}) return jsonify({"status": 200})
elif model_type == "openai": elif model_type == "openai":
api_key = data.get("api_key") api_key = data.get("api_key") # Get API key
thread = threading.Thread( thread = threading.Thread(
target=self.ai.process_openai, target=self.ai.process_openai,
args=(ai_model, messages, self, access_token, api_key), args=(ai_model, messages, self, access_token, api_key),
@ -79,7 +83,7 @@ class API:
sleep(0.5) sleep(0.5)
return jsonify({"status": 200}) return jsonify({"status": 200})
elif model_type == "anthropic": elif model_type == "anthropic":
api_key = data.get("api_key") api_key = data.get("api_key") # Get API key
thread = threading.Thread( thread = threading.Thread(
target=self.ai.process_anthropic, target=self.ai.process_anthropic,
args=(ai_model, messages, self, access_token, api_key), args=(ai_model, messages, self, access_token, api_key),
@ -89,7 +93,7 @@ class API:
sleep(0.5) sleep(0.5)
return jsonify({"status": 200}) return jsonify({"status": 200})
elif model_type == "google": elif model_type == "google":
api_key = data.get("api_key") api_key = data.get("api_key") # Get API key
thread = threading.Thread( thread = threading.Thread(
target=self.ai.process_google, target=self.ai.process_google,
args=(ai_model, messages, self, access_token, api_key), args=(ai_model, messages, self, access_token, api_key),
@ -99,79 +103,85 @@ class API:
sleep(0.5) sleep(0.5)
return jsonify({"status": 200}) 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"]) @self.app.route("/interstellar_ai/api/ai_get", methods=["GET"])
def get_ai(): 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: if data not in self.ai_response:
return jsonify({"status": 401, "error": "Invalid access token"}) return jsonify({"status": 401, "error": "Invalid access token"}) # Token validation
return jsonify({"status": 200, "response": self.ai_response[data]}) return jsonify({"status": 200, "response": self.ai_response[data]}) # Return AI response
# Route for database operations
@self.app.route("/interstellar_ai/db", methods=["POST"]) @self.app.route("/interstellar_ai/db", methods=["POST"])
def db_manipulate(): def db_manipulate():
sent_data = request.get_json() sent_data = request.get_json() # Get JSON data from the request
action = sent_data.get("action") action = sent_data.get("action") # Extract action type
if action == "create_account": 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": elif action == "change_password":
return jsonify( 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": elif action == "get_settings":
return jsonify( 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": elif action == "change_settings":
return jsonify( 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": elif action == "get_history":
return jsonify( 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": elif action == "change_history":
return jsonify( 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": elif action == "check_credentials":
return jsonify( 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": elif action == "delete_account":
return jsonify( 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": elif action == "get_email":
return jsonify( 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": 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"]) @self.app.route("/interstellar_ai/api/voice_recognition", methods=["POST"])
def voice_recognition(): def voice_recognition():
audio = request.files.get("audio") audio = request.files.get("audio") # Get audio file from request
text = self.voice.recognition(audio) text = self.voice.recognition(audio) # Perform voice recognition
return jsonify({"status": 200, "response": text}) return jsonify({"status": 200, "response": text}) # Return recognized text
# Route for weather information retrieval
@self.app.route("/interstellar_ai/api/weather", methods=["POST"]) @self.app.route("/interstellar_ai/api/weather", methods=["POST"])
def get_weather(): def get_weather():
sent_data = request.get_json() sent_data = request.get_json() # Get JSON data from the request
unit_type = sent_data.get("unit_type") unit_type = sent_data.get("unit_type") # Extract unit type (metric, imperial)
city = sent_data.get("city") city = sent_data.get("city") # Extract city name
weather_data = self.weather.getweather(unit_type, city) weather_data = self.weather.getweather(unit_type, city) # Get weather data
return jsonify({"status": 200, "response": 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"]) @self.app.route("/interstellar_ai/api/tts", methods=["POST"])
def tts(): def tts():
text = request.args.get("text") text = request.args.get("text") # Get text from query parameters
return jsonify({"status": 200, "response": self.tts.gen_tts(text)}) 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 = API()
api.run() api.run()