Update flask_Character.py

flask_Character.py

@@ -354,31 +354,41 @@ def _generate_response_internal(
     """
     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."
-    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.
+    
+    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}
+    Create a response to the following email with the following parameters:
+    - Language: {language}
+    - Length: {length}
+    - Style: {style}
+    - Tone: {tone}
+    - Emoji usage: {emoji}
 
-Email:
-{email}
+    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
-    output = chain.invoke({"email": email_text, "language": language, "length": length, "style": style, "tone": tone, "emoji": emoji})
-    return output.content.strip()
+    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
@@ -393,11 +403,13 @@ 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:
-            # If DB is not available, set a specific exception on the future
+            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
@@ -410,10 +422,12 @@ async def handle_single_reply_request(request_data: GenerateReplyRequest, future
             "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"]),
@@ -421,8 +435,10 @@ async def handle_single_reply_request(request_data: GenerateReplyRequest, future
             }
             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,
@@ -433,6 +449,7 @@ async def handle_single_reply_request(request_data: GenerateReplyRequest, future
             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,
@@ -443,11 +460,13 @@ async def handle_single_reply_request(request_data: GenerateReplyRequest, future
             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,
@@ -456,25 +475,31 @@ async def handle_single_reply_request(request_data: GenerateReplyRequest, future
         }
         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:
-        traceback.print_exc()
+        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()
@@ -483,8 +508,11 @@ async def process_reply_batches():
                 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):
@@ -494,18 +522,22 @@ async def process_reply_batches():
                     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]
-            # Use asyncio.gather to run all tasks in the batch concurrently
+            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)
@@ -525,13 +557,14 @@ app = FastAPI(
 @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"Unhandled exception caught by global handler for request: {request.url}")
+    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(
@@ -545,7 +578,7 @@ async def global_exception_handler_wrapper(request, exc):
 @app.on_event("startup")
 async def startup_event():
     global client, db, extracted_emails_collection, generated_replies_collection, batch_processor_task
-    print("FastAPI app startup sequence initiated.")
+    print(f"[{datetime.now()}] FastAPI app startup sequence initiated.")
     try:
         # Connect to MongoDB
         client = MongoClient(MONGO_URI, serverSelectionTimeoutMS=5000)
@@ -553,23 +586,23 @@ async def startup_event():
         db = client[DB_NAME]
         extracted_emails_collection = db[EXTRACTED_EMAILS_COLLECTION]
         generated_replies_collection = db[GENERATED_REPLIES_COLLECTION]
-        print(f"Successfully connected to MongoDB: {DB_NAME}")
+        print(f"[{datetime.now()}] Successfully connected to MongoDB: {DB_NAME}")
 
         # 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("Batch processor task for replies started.")
+            print(f"[{datetime.now()}] Batch processor task for replies started.")
         else:
-            print("Batch processor task for replies is already running or being initialized.")
+            print(f"[{datetime.now()}] Batch processor task for replies is already running or being initialized.")
 
     except (ConnectionFailure, OperationFailure) as e:
-        print(f"ERROR: MongoDB Connection/Operation Failure: {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"ERROR: An unexpected error occurred during MongoDB connection or batch startup: {e}")
+        print(f"[{datetime.now()}] ERROR: An unexpected error occurred during MongoDB connection or batch startup: {e}")
         traceback.print_exc()
         client = None
         db = None
@@ -580,33 +613,34 @@ async def startup_event():
             try:
                 client.admin.command('ping')
             except Exception as e:
-                print(f"MongoDB ping failed after initial connection attempt during finally block: {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("MongoDB client or db object is None after connection attempt in startup. Database likely not connected.")
+            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("FastAPI app startup sequence completed for MongoDB client & Batch Processor initialization.")
+        print(f"[{datetime.now()}] FastAPI app startup sequence completed for MongoDB client & Batch Processor initialization.")
 
 
 @app.on_event("shutdown")
 async def shutdown_event():
     global client, batch_processor_task
-    print("FastAPI app shutting down.")
+    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("Batch processor task for replies cancelled during shutdown.")
+            print(f"[{datetime.now()}] Batch processor task for replies cancelled during shutdown.")
         except Exception as e:
-            print(f"Error during batch processor task shutdown: {e}")
+            print(f"[{datetime.now()}] Error during batch processor task shutdown: {e}")
             traceback.print_exc()
         batch_processor_task = None
 
     if client:
         client.close()
-        print("MongoDB client closed.")
+        print(f"[{datetime.now()}] MongoDB client closed.")
 
 
 # --- API Endpoints ---
@@ -619,6 +653,7 @@ async def health_check():
     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
@@ -653,15 +688,22 @@ 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:
         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):
@@ -673,13 +715,17 @@ async def extract_email_data(request: ProcessEmailRequest):
                 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}")
 
@@ -699,7 +745,9 @@ 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()
@@ -708,25 +756,34 @@ async def generate_email_reply(request: GenerateReplyRequest):
     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:
-        client_timeout = BATCH_TIMEOUT + 10.0 # e.g., 0.5s batch + 10s LLM response buffer
+        # 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()
-        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.")
+            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:
         if isinstance(e, HTTPException):
-            raise e
+            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)}")
+        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.")
 
 
 @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:
@@ -745,13 +802,17 @@ async def query_extracted_emails_endpoint(query_params: ExtractedEmailQuery = De
             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()
@@ -760,15 +821,19 @@ async def query_extracted_emails_endpoint(query_params: ExtractedEmailQuery = De
                 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
@@ -783,14 +848,19 @@ async def query_generated_replies_endpoint(query_params: GeneratedReplyQuery = D
             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}")