from datetime import datetime from typing import List, Dict, Any from motor.motor_asyncio import AsyncIOMotorClient from src.core.config import settings import os class ChatHistoryService: def __init__(self): self._client = None self._db = None self._collection = None @property def collection(self): if self._collection is None: # Initialize client only once self._client = AsyncIOMotorClient(settings.MONGODB_URI) self._db = self._client[settings.DB_NAME] # Use HISTORY_COLLECTION from settings or default to user_chats col_name = getattr(settings, "HISTORY_COLLECTION", "user_chats") self._collection = self._db[col_name] return self._collection async def add_message(self, user_identifier: str, role: str, message: str, full_name: str = None, phone_number: str = None, source: str = "website"): """ Adds a message to the user's chat history with root-level metadata. For WhatsApp: user_identifier is ignored, phone_number is used For Website: user_identifier is the user_id from frontend (converted to ObjectId) """ from bson import ObjectId chat_entry = { "role": role, "parts": message, "timestamp": datetime.utcnow() } # Determine query and update based on source if source == "whatsapp": # WhatsApp: Find by phone_number (contact_id field) query = {"contact_id": phone_number, "source": "whatsapp"} set_doc = { "contact_id": phone_number, "source": "whatsapp" } if full_name: set_doc["full_name"] = full_name else: # Website: Find by user_id (convert string to ObjectId) try: user_id_obj = ObjectId(user_identifier) except: # If invalid ObjectId, create new one user_id_obj = ObjectId() query = {"user_id": user_id_obj, "source": "website"} set_doc = { "user_id": user_id_obj, "source": "website" } if full_name: set_doc["full_name"] = full_name update_doc = { "$push": {"chat_history": chat_entry}, "$set": set_doc } await self.collection.update_one( query, update_doc, upsert=True ) async def get_recent_history(self, user_identifier: str, limit: int = 10, source: str = None) -> List[Dict[str, Any]]: """ Retrieves the last N messages. For WhatsApp: user_identifier is contact_id For Website: user_identifier is user_id (converted to ObjectId) """ from bson import ObjectId # Smart detection: Try both if source is unreliable if source == "whatsapp": query = {"contact_id": user_identifier, "source": "whatsapp"} elif source == "website": # Check if it's actually a phone number (WhatsApp) if user_identifier.isdigit() and len(user_identifier) >= 10: # Try WhatsApp first doc = await self.collection.find_one({"contact_id": user_identifier, "source": "whatsapp"}) if doc and "chat_history" in doc: return doc["chat_history"][-limit:] # Try as ObjectId for website try: user_id_obj = ObjectId(user_identifier) query = {"user_id": user_id_obj, "source": "website"} except: return [] else: # No source specified, try both doc = await self.collection.find_one({"contact_id": user_identifier, "source": "whatsapp"}) if doc and "chat_history" in doc: return doc["chat_history"][-limit:] try: user_id_obj = ObjectId(user_identifier) query = {"user_id": user_id_obj, "source": "website"} except: return [] doc = await self.collection.find_one(query) if not doc or "chat_history" not in doc: return [] # Get last 'limit' messages history = doc["chat_history"][-limit:] return history async def get_full_history(self, user_identifier: str, source: str = None) -> Dict[str, Any]: """ Retrieves complete chat history for a user (for frontend display). For WhatsApp: user_identifier is contact_id For Website: user_identifier is user_id (converted to ObjectId) """ from bson import ObjectId doc = None # Smart detection: Try both if source is unreliable if source == "whatsapp": query = {"contact_id": user_identifier, "source": "whatsapp"} doc = await self.collection.find_one(query) elif source == "website": # Check if it's actually a phone number (WhatsApp) if user_identifier.isdigit() and len(user_identifier) >= 10: # Try WhatsApp first doc = await self.collection.find_one({"contact_id": user_identifier, "source": "whatsapp"}) # If not found, try as ObjectId for website if not doc: try: user_id_obj = ObjectId(user_identifier) query = {"user_id": user_id_obj, "source": "website"} doc = await self.collection.find_one(query) except: pass else: # No source specified, try both doc = await self.collection.find_one({"contact_id": user_identifier, "source": "whatsapp"}) if not doc: try: user_id_obj = ObjectId(user_identifier) query = {"user_id": user_id_obj, "source": "website"} doc = await self.collection.find_one(query) except: pass if not doc: return { "user_id": user_identifier if source != "whatsapp" else None, "contact_id": user_identifier if source == "whatsapp" else None, "source": source, "chat_history": [], "total_messages": 0 } return { "user_id": str(doc.get("user_id")) if doc.get("user_id") else None, "contact_id": doc.get("contact_id"), "source": doc.get("source"), "full_name": doc.get("full_name"), "chat_history": doc.get("chat_history", []), "total_messages": len(doc.get("chat_history", [])) } async def get_all_chats_with_user_details(self, page: int = 1, limit: int = 20) -> Dict[str, Any]: """ Get all chat histories with user details from admins collection. Uses MongoDB $lookup to join user_chats with admins. Supports pagination. """ from bson import ObjectId skip = (page - 1) * limit # Count total documents total_count = await self.collection.count_documents({}) pipeline = [ {"$skip": skip}, {"$limit": limit}, { "$lookup": { "from": "admins", "localField": "user_id", "foreignField": "_id", "as": "user_details" } }, { "$unwind": { "path": "$user_details", "preserveNullAndEmptyArrays": True } } ] cursor = self.collection.aggregate(pipeline) results = await cursor.to_list(length=limit) # Convert ObjectId and datetime to JSON serializable format def convert_doc(doc): if "_id" in doc: doc["_id"] = str(doc["_id"]) if "user_id" in doc and isinstance(doc["user_id"], ObjectId): doc["user_id"] = str(doc["user_id"]) if "chat_history" in doc: for msg in doc["chat_history"]: if "timestamp" in msg and isinstance(msg["timestamp"], datetime): msg["timestamp"] = msg["timestamp"].isoformat() if "user_details" in doc and doc["user_details"]: if "_id" in doc["user_details"]: doc["user_details"]["_id"] = str(doc["user_details"]["_id"]) return doc results = [convert_doc(doc) for doc in results] return { "total": total_count, "page": page, "limit": limit, "total_pages": (total_count + limit - 1) // limit, "chats": results } history_service = ChatHistoryService()