Implemented bot endpoint with DB requests and other tweaks
This commit is contained in:
Logan Cusano
@@ -4,13 +4,14 @@ from uuid import uuid4
|
||||
from typing import Optional, List, Dict, Any
|
||||
from enum import Enum
|
||||
from internal.db_handler import MongoHandler
|
||||
from internal.types import System
|
||||
from internal.types import System, DiscordId
|
||||
|
||||
# Init vars
|
||||
DB_NAME = os.getenv("DB_NAME", "default_db")
|
||||
MONGO_URL = os.getenv("MONGO_URL", "mongodb://10.10.202.4:27017/")
|
||||
|
||||
SYSTEM_DB_COLLECTION_NAME = "radio_systems"
|
||||
DISCORD_ID_DB_COLLECTION_NAME = "discord_bot_ids"
|
||||
|
||||
# --- System class ---
|
||||
class SystemDbController():
|
||||
@@ -189,3 +190,160 @@ class SystemDbController():
|
||||
except Exception as e:
|
||||
print(f"Delete failed: {e}")
|
||||
return None
|
||||
|
||||
|
||||
# --- DiscordIdDbController class ---
|
||||
class DiscordIdDbController():
|
||||
def __init__(self):
|
||||
# Init the handler for Discord IDs
|
||||
self.db_h = MongoHandler(DB_NAME, DISCORD_ID_DB_COLLECTION_NAME, MONGO_URL)
|
||||
|
||||
async def create_discord_id(self, discord_id_data: Dict[str, Any]) -> Optional[DiscordId]:
|
||||
"""
|
||||
Creates a new Discord ID entry in the database.
|
||||
|
||||
Args:
|
||||
discord_id_data: A dictionary containing the data for the new Discord ID.
|
||||
|
||||
Returns:
|
||||
The created DiscordId object if successful, None otherwise.
|
||||
"""
|
||||
print("\n--- Creating a Discord ID document ---")
|
||||
try:
|
||||
if not discord_id_data.get("_id"):
|
||||
discord_id_data['_id'] = str(uuid4()) # Ensure _id is a string
|
||||
|
||||
inserted_id = None
|
||||
async with self.db_h as db:
|
||||
insert_result = await self.db_h.insert_one(discord_id_data)
|
||||
inserted_id = insert_result.inserted_id
|
||||
|
||||
if inserted_id:
|
||||
print(f"Discord ID insert successful with ID: {inserted_id}")
|
||||
query = {"_id": inserted_id}
|
||||
inserted_doc = None
|
||||
async with self.db_h as db:
|
||||
inserted_doc = await db.find_one(query)
|
||||
|
||||
if inserted_doc:
|
||||
return DiscordId.from_dict(inserted_doc)
|
||||
else:
|
||||
print("Discord ID insert acknowledged but no ID returned.")
|
||||
return None
|
||||
|
||||
except Exception as e:
|
||||
print(f"Discord ID create failed: {e}")
|
||||
return None
|
||||
|
||||
async def find_discord_id(self, query: Dict[str, Any], active_only: bool = False) -> Optional[DiscordId]:
|
||||
"""
|
||||
Finds a single Discord ID entry in the database.
|
||||
|
||||
Args:
|
||||
query: A dictionary representing the query criteria.
|
||||
active_only: If True, only returns active Discord IDs.
|
||||
|
||||
Returns:
|
||||
A DiscordId object if found, None otherwise.
|
||||
"""
|
||||
print("\n--- Finding one Discord ID document ---")
|
||||
try:
|
||||
if active_only:
|
||||
query["active"] = True
|
||||
|
||||
found_doc = None
|
||||
async with self.db_h as db:
|
||||
found_doc = await db.find_one(query)
|
||||
|
||||
if found_doc:
|
||||
print("Found Discord ID document (raw dict):", found_doc)
|
||||
return DiscordId.from_dict(found_doc)
|
||||
else:
|
||||
print("Discord ID document not found.")
|
||||
return None
|
||||
except Exception as e:
|
||||
print(f"Discord ID find failed: {e}")
|
||||
return None
|
||||
|
||||
async def find_discord_ids(self, query: Dict[str, Any] = {}, guild_id: Optional[str] = None, active_only: bool = False) -> Optional[List[DiscordId]]:
|
||||
"""
|
||||
Finds one or more Discord ID entries in the database.
|
||||
|
||||
Args:
|
||||
query: A dictionary representing the query criteria.
|
||||
guild_id: Optional. If provided, filters Discord IDs that belong to this guild.
|
||||
active_only: If True, only returns active Discord IDs.
|
||||
|
||||
Returns:
|
||||
A list of DiscordId object(s) if found, None otherwise.
|
||||
"""
|
||||
print("\n--- Finding multiple Discord ID documents ---")
|
||||
try:
|
||||
# Add active filter if requested
|
||||
if active_only:
|
||||
query["active"] = True
|
||||
|
||||
# Add guild_id filter if provided
|
||||
if guild_id:
|
||||
query["guild_ids"] = {"$in": [guild_id]}
|
||||
|
||||
found_docs = None
|
||||
async with self.db_h as db:
|
||||
found_docs = await db.find(query)
|
||||
|
||||
if found_docs:
|
||||
print(f"Found {len(found_docs)} Discord ID documents (raw dicts).")
|
||||
converted_discord_ids = []
|
||||
for doc in found_docs:
|
||||
converted_discord_ids.append(DiscordId.from_dict(doc))
|
||||
|
||||
return converted_discord_ids if len(converted_discord_ids) > 0 else None
|
||||
else:
|
||||
print("Discord ID documents not found.")
|
||||
return None
|
||||
except Exception as e:
|
||||
print(f"Discord ID find failed: {e}")
|
||||
return None
|
||||
|
||||
async def update_discord_id(self, query: Dict[str, Any], update_data: Dict[str, Any]) -> Optional[int]:
|
||||
"""
|
||||
Updates a single Discord ID entry in the database.
|
||||
|
||||
Args:
|
||||
query: A dictionary representing the query criteria to find the document.
|
||||
update_data: A dictionary representing the update operations (e.g., using $set).
|
||||
|
||||
Returns:
|
||||
The number of modified documents if successful, None otherwise.
|
||||
"""
|
||||
print("\n--- Updating a Discord ID document ---")
|
||||
try:
|
||||
update_result = None
|
||||
async with self.db_h as db:
|
||||
update_result = await db.update_one(query, update_data)
|
||||
|
||||
print(f"Discord ID update result: Matched {update_result.matched_count}, Modified {update_result.modified_count}")
|
||||
return update_result.modified_count
|
||||
except Exception as e:
|
||||
print(f"Discord ID update failed: {e}")
|
||||
return None
|
||||
|
||||
async def delete_discord_id(self, query: Dict[str, Any]) -> Optional[int]:
|
||||
"""
|
||||
Deletes a single Discord ID entry from the database.
|
||||
Args:
|
||||
query: A dictionary representing the query criteria to find the document to delete.
|
||||
Returns:
|
||||
The number of deleted documents if successful, None otherwise.
|
||||
"""
|
||||
print("\n--- Deleting a Discord ID document ---")
|
||||
try:
|
||||
delete_result = None
|
||||
async with self.db_h as db:
|
||||
delete_result = await self.db_h.delete_one(query)
|
||||
|
||||
print(f"Discord ID delete result: Deleted count {delete_result.deleted_count}")
|
||||
return delete_result.deleted_count
|
||||
except Exception as e:
|
||||
print(f"Discord ID delete failed: {e}")
|
||||
return None
|
||||
@@ -1,11 +1,24 @@
|
||||
from typing import Optional, List, Dict, Any
|
||||
from enum import Enum
|
||||
|
||||
class ActiveClient:
|
||||
"""
|
||||
The active client model in memory for quicker access
|
||||
"""
|
||||
websocket = None
|
||||
active_token: str = None
|
||||
|
||||
class DemodTypes(str, Enum):
|
||||
P25 = "P25"
|
||||
DMR = "DMR"
|
||||
ANALOG = "NBFM"
|
||||
|
||||
|
||||
class NodeCommands(str, Enum):
|
||||
JOIN = "join_server"
|
||||
LEAVE = "leave_server"
|
||||
|
||||
|
||||
class TalkgroupTag:
|
||||
"""Represents a talkgroup tag."""
|
||||
def __init__(self, talkgroup: str, tagDec: int):
|
||||
@@ -16,6 +29,71 @@ class TalkgroupTag:
|
||||
def to_dict(self) -> Dict[str, Any]:
|
||||
return {"talkgroup": self.talkgroup, "tagDec": self.tagDec}
|
||||
|
||||
|
||||
class DiscordId:
|
||||
"""
|
||||
A data model for a Discord ID entry.
|
||||
"""
|
||||
def __init__(self,
|
||||
_id: str,
|
||||
discord_id: str,
|
||||
name: str,
|
||||
token: str,
|
||||
active: bool,
|
||||
guild_ids: List[str]):
|
||||
"""
|
||||
Initializes a DiscordId object.
|
||||
|
||||
Args:
|
||||
_id: A unique identifier for the entry (e.g., MongoDB ObjectId string).
|
||||
discord_id: The Discord user ID.
|
||||
name: The name associated with the Discord ID.
|
||||
token: The authentication token.
|
||||
active: Boolean indicating if the ID is active.
|
||||
guild_ids: A list of guild IDs the Discord user is part of.
|
||||
"""
|
||||
self._id: str = str(_id)
|
||||
self.discord_id: str = discord_id
|
||||
self.name: str = name
|
||||
self.token: str = token
|
||||
self.active: bool = active
|
||||
self.guild_ids: List[str] = guild_ids
|
||||
|
||||
def __repr__(self) -> str:
|
||||
"""
|
||||
Provides a developer-friendly string representation of the object.
|
||||
"""
|
||||
return (f"DiscordId(_id='{self._id}', discord_id='{self.discord_id}', name='{self.name}', "
|
||||
f"token='{self.token}', active={self.active}, guild_ids={self.guild_ids})")
|
||||
|
||||
def to_dict(self) -> Dict[str, Any]:
|
||||
"""
|
||||
Converts the DiscordId object to a dictionary suitable for MongoDB.
|
||||
"""
|
||||
return {
|
||||
"_id": self._id,
|
||||
"discord_id": self.discord_id,
|
||||
"name": self.name,
|
||||
"token": self.token,
|
||||
"active": self.active,
|
||||
"guild_ids": self.guild_ids,
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: Dict[str, Any]) -> "DiscordId":
|
||||
"""
|
||||
Creates a DiscordId object from a dictionary (e.g., from MongoDB).
|
||||
"""
|
||||
return cls(
|
||||
_id=data.get("_id"),
|
||||
discord_id=data.get("discord_id", ""),
|
||||
name=data.get("name", ""),
|
||||
token=data.get("token", ""),
|
||||
active=data.get("active", False), # Default to False if not present
|
||||
guild_ids=data.get("guild_ids", []), # Default to empty list if not present
|
||||
)
|
||||
|
||||
|
||||
class System:
|
||||
"""
|
||||
A basic data model for a channel/system entry in a radio system.
|
||||
|
||||
Reference in New Issue
Block a user