import torch import numpy as np from transformers import AutoTokenizer, AutoModelForSequenceClassification import torch.nn.functional as F import spacy from typing import List, Dict, Tuple import logging import os import gradio as gr from fastapi.middleware.cors import CORSMiddleware from concurrent.futures import ThreadPoolExecutor from functools import partial import time import pandas as pd from datetime import datetime import threading import random from openpyxl import load_workbook # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Constants MAX_LENGTH = 512 MODEL_NAME = "microsoft/deberta-v3-small" WINDOW_SIZE = 6 WINDOW_OVERLAP = 2 CONFIDENCE_THRESHOLD = 0.65 BATCH_SIZE = 8 # Reduced batch size for CPU MAX_WORKERS = 4 # Number of worker threads for processing class ExcelLogger: def __init__(self, log_dir="logs", excel_file=None): """Initialize the Excel logger. Args: log_dir: Directory to store log files excel_file: Specific Excel file name (defaults to predictions_YYYY-MM.xlsx) """ self.log_dir = log_dir os.makedirs(log_dir, exist_ok=True) # Use monthly Excel files by default if excel_file is None: current_month = datetime.now().strftime('%Y-%m') excel_file = f"predictions_{current_month}.xlsx" self.excel_path = os.path.join(log_dir, excel_file) # Create excel file with headers if it doesn't exist if not os.path.exists(self.excel_path): self._create_excel_file() # Create a lock for thread safety self.file_lock = threading.Lock() def _create_excel_file(self): """Create a new Excel file with appropriate sheets and headers.""" # Create DataFrame for metrics metrics_df = pd.DataFrame(columns=[ 'timestamp', 'word_count', 'mode', 'prediction', 'confidence', 'prediction_time_seconds', 'num_sentences' ]) # Create DataFrame for text storage text_df = pd.DataFrame(columns=[ 'entry_id', 'timestamp', 'text' ]) # Create Excel writer with pd.ExcelWriter(self.excel_path, engine='openpyxl') as writer: metrics_df.to_excel(writer, sheet_name='Metrics', index=False) text_df.to_excel(writer, sheet_name='TextData', index=False) logger.info(f"Created new Excel log file: {self.excel_path}") def log_prediction(self, prediction_data, store_text=True): """Log prediction data to the Excel file. Args: prediction_data: Dictionary containing prediction metrics store_text: Whether to store the full text """ # Generate a unique entry ID entry_id = f"{datetime.now().strftime('%Y%m%d%H%M%S')}_{random.randint(1000, 9999)}" # Extract text if present text = prediction_data.pop('text', None) if store_text else None # Ensure timestamp is present if 'timestamp' not in prediction_data: prediction_data['timestamp'] = datetime.now().isoformat() # Add entry_id to the metrics metrics_data = prediction_data.copy() metrics_data['entry_id'] = entry_id # Start a thread to write data to Excel thread = threading.Thread( target=self._write_to_excel, args=(metrics_data, text, entry_id, store_text) ) thread.daemon = True thread.start() def _write_to_excel(self, metrics_data, text, entry_id, store_text): """Write data to Excel file with retry mechanism for concurrent access.""" max_retries = 5 retry_delay = 0.5 for attempt in range(max_retries): try: with self.file_lock: # Load existing data metrics_df = pd.read_excel(self.excel_path, sheet_name='Metrics') # Append new metrics data new_metrics = pd.DataFrame([metrics_data]) metrics_df = pd.concat([metrics_df, new_metrics], ignore_index=True) # If text storage is requested if store_text and text: try: text_df = pd.read_excel(self.excel_path, sheet_name='TextData') # Append new text data new_text = pd.DataFrame([{ 'entry_id': entry_id, 'timestamp': metrics_data['timestamp'], 'text': text }]) text_df = pd.concat([text_df, new_text], ignore_index=True) except: # If TextData sheet doesn't exist or can't be read text_df = pd.DataFrame([{ 'entry_id': entry_id, 'timestamp': metrics_data['timestamp'], 'text': text }]) # Write back to Excel with pd.ExcelWriter(self.excel_path, engine='openpyxl', mode='a', if_sheet_exists='replace') as writer: metrics_df.to_excel(writer, sheet_name='Metrics', index=False) if store_text and text: text_df.to_excel(writer, sheet_name='TextData', index=False) # Successfully wrote to file break except Exception as e: # If error occurs (likely due to concurrent access), retry after delay logger.warning(f"Error writing to Excel (attempt {attempt+1}/{max_retries}): {e}") time.sleep(retry_delay * (attempt + 1)) # Progressive backoff else: # If all retries fail, log to backup file logger.error(f"Failed to write to Excel after {max_retries} attempts, logging to backup file") self._write_to_backup(metrics_data, text, entry_id, store_text) def _write_to_backup(self, metrics_data, text, entry_id, store_text): """Write to backup CSV files if Excel writing fails.""" timestamp = datetime.now().strftime('%Y%m%d') # Log metrics to CSV metrics_csv = os.path.join(self.log_dir, f"metrics_backup_{timestamp}.csv") pd.DataFrame([metrics_data]).to_csv(metrics_csv, mode='a', header=not os.path.exists(metrics_csv), index=False) # Log text to separate CSV if needed if store_text and text: text_csv = os.path.join(self.log_dir, f"text_backup_{timestamp}.csv") text_data = { 'entry_id': entry_id, 'timestamp': metrics_data['timestamp'], 'text': text } pd.DataFrame([text_data]).to_csv(text_csv, mode='a', header=not os.path.exists(text_csv), index=False) class TextWindowProcessor: def __init__(self): try: self.nlp = spacy.load("en_core_web_sm") except OSError: logger.info("Downloading spacy model...") spacy.cli.download("en_core_web_sm") self.nlp = spacy.load("en_core_web_sm") if 'sentencizer' not in self.nlp.pipe_names: self.nlp.add_pipe('sentencizer') disabled_pipes = [pipe for pipe in self.nlp.pipe_names if pipe != 'sentencizer'] self.nlp.disable_pipes(*disabled_pipes) # Initialize thread pool for parallel processing self.executor = ThreadPoolExecutor(max_workers=MAX_WORKERS) def split_into_sentences(self, text: str) -> List[str]: doc = self.nlp(text) return [str(sent).strip() for sent in doc.sents] def create_windows(self, sentences: List[str], window_size: int, overlap: int) -> List[str]: if len(sentences) < window_size: return [" ".join(sentences)] windows = [] stride = window_size - overlap for i in range(0, len(sentences) - window_size + 1, stride): window = sentences[i:i + window_size] windows.append(" ".join(window)) return windows def create_centered_windows(self, sentences: List[str], window_size: int) -> Tuple[List[str], List[List[int]]]: """Create windows with better boundary handling""" windows = [] window_sentence_indices = [] for i in range(len(sentences)): # Calculate window boundaries centered on current sentence half_window = window_size // 2 start_idx = max(0, i - half_window) end_idx = min(len(sentences), i + half_window + 1) # Create the window window = sentences[start_idx:end_idx] windows.append(" ".join(window)) window_sentence_indices.append(list(range(start_idx, end_idx))) return windows, window_sentence_indices class TextClassifier: def __init__(self): # Set thread configuration before any model loading or parallel work if not torch.cuda.is_available(): torch.set_num_threads(MAX_WORKERS) torch.set_num_interop_threads(MAX_WORKERS) self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu") self.model_name = MODEL_NAME self.tokenizer = None self.model = None self.processor = TextWindowProcessor() self.initialize_model() def initialize_model(self): """Initialize the model and tokenizer.""" logger.info("Initializing model and tokenizer...") from transformers import DebertaV2TokenizerFast self.tokenizer = DebertaV2TokenizerFast.from_pretrained( self.model_name, model_max_length=MAX_LENGTH, use_fast=True ) self.model = AutoModelForSequenceClassification.from_pretrained( self.model_name, num_labels=2 ).to(self.device) model_path = "model_20250209_184929_acc1.0000.pt" if os.path.exists(model_path): logger.info(f"Loading custom model from {model_path}") checkpoint = torch.load(model_path, map_location=self.device) self.model.load_state_dict(checkpoint['model_state_dict']) else: logger.warning("Custom model file not found. Using base model.") self.model.eval() def quick_scan(self, text: str) -> Dict: """Perform a quick scan using simple window analysis.""" if not text.strip(): return { 'prediction': 'unknown', 'confidence': 0.0, 'num_windows': 0 } sentences = self.processor.split_into_sentences(text) windows = self.processor.create_windows(sentences, WINDOW_SIZE, WINDOW_OVERLAP) predictions = [] # Process windows in smaller batches for CPU efficiency for i in range(0, len(windows), BATCH_SIZE): batch_windows = windows[i:i + BATCH_SIZE] inputs = self.tokenizer( batch_windows, truncation=True, padding=True, max_length=MAX_LENGTH, return_tensors="pt" ).to(self.device) with torch.no_grad(): outputs = self.model(**inputs) probs = F.softmax(outputs.logits, dim=-1) for idx, window in enumerate(batch_windows): prediction = { 'window': window, 'human_prob': probs[idx][1].item(), 'ai_prob': probs[idx][0].item(), 'prediction': 'human' if probs[idx][1] > probs[idx][0] else 'ai' } predictions.append(prediction) # Clean up GPU memory if available del inputs, outputs, probs if torch.cuda.is_available(): torch.cuda.empty_cache() if not predictions: return { 'prediction': 'unknown', 'confidence': 0.0, 'num_windows': 0 } avg_human_prob = sum(p['human_prob'] for p in predictions) / len(predictions) avg_ai_prob = sum(p['ai_prob'] for p in predictions) / len(predictions) return { 'prediction': 'human' if avg_human_prob > avg_ai_prob else 'ai', 'confidence': max(avg_human_prob, avg_ai_prob), 'num_windows': len(predictions) } def detailed_scan(self, text: str) -> Dict: """Perform a detailed scan with improved sentence-level analysis.""" # Clean up trailing whitespace text = text.rstrip() if not text.strip(): return { 'sentence_predictions': [], 'highlighted_text': '', 'full_text': '', 'overall_prediction': { 'prediction': 'unknown', 'confidence': 0.0, 'num_sentences': 0 } } sentences = self.processor.split_into_sentences(text) if not sentences: return {} # Create centered windows for each sentence windows, window_sentence_indices = self.processor.create_centered_windows(sentences, WINDOW_SIZE) # Track scores for each sentence sentence_appearances = {i: 0 for i in range(len(sentences))} sentence_scores = {i: {'human_prob': 0.0, 'ai_prob': 0.0} for i in range(len(sentences))} # Process windows in batches for i in range(0, len(windows), BATCH_SIZE): batch_windows = windows[i:i + BATCH_SIZE] batch_indices = window_sentence_indices[i:i + BATCH_SIZE] inputs = self.tokenizer( batch_windows, truncation=True, padding=True, max_length=MAX_LENGTH, return_tensors="pt" ).to(self.device) with torch.no_grad(): outputs = self.model(**inputs) probs = F.softmax(outputs.logits, dim=-1) # Attribute predictions with weighted scoring for window_idx, indices in enumerate(batch_indices): center_idx = len(indices) // 2 center_weight = 0.7 # Higher weight for center sentence edge_weight = 0.3 / (len(indices) - 1) # Distribute remaining weight for pos, sent_idx in enumerate(indices): # Apply higher weight to center sentence weight = center_weight if pos == center_idx else edge_weight sentence_appearances[sent_idx] += weight sentence_scores[sent_idx]['human_prob'] += weight * probs[window_idx][1].item() sentence_scores[sent_idx]['ai_prob'] += weight * probs[window_idx][0].item() # Clean up memory del inputs, outputs, probs if torch.cuda.is_available(): torch.cuda.empty_cache() # Calculate final predictions with boundary smoothing sentence_predictions = [] for i in range(len(sentences)): if sentence_appearances[i] > 0: human_prob = sentence_scores[i]['human_prob'] / sentence_appearances[i] ai_prob = sentence_scores[i]['ai_prob'] / sentence_appearances[i] # Apply minimal smoothing at prediction boundaries if i > 0 and i < len(sentences) - 1: prev_human = sentence_scores[i-1]['human_prob'] / sentence_appearances[i-1] prev_ai = sentence_scores[i-1]['ai_prob'] / sentence_appearances[i-1] next_human = sentence_scores[i+1]['human_prob'] / sentence_appearances[i+1] next_ai = sentence_scores[i+1]['ai_prob'] / sentence_appearances[i+1] # Check if we're at a prediction boundary current_pred = 'human' if human_prob > ai_prob else 'ai' prev_pred = 'human' if prev_human > prev_ai else 'ai' next_pred = 'human' if next_human > next_ai else 'ai' if current_pred != prev_pred or current_pred != next_pred: # Small adjustment at boundaries smooth_factor = 0.1 human_prob = (human_prob * (1 - smooth_factor) + (prev_human + next_human) * smooth_factor / 2) ai_prob = (ai_prob * (1 - smooth_factor) + (prev_ai + next_ai) * smooth_factor / 2) sentence_predictions.append({ 'sentence': sentences[i], 'human_prob': human_prob, 'ai_prob': ai_prob, 'prediction': 'human' if human_prob > ai_prob else 'ai', 'confidence': max(human_prob, ai_prob) }) return { 'sentence_predictions': sentence_predictions, 'highlighted_text': self.format_predictions_html(sentence_predictions), 'full_text': text, 'overall_prediction': self.aggregate_predictions(sentence_predictions) } def format_predictions_html(self, sentence_predictions: List[Dict]) -> str: """Format predictions as HTML with color-coding.""" html_parts = [] for pred in sentence_predictions: sentence = pred['sentence'] confidence = pred['confidence'] if confidence >= CONFIDENCE_THRESHOLD: if pred['prediction'] == 'human': color = "#90EE90" # Light green else: color = "#FFB6C6" # Light red else: if pred['prediction'] == 'human': color = "#E8F5E9" # Very light green else: color = "#FFEBEE" # Very light red html_parts.append(f'{sentence}') return " ".join(html_parts) def aggregate_predictions(self, predictions: List[Dict]) -> Dict: """Aggregate predictions from multiple sentences into a single prediction.""" if not predictions: return { 'prediction': 'unknown', 'confidence': 0.0, 'num_sentences': 0 } total_human_prob = sum(p['human_prob'] for p in predictions) total_ai_prob = sum(p['ai_prob'] for p in predictions) num_sentences = len(predictions) avg_human_prob = total_human_prob / num_sentences avg_ai_prob = total_ai_prob / num_sentences return { 'prediction': 'human' if avg_human_prob > avg_ai_prob else 'ai', 'confidence': max(avg_human_prob, avg_ai_prob), 'num_sentences': num_sentences } # Initialize the logger excel_logger = ExcelLogger(log_dir="prediction_logs") def analyze_text(text: str, mode: str, classifier: TextClassifier) -> tuple: """Analyze text using specified mode and return formatted results.""" # Start timing the prediction 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) prediction = result['prediction'] confidence = result['confidence'] num_windows = result['num_windows'] quick_analysis = f""" PREDICTION: {prediction.upper()} Confidence: {confidence*100:.1f}% Windows analyzed: {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." output = ( text, # No highlighting in quick mode "Quick scan mode - no sentence-level analysis available", quick_analysis ) # End timing end_time = time.time() prediction_time = end_time - start_time # Log the data log_data = { "timestamp": datetime.now().isoformat(), "word_count": word_count, "mode": mode, "prediction": prediction, "confidence": confidence, "prediction_time_seconds": prediction_time, "num_sentences": 0, # No sentence analysis in quick mode "text": text } excel_logger.log_prediction(log_data) else: analysis = classifier.detailed_scan(text) prediction = analysis['overall_prediction']['prediction'] confidence = analysis['overall_prediction']['confidence'] num_sentences = analysis['overall_prediction']['num_sentences'] detailed_analysis = [] for pred in analysis['sentence_predictions']: pred_confidence = pred['confidence'] * 100 detailed_analysis.append(f"Sentence: {pred['sentence']}") detailed_analysis.append(f"Prediction: {pred['prediction'].upper()}") detailed_analysis.append(f"Confidence: {pred_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']} """ output = ( analysis['highlighted_text'], "\n".join(detailed_analysis), overall_result ) # End timing end_time = time.time() prediction_time = end_time - start_time # Log the data log_data = { "timestamp": datetime.now().isoformat(), "word_count": word_count, "mode": mode, "prediction": prediction, "confidence": confidence, "prediction_time_seconds": prediction_time, "num_sentences": num_sentences, "text": text } excel_logger.log_prediction(log_data) return output # Initialize the classifier globally classifier = TextClassifier() # Create Gradio interface with added information about data collection demo = gr.Interface( fn=lambda text, mode: analyze_text(text, mode, classifier), inputs=[ gr.Textbox( lines=8, placeholder="Enter text to analyze...", label="Input Text" ), gr.Radio( choices=["quick", "detailed"], value="quick", label="Analysis Mode", info="Quick mode for faster analysis, Detailed mode for sentence-level analysis" ) ], outputs=[ gr.HTML(label="Highlighted Analysis"), gr.Textbox(label="Sentence-by-Sentence Analysis", lines=10), gr.Textbox(label="Overall Result", lines=4) ], title="AI Text Detector", description="Analyze text to detect if it was written by a human or AI. Choose between quick scan and detailed sentence-level analysis. 200+ words suggested for accurate predictions. Note: For testing purposes, text and analysis data will be recorded.", api_name="predict", flagging_mode="never" ) app = demo.app app.add_middleware( CORSMiddleware, allow_origins=["*"], # For development allow_credentials=True, allow_methods=["GET", "POST", "OPTIONS"], # Explicitly list methods allow_headers=["*"], ) # Ensure CORS is applied before launching if __name__ == "__main__": demo.queue() demo.launch( server_name="0.0.0.0", server_port=7860, share=True )