Update flask_Character.py

flask_Character.py

@@ -1,150 +1,866 @@
+# 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
-import logging
-from fastapi import FastAPI, Response, status
+# 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.
+# Keep them if they serve a specific purpose in your deployment environment.
+os.environ["NUMBA_CACHE_DIR"] = "/tmp/numba_cache"
+os.environ["NUMBA_DISABLE_CACHE"] = "1"
+
+import json
+import re
+from datetime import date, datetime, timedelta
+from typing import List, Optional, Literal, Dict, Any, Tuple
+import traceback
+import asyncio
+
+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
+from typing_extensions import Annotated
+from pydantic_core import core_schema # Import core_schema for direct use in __get_pydantic_json_schema__
+
 from pymongo import MongoClient
-from pymongo.errors import ConfigurationError, ConnectionFailure
-import time # For simulating a batch processor initialization delay
-
-# --- 1. Configuration and Global Variables ---
-# Use a proper logger for better debugging and visibility
-logger = logging.getLogger(__name__)
-logger.setLevel(logging.INFO) # Set desired logging level (INFO, DEBUG, WARNING, ERROR, CRITICAL)
-handler = logging.StreamHandler()
-formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
-handler.setFormatter(formatter)
-logger.addHandler(handler)
-
-# Retrieve MONGO_URI from environment variable. If not set, it will be None.
-MONGO_URI = os.getenv("MONGO_URI")
-DB_NAME = "email_assistant_db" # Replace with your actual database name
-
-# Global variables to hold MongoDB client and database instance
-# These will be initialized in the startup event
-mongo_client = None
-mongo_db = None
-
-# Placeholder for your Batch Processor (if applicable)
-batch_processor_initialized = False
-
-# --- 2. FastAPI Application Instance ---
+from pymongo.errors import ConnectionFailure, OperationFailure
+from bson import ObjectId
+
+# --- 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"
+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
+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
+    @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
+        return json_schema
+
+# Annotated type for convenience in models
+PyObjectId = Annotated[CustomObjectId, BeforeValidator(str)]
+
+
+# ---------------------- Models ----------------------
+class Contact(BaseModel):
+    name: str
+    last_name: str
+    email: Optional[str] = None
+    phone_number: Optional[str] = None
+
+class Appointment(BaseModel):
+    title: str
+    description: str
+    start_date: date
+    start_time: Optional[str] = None
+    end_date: Optional[date] = None
+    end_time: Optional[str] = None
+
+class Task(BaseModel):
+    task_title: str
+    task_description: str
+    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
+    @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):
+                    appt['start_date'] = appt['start_date'].isoformat()
+                if isinstance(appt.get('end_date'), date) and appt.get('end_date') is not None:
+                    appt['end_date'] = appt['end_date'].isoformat()
+        if 'tasks' in data:
+            for task_item in data['tasks']:
+                if isinstance(task_item.get('due_date'), date):
+                    task_item['due_date'] = task_item['due_date'].isoformat()
+        return data
+
+class ProcessEmailRequest(BaseModel):
+    email_text: str = Field(..., example="Oggetto: Follow-up progetto “Delta”...")
+    groq_api_key: str = Field(..., example="YOUR_GROQ_API_KEY")
+
+class GenerateReplyRequest(BaseModel):
+    email_text: str = Field(..., example="Oggetto: Follow-up progetto “Delta”...")
+    groq_api_key: str = Field(..., example="YOUR_GROQ_API_KEY")
+    language: Literal["Italian", "English"] = Field("Italian", examples=["Italian", "English"])
+    length: str = Field("Auto", examples=["Short", "Medium", "Long", "Auto"])
+    style: str = Field("Professional", examples=["Professional", "Casual", "Formal", "Informal"])
+    tone: str = Field("Friendly", examples=["Friendly", "Neutral", "Urgent", "Empathetic"])
+    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
+    language: str
+    length: str
+    style: str
+    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)
+        if "_id" in data and isinstance(data["_id"], ObjectId):
+            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.")
+    cached: bool = Field(..., description="True if the reply was retrieved from cache, False if newly generated.")
+
+# --- 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.")
+
+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.")
+
+# ---------------------- 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
+    match = re.search(r'\{.*\}', text, re.DOTALL)
+    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
+    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
+
+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
+        ))
+    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.")
+
+    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)
+    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_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}")
+        json_data = json.loads(json_str)
+
+        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."
+    
+    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
+        )
+        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
+
+# --- Batching and Caching Configuration ---
+MAX_BATCH_SIZE = 20
+BATCH_TIMEOUT = 0.5 # seconds (Adjust based on expected LLM response time and desired latency)
+
+reply_request_queue: List[Tuple[GenerateReplyRequest, asyncio.Future, float]] = []
+reply_queue_lock = asyncio.Lock()
+reply_queue_condition = asyncio.Condition(lock=reply_queue_lock)
+batch_processor_task: Optional[asyncio.Task] = None
+
+
+# --- Batch Processor and Handler ---
+async def handle_single_reply_request(request_data: GenerateReplyRequest, future: asyncio.Future):
+    """Handles a single request: checks cache, calls LLM, stores result, and sets future."""
+    print(f"[{datetime.now()}] Handle single reply: Starting for email_text_start='{request_data.email_text[:50]}'...")
+    if future.cancelled():
+        print(f"[{datetime.now()}] Handle single reply: Future cancelled. Aborting.")
+        return
+    try:
+        if generated_replies_collection is None:
+            print(f"[{datetime.now()}] Handle single reply: DB collection 'generated_replies_collection' is None.")
+            if not future.done():
+                future.set_exception(HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Database service not available for caching/storage."))
+            return
+
+        cache_query = {
+            "original_email_text": request_data.email_text,
+            "language": request_data.language,
+            "length": request_data.length,
+            "style": request_data.style,
+            "tone": request_data.tone,
+            "emoji": request_data.emoji,
+        }
+        print(f"[{datetime.now()}] Handle single reply: Checking cache for reply...")
+        # Use await 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()}] Handle single reply: Reply found in cache. ID: {str(cached_reply_doc['_id'])}")
+            response = {
+                "reply": cached_reply_doc["generated_reply_text"],
+                "stored_id": str(cached_reply_doc["_id"]),
+                "cached": True
+            }
+            if not future.done():
+                future.set_result(response)
+            print(f"[{datetime.now()}] Handle single reply: Cache result set on future.")
+            return
+
+        print(f"[{datetime.now()}] Handle single reply: Reply not in cache. Calling LLM...")
+        reply_content = await asyncio.to_thread(
+            _generate_response_internal,
+            request_data.email_text,
+            request_data.groq_api_key,
+            request_data.language,
+            request_data.length,
+            request_data.style,
+            request_data.tone,
+            request_data.emoji
+        )
+        print(f"[{datetime.now()}] Handle single reply: LLM call completed. Reply length: {len(reply_content)}.")
+
+        reply_data_to_store = GeneratedReplyData(
+            original_email_text=request_data.email_text,
+            generated_reply_text=reply_content,
+            language=request_data.language,
+            length=request_data.length,
+            style=request_data.style,
+            tone=request_data.tone,
+            emoji=request_data.emoji
+        )
+        print(f"[{datetime.now()}] Handle single reply: Storing reply in DB...")
+        # Use model_dump for Pydantic v2
+        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)
+        print(f"[{datetime.now()}] Handle single reply: Reply stored in DB. ID: {stored_id}")
+
+        final_response = {
+            "reply": reply_content,
+            "stored_id": stored_id,
+            "cached": False
+        }
+        if not future.done():
+            future.set_result(final_response)
+        print(f"[{datetime.now()}] Handle single reply: Final result set on future.")
+
+    except Exception as e:
+        print(f"[{datetime.now()}] Handle single reply: EXCEPTION: {e}")
+        traceback.print_exc() # Print full traceback to logs
+        if not future.done():
+            # Set the exception on the future so the client can catch it
+            future.set_exception(HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to generate reply: {e}"))
+        print(f"[{datetime.now()}] Handle single reply: Exception set on future.")
+
+
+async def process_reply_batches():
+    """Continuously processes requests from the reply_request_queue in batches."""
+    global reply_request_queue
+    print(f"[{datetime.now()}] Batch processor task started.")
+    while True:
+        batch_to_fire: List[Tuple[GenerateReplyRequest, asyncio.Future]] = []
+        async with reply_queue_condition:
+            if not reply_request_queue:
+                print(f"[{datetime.now()}] Batch processor: Queue empty, waiting for requests...")
+                # Wait for new requests or timeout
+                await reply_queue_condition.wait()
+                # After waking up, re-check if queue is still empty
+                if not reply_request_queue:
+                    print(f"[{datetime.now()}] Batch processor: Woke up, queue still empty. Continuing loop.")
+                    continue
+
+            now = asyncio.get_event_loop().time()
+            # Safety check: ensure queue is not empty before accessing index 0
+            if reply_request_queue:
+                oldest_item_timestamp = reply_request_queue[0][2]
+            else:
+                # If queue became empty while waiting, loop again
+                print(f"[{datetime.now()}] Batch processor: Queue became empty before processing. Restarting loop.")
+                continue
+
+            print(f"[{datetime.now()}] Batch processor: Woke up. Queue size: {len(reply_request_queue)}. Oldest item age: {now - oldest_item_timestamp:.2f}s")
+
+            # Condition to trigger batch processing: queue is full OR timeout reached for oldest item
+            if len(reply_request_queue) >= MAX_BATCH_SIZE or \
+               (now - oldest_item_timestamp >= BATCH_TIMEOUT):
+                num_to_take = min(len(reply_request_queue), MAX_BATCH_SIZE)
+                for _ in range(num_to_take):
+                    # Safety check: ensure queue is not empty before popping
+                    if reply_request_queue:
+                        req, fut, _ = reply_request_queue.pop(0)
+                        batch_to_fire.append((req, fut))
+                print(f"[{datetime.now()}] Batch processor: Firing batch of {len(batch_to_fire)} requests.")
+            else:
+                # Calculate time to wait for the next batch or timeout
+                time_to_wait = BATCH_TIMEOUT - (now - oldest_item_timestamp)
+                print(f"[{datetime.now()}] Batch processor: Not enough requests or timeout not reached. Waiting for {time_to_wait:.2f}s.")
+                try:
+                    await asyncio.wait_for(reply_queue_condition.wait(), timeout=time_to_wait)
+                except asyncio.TimeoutError:
+                    print(f"[{datetime.now()}] Batch processor: wait timed out.")
+                    pass # Loop will re-evaluate and likely fire the batch
+
+        if batch_to_fire:
+            tasks = [handle_single_reply_request(req_data, fut) for req_data, fut in batch_to_fire]
+            print(f"[{datetime.now()}] Batch processor: Awaiting completion of {len(tasks)} single reply tasks.")
+            await asyncio.gather(*tasks)
+            print(f"[{datetime.now()}] Batch processor: Batch processing complete.")
+        else:
+            # Short sleep to prevent busy-waiting if queue is empty but not waiting
+            await asyncio.sleep(0.001)
+
+
+# ---------------------- FastAPI Application ----------------------
 app = FastAPI(
-    title="Character API",
-    description="API for managing character data and processing emails.",
-    version="0.0.1",
+    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, batching, and caching.",
+    version="1.1.0",
+    docs_url="/", # Sets Swagger UI to be the root path
+    redoc_url="/redoc"
 )
 
-# --- 3. Application Startup Event Handler ---
-@app.on_event("startup")
-async def startup_event():
-    global mongo_client, mongo_db, batch_processor_initialized
+# --- Global Exception Handler ---
+# Catch Starlette HTTPExceptions (FastAPI uses these internally)
+@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)
 
-    logger.info("Application startup initiated.")
+# 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"
+    )
 
-    # Check for MONGO_URI BEFORE attempting connection
-    if not MONGO_URI:
-        logger.critical("CRITICAL ERROR: MONGO_URI environment variable is NOT SET.")
-        logger.critical("Please set the MONGO_URI environment variable with your MongoDB Atlas connection string.")
-        logger.critical("Example: mongodb+srv://<user>:<pass>@cluster.mongodb.net/dbname?retryWrites=true&w=majority")
-        # Depending on your application's criticality, you might want to exit or raise an exception here
-        # raise ValueError("MONGO_URI environment variable is not set. Cannot connect to MongoDB.")
-        return # Prevent further execution if critical dependency is missing
 
+# --- FastAPI Event Handlers for MongoDB & Batch Processor ---
+@app.on_event("startup")
+async def startup_event():
+    global client, db, extracted_emails_collection, generated_replies_collection, batch_processor_task
+    print(f"[{datetime.now()}] FastAPI app startup sequence initiated.")
     try:
-        logger.info("Attempting to connect to MongoDB...")
-        # Initialize MongoClient
-        client = MongoClient(MONGO_URI, serverSelectionTimeoutMS=5000) # 5-second timeout for server selection
-
-        # The ismaster command is a lightweight way to check connectivity and auth.
-        # It triggers the actual connection attempt and DNS resolution.
-        client.admin.command('ismaster')
-
-        mongo_client = client
-        mongo_db = client[DB_NAME]
-        logger.info(f"SUCCESS: Successfully connected to MongoDB database: '{DB_NAME}'")
-
-        # --- Initialize Batch Processor (if applicable) ---
-        logger.info("Initializing Batch Processor...")
-        # Simulate some work
-        time.sleep(1) # Simulate delay for batch processor setup
-        batch_processor_initialized = True
-        logger.info("Batch Processor initialized successfully.")
-
-    except ConfigurationError as e:
-        logger.critical(f"FATAL ERROR: MongoDB Configuration Error: {e}")
-        logger.critical("Please check your MONGO_URI format and ensure the hostname is correct and resolvable.")
-        mongo_client = None
-        mongo_db = None
-    except ConnectionFailure as e:
-        logger.critical(f"FATAL ERROR: MongoDB Connection Failure: {e}")
-        logger.critical("Check network connectivity, MongoDB Atlas IP access list, and firewall rules.")
-        logger.critical("Also verify that your MongoDB user credentials are correct.")
-        mongo_client = None
-        mongo_db = None
-    except Exception as e:
-        logger.critical(f"FATAL ERROR: An unexpected error occurred during MongoDB connection or batch startup: {e}", exc_info=True)
-        mongo_client = None
-        mongo_db = None
+        # Connect to MongoDB
+        client = MongoClient(MONGO_URI, serverSelectionTimeoutMS=5000)
+        client.admin.command('ping') # Test connection
+        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}")
 
-    if mongo_client is None or mongo_db is None:
-        logger.critical("CRITICAL: MongoDB client or db object is None after connection attempt in startup. Application might not function correctly.")
-    else:
-        logger.info("Application startup complete. MongoDB client & Batch Processor initialization attempted.")
+        # Start the batch processor task if not already running
+        if batch_processor_task is None or batch_processor_task.done():
+            batch_processor_task = asyncio.create_task(process_reply_batches())
+            print(f"[{datetime.now()}] Batch processor task for replies started.")
+        else:
+            print(f"[{datetime.now()}] Batch processor task for replies is already running or being initialized.")
+
+    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
+    except Exception as e:
+        print(f"[{datetime.now()}] ERROR: An unexpected error occurred during MongoDB connection or batch startup: {e}")
+        traceback.print_exc()
+        client = None
+        db = None
+        extracted_emails_collection = None
+        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 & Batch Processor initialization.")
 
 
-# --- 4. Application Shutdown Event Handler (Good Practice) ---
 @app.on_event("shutdown")
 async def shutdown_event():
-    global mongo_client
-    if mongo_client:
-        mongo_client.close()
-        logger.info("MongoDB connection closed.")
-    logger.info("Application shutdown complete.")
+    global client, batch_processor_task
+    print(f"[{datetime.now()}] FastAPI app shutting down.")
+    if batch_processor_task:
+        batch_processor_task.cancel()
+        try:
+            await batch_processor_task
+            print(f"[{datetime.now()}] Batch processor task awaited.")
+        except asyncio.CancelledError:
+            print(f"[{datetime.now()}] Batch processor task for replies cancelled during shutdown.")
+        except Exception as e:
+            print(f"[{datetime.now()}] Error during batch processor task shutdown: {e}")
+            traceback.print_exc()
+        batch_processor_task = None
+
+    if client:
+        client.close()
+        print(f"[{datetime.now()}] MongoDB client closed.")
 
-# --- 5. API Endpoints ---
 
-# Health check endpoint
-@app.get("/health")
+# --- API Endpoints ---
+@app.get("/health", summary="Health Check")
 async def health_check():
-    db_status = "Connected" if mongo_db else "Disconnected"
-    batch_status = "Initialized" if batch_processor_initialized else "Not Initialized"
+    """
+    Checks the health of the API, including MongoDB connection and batch processor status.
+    """
+    db_status = "MongoDB not connected."
+    db_ok = False
+    if client is not None and db is not None:
+        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
+
+    batch_processor_status = "Batch processor not running."
+    if batch_processor_task is not None:
+        if not batch_processor_task.done():
+            batch_processor_status = "Batch processor is running."
+        else:
+            if batch_processor_task.exception():
+                batch_processor_status = f"Batch processor task ended with exception: {batch_processor_task.exception()}"
+            else:
+                batch_processor_status = "Batch processor task is done (may have completed or cancelled)."
+    else:
+        batch_processor_status = "Batch processor task has not been initialized."
 
-    if mongo_db and batch_processor_initialized:
-        return {"status": "ok", "mongodb": db_status, "batch_processor": batch_status}
+    if db_ok:
+        return {"status": "ok", "message": "Email Assistant API is up.", "database": db_status, "batch_processor": batch_processor_status}
     else:
-        # Return 503 Service Unavailable if critical dependencies are not met
-        return Response(
-            content={"status": "degraded", "mongodb": db_status, "batch_processor": batch_status},
+        raise HTTPException(
             status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
-            media_type="application/json"
+            detail={"message": "Service unavailable due to issues.", "database": db_status, "batch_processor": batch_processor_status}
         )
 
-# Example endpoint - requires MongoDB connection
-@app.get("/characters/{character_id}")
-async def get_character(character_id: str):
-    if not mongo_db:
-        logger.error(f"Attempted to access /characters/{character_id} but MongoDB is not connected.")
-        return Response(content="MongoDB is not connected.", status_code=status.HTTP_500_INTERNAL_SERVER_ERROR)
 
+@app.post("/extract-data", response_model=ExtractedData, summary="Extract structured data from an email and store in MongoDB")
+async def extract_email_data(request: ProcessEmailRequest):
+    """
+    Receives an email, extracts contacts, appointments, and tasks using an LLM,
+    and stores the extracted data in MongoDB.
+    """
+    print(f"[{datetime.now()}] /extract-data: Received request.")
+    if extracted_emails_collection is None:
+        print(f"[{datetime.now()}] /extract-data: MongoDB collection is None.")
+        raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="MongoDB not available for extracted email storage. Check server startup logs.")
     try:
-        # Example: Fetch a character from a 'characters' collection
-        # Replace with your actual collection and query
-        character_data = await mongo_db.get_collection("characters").find_one({"_id": character_id})
-        # Note: PyMongo's find_one is synchronous. If using async, you'd need motor or an async wrapper.
-        # For a truly async FastAPI app with MongoDB, consider 'motor' (async pymongo driver).
-        # For now, we'll assume this is fine in an 'await' context for simplicity,
-        # but in a real-world async app, you'd want to run blocking operations in a thread pool.
-        # e.g., result = await loop.run_in_executor(None, mongo_db.collection.find_one, {"_id": character_id})
-
-        if character_data:
-            # MongoDB's _id might be an ObjectId, convert to string for JSON serialization
-            character_data["_id"] = str(character_data["_id"])
-            return character_data
-        else:
-            return Response(content="Character not found.", status_code=status.HTTP_404_NOT_FOUND)
+        current_date_val = date.today()
+        print(f"[{datetime.now()}] /extract-data: Calling internal processing function.")
+        extracted_data = await asyncio.to_thread(
+            _process_email_internal, request.email_text, request.groq_api_key, current_date_val
+        )
+        print(f"[{datetime.now()}] /extract-data: Internal processing complete. Preparing for DB insert.")
+
+        extracted_data_dict = extracted_data.model_dump(by_alias=True, exclude_none=True)
+        # Convert date objects to datetime for MongoDB storage if they are just date objects
+        # Pydantic's default `date` handling might serialize to ISO string, but for
+        # internal MongoDB storage, sometimes `datetime` is preferred for consistency.
+        if 'appointments' in extracted_data_dict:
+            for appt in extracted_data_dict['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 extracted_data_dict:
+            for task_item in extracted_data_dict['tasks']:
+                if isinstance(task_item.get('due_date'), date):
+                    task_item['due_date'] = datetime.combine(task_item['due_date'], datetime.min.time())
+
+        print(f"[{datetime.now()}] /extract-data: Inserting into MongoDB...")
+        result = await asyncio.to_thread(extracted_emails_collection.insert_one, extracted_data_dict)
+        print(f"[{datetime.now()}] /extract-data: Data inserted into MongoDB. ID: {result.inserted_id}")
+
+        extracted_data.id = result.inserted_id
+        return extracted_data
+    except ValueError as e:
+        print(f"[{datetime.now()}] /extract-data: ValueError: {e}")
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
+    except Exception as e:
+        print(f"[{datetime.now()}] /extract-data: Unhandled Exception: {e}")
+        traceback.print_exc()
+        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Internal server error during data extraction: {e}")
+
+
+@app.post("/extract-data-excel", summary="Extract structured data and download as Excel (also stores in MongoDB)")
+async def extract_email_data_excel(request: ProcessEmailRequest):
+    """
+    Placeholder for future functionality to extract data and provide as an Excel download.
+    Currently disabled.
+    """
+    raise HTTPException(status_code=status.HTTP_501_NOT_IMPLEMENTED, detail="Excel functionality is currently disabled.")
+
+
+@app.post("/generate-reply", response_model=GenerateReplyResponse, summary="Generate a smart reply to an email (batched & cached)")
+async def generate_email_reply(request: GenerateReplyRequest):
+    """
+    Generates an intelligent email reply based on specified parameters (language, length, style, tone, emoji).
+    Uses a batch processing system with caching for efficiency.
+    """
+    print(f"[{datetime.now()}] /generate-reply: Received request.")
+    if generated_replies_collection is None or batch_processor_task is None or reply_queue_condition is None:
+        print(f"[{datetime.now()}] /generate-reply: Service not initialized. gen_replies_coll={generated_replies_collection is not None}, batch_task={batch_processor_task is not None}, queue_cond={reply_queue_condition is not None}")
+        raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Reply generation service not fully initialized. Check server logs for database or batch processor issues.")
+
+    future = asyncio.Future()
+    current_time = asyncio.get_event_loop().time()
+
+    async with reply_queue_condition:
+        reply_request_queue.append((request, future, current_time))
+        reply_queue_condition.notify() # Notify the batch processor that a new request is available
+        print(f"[{datetime.now()}] /generate-reply: Request added to queue, notifying batch processor. Queue size: {len(reply_request_queue)}")
+
+    try:
+        # Debugging: Increase timeout significantly to allow full tracing in logs
+        client_timeout = BATCH_TIMEOUT + 60.0 # Example: 0.5s batch + 60s LLM response buffer = 60.5s total timeout
+        print(f"[{datetime.now()}] /generate-reply: Waiting for future result with timeout {client_timeout}s.")
+        result = await asyncio.wait_for(future, timeout=client_timeout)
+        print(f"[{datetime.now()}] /generate-reply: Future result received. Returning data.")
+        return result
+    except asyncio.TimeoutError:
+        print(f"[{datetime.now()}] /generate-reply: Client timeout waiting for future after {client_timeout}s. Future done: {future.done()}")
+        if not future.done():
+            future.cancel() # Cancel if it's still pending
+        raise HTTPException(status_code=status.HTTP_504_GATEWAY_TIMEOUT, detail=f"Request timed out after {client_timeout}s waiting for batch processing. The LLM might be busy or the request queue too long. Check server logs for more details.")
     except Exception as e:
-        logger.error(f"Error fetching character {character_id}: {e}", exc_info=True)
-        return Response(content="Internal server error.", status_code=status.HTTP_500_INTERNAL_SERVER_ERROR)
+        if isinstance(e, HTTPException):
+            print(f"[{datetime.now()}] /generate-reply: Caught HTTPException: {e.status_code} - {e.detail}")
+            raise e # Re-raise FastAPI HTTPExceptions
+        print(f"[{datetime.now()}] /generate-reply: Unhandled Exception: {e}")
+        traceback.print_exc()
+        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Error processing your reply request: {str(e)}. Check server logs for more details.")
 
-# You would add more endpoints here for your application logic
-# e.g., @app.post("/email-process") etc.
 
+@app.get("/query-extracted-emails", response_model=List[ExtractedData], summary="Query extracted emails from MongoDB")
+async def query_extracted_emails_endpoint(query_params: ExtractedEmailQuery = Depends()):
+    print(f"[{datetime.now()}] /query-extracted-emails: Received request with params: {query_params.model_dump_json()}")
+    if extracted_emails_collection is None:
+        print(f"[{datetime.now()}] /query-extracted-emails: MongoDB collection is None.")
+        raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="MongoDB not available for querying extracted emails.")
+    mongo_query: Dict[str, Any] = {}
+    if query_params.contact_name:
+        mongo_query["contacts.name"] = {"$regex": query_params.contact_name, "$options": "i"} # Case-insensitive regex
+    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"}
+
+    if query_params.from_date or query_params.to_date:
+        date_query: Dict[str, datetime] = {}
+        if query_params.from_date:
+            date_query["$gte"] = datetime.combine(query_params.from_date, datetime.min.time())
+        if query_params.to_date:
+            # Query up to the end of the 'to_date' day
+            date_query["$lt"] = datetime.combine(query_params.to_date + timedelta(days=1), datetime.min.time())
+        if date_query :
+            mongo_query["processed_at"] = date_query
+    print(f"[{datetime.now()}] /query-extracted-emails: MongoDB query built: {mongo_query}")
+
+    try:
+        # Use await asyncio.to_thread for blocking MongoDB operations
+        cursor = extracted_emails_collection.find(mongo_query).sort("processed_at", -1).limit(query_params.limit)
+        extracted_docs_raw = await asyncio.to_thread(list, cursor)
+        print(f"[{datetime.now()}] /query-extracted-emails: Found {len(extracted_docs_raw)} documents.")
+
+        results = []
+        for doc_raw in extracted_docs_raw:
+            # Convert datetime objects back to date for Pydantic model validation if necessary
+            if 'appointments' in doc_raw:
+                for appt in doc_raw['appointments']:
+                    if isinstance(appt.get('start_date'), datetime): appt['start_date'] = appt['start_date'].date()
+                    if isinstance(appt.get('end_date'), datetime): appt['end_date'] = appt['end_date'].date()
+            if 'tasks' in doc_raw:
+                for task_item in doc_raw['tasks']:
+                    if isinstance(task_item.get('due_date'), datetime): task_item['due_date'] = task_item['due_date'].date()
+            results.append(ExtractedData(**doc_raw))
+        print(f"[{datetime.now()}] /query-extracted-emails: Returning {len(results)} results.")
+        return results
+    except Exception as e:
+        print(f"[{datetime.now()}] /query-extracted-emails: Unhandled Exception during query: {e}")
+        traceback.print_exc()
+        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Error querying extracted emails: {e}")
+
+
+@app.get("/query-generated-replies", response_model=List[GeneratedReplyData], summary="Query generated replies from MongoDB")
+async def query_generated_replies_endpoint(query_params: GeneratedReplyQuery = Depends()):
+    print(f"[{datetime.now()}] /query-generated-replies: Received request with params: {query_params.model_dump_json()}")
+    if generated_replies_collection is None:
+        print(f"[{datetime.now()}] /query-generated-replies: MongoDB collection is None.")
+        raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="MongoDB not available for querying generated replies.")
+    mongo_query: Dict[str, Any] = {}
+    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
+
+    if query_params.from_date or query_params.to_date:
+        date_query: Dict[str, datetime] = {}
+        if query_params.from_date:
+            date_query["$gte"] = datetime.combine(query_params.from_date, datetime.min.time())
+        if query_params.to_date:
+            date_query["$lt"] = datetime.combine(query_params.to_date + timedelta(days=1), datetime.min.time())
+        if date_query:
+            mongo_query["generated_at"] = date_query
+    print(f"[{datetime.now()}] /query-generated-replies: MongoDB query built: {mongo_query}")
+
+    try:
+        # Use await asyncio.to_thread for blocking MongoDB operations
+        cursor = generated_replies_collection.find(mongo_query).sort("generated_at", -1).limit(query_params.limit)
+        generated_docs_raw = await asyncio.to_thread(list, cursor)
+        print(f"[{datetime.now()}] /query-generated-replies: Found {len(generated_docs_raw)} documents.")
+        results = []
+        for doc_raw in generated_docs_raw:
+            results.append(GeneratedReplyData(**doc_raw))
+        print(f"[{datetime.now()}] /query-generated-replies: Returning {len(results)} results.")
+        return results
+    except Exception as e:
+        print(f"[{datetime.now()}] /query-generated-replies: Unhandled Exception during query: {e}")
+        traceback.print_exc()
+        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Error querying generated replies: {e}")