""" PDF Parser — Extracts structured text, tables, and form fields from text-based (non-scanned) PDF documents using pdfplumber. This is the "Puff n Parse" lane — handles clean, digital PDFs where text is directly embedded in the PDF layers. """ import pdfplumber from pathlib import Path def extract_from_pdf(file_path: str | Path) -> dict: """ Extract all text and table data from a text-based PDF. Returns a dict with: - 'raw_text': Full concatenated text from all pages - 'tables': List of tables (each table is a list of rows) - 'page_count': Number of pages processed - 'pages': Per-page text content """ result = { "raw_text": "", "tables": [], "page_count": 0, "pages": [], } try: with pdfplumber.open(str(file_path)) as pdf: result["page_count"] = len(pdf.pages) for page in pdf.pages: # Extract text page_text = page.extract_text() or "" result["pages"].append(page_text) result["raw_text"] += page_text + "\n\n" # Extract tables page_tables = page.extract_tables() if page_tables: for table in page_tables: # Clean up table: replace None with empty string cleaned_table = [ [cell if cell is not None else "" for cell in row] for row in table if row # Skip empty rows ] if cleaned_table: result["tables"].append(cleaned_table) except Exception as e: result["raw_text"] = f"Error extracting PDF: {str(e)}" return result def extract_tables_as_fields(tables: list[list[list[str]]]) -> list[dict]: """ Convert extracted tables into field-value pairs. Strategy: - If a table has 2 columns, treat col[0] as field name and col[1] as value - If table uses colons inline (Key : Value), extract them logically - Otherwise, generate generic field names (Column_1, Column_2, etc.) """ fields = [] for table_idx, table in enumerate(tables): if not table or len(table) < 1: continue # Check if it's a simple 2-column key-value table if all(len(row) == 2 for row in table): for row in table: name = str(row[0]).strip() value = str(row[1]).strip() if name and name != value: # Skip if name equals value (likely a header repeat) fields.append({ "name": name, "value": value, "field_type": _infer_type(value), "confidence": 0.95, }) continue # Check for inline colon separators (e.g. Key : Value) has_colon_separators = False for row in table: for cell in row: c_str = str(cell).strip() if c_str == ":" or (c_str.endswith(":") and len(c_str) > 1): has_colon_separators = True break if has_colon_separators: break if has_colon_separators: for row in table: fields.extend(_extract_inline_key_values(row)) else: # Multi-column table: use first row as headers headers = [str(h).strip() or f"Column_{i+1}" for i, h in enumerate(table[0])] for row_idx, row in enumerate(table[1:], start=1): non_empty_cells = [str(c).strip() for c in row if c and str(c).strip()] # Heuristic 1: If row has exactly 1 cell containing a colon (e.g. "Renew Mailbox(es): 3") if len(non_empty_cells) == 1 and ":" in non_empty_cells[0]: parts = non_empty_cells[0].split(":", 1) if len(parts[0]) < 50: # Ensures the key isn't a massive paragraph fields.append({ "name": parts[0].strip(), "value": parts[1].strip(), "field_type": _infer_type(parts[1].strip()), "confidence": 0.94, }) continue # Heuristic 2: If row has exactly 2 non-empty cells (e.g. "Sub Total", "$25.13") if len(non_empty_cells) == 2: key_cand, val_cand = non_empty_cells[0], non_empty_cells[1] if len(key_cand) < 50 and not key_cand[0].isdigit(): fields.append({ "name": key_cand, "value": val_cand, "field_type": _infer_type(val_cand), "confidence": 0.96, }) continue # Default: map against column headers for col_idx, cell in enumerate(row): if col_idx < len(headers): val = str(cell).strip() if cell else "" if val: field_name = f"{headers[col_idx]} (Row {row_idx})" fields.append({ "name": field_name, "value": val, "field_type": _infer_type(val), "confidence": 0.90, }) return fields def _extract_inline_key_values(row: list) -> list[dict]: fields = [] cleaned_row = [str(c).strip() if c else "" for c in row] i = 0 n = len(cleaned_row) while i < n: cell = cleaned_row[i] if not cell: i += 1 continue key = None if cell == ":": i += 1 continue if cell.endswith(":"): key = cell[:-1].strip() i += 1 elif i + 1 < n and cleaned_row[i+1] == ":": key = cell i += 2 if key: val_parts = [] while i < n: next_cell = cleaned_row[i] if not next_cell: i += 1 continue if next_cell.endswith(":") or (i + 1 < n and cleaned_row[i+1] == ":"): break val_parts.append(next_cell) i += 1 value_str = " ".join(val_parts).strip() # Fix split AM/PM times value_str = value_str.replace(" P M", " PM").replace(" A M", " AM") # If "P" is at the end and "M" is next, it's joined as "P M" so the replace fixes it # Additional cleanup for things like "12:30:04 P M" -> "12:30:04 PM" import re value_str = re.sub(r'\s+P\s*M\b', ' PM', value_str, flags=re.IGNORECASE) value_str = re.sub(r'\s+A\s*M\b', ' AM', value_str, flags=re.IGNORECASE) fields.append({ "name": key, "value": value_str, "field_type": _infer_type(value_str), "confidence": 0.92 }) else: # If it's not a key and not a value following a key, we'll emit it generically so data isn't lost # But we name it "Orphaned Value" which json_mapper will try to rename semantically fields.append({ "name": "Unknown Field", "value": cell, "field_type": _infer_type(cell), "confidence": 0.85 }) i += 1 return fields def _infer_type(value: str) -> str: """Simple heuristic to infer the data type of a value.""" if not value: return "text" # Check for numbers cleaned = value.replace(",", "").replace(" ", "").replace("R", "").replace("$", "") try: float(cleaned) return "number" except ValueError: pass # Check for dates (simple patterns) date_indicators = ["/", "-"] digit_count = sum(1 for c in value if c.isdigit()) if digit_count >= 4 and any(sep in value for sep in date_indicators): return "date" return "text"