202 lines
6.4 KiB
Python
202 lines
6.4 KiB
Python
import json
|
|
from quart import Blueprint, jsonify, request, abort, current_app
|
|
from werkzeug.exceptions import HTTPException
|
|
from internal.types import ActiveClient, DiscordId
|
|
from internal.db_wrappers import DiscordIdDbController
|
|
|
|
bot_bp = Blueprint('bot', __name__)
|
|
|
|
|
|
# ------- Discord Token Functions
|
|
@bot_bp.route('/request_token', methods=['POST'])
|
|
@jwt_required
|
|
@role_required(UserRoles.MOD)
|
|
async def request_token_route():
|
|
"""
|
|
API endpoint to request a token for a client.
|
|
Expects 'client_id' in the JSON request body.
|
|
"""
|
|
try:
|
|
request_data = await request.get_json()
|
|
if not request_data or 'client_id' not in request_data:
|
|
abort(400, "Missing 'client_id' in request body.")
|
|
|
|
client_id = request_data['client_id']
|
|
print(f"Request received for client_id: {client_id}")
|
|
|
|
# get the available IDs
|
|
active_d_ids = await current_app.d_id_db_h.find_discord_ids(active_only=True)
|
|
|
|
# init available IDs list
|
|
avail_ids = []
|
|
# Init the selected ID
|
|
selected_id = None
|
|
|
|
# Check which IDs are currently in use by other bots
|
|
for d_id in active_d_ids:
|
|
if not find_token_in_active_clients(d_id.token):
|
|
avail_ids.append(d_id)
|
|
|
|
if not avail_ids:
|
|
abort(404, "No available active Discord IDs found.")
|
|
|
|
# --- Logic for selecting a preferred ID based on client_id (TODO) ---
|
|
|
|
selected_id = avail_ids[0]
|
|
current_app.active_clients[client_id].active_token = selected_id
|
|
|
|
# --- End of logic for selecting a preferred ID ---
|
|
|
|
return jsonify(selected_id.to_dict())
|
|
|
|
except Exception as e:
|
|
print(f"Error in request_token_route: {e}")
|
|
abort(500, f"An internal error occurred: {e}")
|
|
|
|
|
|
@bot_bp.route('/tokens/', methods=['GET'])
|
|
@jwt_required
|
|
@role_required(UserRoles.USER)
|
|
async def get_all_discord_tokens():
|
|
"""
|
|
API endpoint to return all discord IDs
|
|
"""
|
|
try:
|
|
# get the available IDs
|
|
d_ids = await current_app.d_id_db_h.find_discord_ids(active_only=False)
|
|
return jsonify([d_id.to_dict() for d_id in d_ids])
|
|
|
|
except Exception as e:
|
|
print(f"Error in request_token_route: {e}")
|
|
abort(500, f"An internal error occurred: {e}")
|
|
|
|
|
|
@bot_bp.route('/token/<string:discord_id_param>', methods=['GET'])
|
|
@jwt_required
|
|
@role_required(UserRoles.MOD)
|
|
async def get_discord_token_by_id(discord_id_param: str):
|
|
"""
|
|
API endpoint to get a single Discord ID by its _id.
|
|
"""
|
|
try:
|
|
query = {"_id": discord_id_param}
|
|
d_id = await current_app.d_id_db_h.find_discord_id(query)
|
|
if d_id:
|
|
return jsonify(d_id.to_dict())
|
|
else:
|
|
abort(404, "Discord ID not found.")
|
|
except Exception as e:
|
|
print(f"Error in get_discord_token_by_id: {e}")
|
|
abort(500, f"An internal error occurred: {e}")
|
|
|
|
|
|
@bot_bp.route('/token', methods=["POST"])
|
|
@jwt_required
|
|
@role_required(UserRoles.MOD)
|
|
async def create_discord_token():
|
|
"""
|
|
API Endpoint to create a discord token (adding a new bot)
|
|
"""
|
|
data = await request.get_json()
|
|
|
|
if '_id' in data:
|
|
id_search_result = await current_app.sys_db_h.find_system({"_id": data["_id"]})
|
|
if id_search_result:
|
|
# If _id is provided and exists, return conflict
|
|
abort(409, f"System with ID '{data['_id']}' already exists")
|
|
|
|
temp_name = data.get("name")
|
|
temp_discord_id = data.get("discord_id")
|
|
temp_token = data.get("token")
|
|
temp_active = bool(data.get("active"))
|
|
temp_guilds = data.get("guild_ids")
|
|
|
|
# Create the discord ID object
|
|
temp_d_id = {
|
|
"discord_id": temp_discord_id,
|
|
"name": temp_name,
|
|
"token": temp_token,
|
|
"active": temp_active,
|
|
"guild_ids": temp_guilds
|
|
}
|
|
|
|
new_d_id = await current_app.d_id_db_h.create_discord_id(temp_d_id)
|
|
|
|
if new_d_id:
|
|
return jsonify(new_d_id.to_dict()), 201
|
|
else:
|
|
abort(500, "Failed to create Discord ID.")
|
|
|
|
|
|
@bot_bp.route('/token/<string:discord_id_param>', methods=['PUT'])
|
|
@jwt_required
|
|
@role_required(UserRoles.MOD)
|
|
async def update_discord_token(discord_id_param: str):
|
|
"""
|
|
API endpoint to update a Discord ID by its _id.
|
|
"""
|
|
try:
|
|
data = await request.get_json()
|
|
if not data:
|
|
abort(400, "No update data provided.")
|
|
|
|
query = {"_id": discord_id_param}
|
|
update_result = await current_app.d_id_db_h.update_discord_id(query, {"$set": data})
|
|
|
|
if update_result is not None:
|
|
if update_result > 0:
|
|
# Optionally, fetch the updated document to return it
|
|
updated_d_id = await current_app.d_id_db_h.find_discord_id(query)
|
|
return jsonify(updated_d_id.to_dict()), 200
|
|
else:
|
|
abort(404, "Discord ID not found or no changes applied.")
|
|
else:
|
|
abort(500, "Failed to update Discord ID.")
|
|
except Exception as e:
|
|
print(f"Error in update_discord_token: {e}")
|
|
abort(500, f"An internal error occurred: {e}")
|
|
|
|
|
|
@bot_bp.route('/token/<string:discord_id_param>', methods=['DELETE'])
|
|
@jwt_required
|
|
@role_required(UserRoles.MOD)
|
|
async def delete_discord_token(discord_id_param: str):
|
|
"""
|
|
API endpoint to delete a Discord ID by its _id.
|
|
"""
|
|
try:
|
|
query = {"_id": discord_id_param}
|
|
delete_count = await current_app.d_id_db_h.delete_discord_id(query)
|
|
|
|
if delete_count is not None:
|
|
if delete_count > 0:
|
|
return jsonify({"message": f"Successfully deleted {delete_count} Discord ID(s)."}), 200
|
|
else:
|
|
abort(404, "Discord ID not found.")
|
|
else:
|
|
abort(500, "Failed to delete Discord ID.")
|
|
except Exception as e:
|
|
print(f"Error in delete_discord_token: {e}")
|
|
abort(500, f"An internal error occurred: {e}")
|
|
|
|
|
|
# ------- Util Functions
|
|
|
|
def find_token_in_active_clients(target_token: str) -> bool:
|
|
"""
|
|
Checks if a target_token exists in the active_token of any ActiveClient object in a list.
|
|
|
|
Args:
|
|
clients: A list of ActiveClient objects.
|
|
target_token: The token string to search for.
|
|
|
|
Returns:
|
|
True if the token is found in any ActiveClient, False otherwise.
|
|
"""
|
|
for client_id in current_app.active_clients:
|
|
try:
|
|
if current_app.active_clients[client_id].active_token.token == target_token:
|
|
return True
|
|
except Exception as e:
|
|
pass
|
|
return False |