Update database_enhanced.py

database_enhanced.py

@@ -1,692 +1,445 @@
 """
-LangGraph Nodes - FINAL WORKING VERSION
-Uses chat_completion() API format + Lazy loading + Fixed alt sentiment
+Enhanced Database Schema for Multi-Stage Review Analysis
+Adds Stage 1-4 columns to existing reviews table
 """
 
-import os
-import json
-import time
-from typing import Dict, Any
+import sqlite3
 from datetime import datetime
-from concurrent.futures import ThreadPoolExecutor, as_completed
-from huggingface_hub import InferenceClient
-import torch
-from transformers import AutoTokenizer, AutoModelForSequenceClassification
-import warnings
-warnings.filterwarnings('ignore')
-
-from langgraph_state import ReviewState, BatchState
-from database_enhanced import EnhancedDatabase
-
-# FIXED: Don't initialize client at module import
-_hf_client = None
-
-def get_hf_client():
-    """Get or initialize HuggingFace client (lazy loading)"""
-    global _hf_client
-    
-    if _hf_client is not None:
-        return _hf_client
-    
-    # Try to get token from environment
-    HF_TOKEN = os.getenv("HUGGINGFACE_API_KEY")
-    
-    if not HF_TOKEN or HF_TOKEN.strip() == "":
-        return None
-    
-    # Initialize client with token
-    print(f"✅ Initializing HF client with token: {HF_TOKEN[:10]}...")
-    _hf_client = InferenceClient(token=HF_TOKEN)
-    return _hf_client
-
-
-# Initialize sentiment models (singleton)
-_sentiment_models_loaded = False
-_best_tokenizer = None
-_best_model = None
-_alt_tokenizer = None
-_alt_model = None
-
-def load_sentiment_models():
-    """Load sentiment models once (singleton pattern)"""
-    global _sentiment_models_loaded, _best_tokenizer, _best_model, _alt_tokenizer, _alt_model
-    
-    if _sentiment_models_loaded:
-        return
-    
-    print("   📦 Loading Twitter-BERT models (one-time)...")
-    
-    # Best Model
-    _best_tokenizer = AutoTokenizer.from_pretrained("cardiffnlp/twitter-roberta-base-sentiment-latest")
-    _best_model = AutoModelForSequenceClassification.from_pretrained("cardiffnlp/twitter-roberta-base-sentiment-latest")
-    _best_model.eval()
-    
-    # Alternate Model - FIXED: Load with low_cpu_mem_usage to avoid meta tensors
-    _alt_tokenizer = AutoTokenizer.from_pretrained("finiteautomata/bertweet-base-sentiment-analysis")
-    _alt_model = AutoModelForSequenceClassification.from_pretrained(
-        "finiteautomata/bertweet-base-sentiment-analysis",
-        low_cpu_mem_usage=False  # FIXED: Don't use meta device
-    )
-    _alt_model.eval()
-    
-    _sentiment_models_loaded = True
-    print("   ✅ Sentiment models loaded!")
-
-
-# ============================================================================
-# STAGE 1: CLASSIFICATION NODE
-# ============================================================================
-
-def llm1_classify(review: Dict[str, Any]) -> Dict[str, Any]:
-    """LLM1: Type, Department, Priority classification"""
-    
-    hf_client = get_hf_client()
-    
-    if hf_client is None:
-        return {
-            'type': 'unknown',
-            'department': 'unknown',
-            'priority': 'medium',
-            'confidence': 0.0,
-            'reasoning': 'HuggingFace API key not set',
-            'model': 'Qwen/Qwen2.5-72B-Instruct'
-        }
-    
-    review_text = review.get('review_text', '')
-    rating = review.get('rating', 3)
-    
-    # FIXED: Use chat format with system + user messages
-    system_prompt = """You are an expert at classifying customer reviews for theme park and attraction apps.
-
-Classify reviews across these dimensions:
-
-1. TYPE: complaint, praise, suggestion, question, or bug_report
-2. DEPARTMENT: engineering, ux, support, or business
-3. PRIORITY: critical, high, medium, or low
-4. CONFIDENCE: 0.0-1.0
-5. REASONING: Brief one-sentence explanation
-
-Respond ONLY in valid JSON format:
-{
-  "type": "complaint/praise/suggestion/question/bug_report",
-  "department": "engineering/ux/support/business",
-  "priority": "critical/high/medium/low",
-  "confidence": 0.0-1.0,
-  "reasoning": "brief explanation"
-}"""
-
-    user_prompt = f"""REVIEW:
-Rating: {rating}/5
-Text: {review_text}
-
-Classify this review:"""
-
-    try:
-        print(f"   🔍 Calling Qwen API...")
-        
-        # FIXED: Use chat_completion instead of text_generation
-        response = hf_client.chat_completion(
-            messages=[
-                {"role": "system", "content": system_prompt},
-                {"role": "user", "content": user_prompt}
-            ],
-            model="Qwen/Qwen2.5-72B-Instruct",
-            max_tokens=200,
-            temperature=0.1
-        )
-        
-        # Extract content from chat response
-        content = response.choices[0].message.content
-        print(f"   ✅ Got response ({len(content)} chars)")
-        
-        # Clean and parse JSON
-        content_clean = content.strip()
-        if content_clean.startswith('```'):
-            content_clean = content_clean.split('```')[1]
-            if content_clean.startswith('json'):
-                content_clean = content_clean[4:]
-        content_clean = content_clean.strip()
-        
-        result = json.loads(content_clean)
-        result['model'] = 'Qwen/Qwen2.5-72B-Instruct'
-        
-        print(f"   ✅ Parsed: {result['type']} → {result['department']}")
-        return result
-        
-    except Exception as e:
-        print(f"❌ LLM1 ERROR: {type(e).__name__}: {str(e)}")
-        
-        return {
-            'type': 'unknown',
-            'department': 'unknown',
-            'priority': 'medium',
-            'confidence': 0.0,
-            'reasoning': f'API Error: {str(e)}',
-            'model': 'Qwen/Qwen2.5-72B-Instruct'
-        }
-
-
-def llm2_analyze(review: Dict[str, Any]) -> Dict[str, Any]:
-    """LLM2: User type, Emotion, Context analysis"""
-    
-    hf_client = get_hf_client()
-    
-    if hf_client is None:
-        return {
-            'user_type': 'unknown',
-            'emotion': 'unknown',
-            'context': 'unknown',
-            'confidence': 0.0,
-            'reasoning': 'HuggingFace API key not set',
-            'model': 'mistralai/Mistral-7B-Instruct-v0.3'
-        }
-    
-    review_text = review.get('review_text', '')
-    rating = review.get('rating', 3)
-    
-    # FIXED: Use chat format
-    system_prompt = """You are an expert at understanding customer psychology and emotional context.
-
-Analyze reviews for:
-1. USER_TYPE: new_user, regular_user, power_user, or churning_user
-2. EMOTION: anger, frustration, joy, satisfaction, disappointment, or confusion
-3. CONTEXT: Brief context (1-2 words)
-4. CONFIDENCE: 0.0-1.0
-5. REASONING: Brief explanation
-
-Respond ONLY in valid JSON format:
-{
-  "user_type": "new_user/regular_user/power_user/churning_user",
-  "emotion": "anger/frustration/joy/satisfaction/disappointment/confusion",
-  "context": "brief context",
-  "confidence": 0.0-1.0,
-  "reasoning": "brief explanation"
-}"""
-
-    user_prompt = f"""REVIEW:
-Rating: {rating}/5
-Text: {review_text}
-
-Analyze this review:"""
-
-    try:
-        print(f"   🔍 Calling Mistral API...")
-        
-        # FIXED: Use chat_completion
-        response = hf_client.chat_completion(
-            messages=[
-                {"role": "system", "content": system_prompt},
-                {"role": "user", "content": user_prompt}
-            ],
-            model="mistralai/Mistral-7B-Instruct-v0.3",
-            max_tokens=200,
-            temperature=0.1
-        )
-        
-        content = response.choices[0].message.content
-        print(f"   ✅ Got response ({len(content)} chars)")
-        
-        # Clean and parse JSON
-        content_clean = content.strip()
-        if content_clean.startswith('```'):
-            content_clean = content_clean.split('```')[1]
-            if content_clean.startswith('json'):
-                content_clean = content_clean[4:]
-        content_clean = content_clean.strip()
-        
-        result = json.loads(content_clean)
-        result['model'] = 'mistralai/Mistral-7B-Instruct-v0.3'
-        
-        print(f"   ✅ Parsed: {result['user_type']}, {result['emotion']}")
-        return result
-        
-    except Exception as e:
-        print(f"❌ LLM2 ERROR: {type(e).__name__}: {str(e)}")
-        
-        return {
-            'user_type': 'unknown',
-            'emotion': 'unknown',
-            'context': 'unknown',
-            'confidence': 0.0,
-            'reasoning': f'API Error: {str(e)}',
-            'model': 'mistralai/Mistral-7B-Instruct-v0.3'
-        }
-
-
-def manager_synthesize(llm1_result: Dict, llm2_result: Dict, review: Dict) -> Dict[str, Any]:
-    """Manager: Synthesize LLM1 and LLM2 results"""
-    
-    hf_client = get_hf_client()
-    
-    if hf_client is None:
-        return {
-            'final_type': llm1_result.get('type', 'unknown'),
-            'final_department': llm1_result.get('department', 'unknown'),
-            'final_priority': llm1_result.get('priority', 'medium'),
-            'synthesis_reasoning': 'HuggingFace API key not set',
-            'model': 'meta-llama/Llama-3.3-70B-Instruct'
-        }
-    
-    review_text = review.get('review_text', '')
-    rating = review.get('rating', 3)
-    
-    # FIXED: Use chat format
-    system_prompt = """You are a synthesis manager evaluating two AI analyses.
-
-Your task:
-1. Validate both analyses
-2. Resolve conflicts
-3. Make final classification decision
-4. Provide synthesis reasoning
-
-Respond ONLY in valid JSON format:
-{
-  "final_type": "from llm1 or adjusted",
-  "final_department": "from llm1 or adjusted",
-  "final_priority": "from llm1 or adjusted",
-  "synthesis_reasoning": "brief explanation"
-}"""
-
-    user_prompt = f"""REVIEW:
-Rating: {rating}/5
-Text: {review_text}
-
-LLM1 ANALYSIS (Type/Dept/Priority):
-{json.dumps(llm1_result, indent=2)}
-
-LLM2 ANALYSIS (User/Emotion/Context):
-{json.dumps(llm2_result, indent=2)}
-
-Synthesize these analyses:"""
+from typing import Dict, List, Any, Optional
+import json
 
-    try:
-        print(f"   🔍 Calling Llama Manager API...")
-        
-        # FIXED: Use chat_completion
-        response = hf_client.chat_completion(
-            messages=[
-                {"role": "system", "content": system_prompt},
-                {"role": "user", "content": user_prompt}
-            ],
-            model="meta-llama/Llama-3.3-70B-Instruct",
-            max_tokens=200,
-            temperature=0.1
+class EnhancedDatabase:
+    """
+    Manages enhanced database schema with Stage 1-4 columns
+    """
+    
+    def __init__(self, db_file: str = "review_database.db"):
+        self.db_file = db_file
+        self.conn = None
+        print(f"📁 Database: {db_file}")
+    
+    def connect(self):
+        """Connect to database"""
+        self.conn = sqlite3.connect(self.db_file, check_same_thread=False)
+        self.conn.row_factory = sqlite3.Row
+        print("✅ Connected to database")
+        return self.conn
+    
+    def close(self):
+        """Close database connection"""
+        if self.conn:
+            self.conn.close()
+            print("✅ Database connection closed")
+    
+    def enhance_schema(self):
+        """
+        Add Stage 1-4 columns to existing reviews table
+        Non-destructive: keeps all existing data
+        """
+        print("\n" + "="*60)
+        print("🔧 ENHANCING DATABASE SCHEMA")
+        print("="*60)
+        
+        cursor = self.conn.cursor()
+        
+        # Get existing columns
+        cursor.execute("PRAGMA table_info(reviews)")
+        existing_columns = [row[1] for row in cursor.fetchall()]
+        print(f"📋 Existing columns: {len(existing_columns)}")
+        
+        # Stage 1: Classification columns
+        stage1_columns = [
+            ("stage1_llm1_type", "TEXT"),
+            ("stage1_llm1_department", "TEXT"),
+            ("stage1_llm1_priority", "TEXT"),
+            ("stage1_llm1_confidence", "REAL"),
+            ("stage1_llm1_reasoning", "TEXT"),
+            ("stage1_llm2_user_type", "TEXT"),
+            ("stage1_llm2_emotion", "TEXT"),
+            ("stage1_llm2_context", "TEXT"),
+            ("stage1_llm2_confidence", "REAL"),
+            ("stage1_llm2_reasoning", "TEXT"),
+            ("stage1_manager_classification", "TEXT"),
+            ("stage1_manager_reasoning", "TEXT"),
+            ("stage1_completed_at", "TIMESTAMP"),
+        ]
+        
+        # Stage 2: Sentiment columns
+        stage2_columns = [
+            ("stage2_best_sentiment", "TEXT"),
+            ("stage2_best_confidence", "REAL"),
+            ("stage2_best_prob_positive", "REAL"),
+            ("stage2_best_prob_neutral", "REAL"),
+            ("stage2_best_prob_negative", "REAL"),
+            ("stage2_alt_sentiment", "TEXT"),
+            ("stage2_alt_confidence", "REAL"),
+            ("stage2_alt_prob_positive", "REAL"),
+            ("stage2_alt_prob_neutral", "REAL"),
+            ("stage2_alt_prob_negative", "REAL"),
+            ("stage2_agreement", "BOOLEAN"),
+            ("stage2_layer_sentiment", "TEXT"),
+            ("stage2_completed_at", "TIMESTAMP"),
+        ]
+        
+        # Stage 3: Finalization columns
+        stage3_columns = [
+            ("stage3_final_sentiment", "TEXT"),
+            ("stage3_confidence", "REAL"),
+            ("stage3_reasoning", "TEXT"),
+            ("stage3_validation_notes", "TEXT"),
+            ("stage3_conflicts_found", "TEXT"),
+            ("stage3_action_recommendation", "TEXT"),
+            ("stage3_needs_human_review", "BOOLEAN"),
+            ("stage3_completed_at", "TIMESTAMP"),
+        ]
+        
+        # Processing metadata
+        metadata_columns = [
+            ("processing_status", "TEXT DEFAULT 'pending'"),
+            ("processing_version", "TEXT DEFAULT 'v1.0'"),
+            ("processing_started_at", "TIMESTAMP"),
+            ("processing_completed_at", "TIMESTAMP"),
+        ]
+        
+        all_new_columns = (
+            stage1_columns + 
+            stage2_columns + 
+            stage3_columns + 
+            metadata_columns
         )
         
-        content = response.choices[0].message.content
-        print(f"   ✅ Got response ({len(content)} chars)")
-        
-        content_clean = content.strip()
-        if content_clean.startswith('```'):
-            content_clean = content_clean.split('```')[1]
-            if content_clean.startswith('json'):
-                content_clean = content_clean[4:]
-        content_clean = content_clean.strip()
-        
-        result = json.loads(content_clean)
-        result['model'] = 'meta-llama/Llama-3.3-70B-Instruct'
-        
-        print(f"   ✅ Manager decision: {result['final_type']} → {result['final_department']}")
-        return result
-        
-    except Exception as e:
-        print(f"❌ MANAGER ERROR: {type(e).__name__}: {str(e)}")
+        # Add columns that don't exist
+        added_count = 0
+        for col_name, col_type in all_new_columns:
+            if col_name not in existing_columns:
+                try:
+                    cursor.execute(f"ALTER TABLE reviews ADD COLUMN {col_name} {col_type}")
+                    added_count += 1
+                    print(f"   ✅ Added column: {col_name}")
+                except sqlite3.OperationalError as e:
+                    if "duplicate column" not in str(e).lower():
+                        print(f"   ⚠️  Error adding {col_name}: {e}")
+        
+        self.conn.commit()
+        print(f"\n✅ Schema enhanced: {added_count} new columns added")
+        
+        # Create logs table for LLM decisions
+        self._create_logs_table(cursor)
+        
+        # Create batch insights table
+        self._create_batch_insights_table(cursor)
+        
+        return added_count
+    
+    def _create_logs_table(self, cursor):
+        """Create table for LLM decision logs"""
+        cursor.execute("""
+            CREATE TABLE IF NOT EXISTS llm_decision_logs (
+                log_id INTEGER PRIMARY KEY AUTOINCREMENT,
+                review_id TEXT NOT NULL,
+                stage TEXT NOT NULL,
+                model_name TEXT NOT NULL,
+                input_prompt TEXT,
+                output_response TEXT,
+                confidence REAL,
+                reasoning TEXT,
+                processing_time_seconds REAL,
+                timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                FOREIGN KEY (review_id) REFERENCES reviews(review_id)
+            )
+        """)
+        
+        cursor.execute("""
+            CREATE INDEX IF NOT EXISTS idx_logs_review_id 
+            ON llm_decision_logs(review_id)
+        """)
+        
+        cursor.execute("""
+            CREATE INDEX IF NOT EXISTS idx_logs_stage 
+            ON llm_decision_logs(stage)
+        """)
+        
+        self.conn.commit()
+        print("   ✅ Created llm_decision_logs table")
+    
+    def _create_batch_insights_table(self, cursor):
+        """Create table for batch analytics"""
+        cursor.execute("""
+            CREATE TABLE IF NOT EXISTS batch_insights (
+                batch_id INTEGER PRIMARY KEY AUTOINCREMENT,
+                analysis_date DATE,
+                total_reviews INTEGER,
+                sentiment_positive INTEGER,
+                sentiment_neutral INTEGER,
+                sentiment_negative INTEGER,
+                priority_critical INTEGER,
+                priority_high INTEGER,
+                priority_medium INTEGER,
+                priority_low INTEGER,
+                dept_engineering INTEGER,
+                dept_ux INTEGER,
+                dept_support INTEGER,
+                dept_business INTEGER,
+                critical_issues TEXT,
+                quick_wins TEXT,
+                recommendations TEXT,
+                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+            )
+        """)
+        
+        self.conn.commit()
+        print("   ✅ Created batch_insights table")
+    
+    def get_pending_reviews(self, limit: Optional[int] = None) -> List[Dict]:
+        """Get reviews that haven't been processed yet"""
+        cursor = self.conn.cursor()
+        
+        query = """
+            SELECT * FROM reviews 
+            WHERE processing_status IS NULL OR processing_status = 'pending'
+            ORDER BY scraped_at DESC
+        """
+        
+        if limit:
+            query += f" LIMIT {limit}"
+        
+        cursor.execute(query)
+        rows = cursor.fetchall()
+        
+        return [dict(row) for row in rows]
+    
+    def update_stage1(self, review_id: str, data: Dict[str, Any]):
+        """Update Stage 1 classification data"""
+        cursor = self.conn.cursor()
+        
+        cursor.execute("""
+            UPDATE reviews SET
+                stage1_llm1_type = ?,
+                stage1_llm1_department = ?,
+                stage1_llm1_priority = ?,
+                stage1_llm1_confidence = ?,
+                stage1_llm1_reasoning = ?,
+                stage1_llm2_user_type = ?,
+                stage1_llm2_emotion = ?,
+                stage1_llm2_context = ?,
+                stage1_llm2_confidence = ?,
+                stage1_llm2_reasoning = ?,
+                stage1_manager_classification = ?,
+                stage1_manager_reasoning = ?,
+                stage1_completed_at = ?,
+                processing_status = 'stage1_complete'
+            WHERE review_id = ?
+        """, (
+            data.get('llm1_type'),
+            data.get('llm1_department'),
+            data.get('llm1_priority'),
+            data.get('llm1_confidence'),
+            data.get('llm1_reasoning'),
+            data.get('llm2_user_type'),
+            data.get('llm2_emotion'),
+            data.get('llm2_context'),
+            data.get('llm2_confidence'),
+            data.get('llm2_reasoning'),
+            data.get('manager_classification'),
+            data.get('manager_reasoning'),
+            datetime.now().isoformat(),
+            review_id
+        ))
+        
+        self.conn.commit()
+    
+    def update_stage2(self, review_id: str, data: Dict[str, Any]):
+        """Update Stage 2 sentiment data"""
+        cursor = self.conn.cursor()
+        
+        cursor.execute("""
+            UPDATE reviews SET
+                stage2_best_sentiment = ?,
+                stage2_best_confidence = ?,
+                stage2_best_prob_positive = ?,
+                stage2_best_prob_neutral = ?,
+                stage2_best_prob_negative = ?,
+                stage2_alt_sentiment = ?,
+                stage2_alt_confidence = ?,
+                stage2_alt_prob_positive = ?,
+                stage2_alt_prob_neutral = ?,
+                stage2_alt_prob_negative = ?,
+                stage2_agreement = ?,
+                stage2_layer_sentiment = ?,
+                stage2_completed_at = ?,
+                processing_status = 'stage2_complete'
+            WHERE review_id = ?
+        """, (
+            data.get('best_sentiment'),
+            data.get('best_confidence'),
+            data.get('best_prob_positive'),
+            data.get('best_prob_neutral'),
+            data.get('best_prob_negative'),
+            data.get('alt_sentiment'),
+            data.get('alt_confidence'),
+            data.get('alt_prob_positive'),
+            data.get('alt_prob_neutral'),
+            data.get('alt_prob_negative'),
+            data.get('agreement'),
+            data.get('layer_sentiment'),
+            datetime.now().isoformat(),
+            review_id
+        ))
+        
+        self.conn.commit()
+    
+    def update_stage3(self, review_id: str, data: Dict[str, Any]):
+        """Update Stage 3 finalization data"""
+        cursor = self.conn.cursor()
+        
+        cursor.execute("""
+            UPDATE reviews SET
+                stage3_final_sentiment = ?,
+                stage3_confidence = ?,
+                stage3_reasoning = ?,
+                stage3_validation_notes = ?,
+                stage3_conflicts_found = ?,
+                stage3_action_recommendation = ?,
+                stage3_needs_human_review = ?,
+                stage3_completed_at = ?,
+                processing_status = 'complete',
+                processing_completed_at = ?
+            WHERE review_id = ?
+        """, (
+            data.get('final_sentiment'),
+            data.get('confidence'),
+            data.get('reasoning'),
+            data.get('validation_notes'),
+            data.get('conflicts_found'),
+            data.get('action_recommendation'),
+            data.get('needs_human_review'),
+            datetime.now().isoformat(),
+            datetime.now().isoformat(),
+            review_id
+        ))
+        
+        self.conn.commit()
+    
+    def log_llm_decision(self, review_id: str, stage: str, model_name: str, 
+                        input_prompt: str, output_response: str, 
+                        confidence: float, reasoning: str, processing_time: float):
+        """Log LLM decision for audit trail"""
+        cursor = self.conn.cursor()
+        
+        cursor.execute("""
+            INSERT INTO llm_decision_logs 
+            (review_id, stage, model_name, input_prompt, output_response, 
+             confidence, reasoning, processing_time_seconds)
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+        """, (
+            review_id, stage, model_name, input_prompt, output_response,
+            confidence, reasoning, processing_time
+        ))
+        
+        self.conn.commit()
+    
+    def get_all_processed_reviews(self) -> List[Dict]:
+        """Get all reviews that have been fully processed"""
+        cursor = self.conn.cursor()
+        
+        cursor.execute("""
+            SELECT * FROM reviews 
+            WHERE processing_status = 'complete'
+            ORDER BY processing_completed_at DESC
+        """)
+        
+        rows = cursor.fetchall()
+        return [dict(row) for row in rows]
+    
+    def save_batch_insights(self, insights: Dict[str, Any]):
+        """Save batch analytics to database"""
+        cursor = self.conn.cursor()
+        
+        cursor.execute("""
+            INSERT INTO batch_insights
+            (analysis_date, total_reviews, sentiment_positive, sentiment_neutral,
+             sentiment_negative, priority_critical, priority_high, priority_medium,
+             priority_low, dept_engineering, dept_ux, dept_support, dept_business,
+             critical_issues, quick_wins, recommendations)
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+        """, (
+            datetime.now().date(),
+            insights.get('total_reviews', 0),
+            insights.get('sentiment_positive', 0),
+            insights.get('sentiment_neutral', 0),
+            insights.get('sentiment_negative', 0),
+            insights.get('priority_critical', 0),
+            insights.get('priority_high', 0),
+            insights.get('priority_medium', 0),
+            insights.get('priority_low', 0),
+            insights.get('dept_engineering', 0),
+            insights.get('dept_ux', 0),
+            insights.get('dept_support', 0),
+            insights.get('dept_business', 0),
+            json.dumps(insights.get('critical_issues', [])),
+            json.dumps(insights.get('quick_wins', [])),
+            json.dumps(insights.get('recommendations', []))
+        ))
+        
+        self.conn.commit()
+        print("   ✅ Batch insights saved to database")
+    
+    def reset_processing_status(self, limit: Optional[int] = None):
+        """Reset processing status to reprocess reviews"""
+        cursor = self.conn.cursor()
+        
+        if limit:
+            # Reset only the most recent N reviews
+            query = """
+                UPDATE reviews 
+                SET processing_status = 'pending',
+                    processing_started_at = NULL,
+                    processing_completed_at = NULL,
+                    stage1_completed_at = NULL,
+                    stage2_completed_at = NULL,
+                    stage3_completed_at = NULL
+                WHERE review_id IN (
+                    SELECT review_id FROM reviews 
+                    ORDER BY scraped_at DESC 
+                    LIMIT ?
+                )
+            """
+            cursor.execute(query, (limit,))
+        else:
+            # Reset all reviews
+            query = """
+                UPDATE reviews 
+                SET processing_status = 'pending',
+                    processing_started_at = NULL,
+                    processing_completed_at = NULL,
+                    stage1_completed_at = NULL,
+                    stage2_completed_at = NULL,
+                    stage3_completed_at = NULL
+            """
+            cursor.execute(query)
         
-        return {
-            'final_type': llm1_result.get('type', 'unknown'),
-            'final_department': llm1_result.get('department', 'unknown'),
-            'final_priority': llm1_result.get('priority', 'medium'),
-            'synthesis_reasoning': f'Manager error: {str(e)}',
-            'model': 'meta-llama/Llama-3.3-70B-Instruct'
-        }
-
-
-def stage1_classification_node(state: ReviewState) -> Dict[str, Any]:
-    """Stage 1 Node: Classification with PARALLEL execution"""
-    print(f"\n      📝 Review ID: {state['review_id']}")
-    print(f"      ⏳ STAGE 1: Classification (Parallel LLM1 + LLM2)...")
-    
-    start_time = time.time()
-    review_dict = dict(state)
-    
-    # PARALLEL EXECUTION
-    with ThreadPoolExecutor(max_workers=2) as executor:
-        future_llm1 = executor.submit(llm1_classify, review_dict)
-        future_llm2 = executor.submit(llm2_analyze, review_dict)
+        affected = cursor.rowcount
+        self.conn.commit()
         
-        llm1_result = future_llm1.result()
-        llm2_result = future_llm2.result()
-    
-    print(f"         ✅ LLM1: {llm1_result['type']} → {llm1_result['department']} (Priority: {llm1_result['priority']})")
-    print(f"         ✅ LLM2: {llm2_result['user_type']}, {llm2_result['emotion']}")
-    
-    # Manager synthesizes
-    print(f"         🤖 Manager synthesizing...")
-    manager_result = manager_synthesize(llm1_result, llm2_result, review_dict)
-    
-    stage1_time = time.time() - start_time
-    print(f"      ✅ Stage 1 complete ({stage1_time:.2f}s)")
-    
-    return {
-        "llm1_result": llm1_result,
-        "llm2_result": llm2_result,
-        "manager_result": manager_result,
-        "classification_type": manager_result['final_type'],
-        "department": manager_result['final_department'],
-        "priority": manager_result['final_priority'],
-        "user_type": llm2_result['user_type'],
-        "emotion": llm2_result['emotion'],
-        "context": llm2_result.get('context', ''),
-        "stage1_completed": True,
-        "stage1_time": stage1_time,
-        "errors": state.get('errors', [])
-    }
-
-
-# ============================================================================
-# STAGE 2: SENTIMENT ANALYSIS
-# ============================================================================
-
-def analyze_best_sentiment(text: str) -> Dict[str, Any]:
-    """Best Model: Twitter-BERT"""
-    load_sentiment_models()
-    
-    try:
-        inputs = _best_tokenizer(text, return_tensors="pt", truncation=True, max_length=512, padding=True)
-        
-        with torch.no_grad():
-            outputs = _best_model(**inputs)
-            probs = torch.nn.functional.softmax(outputs.logits, dim=-1)
-            prediction = torch.argmax(probs, dim=-1).item()
-            confidence = probs[0][prediction].item()
-        
-        label_map = {0: "NEGATIVE", 1: "NEUTRAL", 2: "POSITIVE"}
-        
-        return {
-            'sentiment': label_map[prediction],
-            'confidence': confidence,
-            'prob_negative': probs[0][0].item(),
-            'prob_neutral': probs[0][1].item(),
-            'prob_positive': probs[0][2].item(),
-            'model': 'twitter-roberta-base-sentiment-latest'
-        }
-    except Exception as e:
-        print(f"❌ Best sentiment ERROR: {e}")
-        return {
-            'sentiment': 'NEUTRAL',
-            'confidence': 0.0,
-            'prob_negative': 0.33,
-            'prob_neutral': 0.34,
-            'prob_positive': 0.33,
-            'model': 'error',
-            'error': str(e)
-        }
-
-
-def analyze_alt_sentiment(text: str) -> Dict[str, Any]:
-    """Alternate Model: BERTweet - FIXED"""
-    load_sentiment_models()
-    
-    try:
-        inputs = _alt_tokenizer(text, return_tensors="pt", truncation=True, max_length=512, padding=True)
-        
-        with torch.no_grad():
-            outputs = _alt_model(**inputs)
-            probs = torch.nn.functional.softmax(outputs.logits, dim=-1)
-            prediction = torch.argmax(probs, dim=-1).item()
-            confidence = probs[0][prediction].item()
-        
-        label_map = {0: "NEGATIVE", 1: "NEUTRAL", 2: "POSITIVE"}
-        
-        return {
-            'sentiment': label_map[prediction],
-            'confidence': confidence,
-            'prob_negative': probs[0][0].item(),
-            'prob_neutral': probs[0][1].item(),
-            'prob_positive': probs[0][2].item(),
-            'model': 'bertweet-base-sentiment-analysis'
-        }
-    except Exception as e:
-        print(f"❌ Alt sentiment ERROR: {e}")
-        return {
-            'sentiment': 'NEUTRAL',
-            'confidence': 0.0,
-            'prob_negative': 0.33,
-            'prob_neutral': 0.34,
-            'prob_positive': 0.33,
-            'model': 'error',
-            'error': str(e)
-        }
-
-
-def sentiment_layer(best_result: Dict, alt_result: Dict) -> Dict[str, Any]:
-    """Sentiment Layer: Combine with confidence weighting"""
-    best_sentiment = best_result.get('sentiment')
-    best_confidence = best_result.get('confidence', 0.0)
-    
-    alt_sentiment = alt_result.get('sentiment')
-    alt_confidence = alt_result.get('confidence', 0.0)
-    
-    agreement = (best_sentiment == alt_sentiment)
-    
-    if agreement:
-        final_sentiment = best_sentiment
-        combined_confidence = max(best_confidence, alt_confidence)
-        agreement_strength = "STRONG"
-    else:
-        if best_confidence > alt_confidence:
-            final_sentiment = best_sentiment
-            combined_confidence = best_confidence
-        else:
-            final_sentiment = alt_sentiment
-            combined_confidence = alt_confidence
-        agreement_strength = "WEAK"
-    
-    return {
-        'layer_sentiment': final_sentiment,
-        'combined_confidence': combined_confidence,
-        'agreement': agreement,
-        'agreement_strength': agreement_strength
-    }
-
-
-def stage2_sentiment_node(state: ReviewState) -> Dict[str, Any]:
-    """Stage 2 Node: Sentiment with PARALLEL execution"""
-    print(f"\n      ⏳ STAGE 2: Sentiment Analysis (Parallel Best + Alternate)...")
-    
-    start_time = time.time()
-    review_text = state['review_text']
-    
-    # PARALLEL EXECUTION
-    with ThreadPoolExecutor(max_workers=2) as executor:
-        future_best = executor.submit(analyze_best_sentiment, review_text)
-        future_alt = executor.submit(analyze_alt_sentiment, review_text)
+        if affected > 0:
+            print(f"   🔄 Reset {affected} reviews to pending status")
         
-        best_result = future_best.result()
-        alt_result = future_alt.result()
-    
-    print(f"         ✅ Best: {best_result['sentiment']} ({best_result['confidence']:.3f})")
-    print(f"         ✅ Alt: {alt_result['sentiment']} ({alt_result['confidence']:.3f})")
-    
-    # Sentiment Layer combines results
-    layer_result = sentiment_layer(best_result, alt_result)
-    
-    agreement_icon = "✅" if layer_result['agreement'] else "⚠️ "
-    print(f"         {agreement_icon} Final: {layer_result['layer_sentiment']} (agreement: {layer_result['agreement']})")
-    
-    stage2_time = time.time() - start_time
-    print(f"      ✅ Stage 2 complete ({stage2_time:.2f}s)")
-    
-    return {
-        "best_sentiment_result": best_result,
-        "alt_sentiment_result": alt_result,
-        "sentiment_layer_result": layer_result,
-        "sentiment": layer_result['layer_sentiment'],
-        "sentiment_confidence": layer_result['combined_confidence'],
-        "sentiment_agreement": layer_result['agreement'],
-        "stage2_completed": True,
-        "stage2_time": stage2_time,
-        "errors": state.get('errors', [])
-    }
-
-
-# ============================================================================
-# STAGE 3: FINALIZATION NODE
-# ============================================================================
-
-def stage3_finalization_node(state: ReviewState) -> Dict[str, Any]:
-    """Stage 3 Node: Final synthesis with LLM3"""
-    print(f"\n      ⏳ STAGE 3: Finalization (LLM3)...")
-    
-    start_time = time.time()
-    
-    hf_client = get_hf_client()
-    
-    if hf_client is None:
-        result = {
-            'final_sentiment': state.get('sentiment', 'NEUTRAL'),
-            'confidence': state.get('sentiment_confidence', 0.0),
-            'reasoning': 'Stage 3 skipped - HuggingFace API key not set',
-            'validation_notes': 'API key missing',
-            'conflicts_found': 'none',
-            'action_recommendation': f"Route to {state.get('department', 'support')}",
-            'needs_human_review': True,
-            'model': 'meta-llama/Llama-3.1-70B-Instruct'
-        }
-        
-        stage3_time = 0.00
-        print(f"         ✅ Final: {result['final_sentiment']} ({result.get('confidence', 0):.3f})")
-        print(f"         📋 Needs Review: {result.get('needs_human_review', False)}")
-        print(f"      ✅ Stage 3 complete ({stage3_time:.2f}s)")
-        
-        return {
-            "final_result": result,
-            "final_sentiment": result['final_sentiment'],
-            "final_confidence": result['confidence'],
-            "reasoning": result['reasoning'],
-            "action_recommendation": result['action_recommendation'],
-            "conflicts_found": result['conflicts_found'],
-            "validation_notes": result['validation_notes'],
-            "needs_human_review": result['needs_human_review'],
-            "stage3_completed": True,
-            "stage3_time": stage3_time,
-            "total_time": state.get('stage1_time', 0) + state.get('stage2_time', 0),
-            "processing_completed_at": datetime.now().isoformat(),
-            "errors": state.get('errors', [])
-        }
-    
-    review_text = state['review_text']
-    rating = state['rating']
-    
-    # FIXED: Use chat format
-    system_prompt = """You are a final decision-making AI analyzing customer feedback for a theme park/attraction app.
-
-Your task:
-1. Review all data from previous stages
-2. Make FINAL sentiment decision
-3. Provide comprehensive reasoning
-4. Generate action recommendation
-5. Flag if human review needed
-
-Respond ONLY in valid JSON format:
-{
-  "final_sentiment": "POSITIVE/NEGATIVE/NEUTRAL",
-  "confidence": 0.0-1.0,
-  "reasoning": "Comprehensive explanation",
-  "validation_notes": "Does classification match sentiment?",
-  "conflicts_found": "any conflicts or 'none'",
-  "action_recommendation": "Specific action",
-  "needs_human_review": true/false
-}"""
-
-    user_prompt = f"""REVIEW DATA:
-Rating: {rating}/5
-Text: {review_text}
-
-STAGE 1 CLASSIFICATION:
-- Type: {state.get('classification_type')}
-- Department: {state.get('department')}
-- Priority: {state.get('priority')}
-- User Type: {state.get('user_type')}
-- Emotion: {state.get('emotion')}
-
-STAGE 2 SENTIMENT:
-- Best: {state['best_sentiment_result'].get('sentiment')} ({state['best_sentiment_result'].get('confidence'):.2f})
-- Alternate: {state['alt_sentiment_result'].get('sentiment')} ({state['alt_sentiment_result'].get('confidence'):.2f})
-- Agreement: {state.get('sentiment_agreement')}
+        return affected
 
-Make your final decision:"""
 
-    try:
-        print(f"   🔍 Calling Llama 70B API...")
-        
-        # FIXED: Use chat_completion
-        response = hf_client.chat_completion(
-            messages=[
-                {"role": "system", "content": system_prompt},
-                {"role": "user", "content": user_prompt}
-            ],
-            model="meta-llama/Llama-3.1-70B-Instruct",
-            max_tokens=400,
-            temperature=0.1
-        )
-        
-        content = response.choices[0].message.content
-        print(f"   ✅ Got response ({len(content)} chars)")
-        
-        content_clean = content.strip()
-        if content_clean.startswith('```'):
-            content_clean = content_clean.split('```')[1]
-            if content_clean.startswith('json'):
-                content_clean = content_clean[4:]
-        content_clean = content_clean.strip()
-        
-        result = json.loads(content_clean)
-        result['model'] = 'meta-llama/Llama-3.1-70B-Instruct'
-        
-    except Exception as e:
-        print(f"❌ STAGE 3 ERROR: {type(e).__name__}: {str(e)}")
-        
-        result = {
-            'final_sentiment': state.get('sentiment', 'NEUTRAL'),
-            'confidence': state.get('sentiment_confidence', 0.5),
-            'reasoning': f'Error in LLM3: {str(e)}',
-            'validation_notes': 'Error',
-            'conflicts_found': 'error',
-            'action_recommendation': f"Route to {state.get('department')}",
-            'needs_human_review': True,
-            'model': 'meta-llama/Llama-3.1-70B-Instruct'
-        }
-    
-    stage3_time = time.time() - start_time
+if __name__ == "__main__":
+    # Test database enhancement
+    print("\n" + "="*60)
+    print("🧪 TESTING DATABASE ENHANCEMENT")
+    print("="*60 + "\n")
     
-    print(f"         ✅ Final: {result['final_sentiment']} ({result.get('confidence', 0):.3f})")
-    print(f"         📋 Needs Review: {result.get('needs_human_review', False)}")
-    print(f"      ✅ Stage 3 complete ({stage3_time:.2f}s)")
+    db = EnhancedDatabase()
+    db.connect()
+    db.enhance_schema()
     
-    # Calculate total time
-    total_time = state.get('stage1_time', 0) + state.get('stage2_time', 0) + stage3_time
+    # Get pending reviews
+    pending = db.get_pending_reviews(limit=5)
+    print(f"\n📋 Found {len(pending)} pending reviews")
     
-    return {
-        "final_result": result,
-        "final_sentiment": result['final_sentiment'],
-        "final_confidence": result['confidence'],
-        "reasoning": result['reasoning'],
-        "action_recommendation": result['action_recommendation'],
-        "conflicts_found": result['conflicts_found'],
-        "validation_notes": result['validation_notes'],
-        "needs_human_review": result['needs_human_review'],
-        "stage3_completed": True,
-        "stage3_time": stage3_time,
-        "total_time": total_time,
-        "processing_completed_at": datetime.now().isoformat(),
-        "errors": state.get('errors', [])
-    }
-
-
-if __name__ == "__main__":
-    print("\n✅ LangGraph nodes module loaded!")
-    print("   Nodes available:")
-    print("   - stage1_classification_node (parallel LLM1+LLM2)")
-    print("   - stage2_sentiment_node (parallel Best+Alt)")
-    print("   - stage3_finalization_node (LLM3)")
+    db.close()
+    print("\n✅ Database enhancement test complete!")