"""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)}"], )