"""
Extraction Pipeline V2 - Orchestrator for Docling-based extraction.

This module orchestrates all V2 extraction components:
1. Excel to HTML conversion (with LibreOffice + border detection)
2. HTML table extraction
3. LLM-based header detection
4. Record building

Returns success/failure status with fallback support for V1 pipeline.
"""

import asyncio
from typing import Any
from uuid import UUID
from dataclasses import dataclass

import structlog

from src.extraction_v2.document_converter import DocumentToHtmlConverter
from src.extraction_v2.html_table_extractor import HtmlTableExtractor, HtmlTable
from src.extraction_v2.llm_header_detector import LlmHeaderDetector
from src.extraction_v2.record_builder import RecordBuilder
from src.extraction_v2.semantic_chunker import SemanticChunker
from src.extraction_v2.semantic_tagger import SemanticTagger
from src.extraction_v2.large_table_extractor import (
    LargeTableExtractor,
    LargeTableConfig,
    LargeTableResult
)

logger = structlog.get_logger(__name__)


class _ThreadSafeEventLoopPolicy(asyncio.DefaultEventLoopPolicy):
    """
    Custom event loop policy that automatically creates event loops for threads.

    The default asyncio policy raises RuntimeError when get_event_loop() is called
    from a thread without an event loop. This policy instead creates a new loop,
    which is needed for libraries like Docling that use internal threading.
    """

    def get_event_loop(self):
        """Get or create an event loop for the current thread."""
        try:
            return super().get_event_loop()
        except RuntimeError:
            # No event loop in this thread, create one
            loop = self.new_event_loop()
            self.set_event_loop(loop)
            return loop


def _ensure_event_loop() -> None:
    """
    Ensure an event loop exists and set a thread-safe event loop policy.

    This is needed because Docling uses asyncio internally (including in spawned
    threads), and in forked Celery worker processes, there may not be a valid
    event loop.
    """
    # Install thread-safe policy if not already installed
    current_policy = asyncio.get_event_loop_policy()
    if not isinstance(current_policy, _ThreadSafeEventLoopPolicy):
        asyncio.set_event_loop_policy(_ThreadSafeEventLoopPolicy())
        logger.debug("Installed thread-safe event loop policy")

    try:
        asyncio.get_event_loop()
    except RuntimeError:
        # No event loop exists, create one
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        logger.debug("Created new event loop for current thread")


@dataclass
class ExtractionResult:
    """Result from V2 extraction pipeline."""
    success: bool
    document_id: UUID
    file_path: str
    record_count: int
    records: list[dict[str, Any]]
    table_count: int = 0
    error: str | None = None

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary for JSON serialization."""
        return {
            "success": self.success,
            "document_id": str(self.document_id),
            "file_path": self.file_path,
            "record_count": self.record_count,
            "table_count": self.table_count,
            "error": self.error
        }


class ExtractionPipelineV2:
    """V2 pipeline orchestrator using Docling."""

    def __init__(
        self,
        converter: DocumentToHtmlConverter | None = None,
        extractor: HtmlTableExtractor | None = None,
        detector: LlmHeaderDetector | None = None,
        builder: RecordBuilder | None = None,
        semantic_chunker: SemanticChunker | None = None,
        semantic_tagger: SemanticTagger | None = None,
        large_table_extractor: LargeTableExtractor | None = None,
        large_table_config: LargeTableConfig | None = None
    ):
        """
        Initialize the V2 extraction pipeline.

        Args:
            converter: Optional DocumentToHtmlConverter instance
            extractor: Optional HtmlTableExtractor instance
            detector: Optional LlmHeaderDetector instance
            builder: Optional RecordBuilder instance
            semantic_chunker: Optional SemanticChunker instance for Word documents
            semantic_tagger: Optional SemanticTagger instance for tagging records
            large_table_extractor: Optional LargeTableExtractor for large tables
            large_table_config: Optional configuration for large table extraction
        """
        self.converter = converter or DocumentToHtmlConverter()
        self.extractor = extractor or HtmlTableExtractor()
        self.detector = detector or LlmHeaderDetector()
        self.builder = builder or RecordBuilder()
        self.semantic_chunker = semantic_chunker or SemanticChunker()
        self.semantic_tagger = semantic_tagger or SemanticTagger()
        self.large_table_extractor = large_table_extractor or LargeTableExtractor(
            large_table_config
        )

    async def process_document(
        self,
        document_id: UUID,
        file_path: str
    ) -> ExtractionResult:
        """
        Process document using V2 pipeline.

        Returns ExtractionResult with success=False if conversion fails
        (caller should fallback to V1 pipeline).

        Args:
            document_id: UUID for document tracking
            file_path: Path to Excel file (.xlsx, .xls, or .xlsm)

        Returns:
            ExtractionResult with records or error information
        """
        color_map_path = None

        try:
            logger.info(
                f"Starting V2 extraction pipeline for document {document_id}",
                file_path=file_path
            )

            # Step 1: Excel → HTML (Docling) + Color Extraction
            logger.info("Step 1/4: Converting Excel to HTML and extracting colors...")
            html, color_map_path = self.converter.convert_to_html(file_path)

            if html is None:
                logger.error("Docling conversion failed")
                return ExtractionResult(
                    success=False,
                    document_id=document_id,
                    file_path=file_path,
                    record_count=0,
                    records=[],
                    error="Docling conversion failed"
                )

            # Load color map if available
            color_map = {}
            if color_map_path and os.path.exists(color_map_path):
                try:
                    import json
                    with open(color_map_path, 'r') as f:
                        color_map = json.load(f)
                    logger.info(f"Loaded color map with {sum(len(v) for v in color_map.values())} colored cells")
                except Exception as e:
                    logger.warning(f"Failed to load color map: {e}")
                    color_map = {}

            # Step 2: HTML → Tables (classify into valid tables and raw text)
            logger.info("Step 2/4: Extracting and classifying tables from HTML...")
            valid_tables, raw_text_tables = self.extractor.extract_all_tables(html)

            total_tables = len(valid_tables) + len(raw_text_tables)
            if total_tables == 0:
                logger.warning("No tables found in HTML")
                return ExtractionResult(
                    success=True,  # Not an error, just no tables
                    document_id=document_id,
                    file_path=file_path,
                    record_count=0,
                    records=[],
                    table_count=0
                )

            logger.info(
                f"Found {total_tables} table(s): "
                f"{len(valid_tables)} data tables, {len(raw_text_tables)} raw text"
            )

            # Step 3 & 4: Process each valid table (Headers → Records)
            all_records = []

            for table_idx, table in enumerate(valid_tables, start=1):
                logger.info(
                    f"Processing data table {table_idx}/{len(valid_tables)}: "
                    f"{table.row_count} rows x {table.col_count} cols"
                )

                try:
                    # Get sheet name from converter's mapping
                    # Use table.table_index (HTML extraction order) which matches
                    # Docling's table list order (Docling includes all tables)
                    sheet_name = self.converter.get_sheet_name_for_table(table.table_index)
                    if sheet_name:
                        logger.debug(f"Table {table_idx} (html_idx={table.table_index}) is from sheet: {sheet_name}")
                    else:
                        logger.warning(f"Could not determine sheet name for table {table_idx}, using 'Unknown'")
                        sheet_name = "Unknown"

                    # Step 3: Detect headers with LLM
                    logger.info(
                        f"Step 3/4 (table {table_idx}): "
                        "Detecting headers with LLM..."
                    )
                    # Pass raw HTML to LLM (it will extract first 10 rows internally)
                    headers = self.detector.detect_headers(table.html)

                    logger.info(
                        f"Detected {len(headers.header_rows)} header row(s), "
                        f"{len(headers.headers)} column(s)"
                    )

                    # Step 4: Build records
                    logger.info(
                        f"Step 4/4 (table {table_idx}): Building records..."
                    )
                    records = self.builder.build_records(
                        table=table,
                        headers=headers,
                        document_id=document_id,
                        sheet_name=sheet_name,  # Use mapped sheet name
                        color_map=color_map  # Pass color map for cell color metadata
                    )

                    all_records.extend(records)
                    logger.info(
                        f"Table {table_idx} complete: {len(records)} records"
                    )

                except Exception as e:
                    logger.error(
                        f"Error processing table {table_idx}: {e}",
                        exc_info=True
                    )
                    # Continue with next table
                    continue

            # Process raw text tables (NO LLM header detection - cost savings)
            # Group consecutive raw text tables by sheet to preserve context
            if raw_text_tables:
                logger.info(
                    f"Processing {len(raw_text_tables)} raw text table(s) "
                    "(skipping LLM header detection)"
                )

                # Group raw text tables by sheet
                raw_text_by_sheet = self._group_raw_text_by_sheet(
                    raw_text_tables, valid_tables
                )

                for sheet_name, tables in raw_text_by_sheet.items():
                    try:
                        # Build combined raw text record for this sheet
                        raw_records = self.builder.build_combined_raw_text_records(
                            tables=tables,
                            document_id=document_id,
                            sheet_name=sheet_name,
                            color_map=color_map  # Pass color map for cell color metadata
                        )

                        all_records.extend(raw_records)
                        logger.info(
                            f"Raw text for sheet '{sheet_name}': "
                            f"{len(tables)} table(s) combined into {len(raw_records)} record(s)"
                        )

                    except Exception as e:
                        logger.error(
                            f"Error processing raw text for sheet {sheet_name}: {e}",
                            exc_info=True
                        )
                        continue

            # Step 5: Semantic Tagging (NEW)
            logger.info("Step 5/5: Applying semantic tags to records...")
            if all_records:
                try:
                    # Prepare records for tagging (convert ExtractedRecord to dict format)
                    tagging_records = []
                    for record in all_records:
                        # Extract text content for tagging
                        text_parts = []
                        if hasattr(record, 'content') and record.content:
                            for key, value in record.content.items():
                                if value and str(value).strip():
                                    text_parts.append(f"{key}: {value}")

                        tagging_record = {
                            "text": " | ".join(text_parts) if text_parts else "",
                            "parent_header": record.headers[0] if (hasattr(record, 'headers') and record.headers) else "",
                            "sheet_name": record._source.get("sheet", "") if hasattr(record, '_source') else ""
                        }
                        tagging_records.append(tagging_record)

                    # Tag the records
                    tagged_records = self.semantic_tagger.tag_records(tagging_records)

                    # Apply tags back to original records
                    for original_record, tagged_record in zip(all_records, tagged_records):
                        original_record.tags = tagged_record.get("tags", [])

                    logger.info(f"Tagged {len(all_records)} records")

                except Exception as e:
                    logger.error(
                        f"Tagging failed, continuing with empty tags: {e}",
                        exc_info=True
                    )
                    # Set empty tags on failure
                    for record in all_records:
                        record.tags = []

            logger.info(
                f"V2 pipeline complete: {len(all_records)} total records "
                f"from {len(valid_tables)} data table(s) and "
                f"{len(raw_text_tables)} raw text table(s)"
            )

            return ExtractionResult(
                success=True,
                document_id=document_id,
                file_path=file_path,
                record_count=len(all_records),
                records=all_records,
                table_count=total_tables
            )

        except Exception as e:
            logger.error(
                f"V2 pipeline failed: {e}",
                exc_info=True
            )
            return ExtractionResult(
                success=False,
                document_id=document_id,
                file_path=file_path,
                record_count=0,
                records=[],
                error=str(e)
            )

        finally:
            # Clean up color map temp file
            if color_map_path and os.path.exists(color_map_path):
                try:
                    os.remove(color_map_path)
                    logger.debug(f"Cleaned up color map file: {color_map_path}")
                except Exception as e:
                    logger.warning(f"Failed to cleanup color map file: {e}")

    def _group_raw_text_by_sheet(
        self,
        raw_text_tables: list["HtmlTable"],
        valid_tables: list["HtmlTable"]
    ) -> dict[str, list["HtmlTable"]]:
        """
        Group raw text tables by their sheet name.

        This ensures that multiple raw text blocks on the same sheet
        are combined into a single record for better context.

        Args:
            raw_text_tables: List of single-column raw text tables
            valid_tables: List of valid data tables (for sheet name inference)

        Returns:
            Dictionary mapping sheet_name -> list of raw text tables
        """
        grouped: dict[str, list[HtmlTable]] = {}

        for raw_table in raw_text_tables:
            # Get sheet name using direct lookup or inference
            sheet_name = raw_table.sheet_name
            if not sheet_name:
                sheet_name = self._get_sheet_name_for_raw_text(raw_table, valid_tables)
            if not sheet_name:
                sheet_name = "Unknown"

            if sheet_name not in grouped:
                grouped[sheet_name] = []
            grouped[sheet_name].append(raw_table)

        return grouped

    def _get_sheet_name_for_raw_text(
        self,
        raw_table: "HtmlTable",
        valid_tables: list["HtmlTable"]
    ) -> str | None:
        """
        Get sheet name for a raw text table using its HTML table_index.

        Docling's table list includes ALL tables (both valid and single-column),
        so we can directly use the raw_table.table_index to look up the sheet name.

        If direct lookup fails (table not in Docling's list), fall back to
        finding the nearest valid table's sheet.

        Args:
            raw_table: The raw text HtmlTable to find sheet name for
            valid_tables: List of valid data tables (used as fallback)

        Returns:
            Sheet name or None if unable to determine
        """
        raw_idx = raw_table.table_index

        # Try direct lookup using the HTML table index
        # (Docling's table list includes all tables)
        sheet_name = self.converter.get_sheet_name_for_table(raw_idx)
        if sheet_name:
            logger.debug(
                f"Found sheet '{sheet_name}' for raw text table {raw_idx} via direct lookup"
            )
            return sheet_name

        # Fallback: find nearest valid table's sheet
        if not valid_tables:
            return None

        closest_table = None
        min_distance = float('inf')

        for valid_table in valid_tables:
            distance = abs(valid_table.table_index - raw_idx)
            if distance < min_distance:
                min_distance = distance
                closest_table = valid_table

        if closest_table is None:
            return None

        # Get sheet name from closest valid table using its HTML index
        sheet_name = self.converter.get_sheet_name_for_table(closest_table.table_index)
        if sheet_name:
            logger.debug(
                f"Inferred sheet '{sheet_name}' for raw text table {raw_idx} "
                f"from nearby valid table {closest_table.table_index}"
            )
        return sheet_name

    async def process_document_with_large_table_detection(
        self,
        document_id: UUID,
        file_path: str
    ) -> ExtractionResult:
        """
        Process Excel document with automatic large table detection.

        This method first detects tables using border detection, then routes:
        - Large tables (>1000 rows) → LargeTableExtractor (chunked processing)
        - Normal tables → Standard Docling pipeline

        Args:
            document_id: UUID for document tracking
            file_path: Path to Excel file (.xlsx, .xls, or .xlsm)

        Returns:
            ExtractionResult with records from both extraction methods
        """
        try:
            logger.info(
                f"Starting V2 extraction with large table detection for {document_id}",
                file_path=file_path
            )

            # Step 1: Detect all tables and classify by size
            tables = self.large_table_extractor.detect_tables(file_path)

            if not tables:
                logger.warning("No tables detected via border detection")
                # Fall back to standard processing
                return await self.process_document(document_id, file_path)

            large_tables = [t for t in tables if t.get('is_large', False)]
            normal_tables = [t for t in tables if not t.get('is_large', False)]

            logger.info(
                f"Table classification: {len(large_tables)} large, {len(normal_tables)} normal"
            )

            all_records = []

            # Step 2: Process large tables with chunked extraction
            if large_tables:
                logger.info(f"Processing {len(large_tables)} large table(s) with chunked extraction")

                for table in large_tables:
                    result = self.large_table_extractor.extract_table(
                        file_path, table, document_id
                    )

                    if result.success:
                        # Convert to pipeline format
                        pipeline_records = self.large_table_extractor.to_pipeline_records(
                            result, document_id
                        )
                        all_records.extend(pipeline_records)
                        logger.info(
                            f"Large table extracted: {result.rows_processed} rows from {result.sheet_name}"
                        )
                    else:
                        logger.error(f"Large table extraction failed: {result.error}")

            # Step 3: Process normal tables with standard Docling pipeline
            if normal_tables:
                logger.info(f"Processing {len(normal_tables)} normal table(s) with Docling")

                # Use standard pipeline for normal tables
                standard_result = await self.process_document(document_id, file_path)

                if standard_result.success and standard_result.records:
                    all_records.extend(standard_result.records)

            logger.info(
                f"V2 pipeline with large table detection complete: "
                f"{len(all_records)} total records"
            )

            return ExtractionResult(
                success=True,
                document_id=document_id,
                file_path=file_path,
                record_count=len(all_records),
                records=all_records,
                table_count=len(tables)
            )

        except Exception as e:
            logger.error(
                f"V2 pipeline with large table detection failed: {e}",
                exc_info=True
            )
            return ExtractionResult(
                success=False,
                document_id=document_id,
                file_path=file_path,
                record_count=0,
                records=[],
                error=str(e)
            )

    def process_document_with_large_table_detection_sync(
        self,
        document_id: UUID,
        file_path: str
    ) -> ExtractionResult:
        """
        Synchronous version of process_document_with_large_table_detection.

        Args:
            document_id: UUID for document tracking
            file_path: Path to Excel file

        Returns:
            ExtractionResult with records or error information
        """
        import asyncio

        try:
            loop = asyncio.get_event_loop()
            if loop.is_running():
                return asyncio.run(
                    self.process_document_with_large_table_detection(document_id, file_path)
                )
            else:
                return asyncio.run(
                    self.process_document_with_large_table_detection(document_id, file_path)
                )
        except RuntimeError:
            return asyncio.run(
                self.process_document_with_large_table_detection(document_id, file_path)
            )

    async def process_word_document(
        self,
        document_id: UUID,
        file_path: str
    ) -> ExtractionResult:
        """
        Process Word document using V2 pipeline with semantic chunking.

        Args:
            document_id: UUID for document tracking
            file_path: Path to Word file (.docx, .doc, or .docm)

        Returns:
            ExtractionResult with chunk records or error information
        """
        try:
            logger.info(
                f"Starting V2 Word extraction pipeline for document {document_id}",
                file_path=file_path
            )

            # Step 1: Word → Markdown (via Docling)
            logger.info("Step 1/2: Converting Word to Markdown with preprocessing...")
            markdown = self.converter.convert_to_html(file_path)

            if markdown is None:
                logger.error("Word document conversion failed")
                return ExtractionResult(
                    success=False,
                    document_id=document_id,
                    file_path=file_path,
                    record_count=0,
                    records=[],
                    error="Word document conversion failed"
                )

            logger.info(f"Conversion complete: {len(markdown)} chars")

            # Step 2: Get DoclingDocument for chunking
            # We need to convert again to get the DoclingDocument object
            logger.info("Step 2/2: Chunking document with HybridChunker...")

            # Re-convert to get DoclingDocument (not ideal, but necessary for chunking)
            from pathlib import Path
            from docling.document_converter import DocumentConverter, WordFormatOption
            from docling.datamodel.base_models import InputFormat

            file_ext = Path(file_path).suffix.lower()
            if file_ext in ['.doc', '.docm']:
                # Convert to .docx first
                temp_docx = self.converter._convert_doc_to_docx(Path(file_path))
                if temp_docx is None:
                    return ExtractionResult(
                        success=False,
                        document_id=document_id,
                        file_path=file_path,
                        record_count=0,
                        records=[],
                        error="LibreOffice conversion failed"
                    )
                conversion_path = temp_docx
            else:
                conversion_path = file_path

            # Convert to DoclingDocument
            converter = DocumentConverter(
                format_options={InputFormat.DOCX: WordFormatOption()}
            )
            _ensure_event_loop()  # Ensure event loop exists for Docling
            result = converter.convert(conversion_path)
            doc = result.document

            # Chunk the document
            chunks = self.semantic_chunker.chunk_document(
                doc=doc,
                file_id=document_id,
                original_filename=Path(file_path).name
            )

            logger.info(
                f"V2 Word pipeline complete: {len(chunks)} chunks created"
            )

            # Convert chunks to ExtractionResult format
            return ExtractionResult(
                success=True,
                document_id=document_id,
                file_path=file_path,
                record_count=len(chunks),
                records=chunks,
                table_count=0  # Word docs don't separate tables in V2
            )

        except Exception as e:
            logger.error(
                f"V2 Word pipeline failed: {e}",
                exc_info=True
            )
            return ExtractionResult(
                success=False,
                document_id=document_id,
                file_path=file_path,
                record_count=0,
                records=[],
                error=str(e)
            )

    def process_document_sync(
        self,
        document_id: UUID,
        file_path: str
    ) -> ExtractionResult:
        """
        Synchronous version of process_document.

        Use this if you don't need async processing.

        Args:
            document_id: UUID for document tracking
            file_path: Path to Excel file

        Returns:
            ExtractionResult with records or error information
        """
        # Run in a separate thread to avoid event loop conflicts with Celery workers
        import asyncio
        import concurrent.futures

        def run_async_in_thread():
            """Run async function in a new thread with its own event loop."""
            new_loop = asyncio.new_event_loop()
            asyncio.set_event_loop(new_loop)
            try:
                return new_loop.run_until_complete(
                    self.process_document(document_id, file_path)
                )
            finally:
                # Cancel all pending tasks to avoid "Task exception was never retrieved" warnings
                pending = asyncio.all_tasks(new_loop)
                for task in pending:
                    task.cancel()
                # Allow cancelled tasks to complete
                if pending:
                    new_loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True))
                new_loop.close()

        with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
            future = executor.submit(run_async_in_thread)
            return future.result()

    def process_word_document_sync(
        self,
        document_id: UUID,
        file_path: str
    ) -> ExtractionResult:
        """
        Synchronous version of process_word_document.

        Args:
            document_id: UUID for document tracking
            file_path: Path to Word file (.docx, .doc, or .docm)

        Returns:
            ExtractionResult with chunk records or error information
        """
        # Run in a separate thread to avoid event loop conflicts with Celery workers
        import asyncio
        import concurrent.futures

        def run_async_in_thread():
            """Run async function in a new thread with its own event loop."""
            new_loop = asyncio.new_event_loop()
            asyncio.set_event_loop(new_loop)
            try:
                return new_loop.run_until_complete(
                    self.process_word_document(document_id, file_path)
                )
            finally:
                # Cancel all pending tasks to avoid "Task exception was never retrieved" warnings
                pending = asyncio.all_tasks(new_loop)
                for task in pending:
                    task.cancel()
                # Allow cancelled tasks to complete
                if pending:
                    new_loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True))
                new_loop.close()

        with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
            future = executor.submit(run_async_in_thread)
            return future.result()

    async def process_pdf_document(
        self,
        document_id: UUID,
        file_path: str
    ) -> ExtractionResult:
        """
        Process PDF document using V2 pipeline with semantic chunking.

        Args:
            document_id: UUID for document tracking
            file_path: Path to PDF file

        Returns:
            ExtractionResult with chunk records or error information
        """
        try:
            logger.info(
                f"Starting V2 PDF extraction pipeline for document {document_id}",
                file_path=file_path
            )

            # Step 1: PDF → Markdown (via Docling)
            logger.info("Step 1/2: Converting PDF to Markdown...")
            markdown = self.converter.convert_to_html(file_path)

            if markdown is None:
                logger.error("PDF document conversion failed")
                return ExtractionResult(
                    success=False,
                    document_id=document_id,
                    file_path=file_path,
                    record_count=0,
                    records=[],
                    error="PDF document conversion failed"
                )

            logger.info(f"Conversion complete: {len(markdown)} chars")

            # Step 2: Get DoclingDocument for chunking
            logger.info("Step 2/2: Chunking document with HybridChunker...")

            from pathlib import Path
            from docling.document_converter import DocumentConverter, PdfFormatOption
            from docling.datamodel.base_models import InputFormat
            from docling.datamodel.pipeline_options import PdfPipelineOptions

            # Convert to DoclingDocument
            # Disable OCR by default for performance (can be enabled if needed)
            pipeline_options = PdfPipelineOptions(do_ocr=False)
            converter = DocumentConverter(
                format_options={InputFormat.PDF: PdfFormatOption(pipeline_options=pipeline_options)}
            )
            _ensure_event_loop()  # Ensure event loop exists for Docling
            result = converter.convert(file_path)
            doc = result.document

            # Chunk the document with custom serializers
            # SemanticChunker automatically uses ImgTableAnnotationSerializerProvider
            chunks = self.semantic_chunker.chunk_document(
                doc=doc,
                file_id=document_id,
                original_filename=Path(file_path).name
            )

            logger.info(
                f"V2 PDF pipeline complete: {len(chunks)} chunks created"
            )

            # Convert chunks to ExtractionResult format
            return ExtractionResult(
                success=True,
                document_id=document_id,
                file_path=file_path,
                record_count=len(chunks),
                records=chunks,
                table_count=0  # PDF docs don't separate tables in V2
            )

        except Exception as e:
            logger.error(
                f"V2 PDF pipeline failed: {e}",
                exc_info=True
            )
            return ExtractionResult(
                success=False,
                document_id=document_id,
                file_path=file_path,
                record_count=0,
                records=[],
                error=str(e)
            )

    def process_pdf_document_sync(
        self,
        document_id: UUID,
        file_path: str
    ) -> ExtractionResult:
        """
        Synchronous version of process_pdf_document.

        Args:
            document_id: UUID for document tracking
            file_path: Path to PDF file

        Returns:
            ExtractionResult with chunk records or error information
        """
        # Run in a separate thread to avoid event loop conflicts with Celery workers
        import asyncio
        import concurrent.futures

        def run_async_in_thread():
            """Run async function in a new thread with its own event loop."""
            new_loop = asyncio.new_event_loop()
            asyncio.set_event_loop(new_loop)
            try:
                return new_loop.run_until_complete(
                    self.process_pdf_document(document_id, file_path)
                )
            finally:
                # Cancel all pending tasks to avoid "Task exception was never retrieved" warnings
                pending = asyncio.all_tasks(new_loop)
                for task in pending:
                    task.cancel()
                # Allow cancelled tasks to complete
                if pending:
                    new_loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True))
                new_loop.close()

        with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
            future = executor.submit(run_async_in_thread)
            return future.result()

    async def process_pptx_document(
        self,
        document_id: UUID,
        file_path: str
    ) -> ExtractionResult:
        """
        Process PowerPoint document using V2 pipeline with semantic chunking.

        Speaker notes are embedded during preprocessing.
        Images and tables handled by ImgTableAnnotationSerializerProvider during chunking.

        Args:
            document_id: UUID for document tracking
            file_path: Path to PowerPoint file (.pptx, .ppt, or .pptm)

        Returns:
            ExtractionResult with chunk records or error information
        """
        try:
            logger.info(
                f"Starting V2 PowerPoint extraction pipeline for document {document_id}",
                file_path=file_path
            )

            # Step 1: PowerPoint → Markdown (via Docling)
            # This includes speaker notes if they exist (preprocessed conditionally)
            logger.info("Step 1/2: Converting PowerPoint to Markdown...")
            markdown = self.converter.convert_to_html(file_path)

            if markdown is None:
                logger.error("PowerPoint document conversion failed")
                return ExtractionResult(
                    success=False,
                    document_id=document_id,
                    file_path=file_path,
                    record_count=0,
                    records=[],
                    error="PowerPoint document conversion failed"
                )

            logger.info(f"Conversion complete: {len(markdown)} chars")

            # Step 2: Get DoclingDocument for chunking
            logger.info("Step 2/2: Chunking document with HybridChunker...")

            from pathlib import Path
            from docling.document_converter import DocumentConverter
            from docling.datamodel.base_models import InputFormat

            file_ext = Path(file_path).suffix.lower()

            # Handle legacy formats
            if file_ext in ['.ppt', '.pptm']:
                # Convert to .pptx first
                temp_pptx = self.converter._convert_ppt_to_pptx(Path(file_path))
                if temp_pptx is None:
                    return ExtractionResult(
                        success=False,
                        document_id=document_id,
                        file_path=file_path,
                        record_count=0,
                        records=[],
                        error="LibreOffice conversion failed"
                    )
                conversion_path = temp_pptx
            else:
                conversion_path = file_path

            # Convert to DoclingDocument
            # No format option needed - Docling auto-detects PowerPoint
            converter = DocumentConverter()
            _ensure_event_loop()  # Ensure event loop exists for Docling
            result = converter.convert(conversion_path)
            doc = result.document

            # Chunk the document with custom serializers
            # SemanticChunker automatically uses ImgTableAnnotationSerializerProvider
            chunks = self.semantic_chunker.chunk_document(
                doc=doc,
                file_id=document_id,
                original_filename=Path(file_path).name
            )

            logger.info(
                f"V2 PowerPoint pipeline complete: {len(chunks)} chunks created"
            )

            # Convert chunks to ExtractionResult format
            return ExtractionResult(
                success=True,
                document_id=document_id,
                file_path=file_path,
                record_count=len(chunks),
                records=chunks,
                table_count=0  # PowerPoint docs don't separate tables in V2
            )

        except Exception as e:
            logger.error(
                f"V2 PowerPoint pipeline failed: {e}",
                exc_info=True
            )
            return ExtractionResult(
                success=False,
                document_id=document_id,
                file_path=file_path,
                record_count=0,
                records=[],
                error=str(e)
            )

    def process_pptx_document_sync(
        self,
        document_id: UUID,
        file_path: str
    ) -> ExtractionResult:
        """
        Synchronous version of process_pptx_document.

        Args:
            document_id: UUID for document tracking
            file_path: Path to PowerPoint file (.pptx, .ppt, or .pptm)

        Returns:
            ExtractionResult with chunk records or error information
        """
        # Run in a separate thread to avoid event loop conflicts with Celery workers
        import asyncio
        import concurrent.futures

        def run_async_in_thread():
            """Run async function in a new thread with its own event loop."""
            new_loop = asyncio.new_event_loop()
            asyncio.set_event_loop(new_loop)
            try:
                return new_loop.run_until_complete(
                    self.process_pptx_document(document_id, file_path)
                )
            finally:
                # Cancel all pending tasks to avoid "Task exception was never retrieved" warnings
                pending = asyncio.all_tasks(new_loop)
                for task in pending:
                    task.cancel()
                # Allow cancelled tasks to complete
                if pending:
                    new_loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True))
                new_loop.close()

        with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
            future = executor.submit(run_async_in_thread)
            return future.result()


# Convenience function for easy imports
async def extract_from_excel(
    document_id: UUID,
    file_path: str
) -> ExtractionResult:
    """
    Convenience function to extract records from Excel file.

    Args:
        document_id: UUID for document tracking
        file_path: Path to Excel file

    Returns:
        ExtractionResult with records or error information
    """
    pipeline = ExtractionPipelineV2()
    return await pipeline.process_document(document_id, file_path)


def extract_from_excel_sync(
    document_id: UUID,
    file_path: str
) -> ExtractionResult:
    """
    Synchronous convenience function to extract records from Excel file.

    Args:
        document_id: UUID for document tracking
        file_path: Path to Excel file

    Returns:
        ExtractionResult with records or error information
    """
    pipeline = ExtractionPipelineV2()
    return pipeline.process_document_sync(document_id, file_path)


async def extract_from_word(
    document_id: UUID,
    file_path: str
) -> ExtractionResult:
    """
    Convenience function to extract chunks from Word file.

    Args:
        document_id: UUID for document tracking
        file_path: Path to Word file (.docx, .doc, or .docm)

    Returns:
        ExtractionResult with chunk records or error information
    """
    pipeline = ExtractionPipelineV2()
    return await pipeline.process_word_document(document_id, file_path)


def extract_from_word_sync(
    document_id: UUID,
    file_path: str
) -> ExtractionResult:
    """
    Synchronous convenience function to extract chunks from Word file.

    Args:
        document_id: UUID for document tracking
        file_path: Path to Word file (.docx, .doc, or .docm)

    Returns:
        ExtractionResult with chunk records or error information
    """
    import asyncio

    try:
        loop = asyncio.get_event_loop()
        if loop.is_running():
            return asyncio.run(extract_from_word(document_id, file_path))
        else:
            return asyncio.run(extract_from_word(document_id, file_path))
    except RuntimeError:
        return asyncio.run(extract_from_word(document_id, file_path))


# Large table extraction convenience functions
async def extract_from_excel_with_large_table_detection(
    document_id: UUID,
    file_path: str
) -> ExtractionResult:
    """
    Extract records from Excel file with automatic large table detection.

    Large tables (>1000 rows) are processed with chunked extraction,
    normal tables use standard Docling pipeline.

    Args:
        document_id: UUID for document tracking
        file_path: Path to Excel file

    Returns:
        ExtractionResult with records or error information
    """
    pipeline = ExtractionPipelineV2()
    return await pipeline.process_document_with_large_table_detection(
        document_id, file_path
    )


def extract_from_excel_with_large_table_detection_sync(
    document_id: UUID,
    file_path: str
) -> ExtractionResult:
    """
    Synchronous version of extract_from_excel_with_large_table_detection.

    Args:
        document_id: UUID for document tracking
        file_path: Path to Excel file

    Returns:
        ExtractionResult with records or error information
    """
    pipeline = ExtractionPipelineV2()
    return pipeline.process_document_with_large_table_detection_sync(
        document_id, file_path
    )