"""Symbol resolution for extracted records. This module provides functionality to resolve symbols in extracted content using symbol dictionaries detected from the same document. """ from typing import Any from uuid import UUID import structlog from sqlalchemy.orm import Session from src.db.models import ExtractedRecord as ExtractedRecordORM from src.db.models import SymbolDictionary from src.extraction.models import SymbolResolutionSummary logger = structlog.get_logger(__name__) # Batch size for database updates BATCH_SIZE = 500 # Common symbol characters for heuristic detection SYMBOL_CHARS = set("○×△◎-*●□■◇◆✓✗☐☑") # Context value for custom symbols (must match symbol_service.py) CUSTOM_SYMBOL_CONTEXT = "api_upload" class SymbolResolver: """Resolves symbols in extracted records using detected dictionaries. This class loads symbol dictionaries from the database and resolves symbol values in extracted records, creating resolved_content fields with human-readable meanings. """ def __init__(self): """Initialize the SymbolResolver.""" pass def resolve_document( self, document_id: UUID, session: Session ) -> SymbolResolutionSummary: """Resolve all symbols in extracted records for a document. Args: document_id: UUID of the document to process session: SQLAlchemy database session Returns: SymbolResolutionSummary with resolution statistics """ logger.info("symbol_resolution_started", document_id=str(document_id)) # Step 1: Build symbol lookup from database symbol_lookup = self._build_symbol_lookup(document_id, session) if not symbol_lookup: logger.info( "no_symbols_to_resolve", document_id=str(document_id), ) return SymbolResolutionSummary( document_id=document_id, records_processed=0, symbols_resolved=0, unresolved_count=0, unresolved_symbols=[], warnings=[], ) logger.info( "symbol_lookup_built", document_id=str(document_id), symbol_count=len(symbol_lookup), ) # Step 2: Load extracted records for this document records = ( session.query(ExtractedRecordORM) .filter(ExtractedRecordORM.document_id == document_id) .all() ) if not records: logger.info( "no_records_to_resolve", document_id=str(document_id), ) return SymbolResolutionSummary( document_id=document_id, records_processed=0, symbols_resolved=0, unresolved_count=0, unresolved_symbols=[], warnings=["No extracted records found for document"], ) # Step 3: Resolve symbols in each record updates: list[tuple[UUID, dict]] = [] total_symbols_resolved = 0 all_unresolved: set[str] = set() warned_symbols: set[str] = set() # Track symbols we've warned about for record in records: resolved_content, resolved_count, unresolved = self._resolve_record( record.content, symbol_lookup ) if resolved_content: updates.append((record.id, resolved_content)) total_symbols_resolved += resolved_count # Log warnings for unknown symbols (first occurrence only) for symbol in unresolved: if symbol not in warned_symbols: logger.warning( "unknown_symbol_encountered", document_id=str(document_id), symbol=symbol, ) warned_symbols.add(symbol) all_unresolved.update(unresolved) # Step 4: Batch update database if updates: self._batch_update_resolved_content(updates, session) # Step 5: Build and return summary summary = SymbolResolutionSummary( document_id=document_id, records_processed=len(records), symbols_resolved=total_symbols_resolved, unresolved_count=len(all_unresolved), unresolved_symbols=sorted(list(all_unresolved)), warnings=[], ) logger.info( "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 def _build_symbol_lookup( self, document_id: UUID, session: Session ) -> dict[str, str]: """Build a symbol-to-meaning lookup dictionary from database. Custom symbols (context="api_upload") take precedence over auto-detected symbols. This is achieved by loading auto-detected symbols first, then custom symbols, allowing custom symbols to override. Args: document_id: UUID of the document session: SQLAlchemy database session Returns: Dictionary mapping symbol strings to their meanings """ symbol_lookup: dict[str, str] = {} # Query all symbols for this document, ordered so custom symbols come LAST # This ensures custom symbols override auto-detected ones symbols = ( session.query(SymbolDictionary) .filter(SymbolDictionary.document_id == document_id) .order_by( # Custom symbols (api_upload) should come last to override # NULL context or other context values come first (SymbolDictionary.context == CUSTOM_SYMBOL_CONTEXT).asc() ) .all() ) for symbol_record in symbols: symbol = symbol_record.symbol meaning = symbol_record.meaning is_custom = symbol_record.context == CUSTOM_SYMBOL_CONTEXT # Check if we're overriding an existing symbol if symbol in symbol_lookup: if is_custom: # Custom symbol overriding auto-detected logger.info( "custom_symbol_overriding_auto_detected", document_id=str(document_id), symbol=symbol, auto_meaning=symbol_lookup[symbol], custom_meaning=meaning, ) symbol_lookup[symbol] = meaning else: # Duplicate auto-detected symbol - first wins, skip logger.debug( "duplicate_auto_symbol_in_lookup", document_id=str(document_id), symbol=symbol, ) else: symbol_lookup[symbol] = meaning return symbol_lookup def _resolve_record( self, content: dict[str, Any], symbol_lookup: dict[str, str] ) -> tuple[dict[str, Any] | None, int, set[str]]: """Resolve symbols in a single record's content. Args: content: Dictionary of field name -> value from extracted record symbol_lookup: Dictionary mapping symbols to meanings Returns: Tuple of: - resolved_content: dict with original + resolved fields, or None if no symbols - resolved_count: number of symbols resolved - unresolved: set of symbols encountered but not in lookup """ if not content: return None, 0, set() resolved_content = content.copy() resolved_count = 0 unresolved: set[str] = set() for field, value in content.items(): # Skip None values and non-string values if value is None: continue # Convert to string and strip whitespace value_str = str(value).strip() if not value_str: continue # Check if this value is a known symbol if value_str in symbol_lookup: # Known symbol - add resolved field resolved_field = f"{field}_resolved" resolved_content[resolved_field] = symbol_lookup[value_str] resolved_count += 1 elif self._looks_like_symbol(value_str): # Looks like a symbol but not in dictionary unresolved.add(value_str) if resolved_count > 0: return resolved_content, resolved_count, unresolved return None, 0, unresolved def _looks_like_symbol(self, value: str) -> bool: """Check if a value looks like it might be a symbol. Uses heuristics to identify potential symbols: - Short strings (1-10 characters) - Contains common symbol characters - Short alphanumeric codes (max 4 chars) Args: value: String value to check Returns: True if the value looks like a potential symbol """ if not value or len(value) > 10: return False # Check for common symbol characters if any(c in SYMBOL_CHARS for c in value): return True # Check for short alphanumeric codes (e.g., "07", "A1") if value.isalnum() and len(value) <= 4: return True return False def _batch_update_resolved_content( self, updates: list[tuple[UUID, dict]], session: Session ) -> int: """Update resolved_content for multiple records in batches. Args: updates: List of (record_id, resolved_content) tuples session: SQLAlchemy database session Returns: Total number of records updated """ if not updates: return 0 total_updated = 0 # Process in batches for i in range(0, len(updates), BATCH_SIZE): batch = updates[i : i + BATCH_SIZE] try: # Update each record in the batch for record_id, resolved_content in batch: session.query(ExtractedRecordORM).filter( ExtractedRecordORM.id == record_id ).update( {"resolved_content": resolved_content}, synchronize_session=False, ) session.commit() total_updated += len(batch) logger.debug( "batch_update_completed", batch_size=len(batch), total_updated=total_updated, ) except Exception as e: session.rollback() logger.error( "batch_update_failed", batch_start=i, batch_size=len(batch), error=str(e), ) raise logger.info( "all_batches_updated", total_records=total_updated, ) return total_updated