⚙️ Integrate MongoDB for memory storage and retrieval; add search functionality

app.py

@@ -1,4 +1,6 @@
+from pymongo import MongoClient
 from typing import Literal
+from bson import ObjectId
 from io import StringIO
 import datetime
 import requests
@@ -6,11 +8,23 @@ import random
 import base64
 import gradio
 import qrcode
-import json
 import sys
 import os
 
-MemoryFile = 'Memories.json'
+MongoURI = os.getenv('MONGO_URI')
+DatabaseName = 'MCP-Utilities'
+CollectionName = 'Memories'
+
+try:
+	Client = MongoClient(MongoURI)
+	Database = Client[DatabaseName]
+	Collection = Database[CollectionName]
+	Client.server_info()
+	print('Connected to MongoDB successfully.')
+except Exception as e:
+	print(f'Error connecting to MongoDB: {str(e)}')
+	Collection = None
+
 Theme = gradio.themes.Citrus( # type: ignore
 	primary_hue='blue',
 	secondary_hue='blue',
@@ -175,64 +189,106 @@ def QRCode(Text: str) -> str:
 # ╰────────────────────╯
 
 def SaveMemory(Text: str) -> str:
-	'''Save a memory with the given text.
+	'''Save a memory with the given text to MongoDB.
 	Args:
 		Text (str): The text to save as a memory.
 	Returns:
 		str: A message indicating the result of the save operation.
 	'''
 	try:
-		if os.path.exists(MemoryFile):
-			with open(MemoryFile, 'r') as File:
-				Memories = json.load(File)
-		else:
-			Memories = []
+		if Collection is None:
+			return 'Error: MongoDB connection not available'
 
 		MemoryEntry = {
-			'id': len(Memories) + 1,
 			'text': Text,
-			'timestamp': datetime.datetime.now().isoformat()
+			'timestamp': datetime.datetime.now()
 		}
-		Memories.append(MemoryEntry)
-
-		with open(MemoryFile, 'w') as File:
-			json.dump(Memories, File)
 
-		return f'Memory saved with ID {MemoryEntry["id"]}.'
+		Result = Collection.insert_one(MemoryEntry)
+		return f'Memory saved with ID {str(Result.inserted_id)}'
 	except Exception as e:
 		return f'Error saving memory: {str(e)}'
 
 def ListMemories() -> str:
-	'''List all saved memories.
+	'''List all saved memories from MongoDB.
 	Returns:
 		str: A formatted string of all saved memories.
 	'''
 	try:
-		with open(MemoryFile, 'r') as File:
-			Memories = json.load(File)
-		return '\n'.join([f"ID: {Memory['id']}, Text: {Memory['text']}" for Memory in Memories])
+		if Collection is None:
+			return 'Error: MongoDB connection not available'
+
+		Memories = list(Collection.find().sort('timestamp', -1).limit(50))
+
+		if not Memories:
+			return 'No memories found.'
+
+		FormattedMemories = []
+		for Memory in Memories:
+			Timestamp = Memory['timestamp'].strftime('%Y-%m-%d %H:%M:%S')
+			MemoryId = str(Memory['_id'])
+			FormattedMemories.append(f'ID: {MemoryId} [{Timestamp}]: {Memory["text"]}')
+
+		return '\n'.join(FormattedMemories)
 	except Exception as e:
 		return f'Error listing memories: {str(e)}'
 
-def DeleteMemory(MemoryID: int) -> str:
-	'''Delete a memory by its ID.
+def DeleteMemory(MemoryID: str) -> str:
+	'''Delete a memory by its MongoDB ObjectId.
 	Args:
-		MemoryID (int): The ID of the memory to delete.
+		MemoryID (str): The MongoDB ObjectId of the memory to delete.
 	Returns:
 		str: A message indicating the result of the delete operation.
 	'''
 	try:
-		with open(MemoryFile, 'r') as File:
-			Memories = json.load(File)
-		Memories = [Memory for Memory in Memories if Memory['id'] != MemoryID]
-		for i, Memory in enumerate(Memories):
-			Memory['id'] = i + 1
-		with open(MemoryFile, 'w') as File:
-			json.dump(Memories, File)
-		return f'Memory with ID {MemoryID} deleted.'
+		if Collection is None:
+			return 'Error: MongoDB connection not available'
+
+		if not ObjectId.is_valid(MemoryID):
+			return 'Error: Invalid memory ID format'
+
+		Result = Collection.delete_one({'_id': ObjectId(MemoryID)})
+
+		if Result.deleted_count > 0:
+			return f'Memory with ID {MemoryID} deleted successfully'
+		else:
+			return f'Memory with ID {MemoryID} not found'
 	except Exception as e:
 		return f'Error deleting memory: {str(e)}'
 
+def SearchMemories(Query: str) -> str:
+	'''Search memories by text content.
+	Args:
+		Query (str): The text to search for in memories.
+	Returns:
+		str: A formatted string of matching memories.
+	'''
+	try:
+		if Collection is None:
+			return 'Error: MongoDB connection not available'
+
+		try:
+			Collection.create_index([("text", "text")])
+		except Exception as e:
+			print(f'Error creating text index: {str(e)}')
+
+		Memories = list(Collection.find(
+			{'$text': {'$search': Query}}
+		).sort('timestamp', -1).limit(20))
+
+		if not Memories:
+			return f'No memories found matching "{Query}"'
+
+		FormattedMemories = []
+		for Memory in Memories:
+			Timestamp = Memory['timestamp'].strftime('%Y-%m-%d %H:%M:%S')
+			MemoryId = str(Memory['_id'])
+			FormattedMemories.append(f'ID: {MemoryId} [{Timestamp}]: {Memory["text"]}')
+
+		return '\n'.join(FormattedMemories)
+	except Exception as e:
+		return f'Error searching memories: {str(e)}'
+
 # ╭──────────────────────────────╮
 # │    Gradio Interface Setup    │
 # ╰────────────────────────────��─╯
@@ -302,7 +358,7 @@ with gradio.Blocks(
 						ReverseOutput = gradio.Text(label='Reversed Text ↩️', interactive=False)
 						ReverseBtn = gradio.Button('Reverse Text 🔄', variant='primary')
 						ReverseBtn.click(Reverse, inputs=ReverseInput, outputs=ReverseOutput)
-				
+
 					with gradio.Group():
 						WordCountInput = gradio.Textbox(label='Text for Word Count 📏', placeholder='Enter text to count words', lines=3)
 						WordCountChoice = gradio.Radio(label='Count Type 🔢', choices=['words', 'characters'], value='words', interactive=True)
@@ -334,7 +390,7 @@ with gradio.Blocks(
 						SaveBtn.click(SaveMemory, inputs=SaveInput, outputs=SaveOutput)
 
 					with gradio.Group():
-						DeleteInput = gradio.Number(label='Memory ID to Delete 🗑️', value=1, minimum=1)
+						DeleteInput = gradio.Textbox(label='Memory ID to Delete 🗑️', placeholder='Enter MongoDB ObjectId', lines=1)
 						DeleteOutput = gradio.Text(label='Delete Result ❌', interactive=False)
 						DeleteBtn = gradio.Button('Delete Memory 🗑️', variant='secondary')
 						DeleteBtn.click(DeleteMemory, inputs=DeleteInput, outputs=DeleteOutput)
@@ -345,6 +401,12 @@ with gradio.Blocks(
 						ListBtn = gradio.Button('List Memories 📄', variant='primary')
 						ListBtn.click(ListMemories, outputs=ListOutput)
 
+					with gradio.Group():
+						SearchInput = gradio.Textbox(label='Search Memories 🔍', placeholder='Enter text to search in memories', lines=3)
+						SearchOutput = gradio.Text(label='Search Results 📑', interactive=False)
+						SearchBtn = gradio.Button('Search Memories 🔎', variant='primary')
+						SearchBtn.click(SearchMemories, inputs=SearchInput, outputs=SearchOutput)
+
 	App.launch(
 		mcp_server=True
 	)