# mongo_tools.py try: from .mongochain import MongoDBHandler except ImportError: from mongochain import MongoDBHandler from typing import Dict, List, Any # Global handler instance _mongo_handler = MongoDBHandler() def mongo_configure(uri: str, db: str, coll: str): """Configures the global MongoDB handler.""" global _mongo_handler _mongo_handler = MongoDBHandler(uri=uri, db_name=db, collection_name=coll) return _mongo_handler.connect() def mongo_insert_doc(content: str, metadata: Dict[str, Any] = None) -> str: """ Inserts a document into MongoDB. Args: content: The text content to store. metadata: Optional dictionary of metadata (source, author, etc.). """ success = _mongo_handler.insert_chunk(content, metadata or {}) return "Successfully inserted doc." if success else "Failed to insert doc." def mongo_find_docs(query_json: Dict[str, Any], limit: int = 5) -> List[Dict[str, Any]]: """ Finds documents in MongoDB matching a JSON query. Args: query_json: A MongoDB query dictionary (e.g., {"metadata.source": "tech_specs.pdf"}). limit: Max number of documents to return. """ results = _mongo_handler.find_relevant(query_json, limit=limit) # Convert ObjectId to string for JSON compatibility for res in results: if "_id" in res: res["_id"] = str(res["_id"]) return results def mongo_get_collection_stats() -> Dict[str, Any]: """Returns statistics about the currently connected MongoDB collection.""" return _mongo_handler.get_stats() def mongo_clear_collection() -> str: """Deletes all documents in the current collection.""" success = _mongo_handler.clear() return "Collection cleared." if success else "Failed to clear collection." def mongo_keyword_search(keyword: str, limit: int = 5) -> List[Dict[str, Any]]: """ Performs a simple regex keyword search in the document content. Args: keyword: The string to search for. limit: Max results. """ query = {"content": {"$regex": keyword, "$options": "i"}} return mongo_find_docs(query, limit)