# Extraction V2 Implementation Guide

## 📋 Overview

This directory implements Story 3.4b: Docling-Based Excel to HTML Converter.

**Existing Code:** You have `excel_parser_service.py` in the root directory that already uses Docling. However, that's a full-featured service with image/shape extraction. For the DSOL architecture (as per `docs/architecture.md` and `docs/epics.md`), we need focused, single-responsibility modules.

## 🏗️ Architecture Decision

**Option 1: Adapt existing code (RECOMMENDED)**
- Extract the Docling conversion logic from `excel_parser_service.py`
- Create focused modules for each V2 component
- Keep separation between service layer and extraction layer

**Option 2: Start fresh**
- Implement from scratch following the architecture
- Reference existing code for Docling usage patterns

## 📝 Implementation Checklist

### Module 1: `excel_to_html_converter.py`

**Purpose:** Convert Excel → HTML using Docling (with XLS/XLSM support via LibreOffice)

**Reference:**
- Lines 546-558 in `excel_parser_service.py` (Docling conversion)
- Lines 339-416 in `excel_parser_service.py` (LibreOffice conversion)

```python
import os
import subprocess
import shutil
import tempfile
from pathlib import Path
from docling.datamodel.base_models import InputFormat
from docling.document_converter import DocumentConverter, ExcelFormatOption
from src.commons.handlers.log_handler import get_logger

logger = get_logger(__name__)

class ExcelToHtmlConverter:
    """Converts Excel files to HTML using Docling (with XLS/XLSM preprocessing)."""

    def __init__(self):
        self.converter = DocumentConverter(
            format_options={
                InputFormat.XLSX: ExcelFormatOption()
            }
        )

    def convert_to_html(self, file_path: str) -> str | None:
        """
        Convert Excel to HTML.

        Processing Pipeline:
        1. Convert .xls/.xlsm to .xlsx if needed (via LibreOffice)
        2. Preprocess: Detect table borders and insert separator rows
        3. Convert preprocessed .xlsx to HTML (via Docling)

        Handles .xlsx, .xls, and .xlsm files:
        - .xlsx: Preprocess → Docling
        - .xls/.xlsm: LibreOffice → Preprocess → Docling

        Args:
            file_path: Path to Excel file (.xlsx, .xls, or .xlsm)

        Returns:
            HTML string or None on failure
        """
        temp_xlsx_path = None
        preprocessed_path = None

        try:
            # Step 1: Convert .xls or .xlsm to .xlsx first
            _, ext = os.path.splitext(file_path)
            ext_lower = ext.lower()

            if ext_lower in ['.xls', '.xlsm']:
                logger.info(f"Detected {ext_lower} file, converting to .xlsx via LibreOffice...")
                temp_xlsx_path = self._convert_to_xlsx_via_libreoffice(file_path)

                if temp_xlsx_path is None:
                    logger.error("LibreOffice conversion failed")
                    return None

                file_to_preprocess = temp_xlsx_path
            else:
                # Already .xlsx
                file_to_preprocess = file_path

            # Step 2: Preprocess - Detect borders and insert separator rows
            logger.info(f"Preprocessing: Detecting table borders and inserting separators...")
            preprocessed_path = self._preprocess_with_border_detection(file_to_preprocess)

            if preprocessed_path is None:
                logger.warning("Border detection preprocessing failed, using original file")
                file_to_convert = file_to_preprocess
            else:
                logger.info(f"Preprocessing complete, using separated file")
                file_to_convert = preprocessed_path

            # Step 3: Convert to HTML via Docling
            logger.info(f"Converting Excel to HTML via Docling: {file_to_convert}")
            result = self.converter.convert(file_to_convert)
            doc = result.document

            # Export to HTML format
            html = doc.export_to_html()

            logger.info(f"Successfully converted Excel to HTML ({len(html)} chars)")
            return html

        except Exception as e:
            logger.error(f"Docling conversion failed: {e}")
            return None

        finally:
            # Clean up temporary files
            for temp_file in [temp_xlsx_path, preprocessed_path]:
                if temp_file and temp_file != file_path and os.path.exists(temp_file):
                    try:
                        os.remove(temp_file)
                        logger.debug(f"Cleaned up temporary file: {temp_file}")
                    except Exception as e:
                        logger.warning(f"Failed to remove temp file {temp_file}: {e}")

    def _preprocess_with_border_detection(self, file_path: str) -> str | None:
        """
        Preprocess Excel file by detecting table borders and inserting separator rows.

        This helps Docling identify individual tables more accurately by ensuring
        tables are separated from surrounding content.

        Args:
            file_path: Path to .xlsx file

        Returns:
            Path to preprocessed file with separator rows, or None on failure
        """
        try:
            from src.extraction_v2.detect_border import BorderTableDetector

            detector = BorderTableDetector()

            # Create temporary output path
            temp_dir = tempfile.mkdtemp()
            output_path = os.path.join(
                temp_dir,
                f"{os.path.splitext(os.path.basename(file_path))[0]}_separated.xlsx"
            )

            # Insert separator rows
            logger.info("Running border detection and separator insertion...")
            stats = detector.insert_separator_rows(file_path, output_path)

            logger.info(
                f"Border detection complete: {stats['tables_processed']} tables processed, "
                f"{stats['rows_inserted_above'] + stats['rows_inserted_below']} separator rows inserted"
            )

            return output_path

        except Exception as e:
            logger.warning(f"Border detection preprocessing failed: {e}")
            return None

    def _convert_to_xlsx_via_libreoffice(self, input_file: str) -> str | None:
        """
        Convert .xls or .xlsm to .xlsx using LibreOffice CLI.

        This preserves images, shapes, charts, and formatting.

        Args:
            input_file: Path to .xls or .xlsm file

        Returns:
            Path to converted .xlsx file, or None on failure
        """
        # Find LibreOffice executable
        libreoffice_cmd = self._find_libreoffice_command()

        if not libreoffice_cmd:
            logger.error(
                "LibreOffice not found. Please install LibreOffice for .xls/.xlsm support. "
                "Install: https://www.libreoffice.org/download/"
            )
            return None

        try:
            # Create temp directory for output
            temp_dir = tempfile.mkdtemp()

            # Run LibreOffice conversion
            logger.info(f"Running LibreOffice conversion: {libreoffice_cmd}")
            result = subprocess.run(
                [
                    libreoffice_cmd,
                    '--headless',           # No GUI
                    '--convert-to', 'xlsx', # Target format
                    '--outdir', temp_dir,   # Output directory
                    input_file              # Input file
                ],
                capture_output=True,
                text=True,
                timeout=30  # 30 second timeout
            )

            if result.returncode != 0:
                logger.error(f"LibreOffice conversion failed: {result.stderr}")
                return None

            # Find the output file
            input_basename = os.path.splitext(os.path.basename(input_file))[0]
            output_file = os.path.join(temp_dir, f"{input_basename}.xlsx")

            if not os.path.exists(output_file):
                logger.error(f"Conversion succeeded but output file not found: {output_file}")
                return None

            logger.info(f"Successfully converted to .xlsx: {output_file}")
            return output_file

        except subprocess.TimeoutExpired:
            logger.error("LibreOffice conversion timed out (>30s)")
            return None
        except Exception as e:
            logger.error(f"Error during LibreOffice conversion: {e}")
            return None

    def _find_libreoffice_command(self) -> str | None:
        """
        Find LibreOffice executable on the system.

        Returns:
            Path to LibreOffice command, or None if not found
        """
        # Common locations for LibreOffice
        possible_commands = [
            'libreoffice',
            'soffice',
            '/Applications/LibreOffice.app/Contents/MacOS/soffice',  # macOS
            '/usr/bin/libreoffice',  # Linux
            '/usr/bin/soffice',      # Linux alternative
            'C:\\Program Files\\LibreOffice\\program\\soffice.exe',  # Windows
        ]

        for cmd in possible_commands:
            # Check if it's a full path that exists
            if os.path.exists(cmd):
                return cmd

            # Check if it's in PATH
            if shutil.which(cmd):
                return cmd

        return None
```

**Implementation Steps:**
1. ✅ Create file with class structure above
2. ✅ Add LibreOffice CLI conversion for .xls/.xlsm
3. ✅ Add file extension detection
4. ✅ Add temporary file cleanup
5. ✅ Add timeout handling (30s for LibreOffice, overall ≤10s for Docling)
6. ✅ Add logging for conversion metrics
7. ⏳ Write unit tests with .xlsx, .xls, and .xlsm sample files

**Error Handling:**
- LibreOffice not found → log error with installation instructions
- Conversion timeout → return None (trigger V1 fallback)
- Invalid file → return None
- Docling failure → return None

**Performance Targets:**
- .xlsx direct conversion: ≤10 seconds
- .xls/.xlsm with LibreOffice: ≤40 seconds total (30s LibreOffice + 10s Docling)

---

### Module 2: `html_table_extractor.py`

**Purpose:** Extract `<table>` elements from HTML

**Dependencies:** BeautifulSoup4, lxml

```python
from bs4 import BeautifulSoup
from typing import List
from dataclasses import dataclass

@dataclass
class HtmlTable:
    """Represents an extracted HTML table."""
    html: str
    rows: List[List[str]]
    sheet_name: str | None = None
    table_index: int = 0

class HtmlTableExtractor:
    """Extracts tables from HTML structure."""

    def extract_tables(self, html: str) -> List[HtmlTable]:
        """
        Parse HTML and extract all <table> elements.

        Args:
            html: HTML string from Docling

        Returns:
            List of HtmlTable objects
        """
        soup = BeautifulSoup(html, 'lxml')
        tables = soup.find_all('table')

        extracted = []
        for idx, table_elem in enumerate(tables):
            rows = self._extract_rows(table_elem)

            extracted.append(HtmlTable(
                html=str(table_elem),
                rows=rows,
                table_index=idx
            ))

        return extracted

    def _extract_rows(self, table_elem) -> List[List[str]]:
        """Extract text content from table rows."""
        rows = []
        for tr in table_elem.find_all('tr'):
            cells = [td.get_text(strip=True) for td in tr.find_all(['td', 'th'])]
            rows.append(cells)
        return rows
```

**Implementation Steps:**
1. Create file with class structure above
2. Handle merged cells (rowspan, colspan)
3. Extract first 10 rows for header detection
4. Add table metadata extraction
5. Write unit tests with HTML fixtures

---

### Module 3: `llm_header_detector.py`

**Purpose:** Use LLM to detect multi-level headers

**Dependencies:** Azure OpenAI (from existing `src/llm/`)

```python
from typing import List, Dict
from dataclasses import dataclass

@dataclass
class HeaderStructure:
    """Detected header structure."""
    header_rows: List[int]  # Which rows are headers
    headers: List[Dict[str, any]]  # Column definitions

    # Example: [
    #   {"column": 0, "hierarchy": ["Product"]},
    #   {"column": 1, "hierarchy": ["Sales", "Q1"]},
    # ]

class LlmHeaderDetector:
    """Uses LLM to detect headers from first 10 rows."""

    def __init__(self, llm_client):
        self.llm_client = llm_client

    def detect_headers(self, first_10_rows: List[List[str]]) -> HeaderStructure:
        """
        Analyze first 10 rows and detect hierarchical headers.

        Args:
            first_10_rows: First 10 rows of table

        Returns:
            HeaderStructure with detected headers
        """
        prompt = self._build_prompt(first_10_rows)
        response = self.llm_client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": prompt}],
            temperature=0.1
        )

        # Parse JSON response
        return self._parse_response(response)

    def _build_prompt(self, rows: List[List[str]]) -> str:
        # See src/llm/prompts.py for full template
        pass
```

**Prompt Template Location:** `src/llm/prompts.py`

```python
# Add to src/llm/prompts.py

HEADER_DETECTION_PROMPT = """
Analyze the first 10 rows of this table and identify the header row(s).

Table data:
{table_rows}

Tasks:
1. Identify which row(s) contain headers (rows are 0-indexed)
2. Detect if headers are multi-level (spanning multiple rows)
3. Return hierarchical header structure using "->" separator for levels

Output JSON format:
{
  "header_rows": [0, 1],
  "headers": [
    {"column": 0, "hierarchy": ["Product"]},
    {"column": 1, "hierarchy": ["Sales", "Q1"]},
    {"column": 2, "hierarchy": ["Sales", "Q2"]},
    {"column": 3, "hierarchy": ["Inventory", "Current"]}
  ]
}

Rules:
- Use "->" to separate hierarchy levels (not ">")
- Return empty array [] if no headers detected
- Include ALL columns, even single-level ones
"""
```

**Implementation Steps:**
1. Create file with class structure above
2. Implement prompt template in `src/llm/prompts.py`
3. Add JSON parsing with error handling
4. Handle LLM failures (return simple heuristic fallback)
5. Write unit tests with mock LLM responses

---

### Module 4: `record_builder.py`

**Purpose:** Build JSON records from HTML tables + detected headers

```python
from typing import List, Dict, Any
from uuid import UUID

class RecordBuilder:
    """Builds structured records from HTML tables with detected headers."""

    def build_records(
        self,
        table: HtmlTable,
        headers: HeaderStructure,
        document_id: UUID,
        sheet_name: str
    ) -> List[Dict[str, Any]]:
        """
        Convert table rows to JSON records.

        Args:
            table: Extracted HTML table
            headers: Detected header structure
            document_id: Document UUID
            sheet_name: Sheet name

        Returns:
            List of record dictionaries
        """
        records = []
        data_start_row = max(headers.header_rows) + 1

        for row_idx, row_data in enumerate(table.rows[data_start_row:], start=data_start_row):
            record = {
                "content": {},
                "_source": {
                    "document_id": str(document_id),
                    "sheet": sheet_name,
                    "row": row_idx,
                    "table_id": table.table_index
                }
            }

            # Map headers to cell values
            for col_idx, cell_value in enumerate(row_data):
                if col_idx < len(headers.headers):
                    header_def = headers.headers[col_idx]
                    # Create hierarchical key: "Sales -> Q1"
                    header_key = " -> ".join(header_def["hierarchy"])
                    record["content"][header_key] = cell_value

            records.append(record)

        return records
```

**Implementation Steps:**
1. Create file with class structure above
2. Handle empty cells (null vs empty string)
3. Preserve data types (numbers, dates, booleans)
4. Add source metadata tracking
5. Write unit tests with various table structures

---

### Module 5: `pipeline.py`

**Purpose:** Orchestrate V2 extraction workflow

```python
from typing import List, Dict, Any
from uuid import UUID
from dataclasses import dataclass

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

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

    def __init__(
        self,
        converter: ExcelToHtmlConverter,
        extractor: HtmlTableExtractor,
        detector: LlmHeaderDetector,
        builder: RecordBuilder
    ):
        self.converter = converter
        self.extractor = extractor
        self.detector = detector
        self.builder = builder

    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).
        """
        try:
            # Step 1: Excel → HTML (Docling)
            html = self.converter.convert_to_html(file_path)
            if html is None:
                return ExtractionResult(
                    success=False,
                    document_id=document_id,
                    record_count=0,
                    records=[],
                    error="Docling conversion failed"
                )

            # Step 2: HTML → Tables
            tables = self.extractor.extract_tables(html)

            # Step 3: Detect headers (LLM)
            all_records = []
            for table in tables:
                first_10 = table.rows[:10]
                headers = self.detector.detect_headers(first_10)

                # Step 4: Build records
                records = self.builder.build_records(
                    table, headers, document_id, table.sheet_name or "Sheet1"
                )
                all_records.extend(records)

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

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

**Implementation Steps:**
1. Create file with class structure above
2. Add progress tracking (Redis integration)
3. Add timeout handling per step
4. Add fallback decision logic
5. Write integration tests

---

## 🔗 Integration with Main Pipeline

**Location:** `src/workers/extraction_tasks.py` or similar

```python
from src.extraction_v2.pipeline import ExtractionPipelineV2
from src.extraction.pipeline import ExtractionPipeline  # V1

async def process_document_task(document_id: UUID, file_path: str):
    """Main task that routes to V1 or V2."""

    # Try V2 first
    v2_pipeline = get_v2_pipeline()
    result = await v2_pipeline.process_document(document_id, file_path)

    if result.success:
        logger.info(f"V2 pipeline succeeded: {result.record_count} records")
        return result

    # Fallback to V1
    logger.warning(f"V2 failed ({result.error}), falling back to V1")
    v1_pipeline = get_v1_pipeline()
    v1_result = await v1_pipeline.process_document(document_id, file_path)

    return v1_result
```

---

## 🧪 Testing Strategy

### Unit Tests

```
tests/test_extraction_v2/
├── test_excel_to_html_converter.py
├── test_html_table_extractor.py
├── test_llm_header_detector.py
├── test_record_builder.py
└── test_pipeline.py
```

### Integration Tests

```
tests/integration/test_extraction_v2_integration.py
```

**Test Data:** Use Excel files from `tests/fixtures/excel/`

---

## 📊 Success Criteria (from Story 3.4b)

- [ ] ≤10 seconds conversion time for files < 10MB
- [ ] Graceful fallback to V1 on Docling failure
- [ ] LLM header detection ≥85% accuracy
- [ ] Hierarchical headers use `->` separator
- [ ] JSON output matches V1 format (for consistency)

---

## 🚀 Next Steps

1. **Start with Module 1** (`excel_to_html_converter.py`)
   - Extract logic from lines 546-558 of `excel_parser_service.py`
   - Simplify to focus on Excel → HTML only
   - Add comprehensive error handling

2. **Implement remaining modules in order** (2 → 3 → 4 → 5)

3. **Add integration point** in main pipeline router

4. **Write tests** for each module

5. **Run A/B testing** between V1 and V2 for first 2 weeks

---

**Reference:** See `excel_parser_service.py` in root directory for working Docling example (but note it's a service-level implementation with additional features like image extraction - we need focused, single-purpose modules for the extraction layer).