125 lines
5.2 KiB
Python
125 lines
5.2 KiB
Python
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})
|
|
|
|
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():
|
|
action = request.args.get('action')
|
|
data = request.args.get('data')
|
|
if action == "create_account":
|
|
return jsonify({'status': 200, 'response': self.db.add_user(data)})
|
|
elif action == "change_password":
|
|
return jsonify({'status': 200, 'response': self.db.update_password(data)})
|
|
elif action == "get_data":
|
|
return jsonify({'status': 200, 'response': self.db.get_data(data)})
|
|
elif action == "change_data":
|
|
return jsonify({'status': 200, 'response': self.db.change_data(data)})
|
|
elif action == "check_credentials":
|
|
return jsonify({'status': 200, 'response': self.db.check_credentials(data)})
|
|
|
|
return jsonify({'status': 401, 'response': "Invalid action"})
|
|
|
|
@self.app.route('/interstellar_ai/api/voice_recognition', methods=['POST'])
|
|
def voice_recognition():
|
|
recognition_type = request.args.get('type')
|
|
audio = request.args.get('audio')
|
|
option = request.args.get('option')
|
|
if recognition_type == "basic":
|
|
return jsonify({'status': 200, 'response': self.voice.basic_recognition(audio, option)})
|
|
|
|
return jsonify({'status': 401, 'response': "Invalid type"})
|
|
|
|
@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()
|