Update flask_Character.py

flask_Character.py

@@ -1,6 +1,3 @@
-# This software is licensed under a **dual-license model**
-# For individuals and businesses earning **under $1M per year**, this software is licensed under the **MIT License**
-# Businesses or organizations with **annual revenue of $1,000,000 or more** must obtain permission to use this software commercially.
 import os
 # NUMBA_CACHE_DIR and NUMBA_DISABLE_CACHE are often set for specific environments,
 # e.g., if you're experiencing issues with Numba's caching behavior or in containerized environments.
@@ -14,11 +11,13 @@ from datetime import date, datetime, timedelta
 from typing import List, Optional, Literal, Dict, Any, Tuple
 import traceback
 import asyncio
+from uuid import uuid4, UUID # For unique request IDs
 
 from fastapi import FastAPI, HTTPException, Response, Query, Depends, status
 from fastapi.responses import FileResponse
 from fastapi.exception_handlers import http_exception_handler
 from starlette.exceptions import HTTPException as StarletteHTTPException
+
 from langchain.prompts import PromptTemplate
 from langchain_groq import ChatGroq
 from pydantic import BaseModel, Field, BeforeValidator, model_serializer
@@ -29,15 +28,29 @@ from pymongo import MongoClient
 from pymongo.errors import ConnectionFailure, OperationFailure
 from bson import ObjectId
 
+# --- Batching Configuration ---
+MAX_BATCH_SIZE = 20
+BATCH_INTERVAL_SECONDS = 1.0
+
+# --- Queues and pending request stores for batching ---
+extract_data_queue: Optional[asyncio.Queue] = None
+generate_reply_queue: Optional[asyncio.Queue] = None
+
+# Dictionaries to store futures for pending requests, keyed by unique request ID
+# The value will be an asyncio.Future that the endpoint handler will await
+extract_pending_requests: Dict[UUID, asyncio.Future] = {}
+generate_pending_requests: Dict[UUID, asyncio.Future] = {}
+
+# Shutdown event for worker tasks
+shutdown_event = asyncio.Event()
+
+
 # --- MongoDB Configuration ---
-# IMPORTANT: Use environment variables for your MONGO_URI in production for security.
-# Example: MONGO_URI = os.getenv("MONGO_URI", "mongodb://localhost:27017")
-MONGO_URI = "mongodb+srv://precison9:P1LhtFknkT75yg5L@cluster0.isuwpef.mongodb.net"
+MONGO_URI = "mongodb+srv://precison9:P1LhtFknkT75yg5L@cluster0.isuwpef.mongodb.net" # Use os.getenv in prod
 DB_NAME = "email_assistant_db"
 EXTRACTED_EMAILS_COLLECTION = "extracted_emails"
 GENERATED_REPLIES_COLLECTION = "generated_replies"
 
-# Global variables for MongoDB client and collections
 client: Optional[MongoClient] = None
 db: Optional[Any] = None
 extracted_emails_collection: Optional[Any] = None
@@ -45,50 +58,30 @@ generated_replies_collection: Optional[Any] = None
 
 # --- Pydantic ObjectId Handling ---
 class CustomObjectId(str):
-    """
-    Custom Pydantic type for handling MongoDB ObjectIds.
-    It validates that the input is a valid ObjectId string and
-    ensures it's represented as a string in JSON Schema.
-    """
     @classmethod
     def __get_validators__(cls):
         yield cls.validate
-
     @classmethod
     def validate(cls, v):
-        # Allow None or empty string to pass through for optional fields
-        # This validator is only called if the field is not None
-        # Pydantic's Optional[PyObjectId] handles the None case before this validator
         if v is None or v == "":
             return None
-
         if not isinstance(v, (str, ObjectId)):
             raise ValueError("ObjectId must be a string or ObjectId instance")
-
-        # Convert ObjectId to string if it's already an ObjectId instance
         if isinstance(v, ObjectId):
             return str(v)
-
-        # Validate string format
         if not ObjectId.is_valid(v):
             raise ValueError("Invalid ObjectId format")
-        return cls(v) # Return an instance of CustomObjectId (which is a str subclass)
-
-    # This method is crucial for Pydantic v2 to generate correct OpenAPI schema
+        return cls(v)
     @classmethod
     def __get_pydantic_json_schema__(
         cls, _core_schema: core_schema.CoreSchema, handler
     ) -> Dict[str, Any]:
-        # We tell Pydantic that this custom type should be represented as a standard string
-        # in the generated JSON Schema (OpenAPI documentation).
         json_schema = handler(core_schema.str_schema())
-        json_schema["example"] = "60c728ef238b9c7b9e0f6c2a" # Add an example for clarity
+        json_schema["example"] = "60c728ef238b9c7b9e0f6c2a"
         return json_schema
 
-# Annotated type for convenience in models
 PyObjectId = Annotated[CustomObjectId, BeforeValidator(str)]
 
-
 # ---------------------- Models ----------------------
 class Contact(BaseModel):
     name: str
@@ -110,28 +103,20 @@ class Task(BaseModel):
     due_date: date
 
 class ExtractedData(BaseModel):
-    # Use PyObjectId for the _id field
     id: Optional[PyObjectId] = Field(alias="_id", default=None)
     contacts: List[Contact]
     appointments: List[Appointment]
     tasks: List[Task]
     original_email_text: str
     processed_at: datetime = Field(default_factory=datetime.utcnow)
-
     class Config:
-        populate_by_name = True # Allow setting 'id' or '_id'
-        arbitrary_types_allowed = True # Allow CustomObjectId and ObjectId
-
-    # Custom serializer for JSON output to ensure ObjectId is converted to string
+        populate_by_name = True
+        arbitrary_types_allowed = True
     @model_serializer(when_used='json')
     def serialize_model(self):
         data = self.model_dump(by_alias=True, exclude_none=True)
-        # Ensure _id is a string when serializing to JSON
         if "_id" in data and isinstance(data["_id"], ObjectId):
             data["_id"] = str(data["_id"])
-        # Ensure dates are correctly serialized to ISO format if they are date objects
-        # Pydantic v2 usually handles this automatically for `date` types,
-        # but explicit conversion can be useful if direct manipulation is expected or for specific formats.
         if 'appointments' in data:
             for appt in data['appointments']:
                 if isinstance(appt.get('start_date'), date):
@@ -158,7 +143,6 @@ class GenerateReplyRequest(BaseModel):
     emoji: str = Field("Auto", examples=["Auto", "None", "Occasional", "Frequent"])
 
 class GeneratedReplyData(BaseModel):
-    # Use PyObjectId for the _id field
     id: Optional[PyObjectId] = Field(alias="_id", default=None)
     original_email_text: str
     generated_reply_text: str
@@ -168,11 +152,9 @@ class GeneratedReplyData(BaseModel):
     tone: str
     emoji: str
     generated_at: datetime = Field(default_factory=datetime.utcnow)
-
     class Config:
         populate_by_name = True
         arbitrary_types_allowed = True
-
     @model_serializer(when_used='json')
     def serialize_model(self):
         data = self.model_dump(by_alias=True, exclude_none=True)
@@ -180,7 +162,6 @@ class GeneratedReplyData(BaseModel):
             data["_id"] = str(data["_id"])
         return data
 
-# NEW: Response Model for /generate-reply endpoint
 class GenerateReplyResponse(BaseModel):
     reply: str = Field(..., description="The AI-generated reply text.")
     stored_id: str = Field(..., description="The MongoDB ID of the stored reply.")
@@ -188,519 +169,430 @@ class GenerateReplyResponse(BaseModel):
 
 # --- Query Models for GET Endpoints ---
 class ExtractedEmailQuery(BaseModel):
-    contact_name: Optional[str] = Query(None, description="Filter by contact name (case-insensitive partial match).")
-    appointment_title: Optional[str] = Query(None, description="Filter by appointment title (case-insensitive partial match).")
-    task_title: Optional[str] = Query(None, description="Filter by task title (case-insensitive partial match).")
-    from_date: Optional[date] = Query(None, description="Filter by data processed on or after this date (YYYY-MM-DD).")
-    to_date: Optional[date] = Query(None, description="Filter by data processed on or before this date (YYYY-MM-DD).")
-    limit: int = Query(10, ge=1, le=100, description="Maximum number of results to return.")
+    contact_name: Optional[str] = Query(None)
+    appointment_title: Optional[str] = Query(None)
+    task_title: Optional[str] = Query(None)
+    from_date: Optional[date] = Query(None)
+    to_date: Optional[date] = Query(None)
+    limit: int = Query(10, ge=1, le=100)
 
 class GeneratedReplyQuery(BaseModel):
-    language: Optional[Literal["Italian", "English"]] = Query(None, description="Filter by reply language.")
-    style: Optional[str] = Query(None, description="Filter by reply style (e.g., Professional, Casual).")
-    tone: Optional[str] = Query(None, description="Filter by reply tone (e.g., Friendly, Neutral).")
-    from_date: Optional[date] = Query(None, description="Filter by data generated on or after this date (YYYY-MM-DD).")
-    to_date: Optional[date] = Query(None, description="Filter by data generated on or before this date (YYYY-MM-DD).")
-    limit: int = Query(10, ge=1, le=100, description="Maximum number of results to return.")
+    language: Optional[Literal["Italian", "English"]] = Query(None)
+    style: Optional[str] = Query(None)
+    tone: Optional[str] = Query(None)
+    from_date: Optional[date] = Query(None)
+    to_date: Optional[date] = Query(None)
+    limit: int = Query(10, ge=1, le=100)
 
 # ---------------------- Utility Functions ----------------------
 def extract_last_json_block(text: str) -> Optional[str]:
-    """
-    Extracts the last JSON block enclosed in ```json``` from a string,
-    or a standalone JSON object if no code block is found.
-    """
     pattern = r'```json\s*(.*?)\s*```'
     matches = re.findall(pattern, text, re.DOTALL)
-    if matches:
-        return matches[-1].strip()
-    # Fallback: try to find a standalone JSON object
+    if matches: return matches[-1].strip()
     match = re.search(r'\{.*\}', text, re.DOTALL)
-    if match:
-        return match.group(0)
+    if match: return match.group(0)
     return None
 
 def parse_date(date_str: Optional[str], current_date: date) -> Optional[date]:
-    """
-    Parses a date string, handling 'today', 'tomorrow', and YYYY-MM-DD format.
-    Returns None if input is None or cannot be parsed into a valid date.
-    """
-    if not date_str:
-        return None
+    if not date_str: return None
     date_str_lower = date_str.lower().strip()
-    if date_str_lower == "today":
-        return current_date
-    if date_str_lower == "tomorrow":
-        return current_date + timedelta(days=1)
-    try:
-        return datetime.strptime(date_str_lower, "%Y-%m-%d").date()
-    except ValueError:
-        # If parsing fails, return None. The calling function (normalize_llm_output)
-        # will then decide the default (e.g., current_date).
-        return None
+    if date_str_lower == "today": return current_date
+    if date_str_lower == "tomorrow": return current_date + timedelta(days=1)
+    try: return datetime.strptime(date_str_lower, "%Y-%m-%d").date()
+    except ValueError: return None
 
 def normalize_llm_output(data: dict, current_date: date, original_email_text: str) -> ExtractedData:
-    """
-    Normalizes and validates LLM extracted data into ExtractedData Pydantic model.
-    Handles defaults for dates and name splitting.
-    """
     def split_name(full_name: str) -> tuple[str, str]:
         parts = full_name.strip().split()
         name = parts[0] if parts else ""
         last_name = " ".join(parts[1:]) if len(parts) > 1 else ""
         return name, last_name
-
-    contacts_data = []
-    for c in data.get("contacts", []):
-        name_val, last_name_val = split_name(c.get("name", ""))
-        contacts_data.append(Contact(name=name_val, last_name=last_name_val, email=c.get("email"), phone_number=c.get("phone_number")))
-
-    appointments_data = []
-    for a in data.get("appointments", []):
-        # Default start_date to current_date if not provided or invalid
-        start_date_val = parse_date(a.get("start_date"), current_date) or current_date
-        # end_date remains optional
-        end_date_val = parse_date(a.get("end_date"), current_date)
-
-        appointments_data.append(Appointment(
-            title=a.get("title", "Untitled"), description=a.get("description", "No description"),
-            start_date=start_date_val, start_time=a.get("start_time"),
-            end_date=end_date_val, end_time=a.get("end_time")
-        ))
-
-    tasks_data = []
-    for t in data.get("tasks", []):
-        # Default due_date to current_date if not provided or invalid
-        due_date_val = parse_date(t.get("due_date"), current_date) or current_date
-        tasks_data.append(Task(
-            task_title=t.get("task_title", "Untitled"), task_description=t.get("task_description", "No description"),
-            due_date=due_date_val
-        ))
+    contacts_data = [Contact(name=split_name(c.get("name",""))[0], last_name=split_name(c.get("name",""))[1], email=c.get("email"), phone_number=c.get("phone_number")) for c in data.get("contacts", [])]
+    appointments_data = [Appointment(title=a.get("title", "Untitled"), description=a.get("description", "No description"), start_date=parse_date(a.get("start_date"), current_date) or current_date, start_time=a.get("start_time"), end_date=parse_date(a.get("end_date"), current_date), end_time=a.get("end_time")) for a in data.get("appointments", [])]
+    tasks_data = [Task(task_title=t.get("task_title", "Untitled"), task_description=t.get("task_description", "No description"), due_date=parse_date(t.get("due_date"), current_date) or current_date) for t in data.get("tasks", [])]
     return ExtractedData(contacts=contacts_data, appointments=appointments_data, tasks=tasks_data, original_email_text=original_email_text)
 
 # ---------------------- Core Logic (Internal Functions) ----------------------
 def _process_email_internal(email_text: str, api_key: str, current_date: date) -> ExtractedData:
-    """
-    Internal function to process email text using LLM and extract structured data.
-    """
-    if not email_text:
-        raise ValueError("Email text cannot be empty for processing.")
-
+    if not email_text: raise ValueError("Email text cannot be empty for processing.")
     llm = ChatGroq(model="meta-llama/llama-4-scout-17b-16e-instruct", temperature=0, max_tokens=2000, groq_api_key=api_key)
-
     prompt_today_str = current_date.isoformat()
     prompt_tomorrow_str = (current_date + timedelta(days=1)).isoformat()
-
-    prompt_template_str = f"""
-You are an expert email assistant tasked with extracting structured information from an Italian email.
-
-**Your response MUST be a single, complete JSON object, wrapped in a ```json``` block.**
-**DO NOT include any conversational text, explanations, or preambles outside the JSON block.**
-**The JSON should contain three top-level keys: "contacts", "appointments", and "tasks".**
-If a category has no items, its list should be empty (e.g., "contacts": []).
-
-Here is the required JSON schema for each category:
-
-- **contacts**: List of Contact objects.
-    Each Contact object must have:
-    - `name` (string, full name)
-    - `last_name` (string, last name) - You should infer this from the full name.
-    - `email` (string, optional, null if not present)
-    - `phone_number` (string, optional, null if not present)
-
-- **appointments**: List of Appointment objects.
-    Each Appointment object must have:
-    - `title` (string, short, meaningful title in Italian based on the meeting's purpose)
-    - `description` (string, summary of the meeting's goal)
-    - `start_date` (string, YYYY-MM-DD. If not explicitly mentioned, use "{prompt_today_str}" for "today", or "{prompt_tomorrow_str}" for "tomorrow")
-    - `start_time` (string, optional, e.g., "10:30 AM", null if not present)
-    - `end_date` (string, YYYY-MM-DD, optional, null if unknown or not applicable)
-    - `end_time` (string, optional, e.g., "11:00 AM", null if not present)
-
-- **tasks**: List of Task objects.
-    Each Task object must have:
-    - `task_title` (string, short summary of action item)
-    - `task_description` (string, more detailed explanation)
-    - `due_date` (string, YYYY-MM-DD. Infer from context, e.g., "entro domani" becomes "{prompt_tomorrow_str}", "today" becomes "{prompt_today_str}")
-
----
-
-Email:
-{{email}}
-"""
-    prompt_template = PromptTemplate(input_variables=["email", "prompt_today_str", "prompt_tomorrow_str"], template=prompt_template_str)
+    prompt_template_str = f"""You are an expert email assistant tasked with extracting structured information from an Italian email.**Your response MUST be a single, complete JSON object, wrapped in a ```json``` block.****DO NOT include any conversational text, explanations, or preambles outside the JSON block.****The JSON should contain three top-level keys: "contacts", "appointments", and "tasks".**If a category has no items, its list should be empty (e.g., "contacts": []).Here is the required JSON schema for each category:- **contacts**: List of Contact objects.    Each Contact object must have:    - `name` (string, full name)    - `last_name` (string, last name) - You should infer this from the full name.    - `email` (string, optional, null if not present)    - `phone_number` (string, optional, null if not present)- **appointments**: List of Appointment objects.    Each Appointment object must have:    - `title` (string, short, meaningful title in Italian based on the meeting's purpose)    - `description` (string, summary of the meeting's goal)    - `start_date` (string, YYYY-MM-DD. If not explicitly mentioned, use "{prompt_today_str}" for "today", or "{prompt_tomorrow_str}" for "tomorrow")    - `start_time` (string, optional, e.g., "10:30 AM", null if not present)    - `end_date` (string, YYYY-MM-DD, optional, null if unknown or not applicable)    - `end_time` (string, optional, e.g., "11:00 AM", null if not present)- **tasks**: List of Task objects.    Each Task object must have:    - `task_title` (string, short summary of action item)    - `task_description` (string, more detailed explanation)    - `due_date` (string, YYYY-MM-DD. Infer from context, e.g., "entro domani" becomes "{prompt_tomorrow_str}", "today" becomes "{prompt_today_str}")---Email:{{email}}"""
+    prompt_template = PromptTemplate(input_variables=["email"], template=prompt_template_str) # Removed unused prompt_today_str, prompt_tomorrow_str from input_variables
     chain = prompt_template | llm
     try:
-        llm_output = chain.invoke({"email": email_text, "prompt_today_str": prompt_today_str, "prompt_tomorrow_str": prompt_tomorrow_str})
+        llm_output = chain.invoke({"email": email_text}) # Removed unused variables
         llm_output_str = llm_output.content
-
         json_str = extract_last_json_block(llm_output_str)
-
-        if not json_str:
-            raise ValueError(f"No JSON block found in LLM output. LLM response: {llm_output_str}")
+        if not json_str: raise ValueError(f"No JSON block found in LLM output. LLM response: {llm_output_str}")
         json_data = json.loads(json_str)
+        return normalize_llm_output(json_data, current_date, email_text)
+    except json.JSONDecodeError as e: raise ValueError(f"Failed to parse JSON from LLM output: {e}\nLLM response was:\n{llm_output_str}")
+    except Exception as e: traceback.print_exc(); raise Exception(f"An error occurred during email processing: {e}")
 
-        extracted_data = normalize_llm_output(json_data, current_date, email_text)
-        return extracted_data
-    except json.JSONDecodeError as e:
-        raise ValueError(f"Failed to parse JSON from LLM output: {e}\nLLM response was:\n{llm_output_str}")
-    except Exception as e:
-        traceback.print_exc()
-        raise Exception(f"An error occurred during email processing: {e}")
-
-def _generate_response_internal(
-    email_text: str, api_key: str, language: Literal["Italian", "English"],
-    length: str, style: str, tone: str, emoji: str
-) -> str:
-    """
-    Internal function to generate a reply to an email using LLM.
-    """
-    print(f"[{datetime.now()}] _generate_response_internal: Starting LLM call. API Key starts with: {api_key[:5]}...") # Debug log
-    if not email_text:
-        print(f"[{datetime.now()}] _generate_response_internal: Email text is empty.")
-        return "Cannot generate reply for empty email text."
-
+def _generate_response_internal(email_text: str, api_key: str, language: Literal["Italian", "English"], length: str, style: str, tone: str, emoji: str) -> str:
+    if not email_text: return "Cannot generate reply for empty email text."
     try:
         llm = ChatGroq(model="meta-llama/llama-4-scout-17b-16e-instruct", temperature=0.7, max_tokens=800, groq_api_key=api_key)
-        prompt_template_str="""
-    You are an assistant that helps reply to emails.
-
-    Create a response to the following email with the following parameters:
-    - Language: {language}
-    - Length: {length}
-    - Style: {style}
-    - Tone: {tone}
-    - Emoji usage: {emoji}
-
-    Email:
-    {email}
-
-    Write only the reply body. Do not repeat the email or mention any instruction.
-    """
-        prompt = PromptTemplate(
-            input_variables=["email", "language", "length", "style", "tone", "emoji"],
-            template=prompt_template_str
-        )
+        prompt_template_str="""    You are an assistant that helps reply to emails.    Create a response to the following email with the following parameters:    - Language: {language}    - Length: {length}    - Style: {style}    - Tone: {tone}    - Emoji usage: {emoji}    Email:    {email}    Write only the reply body. Do not repeat the email or mention any instruction.    """
+        prompt = PromptTemplate(input_variables=["email", "language", "length", "style", "tone", "emoji"], template=prompt_template_str)
         chain = prompt | llm
-        print(f"[{datetime.now()}] _generate_response_internal: Invoking LLM chain...") # Debug log
         output = chain.invoke({"email": email_text, "language": language, "length": length, "style": style, "tone": tone, "emoji": emoji})
-        print(f"[{datetime.now()}] _generate_response_internal: LLM chain returned. Content length: {len(output.content)}.") # Debug log
         return output.content.strip()
-    except Exception as e:
-        print(f"[{datetime.now()}] _generate_response_internal: ERROR during LLM invocation: {e}") # Debug log
-        traceback.print_exc() # Print full traceback to logs
-        raise # Re-raise the exception so it can be caught by handle_single_reply_request
-
+    except Exception as e: traceback.print_exc(); raise
 
 # --- FastAPI Application ---
-app = FastAPI(
-    title="Email Assistant API",
-    description="API for extracting structured data from emails and generating intelligent replies using Groq LLMs, with MongoDB integration, dynamic date handling, and caching.",
-    version="1.1.0",
-    docs_url="/", # Sets Swagger UI to be the root path
-    redoc_url="/redoc"
-)
-
-# --- Global Exception Handler ---
-# Catch Starlette HTTPExceptions (FastAPI uses these internally)
+app = FastAPI(title="Email Assistant API", description="API for extracting structured data and generating replies.", version="1.2.0", docs_url="/", redoc_url="/redoc")
+
+# --- Exception Handlers ---
 @app.exception_handler(StarletteHTTPException)
 async def custom_http_exception_handler_wrapper(request, exc):
-    """Handles FastAPI's internal HTTP exceptions."""
-    print(f"[{datetime.now()}] Caught StarletteHTTPException: {exc.status_code} - {exc.detail}")
     return await http_exception_handler(request, exc)
 
-# Catch all other unhandled exceptions
 @app.exception_handler(Exception)
 async def global_exception_handler_wrapper(request, exc):
-    """Handles all unhandled exceptions and returns a consistent JSON error response."""
-    print(f"[{datetime.now()}] Unhandled exception caught by global handler for request: {request.url}")
-    traceback.print_exc() # Print traceback to console for debugging
-    # Return a JSON response for consistency, even for unhandled errors
-    return Response(
-        content=json.dumps({"detail": f"Internal Server Error: {str(exc)}", "type": "unhandled_exception"}),
-        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-        media_type="application/json"
-    )
-
-
-# --- FastAPI Event Handlers for MongoDB ---
+    traceback.print_exc()
+    return Response(content=json.dumps({"detail": f"Internal Server Error: {str(exc)}", "type": "unhandled_exception"}), status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, media_type="application/json")
+
+# --- Batch Worker Functions ---
+async def _execute_single_extract_task(request_item: ProcessEmailRequest, request_id: UUID) -> Tuple[UUID, Any]:
+    """Helper to run a single extract task and return its result or exception."""
+    current_date = date.today()
+    try:
+        # Run blocking LLM call in a thread pool
+        result = await asyncio.to_thread(
+            _process_email_internal, request_item.email_text, request_item.groq_api_key, current_date
+        )
+        return request_id, result
+    except Exception as e:
+        return request_id, e
+
+async def extract_data_batch_worker():
+    global extract_data_queue, extract_pending_requests, shutdown_event
+    print(f"[{datetime.now()}] Extract Data Batch Worker started.")
+    while not shutdown_event.is_set():
+        try:
+            await asyncio.wait_for(asyncio.sleep(BATCH_INTERVAL_SECONDS), timeout=BATCH_INTERVAL_SECONDS + 0.1) # Ensures it runs roughly every interval
+        except asyncio.TimeoutError: # woken up by shutdown
+            if shutdown_event.is_set(): break
+
+        batch_to_process: List[Tuple[ProcessEmailRequest, UUID]] = []
+        while len(batch_to_process) < MAX_BATCH_SIZE:
+            try:
+                request_obj, req_id = extract_data_queue.get_nowait()
+                batch_to_process.append((request_obj, req_id))
+            except asyncio.QueueEmpty:
+                break # No more items for this batch
+
+        if not batch_to_process:
+            continue
+
+        print(f"[{datetime.now()}] Extract Worker: Processing batch of {len(batch_to_process)} requests.")
+        
+        # Concurrently execute all LLM calls for the current batch
+        llm_tasks = [_execute_single_extract_task(req_obj, req_id) for req_obj, req_id in batch_to_process]
+        results = await asyncio.gather(*llm_tasks) # Results are (request_id, result_or_exception)
+
+        for req_id, result_or_exc in results:
+            future = extract_pending_requests.pop(req_id, None) # Get and remove future
+            if future and not future.done():
+                if isinstance(result_or_exc, Exception):
+                    future.set_exception(result_or_exc)
+                else:
+                    future.set_result(result_or_exc) # This is ExtractedData object (pre-DB)
+            elif future and future.done():
+                 print(f"[{datetime.now()}] Extract Worker: Future for {req_id} was already done (e.g. timed out).")
+
+    print(f"[{datetime.now()}] Extract Data Batch Worker shutting down.")
+    # Clear out any remaining requests in the queue by setting exception
+    while not extract_data_queue.empty():
+        try:
+            _, req_id = extract_data_queue.get_nowait()
+            future = extract_pending_requests.pop(req_id, None)
+            if future and not future.done():
+                future.set_exception(HTTPException(status_code=503, detail="Service shutting down, request cancelled."))
+        except asyncio.QueueEmpty:
+            break
+
+
+async def _execute_single_generate_reply_task(request_item: GenerateReplyRequest, request_id: UUID) -> Tuple[UUID, Any]:
+    """Helper to run a single generate reply task and return its result or exception."""
+    try:
+        # Run blocking LLM call in a thread pool
+        result = await asyncio.to_thread(
+            _generate_response_internal,
+            request_item.email_text, request_item.groq_api_key, request_item.language,
+            request_item.length, request_item.style, request_item.tone, request_item.emoji
+        )
+        return request_id, result # This is the reply string
+    except Exception as e:
+        return request_id, e
+
+async def generate_reply_batch_worker():
+    global generate_reply_queue, generate_pending_requests, shutdown_event
+    print(f"[{datetime.now()}] Generate Reply Batch Worker started.")
+    while not shutdown_event.is_set():
+        try:
+            await asyncio.wait_for(asyncio.sleep(BATCH_INTERVAL_SECONDS), timeout=BATCH_INTERVAL_SECONDS + 0.1)
+        except asyncio.TimeoutError:
+            if shutdown_event.is_set(): break
+
+        batch_to_process: List[Tuple[GenerateReplyRequest, UUID]] = []
+        while len(batch_to_process) < MAX_BATCH_SIZE:
+            try:
+                request_obj, req_id = generate_reply_queue.get_nowait()
+                batch_to_process.append((request_obj, req_id))
+            except asyncio.QueueEmpty:
+                break
+
+        if not batch_to_process:
+            continue
+
+        print(f"[{datetime.now()}] Reply Worker: Processing batch of {len(batch_to_process)} requests.")
+        
+        llm_tasks = [_execute_single_generate_reply_task(req_obj, req_id) for req_obj, req_id in batch_to_process]
+        results = await asyncio.gather(*llm_tasks)
+
+        for req_id, result_or_exc in results:
+            future = generate_pending_requests.pop(req_id, None)
+            if future and not future.done():
+                if isinstance(result_or_exc, Exception):
+                    future.set_exception(result_or_exc)
+                else:
+                    future.set_result(result_or_exc) # This is reply string
+            elif future and future.done():
+                 print(f"[{datetime.now()}] Reply Worker: Future for {req_id} was already done (e.g. timed out).")
+
+    print(f"[{datetime.now()}] Generate Reply Batch Worker shutting down.")
+    while not generate_reply_queue.empty():
+        try:
+            _, req_id = generate_reply_queue.get_nowait()
+            future = generate_pending_requests.pop(req_id, None)
+            if future and not future.done():
+                future.set_exception(HTTPException(status_code=503, detail="Service shutting down, request cancelled."))
+        except asyncio.QueueEmpty:
+            break
+
+# --- FastAPI Event Handlers ---
 @app.on_event("startup")
 async def startup_event():
     global client, db, extracted_emails_collection, generated_replies_collection
+    global extract_data_queue, generate_reply_queue # Initialize queues
+
     print(f"[{datetime.now()}] FastAPI app startup sequence initiated.")
     try:
-        # Connect to MongoDB
         client = MongoClient(MONGO_URI, serverSelectionTimeoutMS=5000)
-        client.admin.command('ping') # Test connection
+        client.admin.command('ping')
         db = client[DB_NAME]
         extracted_emails_collection = db[EXTRACTED_EMAILS_COLLECTION]
         generated_replies_collection = db[GENERATED_REPLIES_COLLECTION]
         print(f"[{datetime.now()}] Successfully connected to MongoDB: {DB_NAME}")
 
+        # Initialize queues and start worker tasks
+        extract_data_queue = asyncio.Queue()
+        generate_reply_queue = asyncio.Queue()
+        asyncio.create_task(extract_data_batch_worker())
+        asyncio.create_task(generate_reply_batch_worker())
+        print(f"[{datetime.now()}] Batch processing workers started.")
+
     except (ConnectionFailure, OperationFailure) as e:
         print(f"[{datetime.now()}] ERROR: MongoDB Connection/Operation Failure: {e}")
-        client = None
-        db = None
-        extracted_emails_collection = None
-        generated_replies_collection = None
+        # Critical error, prevent app from fully starting or indicate non-operational state
+        # For simplicity, we'll let it run but endpoints relying on DB will fail
+        client = db = extracted_emails_collection = generated_replies_collection = None
     except Exception as e:
-        print(f"[{datetime.now()}] ERROR: An unexpected error occurred during MongoDB connection startup: {e}")
+        print(f"[{datetime.now()}] ERROR: An unexpected error during startup: {e}")
         traceback.print_exc()
-        client = None
-        db = None
-        extracted_emails_collection = None
-        generated_replies_collection = None
+        client = db = extracted_emails_collection = generated_replies_collection = None
     finally:
-        if client is not None and db is not None:
-            try:
-                client.admin.command('ping')
-            except Exception as e:
-                print(f"[{datetime.now()}] MongoDB ping failed after initial connection attempt during finally block: {e}")
-                client = None; db = None; extracted_emails_collection = None; generated_replies_collection = None
-        else:
-            print(f"[{datetime.now()}] MongoDB client or db object is None after connection attempt in startup. Database likely not connected.")
-            if client is None or db is None:
-                client = None; db = None; extracted_emails_collection = None; generated_replies_collection = None
-        print(f"[{datetime.now()}] FastAPI app startup sequence completed for MongoDB client initialization.")
+         # Simplified check after connection attempt
+        if not (client and db and extracted_emails_collection and generated_replies_collection):
+            print(f"[{datetime.now()}] MongoDB or dependent services (batch queues) might not be fully initialized.")
+        print(f"[{datetime.now()}] FastAPI app startup sequence completed.")
 
 
 @app.on_event("shutdown")
-async def shutdown_event():
-    global client
+async def shutdown_event_handler(): # Renamed to avoid conflict with global shutdown_event
+    global client, shutdown_event # Use the global shutdown_event
     print(f"[{datetime.now()}] FastAPI app shutting down.")
+    
+    # Signal workers to stop
+    shutdown_event.set()
+    
+    # Give workers a moment to process their current items or exit
+    # This timeout should be slightly longer than BATCH_INTERVAL_SECONDS to allow a final batch cycle
+    # Or join the worker tasks if they are stored globally (more robust)
+    await asyncio.sleep(BATCH_INTERVAL_SECONDS + 0.5) 
+
     if client:
         client.close()
         print(f"[{datetime.now()}] MongoDB client closed.")
-
+    print(f"[{datetime.now()}] FastAPI app shutdown sequence completed.")
 
 # --- API Endpoints ---
 @app.get("/health", summary="Health Check")
 async def health_check():
-    """
-    Checks the health of the API, including MongoDB connection.
-    """
     db_status = "MongoDB not connected."
     db_ok = False
-    if client is not None and db is not None:
+    if client and db:
         try:
-            # Use asyncio.to_thread for blocking MongoDB call
             await asyncio.to_thread(db.list_collection_names)
             db_status = "MongoDB connection OK."
             db_ok = True
         except Exception as e:
             db_status = f"MongoDB connection error: {e}"
-            db_ok = False
-
+    
+    queue_status = {
+        "extract_data_queue_size": extract_data_queue.qsize() if extract_data_queue else "N/A",
+        "generate_reply_queue_size": generate_reply_queue.qsize() if generate_reply_queue else "N/A"
+    }
+    
     if db_ok:
-        return {"status": "ok", "message": "Email Assistant API is up.", "database": db_status}
+        return {"status": "ok", "message": "Email Assistant API is up.", "database": db_status, "queues": queue_status}
     else:
-        raise HTTPException(
-            status_code=503,
-            detail={"message": "Service unavailable.", "database": db_status}
-        )
+        raise HTTPException(status_code=503, detail={"message": "Service unavailable.", "database": db_status, "queues": queue_status})
 
+@app.post("/extract-data", response_model=ExtractedData, summary="Extract structured data (batched)")
+async def extract_email_data(request: ProcessEmailRequest):
+    global extract_data_queue, extract_pending_requests, extracted_emails_collection
 
-@app.post("/generate-reply", response_model=GenerateReplyResponse, summary="Generate a smart reply to an email")
-async def generate_email_reply(request: GenerateReplyRequest):
-    """
-    Generates a smart reply to the provided email text using an LLM.
-    The generated reply is also stored in MongoDB for caching and historical purposes.
-    """
-    if generated_replies_collection is None:
-        raise HTTPException(status_code=503, detail="MongoDB not available for generated_replies.")
+    if not extracted_emails_collection or not extract_data_queue:
+        raise HTTPException(status_code=503, detail="Service not available (DB or batch queue).")
+
+    request_id = uuid4()
+    future = asyncio.get_event_loop().create_future()
+    extract_pending_requests[request_id] = future
 
     try:
-        # Check cache first
-        cache_query = {
-            "original_email_text": request.email_text,
-            "language": request.language,
-            "length": request.length,
-            "style": request.style,
-            "tone": request.tone,
-            "emoji": request.emoji,
-        }
-        print(f"[{datetime.now()}] /generate-reply: Checking cache for reply...")
-        # Use asyncio.to_thread for blocking MongoDB operations
-        cached_reply_doc = await asyncio.to_thread(generated_replies_collection.find_one, cache_query)
-
-        if cached_reply_doc:
-            print(f"[{datetime.now()}] /generate-reply: Reply found in cache. ID: {str(cached_reply_doc['_id'])}")
-            return GenerateReplyResponse(
-                reply=cached_reply_doc["generated_reply_text"],
-                stored_id=str(cached_reply_doc["_id"]),
-                cached=True
-            )
-
-        # If not in cache, directly call the internal LLM function
-        print(f"[{datetime.now()}] /generate-reply: Reply not in cache. Calling LLM for generation...")
-        reply_content = await asyncio.to_thread(
-            _generate_response_internal,
-            request.email_text,
-            request.groq_api_key,
-            request.language,
-            request.length,
-            request.style,
-            request.tone,
-            request.emoji
-        )
-        print(f"[{datetime.now()}] /generate-reply: LLM call completed. Storing newly generated reply in MongoDB.")
+        await extract_data_queue.put((request, request_id))
+        print(f"[{datetime.now()}] /extract-data: Queued request {request_id}. Queue size: {extract_data_queue.qsize()}")
 
-        # Prepare data for storage
-        reply_data_to_store = GeneratedReplyData(
-            original_email_text=request.email_text,
-            generated_reply_text=reply_content,
-            language=request.language,
-            length=request.length,
-            style=request.style,
-            tone=request.tone,
-            emoji=request.emoji
-        )
-        # Use model_dump for Pydantic v2. Exclude 'id' as it's generated by MongoDB.
-        reply_data_dict = reply_data_to_store.model_dump(by_alias=True, exclude_none=True, exclude={'id'})
+        # Wait for the future to be resolved by the worker, with a timeout
+        try:
+            # Timeout should be configurable, longer than batch interval + processing time
+            extracted_data_obj = await asyncio.wait_for(future, timeout=60.0)
+        except asyncio.TimeoutError:
+            print(f"[{datetime.now()}] /extract-data: Request {request_id} timed out waiting for worker.")
+            # The future might still be in extract_pending_requests if worker hasn't processed it
+            # Worker will try to pop it; if already popped here, it's fine.
+            extract_pending_requests.pop(request_id, None) # Clean up if timed out
+            raise HTTPException(status_code=504, detail="Request timed out while awaiting processing in batch.")
+        
+        # If here, extracted_data_obj is the ExtractedData model instance from the worker
+        print(f"[{datetime.now()}] /extract-data: Worker processed {request_id}. Inserting to DB.")
+        
+        data_to_insert = extracted_data_obj.model_dump(by_alias=True, exclude_none=True, exclude={'id'})
+        if 'appointments' in data_to_insert:
+            for appt in data_to_insert['appointments']:
+                if isinstance(appt.get('start_date'), date): appt['start_date'] = datetime.combine(appt['start_date'], datetime.min.time())
+                if isinstance(appt.get('end_date'), date) and appt.get('end_date') is not None: appt['end_date'] = datetime.combine(appt['end_date'], datetime.min.time())
+        if 'tasks' in data_to_insert:
+            for task_item in data_to_insert['tasks']:
+                if isinstance(task_item.get('due_date'), date): task_item['due_date'] = datetime.combine(task_item['due_date'], datetime.min.time())
+        
+        insert_result = await asyncio.to_thread(extracted_emails_collection.insert_one, data_to_insert)
+        extracted_data_obj.id = str(insert_result.inserted_id) # Update with DB ID
+        
+        return extracted_data_obj
 
-        # Insert into MongoDB
-        insert_result = await asyncio.to_thread(generated_replies_collection.insert_one, reply_data_dict)
-        stored_id = str(insert_result.inserted_id) # Convert ObjectId to string for the response
+    except HTTPException: # Re-raise HTTPExceptions (like timeout or from future)
+        raise
+    except ValueError as ve: # Typically from Pydantic validation or LLM output parsing
+        raise HTTPException(status_code=400, detail=str(ve))
+    except Exception as e: # Any other exception from the future or this handler
+        traceback.print_exc()
+        raise HTTPException(status_code=500, detail=f"Internal server error during extract data: {e}")
+    finally:
+        # Ensure future is removed if it wasn't already (e.g. successful completion)
+        extract_pending_requests.pop(request_id, None)
 
-        print(f"[{datetime.now()}] /generate-reply: Reply stored in MongoDB. ID: {stored_id}")
 
-        # Return the response as per GenerateReplyResponse model
-        return GenerateReplyResponse(
-            reply=reply_content,
-            stored_id=stored_id,
-            cached=False # Always False since we just generated it
-        )
-    except Exception as e:
-        traceback.print_exc()
-        # Ensure consistent error response
-        raise HTTPException(status_code=500, detail=f"Error generating or storing reply: {str(e)}")
+@app.post("/generate-reply", response_model=GenerateReplyResponse, summary="Generate smart reply (batched)")
+async def generate_email_reply(request: GenerateReplyRequest):
+    global generate_reply_queue, generate_pending_requests, generated_replies_collection
 
-@app.post("/extract-data", response_model=ExtractedData, summary="Extract structured data from an email")
-async def extract_email_data(request: ProcessEmailRequest):
-    """
-    Extracts contacts, appointments, and tasks from the provided email text.
-    """
-    if extracted_emails_collection is None:
-        raise HTTPException(status_code=503, detail="MongoDB not available.")
+    if not generated_replies_collection or not generate_reply_queue:
+        raise HTTPException(status_code=503, detail="Service not available (DB or batch queue).")
 
-    current_date = date.today() # Get current date for context
+    # --- Check cache first (remains outside batching) ---
+    cache_query = {"original_email_text": request.email_text, "language": request.language, "length": request.length, "style": request.style, "tone": request.tone, "emoji": request.emoji}
+    cached_reply_doc = await asyncio.to_thread(generated_replies_collection.find_one, cache_query)
+    if cached_reply_doc:
+        print(f"[{datetime.now()}] /generate-reply: Reply found in cache. ID: {str(cached_reply_doc['_id'])}")
+        return GenerateReplyResponse(reply=cached_reply_doc["generated_reply_text"], stored_id=str(cached_reply_doc["_id"]), cached=True)
 
-    print(f"[{datetime.now()}] /extract-data: Received request.")
-    try:
-        print(f"[{datetime.now()}] /extract-data: Calling internal processing function.")
-        # Run blocking LLM call in a thread pool
-        extracted_data = await asyncio.to_thread(_process_email_internal, request.email_text, request.groq_api_key, current_date)
+    # --- If not cached, queue for generation ---
+    request_id = uuid4()
+    future = asyncio.get_event_loop().create_future()
+    generate_pending_requests[request_id] = future
 
-        print(f"[{datetime.now()}] /extract-data: Internal processing complete. Preparing for DB insert.")
-        # Convert Pydantic model to dictionary for MongoDB insert, handling _id alias
-        # Use model_dump for Pydantic v2
-        data_to_insert = extracted_data.model_dump(by_alias=True, exclude_none=True, exclude={'id'})
+    try:
+        await generate_reply_queue.put((request, request_id))
+        print(f"[{datetime.now()}] /generate-reply: Queued request {request_id}. Queue size: {generate_reply_queue.qsize()}")
 
-        # --- NEW CONVERSION FOR MONGODB ---
-        # MongoDB's BSON doesn't natively support Python's datetime.date type.
-        # It expects datetime.datetime. Convert all date fields to datetime.datetime.
-        if 'appointments' in data_to_insert:
-            for appt in data_to_insert['appointments']:
-                if isinstance(appt.get('start_date'), date):
-                    appt['start_date'] = datetime.combine(appt['start_date'], datetime.min.time())
-                if isinstance(appt.get('end_date'), date) and appt.get('end_date') is not None:
-                    appt['end_date'] = datetime.combine(appt['end_date'], datetime.min.time())
-        if 'tasks' in data_to_insert:
-            for task_item in data_to_insert['tasks']:
-                if isinstance(task_item.get('due_date'), date):
-                    task_item['due_date'] = datetime.combine(task_item['due_date'], datetime.min.time())
-        # --- END NEW CONVERSION ---
+        try:
+            # Timeout should be configurable
+            reply_content_str = await asyncio.wait_for(future, timeout=60.0) 
+        except asyncio.TimeoutError:
+            print(f"[{datetime.now()}] /generate-reply: Request {request_id} timed out waiting for worker.")
+            generate_pending_requests.pop(request_id, None)
+            raise HTTPException(status_code=504, detail="Request timed out while awaiting reply generation in batch.")
 
-        print(f"[{datetime.now()}] /extract-data: Inserting into MongoDB... Data: {data_to_insert}") # Add data logging
-        # Use asyncio.to_thread for blocking MongoDB insert operation
-        insert_result = await asyncio.to_thread(extracted_emails_collection.insert_one, data_to_insert)
+        # If here, reply_content_str is the generated string from the worker
+        print(f"[{datetime.now()}] /generate-reply: Worker generated reply for {request_id}. Storing to DB.")
 
-        # Update the extracted_data object with the MongoDB-generated ID
-        extracted_data.id = str(insert_result.inserted_id)
-        print(f"[{datetime.now()}] /extract-data: Data inserted into MongoDB. ID: {extracted_data.id}")
+        reply_data_to_store = GeneratedReplyData(
+            original_email_text=request.email_text, generated_reply_text=reply_content_str,
+            language=request.language, length=request.length, style=request.style,
+            tone=request.tone, emoji=request.emoji
+        )
+        reply_data_dict = reply_data_to_store.model_dump(by_alias=True, exclude_none=True, exclude={'id'})
+        
+        insert_result = await asyncio.to_thread(generated_replies_collection.insert_one, reply_data_dict)
+        stored_id = str(insert_result.inserted_id)
+        
+        return GenerateReplyResponse(reply=reply_content_str, stored_id=stored_id, cached=False)
 
-        return extracted_data
-    except ValueError as ve:
-        raise HTTPException(status_code=400, detail=str(ve))
+    except HTTPException:
+        raise
     except Exception as e:
-        traceback.print_exc() # Print full traceback for debugging
-        raise HTTPException(status_code=500, detail=f"An internal server error occurred: {e}")
+        traceback.print_exc()
+        raise HTTPException(status_code=500, detail=f"Internal server error during generate reply: {e}")
+    finally:
+        generate_pending_requests.pop(request_id, None)
 
 
 @app.get("/query-extracted-emails", response_model=List[ExtractedData], summary="Query stored extracted email data")
 async def query_extracted_emails(query_params: ExtractedEmailQuery = Depends()):
-    """
-    Queries extracted email data from MongoDB based on various filters.
-    """
-    if extracted_emails_collection is None:
-        raise HTTPException(status_code=503, detail="MongoDB not available.")
-
+    if extracted_emails_collection is None: raise HTTPException(status_code=503, detail="MongoDB not available.")
     mongo_query = {}
-    if query_params.contact_name:
-        # Case-insensitive partial match on contact name or last name
-        mongo_query["$or"] = [
-            {"contacts.name": {"$regex": query_params.contact_name, "$options": "i"}},
-            {"contacts.last_name": {"$regex": query_params.contact_name, "$options": "i"}}
-        ]
-    if query_params.appointment_title:
-        mongo_query["appointments.title"] = {"$regex": query_params.appointment_title, "$options": "i"}
-    if query_params.task_title:
-        mongo_query["tasks.task_title"] = {"$regex": query_params.task_title, "$options": "i"}
-
-    # Date range filtering for processed_at
+    if query_params.contact_name: mongo_query["$or"] = [{"contacts.name": {"$regex": query_params.contact_name, "$options": "i"}}, {"contacts.last_name": {"$regex": query_params.contact_name, "$options": "i"}}]
+    if query_params.appointment_title: mongo_query["appointments.title"] = {"$regex": query_params.appointment_title, "$options": "i"}
+    if query_params.task_title: mongo_query["tasks.task_title"] = {"$regex": query_params.task_title, "$options": "i"}
     date_query = {}
-    if query_params.from_date:
-        date_query["$gte"] = datetime.combine(query_params.from_date, datetime.min.time())
-    if query_params.to_date:
-        date_query["$lte"] = datetime.combine(query_params.to_date, datetime.max.time())
-    if date_query:
-        mongo_query["processed_at"] = date_query
-
+    if query_params.from_date: date_query["$gte"] = datetime.combine(query_params.from_date, datetime.min.time())
+    if query_params.to_date: date_query["$lte"] = datetime.combine(query_params.to_date, datetime.max.time())
+    if date_query: mongo_query["processed_at"] = date_query
     try:
-        # Use asyncio.to_thread for blocking MongoDB find operation
         cursor = await asyncio.to_thread(extracted_emails_collection.find, mongo_query)
-        # Use to_list to limit results and convert to list
         results = await asyncio.to_thread(lambda: list(cursor.limit(query_params.limit)))
-
-        # Convert MongoDB documents to ExtractedData Pydantic models
         return [ExtractedData(**doc) for doc in results]
-    except Exception as e:
-        traceback.print_exc()
-        raise HTTPException(status_code=500, detail=f"Error querying extracted emails: {e}")
-
+    except Exception as e: traceback.print_exc(); raise HTTPException(status_code=500, detail=f"Error querying extracted emails: {e}")
 
 @app.get("/query-generated-replies", response_model=List[GeneratedReplyData], summary="Query stored generated replies")
 async def query_generated_replies(query_params: GeneratedReplyQuery = Depends()):
-    """
-    Queries generated email replies from MongoDB based on various filters.
-    """
-    if generated_replies_collection is None:
-        raise HTTPException(status_code=503, detail="MongoDB not available.")
-
+    if generated_replies_collection is None: raise HTTPException(status_code=503, detail="MongoDB not available.")
     mongo_query = {}
-    if query_params.language:
-        mongo_query["language"] = query_params.language
-    if query_params.style:
-        mongo_query["style"] = query_params.style
-    if query_params.tone:
-        mongo_query["tone"] = query_params.tone
-
-    # Date range filtering for generated_at
+    if query_params.language: mongo_query["language"] = query_params.language
+    if query_params.style: mongo_query["style"] = query_params.style
+    if query_params.tone: mongo_query["tone"] = query_params.tone
     date_query = {}
-    if query_params.from_date:
-        date_query["$gte"] = datetime.combine(query_params.from_date, datetime.min.time())
-    if query_params.to_date:
-        date_query["$lte"] = datetime.combine(query_params.to_date, datetime.max.time())
-    if date_query:
-        mongo_query["generated_at"] = date_query
-
+    if query_params.from_date: date_query["$gte"] = datetime.combine(query_params.from_date, datetime.min.time())
+    if query_params.to_date: date_query["$lte"] = datetime.combine(query_params.to_date, datetime.max.time())
+    if date_query: mongo_query["generated_at"] = date_query
     try:
-        # Use asyncio.to_thread for blocking MongoDB find operation
         cursor = await asyncio.to_thread(generated_replies_collection.find, mongo_query)
-        # Use to_list to limit results and convert to list
         results = await asyncio.to_thread(lambda: list(cursor.limit(query_params.limit)))
-
-        # Convert MongoDB documents to GeneratedReplyData Pydantic models
         return [GeneratedReplyData(**doc) for doc in results]
-    except Exception as e:
-        traceback.print_exc()
-        raise HTTPException(status_code=500, detail=f"Error querying generated replies: {e}")
+    except Exception as e: traceback.print_exc(); raise HTTPException(status_code=500, detail=f"Error querying generated replies: {e}")