Update app.py

app.py

@@ -10,7 +10,10 @@ import gradio as gr
 from fastapi.middleware.cors import CORSMiddleware
 from concurrent.futures import ThreadPoolExecutor
 from functools import partial
-
+import time
+import csv
+import os
+from datetime import datetime
 # Configure logging
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
@@ -24,6 +27,37 @@ CONFIDENCE_THRESHOLD = 0.65
 BATCH_SIZE = 8  # Reduced batch size for CPU
 MAX_WORKERS = 4  # Number of worker threads for processing
 
+
+def log_prediction_data(input_text, word_count, prediction, confidence, execution_time, mode):
+    """Log prediction data to a CSV file in the /tmp directory."""
+    # Define the CSV file path
+    csv_path = "/tmp/prediction_logs.csv"
+   
+    # Check if file exists to determine if we need to write headers
+    file_exists = os.path.isfile(csv_path)
+   
+    try:
+        with open(csv_path, 'a', newline='', encoding='utf-8') as f:
+            writer = csv.writer(f)
+           
+            # Write headers if the file is newly created
+            if not file_exists:
+                writer.writerow(["timestamp", "word_count", "prediction", "confidence", "execution_time_ms", "analysis_mode", "full_text"])
+           
+            # Clean up the input text for CSV storage (replace newlines with spaces)
+            cleaned_text = input_text.replace("\n", " ")
+           
+            # Write the data row with the full text
+            timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+            writer.writerow([timestamp, word_count, prediction, f"{confidence:.2f}", f"{execution_time:.2f}", mode, cleaned_text])
+           
+        logger.info(f"Successfully logged prediction data to {csv_path}")
+        return True
+    except Exception as e:
+        logger.error(f"Error logging prediction data: {str(e)}")
+        return False
+
+
 class TextWindowProcessor:
     def __init__(self):
         try:
@@ -420,57 +454,88 @@ class TextClassifier:
             'num_sentences': num_sentences
         }
 
-def analyze_text(text: str, mode: str, classifier: TextClassifier) -> tuple:
-    """Analyze text using specified mode and return formatted results."""
-    # Count words in the text
-    word_count = len(text.split())
-    
-    # If text is less than 200 words and detailed mode is selected, switch to quick mode
-    original_mode = mode
-    if word_count < 200 and mode == "detailed":
-        mode = "quick"
-    
-    if mode == "quick":
-        result = classifier.quick_scan(text)
-        
-        quick_analysis = f"""
-        PREDICTION: {result['prediction'].upper()}
-        Confidence: {result['confidence']*100:.1f}%
-        Windows analyzed: {result['num_windows']}
-        """
-        
-        # Add note if mode was switched
-        if original_mode == "detailed":
-            quick_analysis += f"\n\nNote: Switched to quick mode because text contains only {word_count} words. Minimum 200 words required for detailed analysis."
-        
-        return (
-            text,  # No highlighting in quick mode
-            "Quick scan mode - no sentence-level analysis available",
-            quick_analysis
-        )
-    else:
-        analysis = classifier.detailed_scan(text)
-        
-        detailed_analysis = []
-        for pred in analysis['sentence_predictions']:
-            confidence = pred['confidence'] * 100
-            detailed_analysis.append(f"Sentence: {pred['sentence']}")
-            detailed_analysis.append(f"Prediction: {pred['prediction'].upper()}")
-            detailed_analysis.append(f"Confidence: {confidence:.1f}%")
-            detailed_analysis.append("-" * 50)
-        
-        final_pred = analysis['overall_prediction']
-        overall_result = f"""
-        FINAL PREDICTION: {final_pred['prediction'].upper()}
-        Overall confidence: {final_pred['confidence']*100:.1f}%
-        Number of sentences analyzed: {final_pred['num_sentences']}
-        """
-        
-        return (
-            analysis['highlighted_text'],
-            "\n".join(detailed_analysis),
-            overall_result
-        )
+
+    def analyze_text(text: str, mode: str, classifier: TextClassifier) -> tuple:
+        """Analyze text using specified mode and return formatted results."""
+        # Start timing
+        start_time = time.time()
+       
+        # Count words in the text
+        word_count = len(text.split())
+       
+        # If text is less than 200 words and detailed mode is selected, switch to quick mode
+        original_mode = mode
+        if word_count < 200 and mode == "detailed":
+            mode = "quick"
+       
+        if mode == "quick":
+            result = classifier.quick_scan(text)
+           
+            quick_analysis = f"""
+            PREDICTION: {result['prediction'].upper()}
+            Confidence: {result['confidence']*100:.1f}%
+            Windows analyzed: {result['num_windows']}
+            """
+           
+            # Add note if mode was switched
+            if original_mode == "detailed":
+                quick_analysis += f"\n\nNote: Switched to quick mode because text contains only {word_count} words. Minimum 200 words required for detailed analysis."
+           
+            # Calculate execution time in milliseconds
+            execution_time = (time.time() - start_time) * 1000
+           
+            # Log the prediction data
+            log_prediction_data(
+                input_text=text,
+                word_count=word_count,
+                prediction=result['prediction'],
+                confidence=result['confidence'],
+                execution_time=execution_time,
+                mode=original_mode
+            )
+           
+            return (
+                text,  # No highlighting in quick mode
+                "Quick scan mode - no sentence-level analysis available",
+                quick_analysis
+            )
+        else:
+            analysis = classifier.detailed_scan(text)
+           
+            detailed_analysis = []
+            for pred in analysis['sentence_predictions']:
+                confidence = pred['confidence'] * 100
+                detailed_analysis.append(f"Sentence: {pred['sentence']}")
+                detailed_analysis.append(f"Prediction: {pred['prediction'].upper()}")
+                detailed_analysis.append(f"Confidence: {confidence:.1f}%")
+                detailed_analysis.append("-" * 50)
+           
+            final_pred = analysis['overall_prediction']
+            overall_result = f"""
+            FINAL PREDICTION: {final_pred['prediction'].upper()}
+            Overall confidence: {final_pred['confidence']*100:.1f}%
+            Number of sentences analyzed: {final_pred['num_sentences']}
+            """
+           
+            # Calculate execution time in milliseconds
+            execution_time = (time.time() - start_time) * 1000
+           
+            # Log the prediction data
+            log_prediction_data(
+                input_text=text,
+                word_count=word_count,
+                prediction=final_pred['prediction'],
+                confidence=final_pred['confidence'],
+                execution_time=execution_time,
+                mode=original_mode
+            )
+           
+            return (
+                analysis['highlighted_text'],
+                "\n".join(detailed_analysis),
+                overall_result
+            )        
+
 
 # Initialize the classifier globally
 classifier = TextClassifier()