"""Extraction pipeline orchestrator for document processing.

This module provides the ExtractionPipeline class that orchestrates the complete
multi-pass extraction workflow from uploaded Excel documents to structured records
in the database.

Multi-Pass Architecture (ADR-003):
    PASS 1: Symbol Detection - Scan sheets for symbol dictionaries before extraction
    PASS 2: Table Extraction - Detect tables, extract headers, resolve merged cells, build records
    PASS 3: Symbol Resolution - Resolve symbols in extracted content using detected dictionaries
    PASS 4: Cross-Reference Detection - Detect and resolve cross-references between records
"""

import json
import time
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional
from uuid import UUID

import openpyxl
import redis as sync_redis
import structlog
from openpyxl.utils.exceptions import InvalidFileException
from sqlalchemy import select
from sqlalchemy.orm import Session

from src.core.exceptions import ExtractionError
from src.db import models as db_models
from src.extraction.header_extractor import HeaderExtractor
from src.extraction.merged_cell_resolver import MergedCellResolver
from src.extraction.models import (
    ExtractedRecord as ExtractionRecord,
    TableBoundary,
    SymbolDetectionSummary,
    SymbolResolutionSummary,
)
from src.extraction.record_builder import RecordBuilder
from src.extraction.table_detector import TableDetector
from src.extraction.symbol_detector import SymbolDetector
from src.extraction.symbol_resolver import SymbolResolver
from src.extraction.cross_ref_detector import CrossRefDetector, CrossRefResolutionSummary

logger = structlog.get_logger(__name__)


@dataclass
class ExtractionSummary:
    """Summary of extraction pipeline execution.

    Includes statistics from all four extraction passes:
    - PASS 1: Symbol detection
    - PASS 2: Table extraction
    - PASS 3: Symbol resolution
    - PASS 4: Cross-reference detection

    Attributes:
        document_id: UUID of processed document
        sheets_processed: Number of sheets processed
        tables_detected: Total number of tables detected across all sheets
        records_extracted: Total number of records extracted
        duration_ms: Total processing time in milliseconds
        warnings: List of non-critical warnings encountered
        symbols_detected: Number of symbol dictionaries detected (PASS 1)
        symbols_resolved: Number of symbol occurrences resolved (PASS 3)
        unresolved_symbols: Number of unresolved symbol occurrences (PASS 3)
        cross_references_found: Number of cross-references detected (PASS 4)
        cross_references_resolved: Number of cross-references resolved (PASS 4)
        cross_references_unresolved: Number of unresolved cross-references (PASS 4)
    """

    document_id: UUID
    sheets_processed: int
    tables_detected: int
    records_extracted: int
    duration_ms: int
    warnings: list[str] = field(default_factory=list)
    # PASS 1: Symbol detection statistics
    symbols_detected: int = 0
    # PASS 3: Symbol resolution statistics
    symbols_resolved: int = 0
    unresolved_symbols: int = 0
    # PASS 4: Cross-reference statistics
    cross_references_found: int = 0
    cross_references_resolved: int = 0
    cross_references_unresolved: int = 0

    def to_dict(self) -> dict:
        """Convert ExtractionSummary to dictionary for JSON serialization.

        Returns:
            Dictionary with all summary fields
        """
        return {
            "document_id": str(self.document_id),
            "sheets_processed": self.sheets_processed,
            "tables_detected": self.tables_detected,
            "records_extracted": self.records_extracted,
            "duration_ms": self.duration_ms,
            "warnings": self.warnings,
            "symbols_detected": self.symbols_detected,
            "symbols_resolved": self.symbols_resolved,
            "unresolved_symbols": self.unresolved_symbols,
            "cross_references_found": self.cross_references_found,
            "cross_references_resolved": self.cross_references_resolved,
            "cross_references_unresolved": self.cross_references_unresolved,
        }


class ExtractionPipeline:
    """Orchestrates the complete multi-pass extraction workflow.

    This class coordinates all extraction components to process Excel documents
    from file storage through to structured records in the database using a
    multi-pass architecture:

    PASS 1: Symbol Detection - Detect symbol dictionaries before table extraction
    PASS 2: Table Extraction - Standard extraction flow (detect → extract → resolve → build)
    PASS 3: Symbol Resolution - Resolve symbols in extracted content
    PASS 4: Cross-Reference Detection - Detect and resolve cross-references
    """

    def __init__(
        self,
        db_session: Session,
        redis_client: Optional[sync_redis.Redis] = None,
    ):
        """Initialize extraction pipeline.

        Args:
            db_session: SQLAlchemy synchronous session for database operations
            redis_client: Optional Redis client for progress tracking
        """
        self.db_session = db_session
        self.redis_client = redis_client

        # PASS 1: Symbol detection components
        self.symbol_detector = SymbolDetector()

        # PASS 2: Table extraction components
        self.table_detector = TableDetector()
        self.header_extractor = HeaderExtractor()
        self.merged_cell_resolver = MergedCellResolver()
        self.record_builder = RecordBuilder()

        # PASS 3: Symbol resolution components
        self.symbol_resolver = SymbolResolver()

        # PASS 4: Cross-reference detection components
        self.cross_ref_detector = CrossRefDetector()

    def process_document(self, document_id: UUID) -> ExtractionSummary:
        """Process a document through the complete multi-pass extraction pipeline.

        Executes four passes in sequence:
            PASS 1: Symbol Detection - Detect symbol dictionaries before extraction
            PASS 2: Table Extraction - Standard extraction flow
            PASS 3: Symbol Resolution - Resolve symbols in extracted content
            PASS 4: Cross-Reference Detection - Detect and resolve cross-references

        Args:
            document_id: UUID of the document to process

        Returns:
            ExtractionSummary with processing results from all passes

        Raises:
            ExtractionError: If critical extraction fails (PASS 2)
        """
        start_time = time.time()
        warnings: list[str] = []

        # Initialize summary statistics
        symbols_detected = 0
        symbols_resolved = 0
        unresolved_symbols = 0
        cross_ref_found = 0
        cross_ref_resolved = 0
        cross_ref_unresolved = 0

        try:
            # Load document metadata
            document = self._load_document(document_id)

            # Update status to processing
            self._update_document_status(document, "processing")

            logger.info(
                "pipeline_started",
                document_id=str(document_id),
                filename=document.filename,
                file_path=document.file_path,
            )

            # Load workbook
            workbook = self._load_workbook(document.file_path)
            sheet_count = len(workbook.worksheets)

            # Handle empty workbook
            if sheet_count == 0:
                logger.warning(
                    "empty_workbook",
                    document_id=str(document_id),
                    filename=document.filename,
                )
                self._update_document_status(
                    document, "completed", sheet_count=0, record_count=0
                )
                duration_ms = int((time.time() - start_time) * 1000)
                return ExtractionSummary(
                    document_id=document_id,
                    sheets_processed=0,
                    tables_detected=0,
                    records_extracted=0,
                    duration_ms=duration_ms,
                    warnings=["Workbook contains no sheets"],
                )

            # ============================================================
            # PASS 1: Symbol Detection (before table extraction)
            # ============================================================
            symbol_detection_summary = self._run_symbol_detection(
                workbook, document_id
            )
            symbols_detected = symbol_detection_summary.total_symbols
            warnings.extend(symbol_detection_summary.warnings)

            # ============================================================
            # PASS 2: Table Extraction (existing flow)
            # ============================================================
            all_records, total_tables, pass2_warnings = self._run_table_extraction(
                workbook, document_id, document.filename, start_time
            )
            warnings.extend(pass2_warnings)

            # Store all records in database
            self._store_records(all_records, document_id)

            # ============================================================
            # PASS 2.5: Inline Legend Detection (after extraction, before resolution)
            # ============================================================
            inline_symbols = self._detect_inline_legends(all_records, document_id)
            symbols_detected += inline_symbols

            # ============================================================
            # PASS 3: Symbol Resolution (after records exist)
            # ============================================================
            symbol_resolution_summary = self._run_symbol_resolution(document_id)
            symbols_resolved = symbol_resolution_summary.symbols_resolved
            unresolved_symbols = symbol_resolution_summary.unresolved_count
            warnings.extend(symbol_resolution_summary.warnings)

            # ============================================================
            # PASS 4: Cross-Reference Detection (after records exist)
            # ============================================================
            cross_ref_summary = self._run_cross_ref_detection(
                document_id, document.api_key_id
            )
            cross_ref_found = cross_ref_summary.references_found
            cross_ref_resolved = cross_ref_summary.references_resolved
            cross_ref_unresolved = cross_ref_summary.unresolved_count
            warnings.extend(cross_ref_summary.warnings)

            # Update document status to completed
            self._update_document_status(
                document,
                "completed",
                sheet_count=sheet_count,
                record_count=len(all_records),
            )

            duration_ms = int((time.time() - start_time) * 1000)

            logger.info(
                "pipeline_completed",
                document_id=str(document_id),
                sheets_processed=sheet_count,
                tables_detected=total_tables,
                records_extracted=len(all_records),
                symbols_detected=symbols_detected,
                symbols_resolved=symbols_resolved,
                cross_references_found=cross_ref_found,
                cross_references_resolved=cross_ref_resolved,
                total_duration_ms=duration_ms,
                warnings_count=len(warnings),
            )

            return ExtractionSummary(
                document_id=document_id,
                sheets_processed=sheet_count,
                tables_detected=total_tables,
                records_extracted=len(all_records),
                duration_ms=duration_ms,
                warnings=warnings,
                symbols_detected=symbols_detected,
                symbols_resolved=symbols_resolved,
                unresolved_symbols=unresolved_symbols,
                cross_references_found=cross_ref_found,
                cross_references_resolved=cross_ref_resolved,
                cross_references_unresolved=cross_ref_unresolved,
            )

        except Exception as e:
            # Update document status to failed
            if "document" in locals():
                self._update_document_status(
                    document, "failed", error_message=str(e)
                )

            logger.error(
                "pipeline_failed",
                document_id=str(document_id),
                error=str(e),
            )

            raise ExtractionError(
                message=f"Extraction pipeline failed for document {document_id}",
                details={"document_id": str(document_id), "error": str(e)},
            )

    def _run_symbol_detection(
        self,
        workbook: openpyxl.Workbook,
        document_id: UUID,
    ) -> SymbolDetectionSummary:
        """PASS 1: Detect symbol dictionaries in workbook before table extraction.

        Scans all sheets for symbol dictionary patterns and stores detected
        symbols in the database for use in PASS 3 resolution.

        Args:
            workbook: openpyxl Workbook to analyze
            document_id: UUID of the document

        Returns:
            SymbolDetectionSummary with detection statistics
        """
        try:
            # Update progress to symbol detection stage
            self._update_stage_progress(document_id, "detecting_symbols")

            logger.info(
                "pass1_symbol_detection_started",
                document_id=str(document_id),
            )

            # Detect symbol dictionaries in all sheets
            detected_dictionaries, summary = self.symbol_detector.detect_dictionaries(
                workbook
            )

            # Store detected symbols in database
            if detected_dictionaries:
                stored_count = self.symbol_detector.store_symbols(
                    detected_dictionaries, document_id, self.db_session
                )
                logger.info(
                    "pass1_symbols_stored",
                    document_id=str(document_id),
                    dictionaries_found=len(detected_dictionaries),
                    symbols_stored=stored_count,
                )

            logger.info(
                "pass1_symbol_detection_completed",
                document_id=str(document_id),
                dictionaries_found=summary.dictionaries_found,
                total_symbols=summary.total_symbols,
            )

            return summary

        except Exception as e:
            # Symbol detection is non-critical - log and continue
            logger.error(
                "pass1_symbol_detection_failed",
                document_id=str(document_id),
                error=str(e),
            )
            return SymbolDetectionSummary(
                dictionaries_found=0,
                total_symbols=0,
                detected_sheets=[],
                warnings=[f"Symbol detection failed: {str(e)}"],
            )

    def _run_table_extraction(
        self,
        workbook: openpyxl.Workbook,
        document_id: UUID,
        filename: str,
        start_time: float,
    ) -> tuple[list[ExtractionRecord], int, list[str]]:
        """PASS 2: Extract tables from workbook sheets.

        Executes the standard extraction flow:
        TableDetector → HeaderExtractor → MergedCellResolver → RecordBuilder

        Args:
            workbook: openpyxl Workbook to process
            document_id: UUID of the document
            filename: Original filename for source tracking
            start_time: Pipeline start time for duration tracking

        Returns:
            Tuple of (records_list, total_tables_count, warnings_list)
        """
        # Update progress to table extraction stage
        self._update_stage_progress(document_id, "extracting_tables")

        logger.info(
            "pass2_table_extraction_started",
            document_id=str(document_id),
        )

        all_records: list[ExtractionRecord] = []
        total_tables = 0
        warnings: list[str] = []
        sheet_count = len(workbook.worksheets)

        for sheet_idx, sheet in enumerate(workbook.worksheets, start=1):
            sheet_name = sheet.title

            try:
                # Detect tables in this sheet
                boundaries = self.table_detector._detect_boundaries_in_sheet(sheet)

                # Convert boundaries to TableBoundary objects
                table_boundaries = []
                for table_idx, boundary in enumerate(boundaries, start=1):
                    table_boundaries.append(
                        TableBoundary(
                            sheet=sheet_name,
                            table_id=table_idx,
                            start_row=boundary["start_row"],
                            end_row=boundary["end_row"],
                            start_col=boundary["start_col"],
                            end_col=boundary["end_col"],
                            title=boundary.get("title"),
                        )
                    )
                boundaries = table_boundaries

                if not boundaries:
                    warning_msg = f"No tables detected in sheet: {sheet_name}"
                    warnings.append(warning_msg)
                    logger.warning(
                        "no_tables_in_sheet",
                        document_id=str(document_id),
                        sheet_name=sheet_name,
                    )
                    self._update_progress(
                        document_id, sheet_idx, sheet_count, len(all_records)
                    )
                    continue

                total_tables += len(boundaries)

                # Process each table in the sheet
                for boundary in boundaries:
                    try:
                        # Extract headers
                        headers = self.header_extractor.extract_headers(
                            sheet, boundary
                        )

                        # Resolve merged cells
                        resolved_grid, merge_metadata = (
                            self.merged_cell_resolver.resolve_merged_cells(
                                sheet, boundary
                            )
                        )

                        # Build records
                        records = self.record_builder.build_records(
                            resolved_grid=resolved_grid,
                            headers=headers,
                            merge_metadata=merge_metadata,
                            table_boundary=boundary,
                            document_id=document_id,
                            filename=filename,
                        )

                        all_records.extend(records)

                        logger.debug(
                            "table_processed",
                            document_id=str(document_id),
                            sheet_name=sheet_name,
                            table_id=boundary.table_id,
                            records_extracted=len(records),
                        )

                    except ExtractionError as e:
                        # Log but continue with other tables
                        warning_msg = f"Failed to process table {boundary.table_id} in sheet {sheet_name}: {e.message}"
                        warnings.append(warning_msg)
                        logger.warning(
                            "table_extraction_failed",
                            document_id=str(document_id),
                            sheet_name=sheet_name,
                            table_id=boundary.table_id,
                            error=e.message,
                            details=e.details,
                        )

                # Update progress after each sheet
                self._update_progress(
                    document_id, sheet_idx, sheet_count, len(all_records)
                )

                sheet_duration_ms = int((time.time() - start_time) * 1000)
                logger.info(
                    "sheet_processed",
                    document_id=str(document_id),
                    sheet_name=sheet_name,
                    tables_detected=len(boundaries),
                    records_extracted=len(
                        [
                            r
                            for r in all_records
                            if r._source["sheet"] == sheet_name
                        ]
                    ),
                    duration_ms=sheet_duration_ms,
                )

            except Exception as e:
                # Log sheet-level errors but continue with other sheets
                warning_msg = f"Failed to process sheet {sheet_name}: {str(e)}"
                warnings.append(warning_msg)
                logger.error(
                    "sheet_processing_failed",
                    document_id=str(document_id),
                    sheet_name=sheet_name,
                    error=str(e),
                )

        logger.info(
            "pass2_table_extraction_completed",
            document_id=str(document_id),
            total_tables=total_tables,
            total_records=len(all_records),
        )

        return all_records, total_tables, warnings

    def _detect_inline_legends(
        self,
        all_records: list[ExtractionRecord],
        document_id: UUID,
    ) -> int:
        """PASS 2.5: Detect inline legends in extracted records.

        Scans extracted records for inline legend patterns like
        "凡例）○：タグあり、－：タグなし" and stores extracted symbols.

        Args:
            all_records: List of ExtractedRecord objects from PASS 2
            document_id: UUID of the document

        Returns:
            Number of symbols detected from inline legends
        """
        try:
            logger.info(
                "pass2_5_inline_legend_detection_started",
                document_id=str(document_id),
            )

            # Convert ExtractedRecord objects to dictionaries for detection
            record_dicts = [{"content": r.content} for r in all_records]

            # Detect inline legends
            symbols = self.symbol_detector.detect_inline_legends(
                record_dicts, sheet_name="inline_legend"
            )

            if symbols:
                # Store inline symbols in database
                records = []
                for symbol_entry in symbols:
                    records.append(
                        {
                            "document_id": document_id,
                            "sheet_name": symbol_entry.context or "inline_legend",
                            "symbol": symbol_entry.symbol,
                            "meaning": symbol_entry.meaning,
                            "context": "inline",
                        }
                    )

                if records:
                    self.db_session.bulk_insert_mappings(
                        db_models.SymbolDictionary, records
                    )
                    self.db_session.commit()

                    logger.info(
                        "pass2_5_inline_symbols_stored",
                        document_id=str(document_id),
                        symbols_stored=len(records),
                    )

            logger.info(
                "pass2_5_inline_legend_detection_completed",
                document_id=str(document_id),
                symbols_detected=len(symbols),
            )

            return len(symbols)

        except Exception as e:
            logger.error(
                "pass2_5_inline_legend_detection_failed",
                document_id=str(document_id),
                error=str(e),
            )
            return 0

    def _run_symbol_resolution(
        self,
        document_id: UUID,
    ) -> SymbolResolutionSummary:
        """PASS 3: Resolve symbols in extracted content.

        Uses symbol dictionaries detected in PASS 1 to resolve symbol values
        in extracted records and update resolved_content field.

        Args:
            document_id: UUID of the document

        Returns:
            SymbolResolutionSummary with resolution statistics
        """
        try:
            # Update progress to symbol resolution stage
            self._update_stage_progress(document_id, "resolving_symbols")

            logger.info(
                "pass3_symbol_resolution_started",
                document_id=str(document_id),
            )

            # Run symbol resolution
            summary = self.symbol_resolver.resolve_document(
                document_id, self.db_session
            )

            logger.info(
                "pass3_symbol_resolution_completed",
                document_id=str(document_id),
                records_processed=summary.records_processed,
                symbols_resolved=summary.symbols_resolved,
                unresolved_count=summary.unresolved_count,
            )

            return summary

        except Exception as e:
            # Symbol resolution is non-critical - log and continue
            logger.error(
                "pass3_symbol_resolution_failed",
                document_id=str(document_id),
                error=str(e),
            )
            return SymbolResolutionSummary(
                document_id=document_id,
                records_processed=0,
                symbols_resolved=0,
                unresolved_count=0,
                unresolved_symbols=[],
                warnings=[f"Symbol resolution failed: {str(e)}"],
            )

    def _load_document(self, document_id: UUID) -> db_models.Document:
        """Load document metadata from database.

        Args:
            document_id: UUID of the document

        Returns:
            Document ORM model

        Raises:
            ExtractionError: If document not found
        """
        stmt = select(db_models.Document).where(
            db_models.Document.id == document_id
        )
        document = self.db_session.scalar(stmt)

        if not document:
            raise ExtractionError(
                message=f"Document not found: {document_id}",
                details={"document_id": str(document_id)},
            )

        return document

    def _load_workbook(self, file_path: str) -> openpyxl.Workbook:
        """Load Excel workbook from file storage.

        Args:
            file_path: Path to the Excel file

        Returns:
            openpyxl Workbook object

        Raises:
            ExtractionError: If file not found or invalid format
        """
        try:
            # Check if file exists
            path = Path(file_path)
            if not path.exists():
                raise ExtractionError(
                    message="File not found",
                    details={
                        "file_path": file_path,
                        "error_code": "FILE_NOT_FOUND",
                    },
                )

            # Load workbook with data_only=True to get calculated values
            workbook = openpyxl.load_workbook(file_path, data_only=True)
            return workbook

        except FileNotFoundError:
            raise ExtractionError(
                message="File not found",
                details={"file_path": file_path, "error_code": "FILE_NOT_FOUND"},
            )
        except InvalidFileException as e:
            raise ExtractionError(
                message="Invalid Excel file format",
                details={
                    "file_path": file_path,
                    "error_code": "INVALID_EXCEL_FORMAT",
                    "error": str(e),
                },
            )
        except Exception as e:
            raise ExtractionError(
                message=f"Failed to load workbook: {str(e)}",
                details={"file_path": file_path, "error": str(e)},
            )

    def _update_document_status(
        self,
        document: db_models.Document,
        status: str,
        sheet_count: Optional[int] = None,
        record_count: Optional[int] = None,
        error_message: Optional[str] = None,
    ) -> None:
        """Update document status in database.

        Args:
            document: Document ORM model
            status: New status value
            sheet_count: Optional sheet count
            record_count: Optional record count
            error_message: Optional error message for failed status
        """
        document.status = status

        if status == "completed":
            document.processed_at = datetime.now(timezone.utc)
            if sheet_count is not None:
                document.sheet_count = sheet_count
            if record_count is not None:
                document.record_count = record_count

        if status == "failed" and error_message:
            document.error_message = error_message

        self.db_session.commit()

        logger.debug(
            "document_status_updated",
            document_id=str(document.id),
            status=status,
            sheet_count=sheet_count,
            record_count=record_count,
        )

    def _store_records(
        self, records: list[ExtractionRecord], document_id: UUID
    ) -> None:
        """Store extracted records in database using batch insert.

        Args:
            records: List of ExtractedRecord objects
            document_id: UUID of the document
        """
        if not records:
            logger.debug(
                "no_records_to_store",
                document_id=str(document_id),
            )
            return

        try:
            # Convert ExtractionRecord objects to ORM model dictionaries
            batch_size = 500
            total_inserted = 0

            for i in range(0, len(records), batch_size):
                batch = records[i : i + batch_size]

                # Prepare batch data for bulk insert
                batch_data = []
                for record in batch:
                    batch_data.append(
                        {
                            "document_id": document_id,
                            "sheet_name": record._source["sheet"],
                            "table_title": record._source.get("table_title"),
                            "row_number": record._source["row"],
                            "col_range": record._source["col_range"],
                            "content": record.content,
                            "headers": record.headers,
                            "resolved_content": record.resolved_content,
                        }
                    )

                # Perform bulk insert
                self.db_session.bulk_insert_mappings(
                    db_models.ExtractedRecord, batch_data
                )
                self.db_session.commit()

                total_inserted += len(batch)

                logger.debug(
                    "records_batch_inserted",
                    document_id=str(document_id),
                    batch_size=len(batch),
                    total_inserted=total_inserted,
                    total_records=len(records),
                )

            logger.info(
                "records_stored",
                document_id=str(document_id),
                total_records=len(records),
            )

        except Exception as e:
            self.db_session.rollback()
            logger.error(
                "records_storage_failed",
                document_id=str(document_id),
                record_count=len(records),
                error=str(e),
            )
            raise ExtractionError(
                message=f"Failed to store extracted records: {str(e)}",
                details={"document_id": str(document_id), "error": str(e)},
            )

    def _update_progress(
        self,
        document_id: UUID,
        current_sheet: int,
        total_sheets: int,
        records_so_far: int,
    ) -> None:
        """Update progress tracking in Redis.

        Args:
            document_id: UUID of the document
            current_sheet: Current sheet number (1-indexed)
            total_sheets: Total number of sheets
            records_so_far: Number of records extracted so far
        """
        if not self.redis_client:
            return

        try:
            key = f"processing:documents:{document_id}"
            progress_data = {
                "status": "processing",
                "current_sheet": current_sheet,
                "total_sheets": total_sheets,
                "records": records_so_far,
            }

            # Store as JSON string
            self.redis_client.setex(
                key,
                3600,  # TTL: 1 hour
                json.dumps(progress_data),
            )

            logger.debug(
                "progress_updated",
                document_id=str(document_id),
                current_sheet=current_sheet,
                total_sheets=total_sheets,
                records=records_so_far,
            )

        except Exception as e:
            # Log but don't fail - progress tracking is non-critical
            logger.warning(
                "progress_update_failed",
                document_id=str(document_id),
                error=str(e),
            )

    def _update_stage_progress(
        self,
        document_id: UUID,
        stage: str,
        extra_data: Optional[dict] = None,
    ) -> None:
        """Update processing stage progress in Redis.

        Stages:
            - detecting_symbols: PASS 1 - Symbol detection
            - extracting_tables: PASS 2 - Table extraction
            - resolving_symbols: PASS 3 - Symbol resolution
            - detecting_references: PASS 4 - Cross-reference detection

        Args:
            document_id: UUID of the document
            stage: Current processing stage
            extra_data: Optional additional data to include in progress
        """
        if not self.redis_client:
            return

        try:
            key = f"processing:documents:{document_id}"
            progress_data = {
                "status": "processing",
                "stage": stage,
            }

            if extra_data:
                progress_data.update(extra_data)

            # Store as JSON string
            self.redis_client.setex(
                key,
                3600,  # TTL: 1 hour
                json.dumps(progress_data),
            )

            logger.debug(
                "stage_progress_updated",
                document_id=str(document_id),
                stage=stage,
            )

        except Exception as e:
            # Log but don't fail - progress tracking is non-critical
            logger.warning(
                "stage_progress_update_failed",
                document_id=str(document_id),
                stage=stage,
                error=str(e),
            )

    def _run_cross_ref_detection(
        self,
        document_id: UUID,
        api_key_id: Optional[UUID] = None,
    ) -> CrossRefResolutionSummary:
        """PASS 4: Run cross-reference detection and resolution for extracted records.

        Detects cross-reference patterns in extracted records, attempts to resolve
        them to target records, and stores the results in the database.

        Args:
            document_id: UUID of the document to process
            api_key_id: Optional API key ID for cross-document resolution

        Returns:
            CrossRefResolutionSummary with detection and resolution statistics
        """
        try:
            # Update progress to cross-reference detection stage
            self._update_stage_progress(document_id, "detecting_references")

            logger.info(
                "pass4_cross_ref_detection_started",
                document_id=str(document_id),
            )

            # Run cross-reference detection
            summary = self.cross_ref_detector.process_document(
                document_id, self.db_session, api_key_id
            )

            logger.info(
                "pass4_cross_ref_detection_completed",
                document_id=str(document_id),
                references_found=summary.references_found,
                references_resolved=summary.references_resolved,
                unresolved_count=summary.unresolved_count,
            )

            return summary

        except Exception as e:
            # Log error but don't fail the pipeline - cross-ref detection is non-critical
            logger.error(
                "pass4_cross_ref_detection_failed",
                document_id=str(document_id),
                error=str(e),
            )

            # Return empty summary on failure
            return CrossRefResolutionSummary(
                document_id=document_id,
                references_found=0,
                references_resolved=0,
                unresolved_count=0,
                unresolved_refs=[],
                warnings=[f"Cross-reference detection failed: {str(e)}"],
            )
