""" JSON Mapper — Structures raw extracted text into field:value JSON pairs. Uses heuristic pattern matching to identify: - Key: Value patterns (colon-separated) - Key = Value patterns - Tabular data (from PDF table extraction) - Labelled form fields - Multi-line grouped content Designed to handle diverse document formats common in African government and business contexts: invoices, certificates, forms, applications. """ import re from models.schemas import ExtractedField def map_text_to_fields(raw_text: str, tables: list | None = None, ocr_blocks: list | None = None) -> list[ExtractedField]: """ Convert raw extracted text (and optional table/block data) into structured ExtractedField objects. Args: raw_text: The full text extracted from the document. tables: Optional list of table field dicts from pdf_parser. ocr_blocks: Optional list of OCR detection blocks with confidence scores. Returns: List of ExtractedField objects ready for the frontend. """ fields: list[ExtractedField] = [] seen_names: set[str] = set() # Priority 1: Use table data if available (most structured) if tables: for table_field in tables: name = table_field.get("name", "").strip() if name and name not in seen_names: fields.append(ExtractedField( name=name, value=table_field.get("value", ""), field_type=table_field.get("field_type", "text"), confidence=table_field.get("confidence", 0.9), )) seen_names.add(name) # Priority 2: Parse raw text for key-value patterns text_fields = _extract_key_value_pairs(raw_text) for field in text_fields: if field.name not in seen_names: fields.append(field) seen_names.add(field.name) # Priority 3: If we have OCR blocks with confidence, enhance field confidence if ocr_blocks and fields: _enhance_confidence(fields, ocr_blocks) # If no structured fields found, create line-by-line fields if not fields: fields = _fallback_line_fields(raw_text) # Priority 4: Semantic Refinement Pass # Rename generic names (like "Column_1 (Row 1)" or "Line X") based on the value's content for field in fields: _refine_field_name_by_value(field) # Deduplicate after refinement final_fields = [] seen_final_names = set() for f in fields: # If the name is duplicated, append a counter original_name = f.name counter = 1 while f.name in seen_final_names: f.name = f"{original_name} {counter}" counter += 1 final_fields.append(f) seen_final_names.add(f.name) return final_fields def _extract_key_value_pairs(text: str) -> list[ExtractedField]: """ Extract key-value pairs from text using multiple pattern strategies. """ fields = [] # Strategy 1: "Key: Value" patterns (most common in forms) colon_pattern = re.compile( r'^[\s]*([A-Za-z][A-Za-z0-9\s/\-_\(\)\.]{1,50})\s*[:]\s*(.+)$', re.MULTILINE, ) for match in colon_pattern.finditer(text): name = match.group(1).strip() value = match.group(2).strip() if len(name) >= 2 and len(value) >= 1 and not _is_noise(name): fields.append(ExtractedField( name=name, value=value, field_type=_infer_type(value), confidence=0.85, )) # Strategy 2: "Key = Value" patterns equals_pattern = re.compile( r'^[\s]*([A-Za-z][A-Za-z0-9\s/\-_]{1,40})\s*[=]\s*(.+)$', re.MULTILINE, ) for match in equals_pattern.finditer(text): name = match.group(1).strip() value = match.group(2).strip() if len(name) >= 2 and len(value) >= 1 and not _is_noise(name): fields.append(ExtractedField( name=name, value=value, field_type=_infer_type(value), confidence=0.80, )) # Strategy 3: Tab-separated fields (common in printed forms) tab_pattern = re.compile( r'^[\s]*([A-Za-z][A-Za-z0-9\s]{1,40})\t+(.+)$', re.MULTILINE, ) for match in tab_pattern.finditer(text): name = match.group(1).strip() value = match.group(2).strip() if len(name) >= 2 and len(value) >= 1 and not _is_noise(name): fields.append(ExtractedField( name=name, value=value, field_type=_infer_type(value), confidence=0.75, )) return fields def _fallback_line_fields(text: str) -> list[ExtractedField]: """ When no key-value patterns are found, create fields from non-empty lines of text. Each line becomes a "Line N" field. """ fields = [] lines = [line.strip() for line in text.split("\n") if line.strip()] for i, line in enumerate(lines, start=1): # Skip very short lines (likely noise) or page separators if len(line) < 3 or line.startswith("---"): continue fields.append(ExtractedField( name=f"Line {i}", value=line, field_type="text", confidence=0.60, )) return fields def _enhance_confidence(fields: list[ExtractedField], ocr_blocks: list[dict]) -> None: """ If OCR blocks contain confidence scores, use them to update the confidence of matching fields. """ # Build a lookup of text → confidence from OCR blocks block_confidences: dict[str, float] = {} for block in ocr_blocks: text = block.get("text", "").strip().lower() conf = block.get("confidence", 0.0) if text: block_confidences[text] = conf # Match fields to OCR blocks by checking if field value appears in blocks for field in fields: value_lower = field.value.strip().lower() if value_lower in block_confidences: field.confidence = block_confidences[value_lower] def _infer_type(value: str) -> str: """Heuristic type inference for extracted values.""" if not value: return "text" cleaned = value.replace(",", "").replace(" ", "").replace("R", "").replace("$", "").replace("€", "") # Number check try: float(cleaned) return "number" except ValueError: pass # Date check date_patterns = [ r'\d{1,2}[/\-]\d{1,2}[/\-]\d{2,4}', # DD/MM/YYYY or similar r'\d{4}[/\-]\d{1,2}[/\-]\d{1,2}', # YYYY-MM-DD ] for pattern in date_patterns: if re.match(pattern, value.strip()): return "date" # Email check if re.match(r'[^@]+@[^@]+\.[^@]+', value.strip()): return "email" # Phone check if re.match(r'^[\+]?[\d\s\-\(\)]{7,15}$', value.strip()): return "phone" return "text" def _is_noise(text: str) -> bool: """Check if a detected field name is likely noise or a false positive.""" noise_words = { "page", "date", "time", "http", "www", "copyright", "all rights", "reserved", "confidential", } lower = text.lower().strip() return lower in noise_words or len(lower) < 2 def _refine_field_name_by_value(field: ExtractedField) -> None: """ If the field name is generic (e.g. Column_1, Line 3, RECEIPT) and the value matches known semantic patterns, rename the field to something intelligent. """ # Check if the name looks generic or noisy generic_patterns = [ r'^column_\d+', r'^line \d+', r'row \d+', r'^receipt$', r'^invoice$', r'^unknown$' ] name_lower = field.name.lower().strip() is_generic = any(re.search(p, name_lower) for p in generic_patterns) # Or if the name is unusually long (which means the parser mapped a whole text block as the name) is_long_name = len(field.name) > 40 # We always try to extract 'Order #' if the value has it, regardless of how generic the name is. value_lower = str(field.value).lower().strip() # 1. Order Number if 'order #' in value_lower or 'order no' in value_lower: field.name = 'Order Number' return # 2. Invoice Number if 'invoice #' in value_lower or 'invoice no' in value_lower: field.name = 'Invoice Number' return # If the name isn't generic and isn't super long, we trust it. if not is_generic and not is_long_name: return val = str(field.value).strip() # 3. Website / URL if re.match(r'^(https?://)?(www\.)?[a-zA-Z0-9-]+\.[a-zA-Z]{2,}(/.*)?$', val): field.name = 'Website' return # 4. Email Address if re.match(r'^[^@\s]+@[^@\s]+\.[^@\s]+$', val): field.name = 'Email Address' return # 5. Date & Time combined if re.match(r'^\d{1,2}[/\-]\d{1,2}[/\-]\d{2,4}\s+\d{1,2}:\d{2}(:\d{2})?\s*(AM|PM|am|pm)?$', val): field.name = 'Date & Time' return # 6. Date only date_patterns = [ r'^\d{1,2}[/\-]\d{1,2}[/\-]\d{2,4}$', r'^\d{4}[/\-]\d{1,2}[/\-]\d{1,2}$', r'^[a-zA-Z]{3,9} \d{1,2},? \d{4}$' # e.g. Jan 1, 2023 ] if any(re.match(p, val) for p in date_patterns): field.name = 'Date' return # 7. Time only if re.match(r'^\d{1,2}:\d{2}(:\d{2})?\s*(AM|PM|am|pm)?$', val): field.name = 'Time' return # 7. Physical Address (heuristic: contains street/st/ave/blvd/suite/city/state/zip) address_keywords = ['street', 'st', 'st.', 'avenue', 'ave', 'boulevard', 'blvd', 'suite', 'road', 'rd', 'drive', 'dr'] if len(val) > 15 and any(kw in value_lower for kw in address_keywords) and any(c.isdigit() for c in val): field.name = 'Address' return # 8. Phone Number if re.match(r'^[\+]?[\d\s\-\(\)]{10,15}$', val): field.name = 'Phone Number' return # 9. Amount / Currency if re.match(r'^[\$\£\€R]?\s*\d{1,3}(,\d{3})*(\.\d{2})?$', val) and any(c in val for c in '$£€R.'): field.name = 'Amount' return # If it was a long name, and we couldn't classify it, we just call it "Text Block" if is_long_name: field.name = 'Text Block'