content_analyzer/document_parser.py

import pickle
import hashlib
import logging
from pathlib import Path
from typing import List, Optional
from datetime import datetime, timedelta
from langchain_core.documents import Document
from langchain_text_splitters import RecursiveCharacterTextSplitter
from configuration.parameters import parameters
from configuration.definitions import MAX_TOTAL_SIZE, ALLOWED_TYPES
import concurrent.futures
from PIL import Image
import gc
from google.genai import types

logger = logging.getLogger(__name__)

def preprocess_image(image, max_dim=1000):
    """Downscale image to max_dim before OpenCV processing."""
    if max(image.size) > max_dim:
        ratio = max_dim / max(image.size)
        new_size = tuple(int(dim * ratio) for dim in image.size)
        return image.resize(new_size, Image.Resampling.LANCZOS)
    return image

def detect_chart_on_page(args):
    """
    Top-level function for parallel local chart detection (required for ProcessPoolExecutor).
    Returns the page number, the PIL image, and the detection result.
    """
    page_num, image = args
    from content_analyzer.visual_detector import LocalChartDetector
    # Downscale image before detection to save memory
    image = preprocess_image(image, max_dim=1000)
    detection_result = LocalChartDetector.detect_charts(image)
    return (page_num, image, detection_result)

def analyze_batch(batch_tuple):
    """
    Top-level function for parallel Gemini batch analysis (future-proof for process pools).
    """
    batch, batch_num, total_batches, gemini_client, file_path, parameters, stats = batch_tuple
    try:
        import logging
        logger = logging.getLogger(__name__)
        from PIL import Image
        from google.genai import types
        images = [Image.open(image_path) for _, image_path, _ in batch]
        prompt = f"""
Analyze the following {len(batch)} chart(s)/graph(s) in order.

For EACH chart, provide comprehensive analysis separated by the marker "---CHART N---".

For each chart include:
**Chart Type**: [line/bar/pie/bubble/scatter/etc]
**Title**: [chart title]
**X-axis**: [label and units]
**Y-axis**: [label and units]
**Data Points**: [extract ALL visible data with exact values]
**Legend**: [list all series/categories]
**Trends**: [key patterns, trends, insights]
**Key Values**: [maximum, minimum, significant values]
**Context**: [any annotations or notes]

Format exactly as:
---CHART 1---
[analysis]

---CHART 2---
[analysis]

---CHART 3---
[analysis]
"""
        # For batch analysis:
        chart_response = gemini_client.models.generate_content(
            model=parameters.CHART_VISION_MODEL,
            contents=[prompt] + images,
            config=types.GenerateContentConfig(
                max_output_tokens=parameters.CHART_MAX_TOKENS * len(batch)
            )
        )
        stats['batch_api_calls'] += 1
        response_text = chart_response.text
        parts = response_text.split('---CHART ')
        batch_docs = []
        for idx, (page_num, image_path, detection_result) in enumerate(batch):
            if idx + 1 < len(parts):
                analysis_text = parts[idx + 1]
                if '---CHART' in analysis_text:
                    analysis_text = analysis_text.split('---CHART')[0]
                lines = analysis_text.split('\n')
                if lines and '---' in lines[0]:
                    lines = lines[1:]
                analysis = '\n'.join(lines).strip()
            else:
                analysis = "Analysis unavailable (parsing error)"
            chart_types_str = ", ".join(detection_result['chart_types']) or "Unknown"
            confidence = detection_result['confidence']
            chart_doc = Document(
                page_content=f"""### 📊 Chart Analysis (Page {page_num})\n\n**Detection Method**: Hybrid (Local OpenCV + Gemini Batch Analysis)\n**Local Confidence**: {confidence:.0%}\n**Detected Types**: {chart_types_str}\n**Batch Size**: {len(batch)} charts analyzed together\n\n---\n\n{analysis}\n""",
                metadata={
                    "source": file_path,
                    "page": page_num,
                    "type": "chart",
                    "extraction_method": "hybrid_batch",
                    "detection_confidence": confidence,
                    "batch_size": len(batch)
                }
            )
            batch_docs.append(chart_doc)
            stats['charts_analyzed_gemini'] += 1
        for img in images:
            img.close()
        logger.info(f"✅ Batch {batch_num} complete ({len(batch)} charts analyzed)")
        return (batch_num - 1, batch_docs)
    except Exception as e:
        logger = logging.getLogger(__name__)
        logger.error(f"Batch analysis failed: {e}, trying sequential fallback...")
        return (batch_num - 1, [])

class DocumentProcessor:
    """
    Processes documents by splitting them into manageable chunks and caching
    the results to avoid reprocessing. Handles chart extraction using local
    OpenCV detection and Gemini Vision API with parallelization for speed.
    """
    # Cache metadata version - increment when cache format changes
    CACHE_VERSION = 4  # Incremented for chart extraction support

    def __init__(self):
        """Initialize the document processor with cache directory and splitter configuration."""
        self.cache_dir = Path(parameters.CACHE_DIR)
        self.cache_dir.mkdir(parents=True, exist_ok=True)
        self.splitter = RecursiveCharacterTextSplitter(
            chunk_size=parameters.CHUNK_SIZE,
            chunk_overlap=parameters.CHUNK_OVERLAP,
            length_function=len,
            is_separator_regex=False,
        )
        self.gemini_client = None
        self.genai_module = None  # Store the module reference
        # Instance-level flag instead of modifying global parameters
        self.chart_extraction_enabled = parameters.ENABLE_CHART_EXTRACTION
        if self.chart_extraction_enabled:
            self._init_gemini_vision()
        logger.debug(f"DocumentProcessor initialized with cache dir: {self.cache_dir}")
        logger.debug(f"Chunk size: {parameters.CHUNK_SIZE}, Chunk overlap: {parameters.CHUNK_OVERLAP}")
        logger.debug(f"Chart extraction: {'enabled' if self.chart_extraction_enabled else 'disabled'}")

    def _init_gemini_vision(self):
        """Initialize Gemini Vision client for chart analysis."""
        genai = None
        try:
            # Use the new google.genai package
            import google.genai as genai
            logger.debug("✅ Loaded google.genai (new package)")
        except ImportError as e:
            logger.warning(f"google-genai not installed: {e}")
            logger.info("Install with: pip install google-genai")
            self.chart_extraction_enabled = False  # Instance-level, not global
            return
        self.genai_module = genai
        try:
            from google import genai
            self.gemini_client = genai.Client(api_key=parameters.GOOGLE_API_KEY)
            logger.info(f"✅ Gemini Vision client initialized")
        except Exception as e:
            logger.error(f"❌ Failed to initialize Gemini Vision client: {e}")
            self.chart_extraction_enabled = False  # Instance-level, not global

    def validate_files(self, files: List) -> bool:
        """
        Validate that uploaded files meet size and type requirements.
        
        Args:
            files: List of uploaded file objects
            
        Returns:
            bool: True if all validations pass
            
        Raises:
            ValueError: If validation fails
        """
        if not files:
            raise ValueError("No files provided")
        
        total_size = 0
        for file in files:
            # Get file size
            if hasattr(file, 'size'):
                file_size = file.size
            else:
                # Fallback: read file to get size
                try:
                    with open(file.name, 'rb') as f:
                        file_size = len(f.read())
                except Exception as e:
                    logger.error(f"Failed to determine file size for {file.name}: {e}")
                    raise ValueError(f"Cannot read file: {file.name}")
            
            # Check individual file size
            if file_size > parameters.MAX_FILE_SIZE:
                raise ValueError(
                    f"File {file.name} exceeds maximum size "
                    f"({file_size / 1024 / 1024:.2f}MB > {parameters.MAX_FILE_SIZE / 1024 / 1024:.2f}MB)"
                )
            
            # Check file type
            file_ext = Path(file.name).suffix.lower()
            if file_ext not in ALLOWED_TYPES:
                raise ValueError(
                    f"File type {file_ext} not supported. Allowed types: {ALLOWED_TYPES}"
                )
            
            total_size += file_size
        
        # Check total size
        if total_size > parameters.MAX_TOTAL_SIZE:
            raise ValueError(
                f"Total file size exceeds maximum "
                f"({total_size / 1024 / 1024:.2f}MB > {parameters.MAX_TOTAL_SIZE / 1024 / 1024:.2f}MB)"
            )
        
        logger.info(f"Validation passed for {len(files)} files (total: {total_size / 1024 / 1024:.2f}MB)")
        return True
    
    def _generate_hash(self, content: bytes) -> str:
        """Generate SHA-256 hash of file content."""
        return hashlib.sha256(content).hexdigest()
    
    def _is_cache_valid(self, cache_path: Path) -> bool:
        """Check if a cache file exists and is still valid (not expired)."""
        if not cache_path.exists():
            logger.debug(f"Cache miss: {cache_path.name}")
            return False
        
        file_age = datetime.now() - datetime.fromtimestamp(cache_path.stat().st_mtime)
        if file_age > timedelta(days=parameters.CACHE_EXPIRE_DAYS):
            logger.info(f"Cache expired (age: {file_age.days} days): {cache_path.name}")
            cache_path.unlink()
            return False
        
        logger.debug(f"Cache hit: {cache_path.name} (age: {file_age.days} days)")
        return True
    
    def _load_from_cache(self, cache_path: Path) -> List:
        """Loads chunks from a pickle file, handling potential corruption."""
        try:
            with open(cache_path, "rb") as f:
                data = pickle.load(f)
            
            if "chunks" not in data or "timestamp" not in data:
                raise KeyError("Cache file missing 'chunks' or 'timestamp' key.")

            logger.info(f"Loaded {len(data['chunks'])} chunks from cache: {cache_path.name}")
            return data["chunks"]
        except (pickle.UnpicklingError, KeyError, EOFError) as e:
            logger.warning(f"Cache corruption detected in {cache_path.name}: {e}. Deleting cache.")
            cache_path.unlink()
            return []
        except Exception as e:
            logger.error(f"Unexpected error loading cache {cache_path.name}: {e}", exc_info=True)
            if cache_path.exists():
                cache_path.unlink()
            return []

    def _save_to_cache(self, chunks: List, cache_path: Path):
        """Saves chunks to a pickle file."""
        try:
            with open(cache_path, "wb") as f:
                pickle.dump({
                    "timestamp": datetime.now().timestamp(),
                    "chunks": chunks
                }, f)
            logger.info(f"Successfully cached {len(chunks)} chunks to {cache_path.name}")
        except Exception as e:
            logger.error(f"Failed to save cache to {cache_path.name}: {e}", exc_info=True)
    
    def _process_file(self, file) -> List[Document]:
        file_ext = Path(file.name).suffix.lower()
        if file_ext not in ALLOWED_TYPES:
            logger.warning(f"Skipping unsupported file type: {file.name}")
            return []
        try:
            documents = []
            if file_ext == '.pdf':
                import concurrent.futures
                results = {}
                def run_pdfplumber():
                    return self._load_pdf_with_pdfplumber(file.name)
                def run_charts():
                    logger.info(f"chart_extraction_enabled={self.chart_extraction_enabled}, gemini_client={self.gemini_client is not None}")
                    if self.chart_extraction_enabled and self.gemini_client:
                        return self._extract_charts_from_pdf(file.name)
                    return []
                try:
                    with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
                        future_pdf = executor.submit(run_pdfplumber)
                        future_charts = executor.submit(run_charts)
                        try:
                            docs = future_pdf.result()
                        except MemoryError as e:
                            logger.error(f"Out of memory in PDFPlumber thread: {e}. Falling back to sequential.")
                            docs = self._load_pdf_with_pdfplumber(file.name)
                        try:
                            chart_docs = future_charts.result()
                        except MemoryError as e:
                            logger.error(f"Out of memory in chart extraction thread: {e}. Falling back to sequential.")
                            chart_docs = self._extract_charts_from_pdf(file.name)
                        documents = docs or []
                        if chart_docs:
                            documents.extend(chart_docs)
                            logger.info(f"📊 Added {len(chart_docs)} chart descriptions to {file.name}")
                except MemoryError as e:
                    logger.error(f"Out of memory in parallel PDF processing: {e}. Falling back to sequential.")
                    documents = self._load_pdf_with_pdfplumber(file.name)
                    if self.chart_extraction_enabled and self.gemini_client:
                        chart_docs = self._extract_charts_from_pdf(file.name)
                        if chart_docs:
                            documents.extend(chart_docs)
                            logger.info(f"📊 Added {len(chart_docs)} chart descriptions to {file.name}")
            else:
                from langchain_community.document_loaders import (
                    Docx2txtLoader,
                    TextLoader,
                )
                loader_map = {
                    '.docx': Docx2txtLoader,
                    '.txt': TextLoader,
                    '.md': TextLoader,
                }
                loader_class = loader_map.get(file_ext)
                if not loader_class:
                    logger.warning(f"No loader found for {file_ext}")
                    return []
                logger.info(f"Loading {file_ext} file: {file.name}")
                loader = loader_class(file.name)
                documents = loader.load()
            if not documents:
                logger.warning(f"No content extracted from {file.name}")
                return []
            all_chunks = []
            total_docs = len(documents)
            # --- STABLE FILE HASHING ---
            with open(file.name, 'rb') as f:
                file_bytes = f.read()
            file_hash = self._generate_hash(file_bytes)  # Stable hash by file content
            stable_source = f"{Path(file.name).name}::{file_hash}"
            for i, doc in enumerate(documents):
                page_chunks = self.splitter.split_text(doc.page_content)
                total_chunks = len(page_chunks)
                for j, chunk in enumerate(page_chunks):
                    chunk_id = f"txt_{file_hash}_{doc.metadata.get('page', i + 1)}_{j}"
                    chunk_doc = Document(
                        page_content=chunk,
                        metadata={
                            "source": stable_source,                            
                            "page": doc.metadata.get("page", i + 1),
                            "type": doc.metadata.get("type", "text"),
                            "chunk_id": chunk_id
                        }
                    )
                    all_chunks.append(chunk_doc)
                 
            logger.info(f"Processed {file.name}: {len(documents)} page(s) → {len(all_chunks)} chunk(s)")
            return all_chunks
        except ImportError as e:
            logger.error(f"Required loader not installed for {file_ext}: {e}")
            return []
        except Exception as e:
            logger.error(f"Failed to process {file.name}: {e}", exc_info=True)
            raise

    def _extract_charts_from_pdf(self, file_path: str) -> List[Document]:
        """
        Extract and analyze charts/graphs from PDF with true batch processing and parallelism.
        PHASE 1: Parallel local chart detection (CPU-bound, uses ProcessPoolExecutor)
        PHASE 2: Parallel Gemini batch analysis (I/O-bound, uses ThreadPoolExecutor)
        """
        file_bytes = Path(file_path).read_bytes()
        file_hash = self._generate_hash(file_bytes)
        stable_source = f"{Path(file_path).name}::{file_hash}"
        def deduplicate_charts_by_title(chart_chunks):
            seen_titles = set()
            unique_chunks = []
            import re
            for chunk in chart_chunks:
                match = re.search(r"\*\*Title\*\*:\s*(.+)", chunk.page_content)
                title = match.group(1).strip() if match else None
                if title and title not in seen_titles:
                    seen_titles.add(title)
                    unique_chunks.append(chunk)
                elif not title:
                    unique_chunks.append(chunk)
            return unique_chunks
        try:
            from pdf2image import convert_from_path
            from PIL import Image
            import pdfplumber
            import tempfile
            import os
            
            # Import local detector if enabled
            use_local = parameters.CHART_USE_LOCAL_DETECTION
            if use_local:
                try:
                    from content_analyzer.visual_detector import LocalChartDetector
                    logger.info(f"📊 [BATCH MODE] Local detection → Temp cache → Batch analysis")
                except ImportError:
                    logger.warning("Local chart detector not available, falling back to Gemini")
                    use_local = False
            
            # Track statistics
            stats = {
                'pages_scanned': 0,
                'charts_detected_local': 0,
                'charts_analyzed_gemini': 0,
                'api_calls_saved': 0,
                'batch_api_calls': 0
            }
            
            # Get PDF page count
            with pdfplumber.open(file_path) as pdf:
                total_pages = len(pdf.pages)
            
            logger.info(f"Processing {total_pages} pages for chart detection...")
            
            # Create temp directory for chart images
            temp_dir = tempfile.mkdtemp(prefix='charts_')
            detected_charts = []  # [(page_num, image_path, detection_result), ...]
            
            try:
                # === PHASE 1: PARALLEL LOCAL CHART DETECTION (CPU-BOUND) ===
                logger.info("Phase 1: Detecting charts and caching to disk...")
                batch_size = parameters.CHART_BATCH_SIZE
                
                detected_charts = []
                if use_local and parameters.CHART_SKIP_GEMINI_DETECTION:
                    logger.info("Parallel local chart detection using ProcessPoolExecutor...")
                    # Use optimal worker count: min of CPU count or 4 to avoid memory issues
                    import os
                    max_workers = min(os.cpu_count() or 2, 4)
                    logger.info(f"Using {max_workers} workers for parallel chart detection")
                    
                    # MEMORY OPTIMIZATION: Process pages in streaming batches instead of loading all at once
                    # This reduces peak memory by 60-80% for large PDFs
                    detection_batch_size = 20  # Process 20 pages at a time to limit memory
                    
                    for batch_start in range(1, total_pages + 1, detection_batch_size):
                        batch_end = min(batch_start + detection_batch_size - 1, total_pages)
                        logger.debug(f"Processing detection batch: pages {batch_start}-{batch_end}")
                        
                        # Load only this batch of pages into memory
                        page_image_tuples = []
                        try:
                            images = convert_from_path(
                                file_path,
                                dpi=parameters.CHART_DPI,
                                first_page=batch_start,
                                last_page=batch_end,
                                fmt='jpeg',
                                jpegopt={'quality': 85, 'optimize': True}
                            )
                            for idx, image in enumerate(images):
                                page_num = batch_start + idx
                                stats['pages_scanned'] += 1
                                # Resize if needed
                                max_dimension = parameters.CHART_MAX_IMAGE_SIZE
                                if max(image.size) > max_dimension:
                                    ratio = max_dimension / max(image.size)
                                    new_size = tuple(int(dim * ratio) for dim in image.size)
                                    image = image.resize(new_size, Image.Resampling.LANCZOS)
                                page_image_tuples.append((page_num, image))
                            del images
                        except Exception as e:
                            logger.warning(f"Failed to process pages {batch_start}-{batch_end}: {e}")
                            continue
                        
                        # Process this batch with parallel detection
                        if page_image_tuples:
                            with concurrent.futures.ProcessPoolExecutor(max_workers=max_workers) as executor:
                                results = list(executor.map(detect_chart_on_page, page_image_tuples))
                            
                            # Process detection results and save charts to disk
                            for page_num, image, detection_result in results:
                                if not detection_result['has_chart']:
                                    logger.debug(f"Page {page_num}: No chart detected (skipping)")
                                    stats['api_calls_saved'] += 1
                                    continue
                                confidence = detection_result['confidence']
                                if confidence < parameters.CHART_MIN_CONFIDENCE:
                                    logger.debug(f"Page {page_num}: Low confidence ({confidence:.0%}), skipping")
                                    stats['api_calls_saved'] += 1
                                    continue
                                logger.info(f"📈 Chart detected on page {page_num} (confidence: {confidence:.0%})")
                                stats['charts_detected_local'] += 1
                                image_path = os.path.join(temp_dir, f'chart_page_{page_num}.jpg')
                                image.save(image_path, 'JPEG', quality=90)
                                detected_charts.append((page_num, image_path, detection_result))
                                # Release memory immediately
                                del image
                            
                            # Clean up batch memory
                            del page_image_tuples
                            del results
                            gc.collect()
                            logger.debug(f"Batch {batch_start}-{batch_end} complete, memory released")
                else:
                    # Fallback: sequential detection
                    for page_num, image in page_image_tuples:
                        if use_local and parameters.CHART_SKIP_GEMINI_DETECTION:
                            detection_result = LocalChartDetector.detect_charts(image)
                            if not detection_result['has_chart']:
                                logger.debug(f"Page {page_num}: No chart detected (skipping)")
                                stats['api_calls_saved'] += 1
                                continue
                            confidence = detection_result['confidence']
                            if confidence < parameters.CHART_MIN_CONFIDENCE:
                                logger.debug(f"Page {page_num}: Low confidence ({confidence:.0%}), skipping")
                                stats['api_calls_saved'] += 1
                                continue
                            logger.info(f"📈 Chart detected on page {page_num} (confidence: {confidence:.0%})")
                            stats['charts_detected_local'] += 1
                            image_path = os.path.join(temp_dir, f'chart_page_{page_num}.jpg')
                            image.save(image_path, 'JPEG', quality=90)
                            detected_charts.append((page_num, image_path, detection_result))

                logger.info(f"Phase 1 complete: {len(detected_charts)} charts detected and cached")
                
                # === PHASE 2: PARALLEL GEMINI BATCH ANALYSIS (I/O-BOUND) ===
                if not detected_charts or not self.gemini_client:
                    return []
                
                logger.info(f"Phase 2: Batch analyzing {len(detected_charts)} charts...")
                chart_documents = []
                
                if parameters.CHART_ENABLE_BATCH_ANALYSIS and len(detected_charts) > 1:
                    # Batch processing with parallel Gemini API calls
                    gemini_batch_size = parameters.CHART_GEMINI_BATCH_SIZE
                    batches = [detected_charts[i:i + gemini_batch_size] for i in range(0, len(detected_charts), gemini_batch_size)]

                    # Prepare batch tuples with batch_num and total_batches
                    batch_tuples = [
                        (batch, idx + 1, len(batches), self.gemini_client, file_path, parameters, stats)
                        for idx, batch in enumerate(batches)
                    ]
                    results = [None] * len(batches)
                    with concurrent.futures.ThreadPoolExecutor(max_workers=20) as executor:
                        future_to_idx = {executor.submit(analyze_batch, batch_tuple): idx for idx, batch_tuple in enumerate(batch_tuples)}
                        for future in concurrent.futures.as_completed(future_to_idx):
                            idx = future_to_idx[future]
                            try:
                                batch_idx, batch_docs = future.result()
                                results[batch_idx] = batch_docs
                            except Exception as exc:
                                logger.error(f"Batch {idx} generated an exception: {exc}")
                    # Flatten results and filter out None
                    chart_index = 0
                    for batch_docs in results:
                        if batch_docs:
                            for doc in batch_docs:
                                doc.metadata["chunk_id"] = f"{file_hash}_{doc.metadata.get('page', 0)}_{chart_index}"
                                chart_documents.append(doc)
                                chart_index += 1
                else:
                    # Sequential processing (batch disabled or single chart)
                    for chart_index, (page_num, image_path, detection_result) in enumerate(detected_charts):
                        try:
                            img = Image.open(image_path)
                            extraction_prompt = """Analyze this chart/graph in comprehensive detail:
                            **Chart Type**: [type]
                            **Title**: [title]
                            **Axes**: [X and Y labels/units]
                            **Data Points**: [extract all visible data]
                            **Legend**: [series/categories]
                            **Trends**: [key patterns and insights]
                            **Key Values**: [max, min, significant]
                            **Context**: [annotations or notes]
                            """
                            chart_response = self.gemini_client.models.generate_content(
                                model=parameters.CHART_VISION_MODEL,
                                contents=[extraction_prompt, img],
                                config=types.GenerateContentConfig(
                                    max_output_tokens=parameters.CHART_MAX_TOKENS
                                )
                            )
                            chart_types_str = ", ".join(detection_result['chart_types']) or "Unknown"
                            chart_doc = Document(
                                page_content=f"""### \U0001F4CA Chart Analysis (Page {page_num})\n\n**Detection Method**: Hybrid (Local OpenCV + Gemini Sequential)\n**Local Confidence**: {detection_result['confidence']:.0%}\n**Detected Types**: {chart_types_str}\n\n---\n\n{chart_response.text}\n""",
                                metadata={
                                    "source": file_path,
                                    "page": page_num,
                                    "type": "chart",
                                    "extraction_method": "hybrid_sequential",
                                    "chunk_id": f"{file_hash}_{page_num}_{chart_index}"
                                }
                            )
                            chart_documents.append(chart_doc)
                            stats['charts_analyzed_gemini'] += 1
                            img.close()
                            logger.info(f"✅ Analyzed chart on page {page_num}")
                        except Exception as e:
                            logger.error(f"Failed to analyze page {page_num}: {e}")
                
                # Log statistics
                if use_local and parameters.CHART_SKIP_GEMINI_DETECTION:
                    cost_saved = stats['api_calls_saved'] * 0.0125
                    actual_cost = stats['batch_api_calls'] * 0.0125 if stats['batch_api_calls'] > 0 else stats['charts_analyzed_gemini'] * 0.0125
                    
                    if stats['batch_api_calls'] > 0:
                        efficiency = stats['charts_analyzed_gemini'] / stats['batch_api_calls']
                    else:
                        efficiency = 1.0
                    
                    logger.info(f"""
📊 Chart Extraction Complete (HYBRID + BATCH MODE):
   Pages scanned: {stats['pages_scanned']}
   Charts detected (local): {stats['charts_detected_local']}
   Charts analyzed (Gemini): {stats['charts_analyzed_gemini']}
   Batch API calls: {stats['batch_api_calls']}
   Charts per API call: {efficiency:.1f}
   API calls saved (detection): {stats['api_calls_saved']}
   Estimated cost savings: ${cost_saved:.3f}
   Actual API cost: ${actual_cost:.3f}
""")
                
                # After chart_documents is created (batch or sequential), deduplicate by title:
                chart_documents = deduplicate_charts_by_title(chart_documents)
                
                return chart_documents
            
            finally:
                # Only clean up after all analysis is done
                try:
                    import shutil
                    shutil.rmtree(temp_dir)
                    logger.debug(f"Cleaned up temp directory: {temp_dir}")
                except Exception as e:
                    logger.warning(f"Failed to clean temp directory {temp_dir}: {e}")
        except ImportError as e:
            logger.warning(f"Dependencies missing for chart extraction: {e}")
            return []
        except MemoryError as e:
            logger.error(f"Out of memory while processing {file_path}. Try reducing DPI or batch size.")
            return []
        except Exception as e:
            logger.error(f"Chart extraction failed for {file_path}: {e}", exc_info=True)
            return []

    def _load_pdf_with_pdfplumber(self, file_path: str) -> List[Document]:
        """
        Load PDF using pdfplumber for text and table extraction.
        
        Uses multiple table detection strategies for complex tables.
        """
        import pdfplumber
        
        logger.info(f"[PDFPLUMBER] Processing: {file_path}")
        file_bytes = Path(file_path).read_bytes()
        file_hash = self._generate_hash(file_bytes)
        stable_source = f"{Path(file_path).name}::{file_hash}"
        
        # Strategy 1: Line-based (default) - for tables with visible borders
        default_parameters = {}
        
        # Strategy 2: Text-based - for borderless tables with aligned text
        text_parameters = {
            "vertical_strategy": "text",
            "horizontal_strategy": "text",
            "snap_tolerance": 5,
            "join_tolerance": 5,
            "edge_min_length": 3,
            "min_words_vertical": 2,
            "min_words_horizontal": 1,
            "text_tolerance": 3,
            "intersection_tolerance": 5,
        }
        
        # Strategy 3: Lines + text hybrid - for complex tables
        hybrid_parameters = {
            "vertical_strategy": "lines_strict",
            "horizontal_strategy": "text",
            "snap_tolerance": 5,
            "join_tolerance": 5,
            "min_words_horizontal": 1,
        }
        
        all_content = []
        total_tables = 0
        with pdfplumber.open(file_path) as pdf:
            for page_num, page in enumerate(pdf.pages, 1):
                page_content = [f"## Page {page_num}"]
                page_tables = []
                table_hashes = set()  # Track unique tables
                
                def add_table_if_unique(table, strategy_name):
                    """Add table if not already found."""
                    if not table or len(table) < 2:
                        return False
                    # Create hash of table content
                    table_str = str(table)
                    table_hash = hash(table_str)
                    if table_hash not in table_hashes:
                        table_hashes.add(table_hash)
                        page_tables.append((table, strategy_name))
                        return True
                    return False
                
                # --- Robust per-page error handling ---
                try:
                    # Strategy 1: Default line-based detection
                    try:
                        default_tables = page.extract_tables()
                        if default_tables:
                            for t in default_tables:
                                add_table_if_unique(t, "default")
                    except Exception as e:
                        logger.warning(f"Default strategy failed on page {page_num}: {e}")
                    # Strategy 2: Text-based detection for borderless tables
                    try:
                        text_tables = page.extract_tables(text_parameters)
                        if text_tables:
                            for t in text_tables:
                                add_table_if_unique(t, "text")
                    except Exception as e:
                        logger.warning(f"Text strategy failed on page {page_num}: {e}")
                    # Strategy 3: Hybrid detection
                    try:
                        hybrid_tables = page.extract_tables(hybrid_parameters)
                        if hybrid_tables:
                            for t in hybrid_tables:
                                add_table_if_unique(t, "hybrid")
                    except Exception as e:
                        logger.warning(f"Hybrid strategy failed on page {page_num}: {e}")
                    # Strategy 4: Use find_tables() for more control
                    try:
                        found_tables = page.find_tables(text_parameters)
                        if found_tables:
                            for ft in found_tables:
                                t = ft.extract()
                                add_table_if_unique(t, "find_tables")
                    except Exception as e:
                        logger.warning(f"find_tables() failed on page {page_num}: {e}")
                    
                    # Convert tables to markdown
                    for table, strategy in page_tables:
                        total_tables += 1
                        md_table = self._table_to_markdown(table, page_num, total_tables)
                        if md_table:
                            page_content.append(md_table)
                
                    # Extract text
                    try:
                        text = page.extract_text()
                        if text:
                            page_content.append(text.strip())
                    except Exception as e:
                        logger.warning(f"Text extraction failed on page {page_num}: {e}")
                    
                    if len(page_content) > 1:
                        combined = "\n\n".join(page_content)
                        chunk_id = f"txt_{file_hash}_{page_num}_0"
                        doc = Document(
                            page_content=combined,
                            metadata={
                                "source": stable_source,                                
                                "page": page_num,
                                "loader": "pdfplumber",
                                "tables_count": total_tables,
                                "type": "text",
                                "chunk_id": chunk_id
                            }
                        )
                        all_content.append(doc)
                except Exception as e:
                    logger.warning(f"Skipping page {page_num} due to error: {e}")
                    continue
        
        logger.info(f"[PDFPLUMBER] Extracted {len(all_content)} chunks, {total_tables} tables")
        return all_content
    
    def _table_to_markdown(self, table: List[List], page_num: int, table_idx: int) -> str:
        """Convert a table (list of rows) to markdown format."""
        if not table or len(table) < 1:
            return ""
        
        # Clean up cells
        cleaned_table = []
        for row in table:
            if row:
                cleaned_row = []
                for cell in row:
                    if cell:
                        cell_text = str(cell).replace('\n', ' ').replace('\r', ' ').replace('|', '\\|').strip()
                        cleaned_row.append(cell_text)
                    else:
                        cleaned_row.append("")
                if any(cleaned_row):
                    cleaned_table.append(cleaned_row)
        
        if len(cleaned_table) < 1:
            return ""
        
        # Determine max columns and pad rows
        max_cols = max(len(row) for row in cleaned_table)
        for row in cleaned_table:
            while len(row) < max_cols:
                row.append("")
        
        # Build markdown table
        md_lines = [f"### Table {table_idx} (Page {page_num})"]
        md_lines.append("| " + " | ".join(cleaned_table[0]) + " |")
        md_lines.append("| " + " | ".join(["---"] * max_cols) + " |")
        
        for row in cleaned_table[1:]:
            md_lines.append("| " + " | ".join(row) + " |")
        
        return "\n".join(md_lines)                        

def run_pdfplumber(file_name):
    from content_analyzer.document_parser import DocumentProcessor
    processor = DocumentProcessor()
    return processor._load_pdf_with_pdfplumber(file_name)

def run_charts(file_name, enable_chart_extraction, gemini_client):
    from content_analyzer.document_parser import DocumentProcessor
    processor = DocumentProcessor()
    processor.gemini_client = gemini_client
    if enable_chart_extraction and gemini_client:
        return processor._extract_charts_from_pdf(file_name)
    return []