Upload folder using huggingface_hub

modal_backend.py

@@ -1,19 +1,20 @@
 """
-Modal Backend for Restaurant Intelligence Agent
-With TRUE MCP Server Integration
+Modal Backend for Restaurant Intelligence Agent - PARALLEL OPTIMIZED
+Version 3.0 - Uses Modal's parallel processing for 5x speed improvement
 
-UPDATED: Now sends slim trend_data instead of full raw_reviews
-- Reduces response size by ~97%
-- Pre-calculates sentiment in backend
-- Fixes HuggingFace timeout issues
+KEY OPTIMIZATIONS:
+1. Parallel batch processing with .map() - Process all batches simultaneously
+2. Parallel insights generation - Chef + Manager at same time
+3. Larger batch sizes (30 reviews instead of 20)
+4. Reduced timeout since parallel is faster
 
-Deploys:
-1. Analysis API endpoint
-2. MCP Server endpoint
+TARGET: 1000 reviews in ~5 minutes (down from 15+ minutes)
 """
 
 import modal
 from typing import Dict, Any, List
+import os
+import json
 
 # Create Modal app
 app = modal.App("restaurant-intelligence")
@@ -33,14 +34,14 @@ image = (
         "matplotlib",
         "fastapi[standard]",
         "httpx",
-        "fastmcp",  # Required by src/mcp_integrations modules
+        "fastmcp",
     )
     .add_local_python_source("src")
 )
 
 
 # ============================================================================
-# HELPER FUNCTION - Calculate sentiment from text
+# HELPER FUNCTIONS
 # ============================================================================
 
 def calculate_sentiment(text: str) -> float:
@@ -65,186 +66,277 @@ def calculate_sentiment(text: str) -> float:
 
 
 # ============================================================================
-# MCP SERVER (TRUE MCP INTEGRATION)
+# PARALLEL BATCH PROCESSOR - The key optimization!
 # ============================================================================
 
-# In-memory storage for MCP
-REVIEW_INDEX: Dict[str, List[str]] = {}
-ANALYSIS_CACHE: Dict[str, Dict[str, Any]] = {}
-
-
-@app.function(image=image, timeout=300)
-@modal.asgi_app()
-def mcp_server():
-    """
-    TRUE MCP Server - exposes tools via MCP protocol over HTTP.
+@app.function(
+    image=image,
+    secrets=[modal.Secret.from_name("anthropic-api-key")],
+    timeout=120,  # 2 min per batch is plenty
+    retries=2,
+)
+def process_batch(batch_data: Dict[str, Any]) -> Dict[str, Any]:
     """
-    from fastapi import FastAPI, HTTPException
-    from pydantic import BaseModel
-    from datetime import datetime
+    Process a single batch of reviews - runs in PARALLEL across containers!
     
-    mcp_api = FastAPI(title="Restaurant Intelligence MCP Server")
+    This function is called via .map() to process all batches simultaneously.
+    Modal will spin up multiple containers to handle batches in parallel.
+    """
+    from anthropic import Anthropic
+    import os
     
-    class ToolRequest(BaseModel):
-        tool_name: str
-        arguments: Dict[str, Any] = {}
+    reviews = batch_data["reviews"]
+    restaurant_name = batch_data["restaurant_name"]
+    batch_index = batch_data["batch_index"]
+    start_index = batch_data["start_index"]
     
-    class IndexReviewsRequest(BaseModel):
-        restaurant_name: str
-        reviews: List[str]
+    print(f"🔄 Processing batch {batch_index} ({len(reviews)} reviews)...")
     
-    class QueryReviewsRequest(BaseModel):
-        restaurant_name: str
-        question: str
-        top_k: int = 5
+    client = Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
     
-    # MCP Tools
-    def index_reviews(restaurant_name: str, reviews: List[str]) -> Dict[str, Any]:
-        REVIEW_INDEX[restaurant_name] = reviews
-        return {
-            "success": True,
-            "restaurant": restaurant_name,
-            "indexed_count": len(reviews),
-            "message": f"Indexed {len(reviews)} reviews for {restaurant_name}"
-        }
+    # Build extraction prompt
+    numbered_reviews = []
+    for i, review in enumerate(reviews):
+        numbered_reviews.append(f"[Review {i}]: {review}")
+    reviews_text = "\n\n".join(numbered_reviews)
     
-    def query_reviews(restaurant_name: str, question: str, top_k: int = 5) -> Dict[str, Any]:
-        reviews = REVIEW_INDEX.get(restaurant_name, [])
-        if not reviews:
-            return {"success": False, "error": f"No reviews indexed for {restaurant_name}"}
+    prompt = f"""You are analyzing customer reviews for {restaurant_name}. Extract BOTH menu items AND aspects in ONE PASS.
+
+REVIEWS:
+{reviews_text}
+
+YOUR TASK - Extract THREE things simultaneously:
+1. **MENU ITEMS** (food & drinks mentioned)
+2. **ASPECTS** (what customers care about: service, ambience, etc.)
+3. **SENTIMENT** for each
+
+SENTIMENT SCALE (IMPORTANT):
+- **Positive (0.6 to 1.0):** Customer clearly enjoyed/praised this item or aspect
+- **Neutral (0.0 to 0.59):** Mixed feelings, okay but not exceptional, or simply mentioned without strong opinion
+- **Negative (-1.0 to -0.01):** Customer complained, criticized, or expressed disappointment
+
+RULES:
+- Specific items only: "salmon sushi", "miso soup", "sake"
+- Separate food from drinks
+- Lowercase names
+- For EACH item/aspect, list which review NUMBERS mention it (just indices, not text)
+
+OUTPUT (JSON):
+{{
+  "food_items": [
+    {{"name": "item name", "mention_count": 2, "sentiment": 0.85, "category": "type", "related_reviews": [0, 5]}}
+  ],
+  "drinks": [
+    {{"name": "drink name", "mention_count": 1, "sentiment": 0.7, "category": "alcohol", "related_reviews": [3]}}
+  ],
+  "aspects": [
+    {{"name": "service speed", "mention_count": 3, "sentiment": 0.65, "description": "brief desc", "related_reviews": [1, 2, 7]}}
+  ]
+}}
+
+CRITICAL: Output ONLY valid JSON, no other text. Use sentiment scale: >= 0.6 positive, 0-0.59 neutral, < 0 negative
+
+Extract everything:"""
+
+    try:
+        response = client.messages.create(
+            model="claude-sonnet-4-20250514",
+            max_tokens=4000,
+            temperature=0.3,
+            messages=[{"role": "user", "content": prompt}]
+        )
         
-        question_words = set(question.lower().split())
-        scored = [(len(question_words & set(r.lower().split())), r) for r in reviews]
-        scored.sort(reverse=True, key=lambda x: x[0])
+        result_text = response.content[0].text
+        result_text = result_text.replace('```json', '').replace('```', '').strip()
         
-        return {
-            "success": True,
-            "restaurant": restaurant_name,
-            "question": question,
-            "relevant_reviews": [r[1] for r in scored[:top_k]],
-            "review_count": min(top_k, len(reviews))
-        }
-    
-    def save_report(restaurant_name: str, report_data: Dict, report_type: str = "analysis") -> Dict[str, Any]:
-        report_id = f"{restaurant_name}_{report_type}_{datetime.now().isoformat()}"
-        ANALYSIS_CACHE[report_id] = {"restaurant": restaurant_name, "type": report_type, "data": report_data}
-        return {"success": True, "report_id": report_id}
-    
-    def list_tools() -> Dict[str, Any]:
-        return {
-            "success": True,
-            "tools": [
-                {"name": "index_reviews", "description": "Index reviews for RAG Q&A"},
-                {"name": "query_reviews", "description": "Answer questions about reviews"},
-                {"name": "save_report", "description": "Save analysis report"},
-            ]
-        }
-    
-    @mcp_api.get("/")
-    async def root():
-        return {"name": "Restaurant Intelligence MCP Server", "protocol": "MCP", "version": "1.0"}
-    
-    @mcp_api.get("/health")
-    async def health():
-        return {"status": "healthy", "mcp": "enabled"}
-    
-    @mcp_api.get("/tools")
-    async def get_tools():
-        return list_tools()
-    
-    @mcp_api.post("/mcp/call")
-    async def call_tool(request: ToolRequest):
-        """TRUE MCP interface - agent calls tools via this endpoint."""
-        tool_map = {
-            "index_reviews": lambda args: index_reviews(args["restaurant_name"], args["reviews"]),
-            "query_reviews": lambda args: query_reviews(args["restaurant_name"], args["question"], args.get("top_k", 5)),
-            "save_report": lambda args: save_report(args["restaurant_name"], args["report_data"], args.get("report_type", "analysis")),
-            "list_tools": lambda args: list_tools()
-        }
+        data = json.loads(result_text)
         
-        if request.tool_name not in tool_map:
-            raise HTTPException(status_code=404, detail=f"Tool '{request.tool_name}' not found")
+        # Map review indices back to full text
+        for item in data.get('food_items', []):
+            indices = item.get('related_reviews', [])
+            item['related_reviews'] = []
+            for idx in indices:
+                if isinstance(idx, int) and 0 <= idx < len(reviews):
+                    item['related_reviews'].append({
+                        'review_index': start_index + idx,
+                        'review_text': reviews[idx]
+                    })
+            if 'name' in item:
+                item['name'] = item['name'].lower()
         
-        try:
-            result = tool_map[request.tool_name](request.arguments)
-            return {"success": True, "tool": request.tool_name, "result": result}
-        except Exception as e:
-            raise HTTPException(status_code=500, detail=str(e))
-    
-    @mcp_api.post("/tools/index_reviews")
-    async def api_index_reviews(request: IndexReviewsRequest):
-        return index_reviews(request.restaurant_name, request.reviews)
-    
-    @mcp_api.post("/tools/query_reviews")
-    async def api_query_reviews(request: QueryReviewsRequest):
-        return query_reviews(request.restaurant_name, request.question, request.top_k)
+        for item in data.get('drinks', []):
+            indices = item.get('related_reviews', [])
+            item['related_reviews'] = []
+            for idx in indices:
+                if isinstance(idx, int) and 0 <= idx < len(reviews):
+                    item['related_reviews'].append({
+                        'review_index': start_index + idx,
+                        'review_text': reviews[idx]
+                    })
+            if 'name' in item:
+                item['name'] = item['name'].lower()
+        
+        for aspect in data.get('aspects', []):
+            indices = aspect.get('related_reviews', [])
+            aspect['related_reviews'] = []
+            for idx in indices:
+                if isinstance(idx, int) and 0 <= idx < len(reviews):
+                    aspect['related_reviews'].append({
+                        'review_index': start_index + idx,
+                        'review_text': reviews[idx]
+                    })
+            if 'name' in aspect:
+                aspect['name'] = aspect['name'].lower()
+        
+        print(f"✅ Batch {batch_index} complete: {len(data.get('food_items', []))} food, {len(data.get('drinks', []))} drinks, {len(data.get('aspects', []))} aspects")
+        return {"success": True, "batch_index": batch_index, "data": data}
+        
+    except json.JSONDecodeError as e:
+        print(f"⚠️ Batch {batch_index} JSON error: {e}")
+        return {"success": False, "batch_index": batch_index, "data": {"food_items": [], "drinks": [], "aspects": []}}
+    except Exception as e:
+        print(f"❌ Batch {batch_index} error: {e}")
+        return {"success": False, "batch_index": batch_index, "data": {"food_items": [], "drinks": [], "aspects": []}}
+
+
+@app.function(
+    image=image,
+    secrets=[modal.Secret.from_name("anthropic-api-key")],
+    timeout=180,  # 3 min for insights
+)
+def generate_insights_parallel(analysis_data: Dict[str, Any], restaurant_name: str, role: str) -> Dict[str, Any]:
+    """Generate insights for a single role - runs in parallel with other insights."""
+    from anthropic import Anthropic
+    import os
+    import re
+    
+    print(f"🧠 Generating {role} insights...")
+    
+    client = Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
+    
+    # Build prompt based on role
+    menu_items = analysis_data.get('menu_analysis', {}).get('food_items', [])[:20]
+    drinks = analysis_data.get('menu_analysis', {}).get('drinks', [])[:10]
+    aspects = analysis_data.get('aspect_analysis', {}).get('aspects', [])[:20]
+    
+    # Format menu summary
+    menu_lines = ["TOP MENU ITEMS:"]
+    for item in menu_items:
+        s = item.get('sentiment', 0)
+        emoji = "🟢" if s >= 0.6 else "🟡" if s >= 0 else "🔴"
+        menu_lines.append(f"  {emoji} {item.get('name', '?')}: sentiment {s:+.2f}, {item.get('mention_count', 0)} mentions")
+    menu_summary = "\n".join(menu_lines)
+    
+    # Format aspect summary
+    aspect_lines = ["TOP ASPECTS:"]
+    for a in aspects:
+        s = a.get('sentiment', 0)
+        emoji = "🟢" if s >= 0.6 else "🟡" if s >= 0 else "🔴"
+        aspect_lines.append(f"  {emoji} {a.get('name', '?')}: sentiment {s:+.2f}, {a.get('mention_count', 0)} mentions")
+    aspect_summary = "\n".join(aspect_lines)
+    
+    if role == 'chef':
+        focus = "Focus on: Food quality, menu items, ingredients, presentation, portions, consistency"
+        topic_filter = "ONLY on food/kitchen topics"
+    else:
+        focus = "Focus on: Service, staff, wait times, ambience, value, cleanliness"
+        topic_filter = "ONLY on operations/service topics"
     
-    return mcp_api
+    prompt = f"""You are an expert restaurant consultant analyzing feedback for {restaurant_name}.
 
+{menu_summary}
 
-# ============================================================================
-# MAIN ANALYSIS FUNCTIONS
-# ============================================================================
+{aspect_summary}
 
-@app.function(image=image)
-def hello() -> Dict[str, Any]:
-    return {"status": "Modal is working!", "mcp": "enabled"}
+SENTIMENT SCALE:
+- 🟢 POSITIVE (>= 0.6): Highlight as STRENGTH
+- 🟡 NEUTRAL (0 to 0.59): Room for improvement
+- 🔴 NEGATIVE (< 0): Flag as CONCERN
 
+YOUR TASK: Generate insights for the {"HEAD CHEF" if role == "chef" else "RESTAURANT MANAGER"}.
+{focus}
 
-@app.function(image=image, timeout=600)
-def scrape_restaurant_modal(url: str, max_reviews: int = 100) -> Dict[str, Any]:
-    """Scrape reviews from OpenTable or Google Maps."""
-    
-    # Detect platform
-    url_lower = url.lower()
-    if 'opentable' in url_lower:
-        from src.scrapers.opentable_scraper import scrape_opentable
-        result = scrape_opentable(url=url, max_reviews=max_reviews, headless=True)
-    elif any(x in url_lower for x in ['google.com/maps', 'goo.gl/maps', 'maps.google', 'maps.app.goo.gl']):
-        from src.scrapers.google_maps_scraper import scrape_google_maps
-        result = scrape_google_maps(url=url, max_reviews=max_reviews, headless=True)
-    else:
-        return {"success": False, "error": "Unsupported platform. Use OpenTable or Google Maps."}
-    
-    if not result.get("success"):
-        return {"success": False, "error": result.get("error")}
-    
-    from src.data_processing import process_reviews, clean_reviews_for_ai
-    
-    df = process_reviews(result)
-    reviews = clean_reviews_for_ai(df["review_text"].tolist(), verbose=False)
-    
-    # Create SLIM trend_data (pre-calculate sentiment, no text!)
-    trend_data = []
-    for _, row in df.iterrows():
-        text = str(row.get("review_text", ""))
-        trend_data.append({
-            "date": str(row.get("date", "")),
-            "rating": float(row.get("overall_rating", 0) or 0),
-            "sentiment": calculate_sentiment(text)  # Pre-calculate!
-        })
-    
+RULES:
+1. Focus {topic_filter}
+2. STRENGTHS from items with sentiment >= 0.6
+3. CONCERNS from items with sentiment < 0
+4. Output ONLY valid JSON
+
+OUTPUT:
+{{
+  "summary": "2-3 sentence executive summary",
+  "strengths": ["strength 1", "strength 2", "strength 3", "strength 4", "strength 5"],
+  "concerns": ["concern 1", "concern 2", "concern 3"],
+  "recommendations": [
+    {{"priority": "high", "action": "action", "reason": "why", "evidence": "data"}},
+    {{"priority": "medium", "action": "action", "reason": "why", "evidence": "data"}},
+    {{"priority": "low", "action": "action", "reason": "why", "evidence": "data"}}
+  ]
+}}
+
+Generate {role} insights:"""
+
+    try:
+        response = client.messages.create(
+            model="claude-sonnet-4-20250514",
+            max_tokens=2000,
+            temperature=0.4,
+            messages=[{"role": "user", "content": prompt}]
+        )
+        
+        result_text = response.content[0].text.strip()
+        result_text = result_text.replace('```json', '').replace('```', '').strip()
+        
+        # Find JSON in response
+        match = re.search(r'\{[\s\S]*\}', result_text)
+        if match:
+            insights = json.loads(match.group())
+            print(f"✅ {role.title()} insights generated")
+            return {"role": role, "insights": insights}
+        else:
+            print(f"⚠️ No JSON found in {role} response")
+            return {"role": role, "insights": _fallback_insights(role)}
+            
+    except Exception as e:
+        print(f"❌ Error generating {role} insights: {e}")
+        return {"role": role, "insights": _fallback_insights(role)}
+
+
+def _fallback_insights(role: str) -> Dict[str, Any]:
+    """Fallback insights if generation fails."""
     return {
-        "success": True,
-        "total_reviews": len(reviews),
-        "reviews": reviews,
-        "trend_data": trend_data,  # Slim version, no text!
-        "metadata": result.get("metadata", {}),
+        "summary": f"Analysis complete. See data for {role} insights.",
+        "strengths": ["Data available in charts"],
+        "concerns": ["Review individual items for details"],
+        "recommendations": [{"priority": "medium", "action": "Review data", "reason": "Auto-generated", "evidence": "N/A"}]
     }
 
 
+# ============================================================================
+# MAIN ANALYSIS FUNCTION - PARALLEL OPTIMIZED
+# ============================================================================
+
 @app.function(
     image=image,
     secrets=[modal.Secret.from_name("anthropic-api-key")],
-    timeout=2400,
+    timeout=600,  # 10 min max (down from 40 min)
 )
-def full_analysis_modal(url: str, max_reviews: int = 100) -> Dict[str, Any]:
+def full_analysis_parallel(url: str, max_reviews: int = 100) -> Dict[str, Any]:
     """
-    Complete end-to-end analysis with MCP integration.
+    PARALLEL OPTIMIZED analysis pipeline.
+    
+    Speed improvements:
+    1. Batches processed in PARALLEL via .map()
+    2. Chef + Manager insights generated in PARALLEL
+    3. Larger batch size (30 reviews)
     
-    UPDATED: Returns slim trend_data instead of full raw_reviews.
-    This reduces response size by ~97% and fixes timeout issues.
+    Target: 1000 reviews in ~5 minutes
     """
+    import time
+    start_time = time.time()
+    
+    print(f"🚀 Starting PARALLEL analysis for {url}")
+    print(f"📊 Max reviews: {max_reviews}")
     
     # Detect platform
     url_lower = url.lower()
@@ -253,7 +345,10 @@ def full_analysis_modal(url: str, max_reviews: int = 100) -> Dict[str, Any]:
     if platform == "unknown":
         return {"success": False, "error": "Unsupported platform. Use OpenTable or Google Maps."}
     
-    # Import scrapers
+    # Phase 1: Scrape reviews
+    print("📥 Phase 1: Scraping reviews...")
+    scrape_start = time.time()
+    
     if platform == "opentable":
         from src.scrapers.opentable_scraper import scrape_opentable
         result = scrape_opentable(url=url, max_reviews=max_reviews, headless=True)
@@ -262,52 +357,181 @@ def full_analysis_modal(url: str, max_reviews: int = 100) -> Dict[str, Any]:
         result = scrape_google_maps(url=url, max_reviews=max_reviews, headless=True)
     
     if not result.get("success"):
-        return {"success": False, "error": result.get("error")}
+        return {"success": False, "error": result.get("error", "Scraping failed")}
     
+    print(f"✅ Scraping complete in {time.time() - scrape_start:.1f}s")
+    
+    # Process reviews
     from src.data_processing import process_reviews, clean_reviews_for_ai
-    from src.agent.base_agent import RestaurantAnalysisAgent
     
     df = process_reviews(result)
     reviews = clean_reviews_for_ai(df["review_text"].tolist(), verbose=False)
     
-    # Create SLIM trend_data (pre-calculate sentiment in backend!)
-    # This is ~97% smaller than sending full review text
+    print(f"📊 Total reviews: {len(reviews)}")
+    
+    # Create trend data
     trend_data = []
     for _, row in df.iterrows():
         text = str(row.get("review_text", ""))
         trend_data.append({
             "date": str(row.get("date", "")),
             "rating": float(row.get("overall_rating", 0) or 0),
-            "sentiment": calculate_sentiment(text)  # Pre-calculated!
+            "sentiment": calculate_sentiment(text)
         })
     
-    # Extract restaurant name from URL
+    # Extract restaurant name
     if platform == "opentable":
         restaurant_name = url.split("/")[-1].split("?")[0].replace("-", " ").title()
     else:
-        # Google Maps
         if '/place/' in url:
             restaurant_name = url.split('/place/')[1].split('/')[0].replace('+', ' ').replace('%20', ' ')
         else:
             restaurant_name = "Restaurant"
     
-    # Run analysis
-    agent = RestaurantAnalysisAgent()
-    analysis = agent.analyze_restaurant(
-        restaurant_url=url,
-        restaurant_name=restaurant_name,
-        reviews=reviews,
-    )
+    # Phase 2: PARALLEL batch extraction
+    print("🔄 Phase 2: PARALLEL batch extraction...")
+    extract_start = time.time()
+    
+    BATCH_SIZE = 30  # Larger batches = fewer API calls
+    batches = []
+    for i in range(0, len(reviews), BATCH_SIZE):
+        batch_reviews = reviews[i:i+BATCH_SIZE]
+        batches.append({
+            "reviews": batch_reviews,
+            "restaurant_name": restaurant_name,
+            "batch_index": len(batches) + 1,
+            "start_index": i
+        })
+    
+    print(f"📦 Created {len(batches)} batches of ~{BATCH_SIZE} reviews each")
+    print(f"🚀 Processing ALL batches in PARALLEL...")
+    
+    # THIS IS THE KEY: Process all batches in parallel!
+    batch_results = list(process_batch.map(batches))
     
-    # Store in MCP cache for Q&A
-    REVIEW_INDEX[restaurant_name] = reviews
+    print(f"✅ All batches complete in {time.time() - extract_start:.1f}s")
     
-    # Add slim trend_data (NOT full raw_reviews!)
-    analysis['trend_data'] = trend_data
-    analysis['source'] = platform
+    # Merge results from all batches
+    all_food_items = {}
+    all_drinks = {}
+    all_aspects = {}
     
-    # Log response size for debugging
-    import json
+    for batch_result in batch_results:
+        if not batch_result.get("success"):
+            continue
+        
+        data = batch_result.get("data", {})
+        
+        # Merge food items
+        for item in data.get('food_items', []):
+            name = item.get('name', '').lower()
+            if not name:
+                continue
+            if name in all_food_items:
+                all_food_items[name]['mention_count'] += item.get('mention_count', 1)
+                all_food_items[name]['related_reviews'].extend(item.get('related_reviews', []))
+                # Weighted average sentiment
+                old_count = all_food_items[name]['mention_count'] - item.get('mention_count', 1)
+                new_count = item.get('mention_count', 1)
+                if old_count + new_count > 0:
+                    old_sent = all_food_items[name]['sentiment']
+                    new_sent = item.get('sentiment', 0)
+                    all_food_items[name]['sentiment'] = (old_sent * old_count + new_sent * new_count) / (old_count + new_count)
+            else:
+                all_food_items[name] = item
+        
+        # Merge drinks
+        for item in data.get('drinks', []):
+            name = item.get('name', '').lower()
+            if not name:
+                continue
+            if name in all_drinks:
+                all_drinks[name]['mention_count'] += item.get('mention_count', 1)
+                all_drinks[name]['related_reviews'].extend(item.get('related_reviews', []))
+                old_count = all_drinks[name]['mention_count'] - item.get('mention_count', 1)
+                new_count = item.get('mention_count', 1)
+                if old_count + new_count > 0:
+                    old_sent = all_drinks[name]['sentiment']
+                    new_sent = item.get('sentiment', 0)
+                    all_drinks[name]['sentiment'] = (old_sent * old_count + new_sent * new_count) / (old_count + new_count)
+            else:
+                all_drinks[name] = item
+        
+        # Merge aspects
+        for aspect in data.get('aspects', []):
+            name = aspect.get('name', '').lower()
+            if not name:
+                continue
+            if name in all_aspects:
+                all_aspects[name]['mention_count'] += aspect.get('mention_count', 1)
+                all_aspects[name]['related_reviews'].extend(aspect.get('related_reviews', []))
+                old_count = all_aspects[name]['mention_count'] - aspect.get('mention_count', 1)
+                new_count = aspect.get('mention_count', 1)
+                if old_count + new_count > 0:
+                    old_sent = all_aspects[name]['sentiment']
+                    new_sent = aspect.get('sentiment', 0)
+                    all_aspects[name]['sentiment'] = (old_sent * old_count + new_sent * new_count) / (old_count + new_count)
+            else:
+                all_aspects[name] = aspect
+    
+    # Sort by mention count
+    food_list = sorted(all_food_items.values(), key=lambda x: x.get('mention_count', 0), reverse=True)
+    drinks_list = sorted(all_drinks.values(), key=lambda x: x.get('mention_count', 0), reverse=True)
+    aspects_list = sorted(all_aspects.values(), key=lambda x: x.get('mention_count', 0), reverse=True)
+    
+    print(f"📊 Discovered: {len(food_list)} food + {len(drinks_list)} drinks + {len(aspects_list)} aspects")
+    
+    # Build analysis data
+    analysis_data = {
+        "menu_analysis": {
+            "food_items": food_list,
+            "drinks": drinks_list
+        },
+        "aspect_analysis": {
+            "aspects": aspects_list
+        }
+    }
+    
+    # Phase 3: PARALLEL insights generation
+    print("🧠 Phase 3: PARALLEL insights generation...")
+    insights_start = time.time()
+    
+    # Generate both insights in parallel!
+    insight_inputs = [
+        (analysis_data, restaurant_name, "chef"),
+        (analysis_data, restaurant_name, "manager")
+    ]
+    
+    insight_results = list(generate_insights_parallel.starmap(insight_inputs))
+    
+    insights = {}
+    for result in insight_results:
+        insights[result["role"]] = result["insights"]
+    
+    print(f"✅ Insights complete in {time.time() - insights_start:.1f}s")
+    
+    # Build final response
+    total_time = time.time() - start_time
+    print(f"🎉 TOTAL TIME: {total_time:.1f}s ({total_time/60:.1f} min)")
+    
+    analysis = {
+        "success": True,
+        "restaurant_name": restaurant_name,
+        "menu_analysis": analysis_data["menu_analysis"],
+        "aspect_analysis": analysis_data["aspect_analysis"],
+        "insights": insights,
+        "trend_data": trend_data,
+        "source": platform,
+        "stats": {
+            "total_reviews": len(reviews),
+            "food_items": len(food_list),
+            "drinks": len(drinks_list),
+            "aspects": len(aspects_list),
+            "processing_time_seconds": round(total_time, 1)
+        }
+    }
+    
+    # Log response size
     response_size = len(json.dumps(analysis))
     print(f"[MODAL] Response size: {response_size / 1024:.1f} KB")
     
@@ -315,113 +539,61 @@ def full_analysis_modal(url: str, max_reviews: int = 100) -> Dict[str, Any]:
 
 
 # ============================================================================
-# FASTAPI APP (serves both analysis and MCP)
+# FASTAPI APP - Updated to use parallel function
 # ============================================================================
 
 @app.function(
     image=image, 
     secrets=[modal.Secret.from_name("anthropic-api-key")],
-    timeout=2400,
+    timeout=900,  # 15 min timeout for the API endpoint
 )
 @modal.asgi_app()
 def fastapi_app():
-    """Main API with MCP integration."""
+    """Main API - uses parallel processing for speed."""
     from fastapi import FastAPI, HTTPException
     from pydantic import BaseModel
     
-    web_app = FastAPI(title="Restaurant Intelligence API with MCP")
+    web_app = FastAPI(title="Restaurant Intelligence API - PARALLEL OPTIMIZED")
     
     class AnalyzeRequest(BaseModel):
         url: str
         max_reviews: int = 100
     
-    class MCPCallRequest(BaseModel):
-        tool_name: str
-        arguments: Dict[str, Any] = {}
-    
     @web_app.get("/")
     async def root():
         return {
             "name": "Restaurant Intelligence API",
-            "version": "3.0",
-            "mcp": "enabled",
-            "optimizations": ["slim_trend_data", "pre_calculated_sentiment"],
-            "endpoints": {
-                "analyze": "/analyze",
-                "mcp_tools": "/mcp/call",
-                "mcp_list": "/mcp/tools"
-            }
+            "version": "3.0-parallel",
+            "optimizations": ["parallel_batches", "parallel_insights", "larger_batch_size"],
+            "target": "1000 reviews in ~5 minutes"
         }
     
     @web_app.get("/health")
     async def health():
-        return {"status": "healthy", "mcp": "enabled"}
+        return {"status": "healthy", "version": "parallel"}
     
     @web_app.post("/analyze")
     async def analyze(request: AnalyzeRequest):
         try:
-            result = full_analysis_modal.remote(url=request.url, max_reviews=request.max_reviews)
+            result = full_analysis_parallel.remote(url=request.url, max_reviews=request.max_reviews)
             return result
         except Exception as e:
+            import traceback
+            traceback.print_exc()
             raise HTTPException(status_code=500, detail=str(e))
     
-    # MCP Endpoints
-    @web_app.get("/mcp/tools")
-    async def mcp_list_tools():
-        return {
-            "tools": [
-                {"name": "index_reviews", "description": "Index reviews for RAG Q&A"},
-                {"name": "query_reviews", "description": "Answer questions about reviews"},
-                {"name": "save_report", "description": "Save analysis report"},
-            ]
-        }
-    
-    @web_app.post("/mcp/call")
-    async def mcp_call(request: MCPCallRequest):
-        """TRUE MCP interface."""
-        # For now, this delegates to local functions
-        if request.tool_name == "index_reviews":
-            args = request.arguments
-            REVIEW_INDEX[args["restaurant_name"]] = args["reviews"]
-            return {"success": True, "indexed": len(args["reviews"])}
-        
-        elif request.tool_name == "query_reviews":
-            args = request.arguments
-            reviews = REVIEW_INDEX.get(args["restaurant_name"], [])
-            if not reviews:
-                return {"success": False, "error": "No reviews indexed"}
-            
-            question_words = set(args["question"].lower().split())
-            scored = [(len(question_words & set(r.lower().split())), r) for r in reviews]
-            scored.sort(reverse=True, key=lambda x: x[0])
-            top_k = args.get("top_k", 5)
-            
-            return {
-                "success": True,
-                "relevant_reviews": [r[1] for r in scored[:top_k]]
-            }
-        
-        return {"success": False, "error": f"Unknown tool: {request.tool_name}"}
-    
     return web_app
 
 
+# ============================================================================
+# LOCAL ENTRYPOINT FOR TESTING
+# ============================================================================
+
 @app.local_entrypoint()
 def main():
-    print("🧪 Testing Modal deployment with MCP...\n")
-    
-    print("1️⃣ Testing connection...")
-    result = hello.remote()
-    print(f"✅ {result}\n")
-    
-    print("2️⃣ MCP Server deployed at:")
-    print("   https://tushar-pingle--restaurant-intelligence-mcp-server.modal.run")
+    print("🧪 Testing PARALLEL Modal deployment...\n")
     
-    print("\n3️⃣ Analysis API deployed at:")
+    print("1️⃣ API will be deployed at:")
     print("   https://tushar-pingle--restaurant-intelligence-fastapi-app.modal.run")
     
-    print("\n✅ Both endpoints ready!")
-    print("\n📊 Optimizations enabled:")
-    print("   - Slim trend_data (no full review text)")
-    print("   - Pre-calculated sentiment in backend")
-    print("   - ~97% smaller response size")
+    print("\n✅ Deploy with: modal deploy modal_backend.py")

src/data_processing/review_cleaner.py

@@ -1,113 +1,151 @@
 """
-Review Text Cleaner
-Sanitizes review text before sending to AI to prevent JSON parsing errors.
+Review Text Cleaner - FIXED VERSION
+Less aggressive cleaning that preserves more reviews.
+
+FIXES:
+1. Don't discard reviews just because they're short
+2. Keep reviews with minimal cleaning
+3. Better handling of special characters
+4. Log what's being cleaned for debugging
+
+Author: Tushar Pingle
+Updated: Nov 2024
 """
 
 import re
 import unicodedata
-from typing import List
+from typing import List, Tuple
 
 
 class ReviewCleaner:
     """
-    Cleans review text to prevent JSON parsing errors and reduce tokens.
+    Cleans review text while preserving as much content as possible.
     """
     
-    def __init__(self):
-        pass
+    # Minimum length for a valid review (characters)
+    MIN_REVIEW_LENGTH = 10  # Very permissive - was effectively 0 but cleaned to nothing
+    
+    def __init__(self, verbose: bool = False):
+        self.verbose = verbose
+        self.stats = {
+            'total': 0,
+            'kept': 0,
+            'removed_empty': 0,
+            'removed_short': 0,
+            'chars_original': 0,
+            'chars_cleaned': 0
+        }
     
     def clean_review(self, text: str) -> str:
         """
         Clean a single review text.
         
-        Args:
-            text: Raw review text
-            
-        Returns:
-            Cleaned text safe for AI processing
+        FIXED: Less aggressive cleaning, preserves more content.
         """
         if not text or not isinstance(text, str):
             return ""
         
-        # 1. Remove excessive whitespace
+        original_len = len(text)
+        
+        # 1. Basic whitespace normalization (gentle)
         text = ' '.join(text.split())
         
-        # 2. Remove emojis and special unicode
+        # 2. Remove only truly problematic emojis (keep basic punctuation)
         text = self._remove_emojis(text)
         
-        # 3. Fix quotes - replace smart quotes with straight quotes
+        # 3. Normalize quotes (don't remove them)
         text = text.replace('"', '"').replace('"', '"')
         text = text.replace("'", "'").replace("'", "'")
         
-        # 4. Remove or escape problematic characters
-        text = text.replace('\n', ' ')  # Remove newlines
-        text = text.replace('\r', ' ')  # Remove carriage returns
-        text = text.replace('\t', ' ')  # Remove tabs
+        # 4. Remove control characters only (keep newlines as spaces)
+        text = text.replace('\n', ' ').replace('\r', ' ').replace('\t', ' ')
+        text = ''.join(char for char in text if unicodedata.category(char)[0] != 'C' or char == ' ')
         
-        # 5. Remove control characters
-        text = ''.join(char for char in text if unicodedata.category(char)[0] != 'C')
-        
-        # 6. Normalize multiple spaces
+        # 5. Normalize multiple spaces
         text = re.sub(r'\s+', ' ', text)
         
-        # 7. Truncate very long reviews (>1000 chars)
-        if len(text) > 1000:
-            text = text[:997] + "..."
+        # 6. Truncate very long reviews (>1500 chars) - increased limit
+        if len(text) > 1500:
+            text = text[:1497] + "..."
         
-        # 8. Strip leading/trailing whitespace
+        # 7. Strip whitespace
         text = text.strip()
         
+        # Track stats
+        self.stats['chars_original'] += original_len
+        self.stats['chars_cleaned'] += len(text)
+        
         return text
     
     def _remove_emojis(self, text: str) -> str:
-        """Remove emojis and other pictographic characters."""
-        # Emoji pattern
+        """
+        Remove emojis but keep more unicode characters.
+        FIXED: Less aggressive pattern.
+        """
+        # Only remove actual emoji pictographs, not all unicode
         emoji_pattern = re.compile(
             "["
             "\U0001F600-\U0001F64F"  # emoticons
-            "\U0001F300-\U0001F5FF"  # symbols & pictographs
+            "\U0001F300-\U0001F5FF"  # symbols & pictographs  
             "\U0001F680-\U0001F6FF"  # transport & map symbols
             "\U0001F1E0-\U0001F1FF"  # flags
-            "\U00002702-\U000027B0"
-            "\U000024C2-\U0001F251"
+            "\U0001F900-\U0001F9FF"  # supplemental symbols
+            "\U0001FA00-\U0001FA6F"  # chess symbols
+            "\U0001FA70-\U0001FAFF"  # symbols extended
+            "\U00002702-\U000027B0"  # dingbats
             "]+",
             flags=re.UNICODE
         )
-        return emoji_pattern.sub(r'', text)
+        return emoji_pattern.sub('', text)
     
     def clean_reviews(self, reviews: List[str]) -> List[str]:
         """
         Clean a list of reviews.
         
-        Args:
-            reviews: List of raw review texts
-            
-        Returns:
-            List of cleaned review texts
+        FIXED: Only removes truly empty reviews, not short ones.
         """
+        self.stats = {
+            'total': len(reviews),
+            'kept': 0,
+            'removed_empty': 0,
+            'removed_short': 0,
+            'chars_original': 0,
+            'chars_cleaned': 0
+        }
+        
         cleaned = []
         for i, review in enumerate(reviews):
+            # Clean the review
             cleaned_text = self.clean_review(review)
-            if cleaned_text:  # Only include non-empty reviews
-                cleaned.append(cleaned_text)
-            else:
-                print(f"   ⚠️  Review {i} became empty after cleaning, skipping")
+            
+            # Check if it's still valid
+            if not cleaned_text:
+                self.stats['removed_empty'] += 1
+                if self.verbose:
+                    print(f"   ⚠️  Review {i} was empty/None, skipping")
+                continue
+            
+            if len(cleaned_text) < self.MIN_REVIEW_LENGTH:
+                self.stats['removed_short'] += 1
+                if self.verbose:
+                    print(f"   ⚠️  Review {i} too short ({len(cleaned_text)} chars): '{cleaned_text[:50]}'")
+                continue
+            
+            cleaned.append(cleaned_text)
+            self.stats['kept'] += 1
         
         return cleaned
     
-    def get_cleaning_stats(self, original: List[str], cleaned: List[str]) -> dict:
+    def get_cleaning_stats(self) -> dict:
         """Get statistics about the cleaning process."""
-        original_chars = sum(len(r) for r in original)
-        cleaned_chars = sum(len(r) for r in cleaned)
-        
         return {
-            "original_count": len(original),
-            "cleaned_count": len(cleaned),
-            "removed_count": len(original) - len(cleaned),
-            "original_chars": original_chars,
-            "cleaned_chars": cleaned_chars,
-            "chars_saved": original_chars - cleaned_chars,
-            "reduction_pct": round((1 - cleaned_chars / original_chars) * 100, 1) if original_chars > 0 else 0
+            "original_count": self.stats['total'],
+            "cleaned_count": self.stats['kept'],
+            "removed_empty": self.stats['removed_empty'],
+            "removed_short": self.stats['removed_short'],
+            "original_chars": self.stats['chars_original'],
+            "cleaned_chars": self.stats['chars_cleaned'],
+            "retention_rate": round(self.stats['kept'] / max(self.stats['total'], 1) * 100, 1)
         }
 
 
@@ -115,39 +153,85 @@ def clean_reviews_for_ai(reviews: List[str], verbose: bool = True) -> List[str]:
     """
     Convenience function to clean reviews.
     
-    Args:
-        reviews: Raw review texts
-        verbose: Print cleaning stats
-        
-    Returns:
-        Cleaned review texts
+    FIXED: Better stats reporting, less aggressive cleaning.
     """
-    cleaner = ReviewCleaner()
+    cleaner = ReviewCleaner(verbose=False)  # Don't spam individual messages
     cleaned = cleaner.clean_reviews(reviews)
     
     if verbose:
-        stats = cleaner.get_cleaning_stats(reviews, cleaned)
+        stats = cleaner.get_cleaning_stats()
         print(f"🧹 Cleaned {stats['original_count']} reviews:")
-        print(f"   Removed: {stats['removed_count']} empty reviews")
-        print(f"   Characters: {stats['original_chars']:,} → {stats['cleaned_chars']:,}")
-        print(f"   Saved: {stats['chars_saved']:,} chars ({stats['reduction_pct']}% reduction)")
+        print(f"   ✅ Kept: {stats['cleaned_count']} ({stats['retention_rate']}%)")
+        if stats['removed_empty'] > 0:
+            print(f"   ❌ Empty: {stats['removed_empty']}")
+        if stats['removed_short'] > 0:
+            print(f"   ❌ Too short: {stats['removed_short']}")
+        
+        # Warn if we're losing too many reviews
+        if stats['retention_rate'] < 50:
+            print(f"   ⚠️  WARNING: Only {stats['retention_rate']}% retention! Check scraper.")
     
     return cleaned
 
 
+# Also add a debug function
+def analyze_review_loss(reviews: List[str]) -> None:
+    """
+    Debug function to understand why reviews are being lost.
+    """
+    print(f"\n{'='*60}")
+    print("REVIEW LOSS ANALYSIS")
+    print(f"{'='*60}\n")
+    
+    empty_count = 0
+    short_count = 0
+    valid_count = 0
+    
+    print("Sample of problematic reviews:\n")
+    
+    for i, review in enumerate(reviews):
+        if not review or not isinstance(review, str):
+            empty_count += 1
+            if empty_count <= 3:
+                print(f"  [{i}] EMPTY: {repr(review)}")
+        elif len(review.strip()) < 10:
+            short_count += 1
+            if short_count <= 3:
+                print(f"  [{i}] SHORT ({len(review)} chars): '{review[:50]}'")
+        else:
+            valid_count += 1
+    
+    print(f"\n{'='*60}")
+    print(f"SUMMARY:")
+    print(f"  Total: {len(reviews)}")
+    print(f"  Valid: {valid_count} ({valid_count/len(reviews)*100:.1f}%)")
+    print(f"  Empty: {empty_count}")
+    print(f"  Short: {short_count}")
+    print(f"{'='*60}\n")
+
+
 if __name__ == "__main__":
     # Test the cleaner
     test_reviews = [
-        'This place is "amazing"! ��😍😍',
-        "The food was great\n\nbut service was slow",
-        'Chef said "it\'s the best" and I agree! \t\t\t',
-        "🍕🍝🍷 Loved everything!!!",
-        "A" * 1500  # Very long review
+        'This place is "amazing"! 😍😍😍 The food was incredible.',
+        "The food was great\n\nbut service was slow. Would come back!",
+        'Chef said "it\'s the best" and I agree! Great experience.',
+        "🍕🍝🍷 Loved everything!!! Best Italian in town.",
+        "",  # Empty
+        "Good",  # Too short
+        "   ",  # Just whitespace
+        None,  # None
+        "The pasta was perfectly cooked, al dente just how I like it.",
     ]
     
-    cleaner = ReviewCleaner()
-    for i, review in enumerate(test_reviews):
-        cleaned = cleaner.clean_review(review)
-        print(f"Original {i+1}: {review[:50]}...")
-        print(f"Cleaned {i+1}:  {cleaned[:50]}...")
-        print()
+    print("Testing review cleaner...\n")
+    
+    # First analyze
+    analyze_review_loss(test_reviews)
+    
+    # Then clean
+    cleaned = clean_reviews_for_ai(test_reviews, verbose=True)
+    
+    print(f"\nCleaned reviews ({len(cleaned)}):")
+    for i, review in enumerate(cleaned):
+        print(f"  {i+1}. {review[:60]}...")

src/scrapers/opentable_scraper.py

@@ -1,33 +1,42 @@
 """
-OpenTable Review Scraper - 2025 Production Version
-Fixed review text extraction based on actual DOM structure.
+OpenTable Scraper - FIXED VERSION
+Production-ready scraper that doesn't lose reviews.
+
+FIXES:
+1. Only counts reviews that have actual text
+2. Better selector specificity
+3. Logs empty vs real reviews for debugging
+4. Continues even if individual reviews fail
+
+Author: Tushar Pingle
+Updated: Nov 2024
 """
 
 import time
-from typing import List, Dict, Any, Optional, Callable
+from typing import Dict, Any, List, Optional, Callable
 from selenium import webdriver
-from selenium.common.exceptions import (
-    NoSuchElementException,
-    StaleElementReferenceException,
-    TimeoutException
-)
+from selenium.webdriver.chrome.options import Options
+from selenium.webdriver.chrome.service import Service
 from selenium.webdriver.common.by import By
 from selenium.webdriver.support.ui import WebDriverWait
 from selenium.webdriver.support import expected_conditions as EC
-from selenium.webdriver.chrome.options import Options
-from selenium.webdriver.chrome.service import Service
+from selenium.common.exceptions import TimeoutException, StaleElementReferenceException
 
 
 class OpenTableScraper:
-    """Scrapes restaurant reviews from OpenTable using production-tested selectors."""
+    """
+    Production OpenTable scraper with improved review extraction.
+    """
     
-    # Production selectors discovered from live DOM inspection (Jan 2025)
+    # Updated selectors - more specific for actual review cards
     SELECTORS = {
         "review_cards": [
+            # Most specific first - only match list items that contain actual review content
             "//li[@data-test='reviews-list-item']",
-            "//section[@id='reviews']//li[contains(., 'Dined')]",
-            "//section[.//h2[contains(., 'people are saying') or contains(., 'Reviews')]]//li[.//p or .//span or .//time]",
-            "//li[@data-test='review']"
+            # Fallback: items in reviews section that have both date AND substantial text
+            "//section[@id='reviews']//li[contains(., 'Dined') and .//span[string-length(normalize-space()) > 30]]",
+            # Generic fallback
+            "//section[.//h2[contains(., 'people are saying') or contains(., 'Reviews')]]//li[.//p[string-length(normalize-space()) > 30] or .//span[string-length(normalize-space()) > 30]]",
         ],
         "next_button": [
             "//a[@aria-label='Go to the next page']",
@@ -65,15 +74,16 @@ class OpenTableScraper:
             ".//li[contains(., 'Ambience')]//span"
         ],
         "review_text": [
-            # NEW: Based on actual DOM structure from screenshot
+            # Priority order - most specific first
             ".//span[@data-test='wrapper-tag']",
             ".//div[@data-test='wrapper-tag']",
             ".//p[@data-test='review-text']",
-            ".//div[contains(@class,'review')]/p",
-            ".//div[contains(@class,'review')]/span",
-            # Fallback: Get any paragraph/span with substantial text, excluding ratings/dates
-            ".//p[not(contains(., 'Dined')) and not(.//*) and string-length(normalize-space())>20]",
-            ".//span[not(contains(., 'Dined')) and not(.//*) and string-length(normalize-space())>20]"
+            # Get text content from review body
+            ".//div[contains(@class,'review')]//p[string-length(normalize-space()) > 20]",
+            ".//div[contains(@class,'review')]//span[string-length(normalize-space()) > 20]",
+            # Fallback: any paragraph/span with substantial text that's not date/rating
+            ".//p[not(contains(., 'Dined')) and not(contains(., 'Overall')) and not(contains(., 'Food')) and not(contains(., 'Service')) and not(contains(., 'Ambience')) and string-length(normalize-space()) > 20]",
+            ".//span[not(contains(., 'Dined')) and not(ancestor::li[contains(., 'Overall')]) and string-length(normalize-space()) > 20]",
         ]
     }
     
@@ -82,6 +92,7 @@ class OpenTableScraper:
         self.page_load_strategy = page_load_strategy
         self.driver = None
         self.wait = None
+        self.empty_count = 0  # Track empty reviews for debugging
     
     def scrape_reviews(
         self,
@@ -89,8 +100,11 @@ class OpenTableScraper:
         max_reviews: Optional[int] = None,
         progress_callback: Optional[Callable[[str], None]] = None
     ) -> Dict[str, Any]:
-        """Scrape reviews from OpenTable restaurant page."""
+        """
+        Scrape reviews from OpenTable restaurant page.
         
+        FIXED: Only counts and returns reviews that have actual text content.
+        """
         if not self._validate_url(url):
             return {'success': False, 'error': 'Invalid OpenTable URL', 'reviews': []}
         
@@ -116,7 +130,8 @@ class OpenTableScraper:
             reviews = []
             
             page_count = 0
-            review_count = 0
+            review_count = 0  # Only counts VALID reviews with text
+            self.empty_count = 0  # Track skipped empty reviews
             
             while True:
                 page_count += 1
@@ -132,32 +147,44 @@ class OpenTableScraper:
                     self._log_progress("⚠️  No reviews found on page.", progress_callback)
                     if page_count == 1:
                         # Save page source for debugging
-                        with open('debug_page_source.html', 'w', encoding='utf-8') as f:
-                            f.write(self.driver.page_source)
-                        self._log_progress("💾 Saved page source to debug_page_source.html", progress_callback)
+                        try:
+                            with open('debug_page_source.html', 'w', encoding='utf-8') as f:
+                                f.write(self.driver.page_source)
+                            self._log_progress("💾 Saved page source to debug_page_source.html", progress_callback)
+                        except:
+                            pass
                     break
                 
-                self._log_progress(f"✅ Found {len(review_elements)} review cards", progress_callback)
+                self._log_progress(f"📋 Found {len(review_elements)} review cards on page", progress_callback)
                 
                 # Extract data from each review
+                page_valid = 0
+                page_empty = 0
+                
                 for idx, review in enumerate(review_elements):
                     if max_reviews and review_count >= max_reviews:
                         self._log_progress(f"🎯 Reached max reviews ({max_reviews}).", progress_callback)
                         break
                     
                     try:
+                        # Extract review text FIRST - this is the critical field
+                        review_text = self._extract_review_text(review)
+                        
+                        # FIXED: Skip reviews without actual text content
+                        if not review_text or len(review_text.strip()) < 10:
+                            page_empty += 1
+                            self.empty_count += 1
+                            continue  # Don't append, don't count
+                        
+                        # Now extract other fields
                         name = self._extract_text_with_fallback(review, self.SELECTORS["name"])
                         date = self._extract_text_with_fallback(review, self.SELECTORS["date"])
                         overall_rating = self._extract_text_with_fallback(review, self.SELECTORS["overall_rating"])
                         food_rating = self._extract_text_with_fallback(review, self.SELECTORS["food_rating"])
                         service_rating = self._extract_text_with_fallback(review, self.SELECTORS["service_rating"])
                         ambience_rating = self._extract_text_with_fallback(review, self.SELECTORS["ambience_rating"])
-                        review_text = self._extract_text_with_fallback(review, self.SELECTORS["review_text"])
-                        
-                        # Clean review text (remove date if it leaked in)
-                        if review_text and "Dined on" in review_text:
-                            review_text = ""
                         
+                        # Append valid review
                         names.append(name)
                         dates.append(date)
                         overall_ratings.append(overall_rating)
@@ -167,14 +194,18 @@ class OpenTableScraper:
                         reviews.append(review_text)
                         
                         review_count += 1
+                        page_valid += 1
                         
-                        if review_count % 10 == 0:
-                            self._log_progress(f"📊 Extracted {review_count} reviews so far...", progress_callback)
+                        if review_count % 50 == 0:
+                            self._log_progress(f"📊 Extracted {review_count} valid reviews so far...", progress_callback)
                         
                     except Exception as e:
                         self._log_progress(f"⚠️  Error on review {idx + 1}: {str(e)}", progress_callback)
                         continue
                 
+                # Log page summary
+                self._log_progress(f"   ✅ Page {page_count}: {page_valid} valid, {page_empty} empty", progress_callback)
+                
                 if max_reviews and review_count >= max_reviews:
                     break
                 
@@ -185,64 +216,130 @@ class OpenTableScraper:
                 
                 time.sleep(3)  # Wait for new page to load
             
-            self._log_progress(f"✅ DONE! Scraped {review_count} reviews from {page_count} pages", progress_callback)
+            self._log_progress(f"✅ DONE! Scraped {review_count} valid reviews from {page_count} pages", progress_callback)
+            if self.empty_count > 0:
+                self._log_progress(f"   ℹ️  Skipped {self.empty_count} empty/invalid review cards", progress_callback)
+            
+            # Extract restaurant metadata
+            metadata = self._extract_metadata()
             
             return {
                 'success': True,
-                'total_reviews': review_count,
-                'total_pages': page_count,
-                'reviews': {
-                    'names': names,
-                    'dates': dates,
-                    'overall_ratings': overall_ratings,
-                    'food_ratings': food_ratings,
-                    'service_ratings': service_ratings,
-                    'ambience_ratings': ambience_ratings,
-                    'review_texts': reviews
+                'total_reviews': review_count,  # Now correctly represents VALID reviews
+                'names': names,
+                'dates': dates,
+                'overall_ratings': overall_ratings,
+                'food_ratings': food_ratings,
+                'service_ratings': service_ratings,
+                'ambience_ratings': ambience_ratings,
+                'reviews': reviews,
+                'metadata': metadata,
+                'stats': {
+                    'pages_scraped': page_count,
+                    'valid_reviews': review_count,
+                    'empty_skipped': self.empty_count
                 }
             }
             
         except Exception as e:
-            self._log_progress(f"❌ Fatal error: {str(e)}", progress_callback)
-            return {'success': False, 'error': str(e), 'reviews': []}
-        
+            import traceback
+            error_msg = f"Scraping error: {str(e)}\n{traceback.format_exc()}"
+            self._log_progress(f"❌ {error_msg}", progress_callback)
+            return {'success': False, 'error': error_msg, 'reviews': []}
         finally:
             self._cleanup()
     
-    def _click_next(self) -> bool:
-        """Click 'Next' button with robust error handling."""
-        xpaths = self.SELECTORS["next_button"]
+    def _extract_review_text(self, review_element) -> str:
+        """
+        Extract review text with multiple fallback strategies.
+        Returns empty string if no valid text found.
+        """
+        # Try each selector
+        for selector in self.SELECTORS["review_text"]:
+            try:
+                elements = review_element.find_elements(By.XPATH, selector)
+                for elem in elements:
+                    text = elem.text.strip()
+                    # Validate it's actual review content
+                    if text and len(text) > 20:
+                        # Filter out dates and ratings that might have leaked
+                        if "Dined on" in text or text.startswith("Overall") or text.startswith("Food"):
+                            continue
+                        # Filter out very short generic text
+                        if text in ["See more", "Read more", "Show more"]:
+                            continue
+                        return text
+            except:
+                continue
         
-        for xp in xpaths:
+        # Last resort: try to get all text from the review card and extract the main content
+        try:
+            full_text = review_element.text
+            # Split by newlines and find the longest substantial text
+            lines = [line.strip() for line in full_text.split('\n') if line.strip()]
+            # Filter out dates, ratings, names
+            content_lines = []
+            for line in lines:
+                if len(line) > 30:  # Substantial text
+                    if not any(skip in line for skip in ['Dined on', 'Overall', 'Food', 'Service', 'Ambience', 'VIP']):
+                        content_lines.append(line)
+            
+            if content_lines:
+                # Return the longest line as the review
+                return max(content_lines, key=len)
+        except:
+            pass
+        
+        return ""
+    
+    def _extract_text_with_fallback(self, parent_element, selectors: List[str]) -> str:
+        """Extract text using fallback XPath selectors."""
+        for selector in selectors:
             try:
-                # Wait until present & visible
-                btn = self.wait.until(EC.presence_of_element_located((By.XPATH, xp)))
-                
-                # If we matched the inner <div data-test="pagination-next">, climb to <a>
-                if btn.tag_name.lower() != "a":
-                    try:
-                        btn = btn.find_element(By.XPATH, "ancestor::a[1]")
-                    except Exception:
-                        pass
+                element = parent_element.find_element(By.XPATH, selector)
+                text = element.text.strip()
+                if text:
+                    return text
+            except:
+                continue
+        return ""
+    
+    def _find_elements_with_fallback(self, selectors: List[str], by: By) -> List:
+        """Try multiple selectors until one works."""
+        for selector in selectors:
+            try:
+                elements = self.driver.find_elements(by, selector)
+                if elements:
+                    return elements
+            except:
+                continue
+        return []
+    
+    def _click_next(self) -> bool:
+        """Click the next page button."""
+        for xp in self.SELECTORS["next_button"]:
+            try:
+                btn = WebDriverWait(self.driver, 3).until(
+                    EC.presence_of_element_located((By.XPATH, xp))
+                )
                 
-                # Guard: disabled?
+                # Check if disabled
                 aria_disabled = (btn.get_attribute("aria-disabled") or "").lower()
                 if aria_disabled in ("true", "1"):
                     return False
                 
-                # Bring into view
+                # Scroll into view
                 try:
                     self.driver.execute_script("arguments[0].scrollIntoView({block:'center'});", btn)
                     time.sleep(0.15)
-                except Exception:
+                except:
                     pass
                 
-                # Try native click first
+                # Try clicking
                 try:
                     WebDriverWait(self.driver, 5).until(EC.element_to_be_clickable((By.XPATH, xp)))
                     btn.click()
-                except Exception:
-                    # JS fallback (needed in headless mode)
+                except:
                     self.driver.execute_script("arguments[0].click();", btn)
                 
                 return True
@@ -250,55 +347,59 @@ class OpenTableScraper:
             except TimeoutException:
                 continue
             except StaleElementReferenceException:
-                # Re-find once
                 try:
                     btn = self.driver.find_element(By.XPATH, xp)
                     self.driver.execute_script("arguments[0].scrollIntoView({block:'center'});", btn)
                     self.driver.execute_script("arguments[0].click();", btn)
                     return True
-                except Exception:
+                except:
                     continue
-            except Exception:
+            except:
                 continue
         
         return False
     
-    def _find_elements_with_fallback(self, selectors: List[str], by: By) -> List:
-        """Try multiple selectors until one works."""
-        for selector in selectors:
-            try:
-                elements = self.driver.find_elements(by, selector)
-                if elements:
-                    return elements
-            except:
-                continue
-        return []
-    
-    def _find_element_with_fallback(self, selectors: List[str], by: By):
-        """Try multiple selectors until one works."""
-        for selector in selectors:
-            try:
-                element = self.driver.find_element(by, selector)
-                if element:
-                    return element
-            except:
-                continue
-        return None
-    
-    def _extract_text_with_fallback(self, parent_element, selectors: List[str]) -> str:
-        """Extract text using fallback XPath selectors."""
-        for selector in selectors:
-            try:
-                element = parent_element.find_element(By.XPATH, selector)
-                text = element.text.strip()
-                if text:
-                    return text
-            except:
-                continue
-        return ""
+    def _extract_metadata(self) -> Dict[str, Any]:
+        """Extract restaurant metadata from page."""
+        metadata = {}
+        try:
+            # Restaurant name
+            name_selectors = [
+                "//h1",
+                "//h1[@data-test='restaurant-name']",
+                "//div[contains(@class,'restaurant-name')]//h1"
+            ]
+            for sel in name_selectors:
+                try:
+                    elem = self.driver.find_element(By.XPATH, sel)
+                    if elem.text.strip():
+                        metadata['restaurant_name'] = elem.text.strip()
+                        break
+                except:
+                    continue
+            
+            # Cuisine type
+            cuisine_selectors = [
+                "//span[contains(@class,'cuisine')]",
+                "//p[contains(@class,'cuisine')]",
+                "//div[contains(@class,'cuisine')]"
+            ]
+            for sel in cuisine_selectors:
+                try:
+                    elem = self.driver.find_element(By.XPATH, sel)
+                    if elem.text.strip():
+                        metadata['cuisine'] = elem.text.strip()
+                        break
+                except:
+                    continue
+                    
+        except:
+            pass
+        
+        return metadata
     
     def _init_driver(self):
-        """Initialize Chrome WebDriver with production settings."""
+        """Initialize Chrome WebDriver."""
         chrome_options = Options()
         chrome_options.page_load_strategy = self.page_load_strategy
         
@@ -308,18 +409,13 @@ class OpenTableScraper:
             chrome_options.add_argument('--disable-dev-shm-usage')
             chrome_options.add_argument('--disable-gpu')
         
-        # Realistic user agent to avoid bot detection
         chrome_options.add_argument('--user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36')
-        
-        # Additional anti-detection measures
         chrome_options.add_experimental_option("excludeSwitches", ["enable-automation"])
         chrome_options.add_experimental_option('useAutomationExtension', False)
         
         service = Service('/usr/local/bin/chromedriver')
         self.driver = webdriver.Chrome(service=service, options=chrome_options)
         self.driver.set_page_load_timeout(30)
-        
-        # Initialize WebDriverWait
         self.wait = WebDriverWait(self.driver, 10)
     
     def _cleanup(self):
@@ -336,7 +432,7 @@ class OpenTableScraper:
         return 'opentable.c' in url.lower()
     
     def _log_progress(self, message: str, callback: Optional[Callable]):
-        """Log progress with emoji indicators."""
+        """Log progress."""
         print(message)
         if callback:
             callback(message)
@@ -347,49 +443,23 @@ class OpenTableScraper:
 
 def scrape_opentable(url: str, max_reviews: Optional[int] = None, headless: bool = True) -> Dict[str, Any]:
     """
-    Scrape reviews from OpenTable.
-    
-    Args:
-        url: OpenTable restaurant URL
-        max_reviews: Maximum number of reviews to scrape (None = all)
-        headless: Run browser in headless mode
+    Convenience function to scrape OpenTable reviews.
     
-    Returns:
-        Dict with 'success', 'total_reviews', 'total_pages', and 'reviews' data
+    FIXED: Only returns reviews with actual text content.
     """
     scraper = OpenTableScraper(headless=headless)
-    return scraper.scrape_reviews(url, max_reviews=max_reviews)
+    return scraper.scrape_reviews(url, max_reviews)
 
 
 if __name__ == "__main__":
-    print("=" * 80)
-    print("🍽️  OpenTable Review Scraper - Production Test (2025)")
-    print("=" * 80 + "\n")
-    
-    test_url = "https://www.opentable.ca/r/miku-restaurant-vancouver"
+    # Test the scraper
+    test_url = "https://www.opentable.ca/r/dockside-restaurant-vancouver-vancouver"
+    result = scrape_opentable(test_url, max_reviews=50)
     
-    print(f"🎯 Target: {test_url}")
-    print("📊 Limit: 20 reviews (test mode)")
-    print("🤖 Mode: HEADLESS\n")
-    
-    result = scrape_opentable(test_url, max_reviews=20, headless=True)
-    
-    print("\n" + "=" * 80)
-    if result['success']:
-        print("✅ SUCCESS!")
-        print(f"   📊 Total reviews scraped: {result['total_reviews']}")
-        print(f"   📄 Total pages visited: {result['total_pages']}")
-        
-        if result['total_reviews'] > 0:
-            print(f"\n   🔍 Sample (first review):")
-            print(f"   👤 Name: {result['reviews']['names'][0]}")
-            print(f"   📅 Date: {result['reviews']['dates'][0]}")
-            print(f"   ⭐ Overall: {result['reviews']['overall_ratings'][0]}")
-            print(f"   🍜 Food: {result['reviews']['food_ratings'][0]}")
-            print(f"   💁 Service: {result['reviews']['service_ratings'][0]}")
-            print(f"   🏮 Ambience: {result['reviews']['ambience_ratings'][0]}")
-            print(f"   💬 Review: {result['reviews']['review_texts'][0][:150]}...")
-    else:
-        print("❌ FAILED")
-        print(f"   Error: {result.get('error', 'Unknown error')}")
-    print("=" * 80)
+    print(f"\n{'='*60}")
+    print(f"Results:")
+    print(f"  Success: {result.get('success')}")
+    print(f"  Total valid reviews: {result.get('total_reviews')}")
+    if result.get('stats'):
+        print(f"  Empty skipped: {result['stats'].get('empty_skipped', 0)}")
+    print(f"{'='*60}")