# _1099_int_2024_parser.py """ Form 1099-INT (2024 revision) parser. Consumes page dicts from extract_pages() in app.py: {page, text, text_alt, width, height, source, image} Returns a flat dict compatible with FIELDS_1099_INT_2024. Only page 1 is used. """ from __future__ import annotations import re from typing import Any # --------------------------------------------------------------------- # Regex helpers # --------------------------------------------------------------------- _EIN_RE = re.compile(r"(?(?:\d{1,3}(?:,\d{3})+|\d+)(?:\.\d{1,2})?)" r"(?![\w./-])" ) _DOLLAR_AMOUNT_RE = re.compile(r"\$[ \t]*([\d,]+(?:\.\d{1,2})?)") _BOX_NUMBER_FOLLOW_RE = re.compile( r"^\s*(?:interest\s+income|early\s+withdrawal|interest\s+on\s+u\.?s\.?|" r"federal\s+income\s+tax|investment\s+interest|foreign\s+tax|" r"foreign\s+country|tax-exempt\s+interest|specified\s+private|" r"market\s+discount|bond\s+premium|" r"tax-exempt\s+and\s+tax\s+credit|state\s+tax|" r"state\s+identification|state\b)", re.IGNORECASE, ) _BOX_ANCHORS = { "box1_interest_income": [r"\b1\s*[.\-:]?\s*Interest\s+income\b"], "box2_early_withdrawal_penalty": [r"\b2\s*[.\-:]?\s*Early\s+withdrawal\s+penalty\b"], "box3_us_savings_bonds_treasury": [ r"\b3\s*[.\-:]?\s*Interest\s+on\s+U\.?S\.?\s+Savings\s+Bonds", r"\b3\s*[.\-:]?\s*Interest\s+on\s+U\.?S\.?\s+Treasury", ], "box4_fed_tax_withheld": [r"\b4\s*[.\-:]?\s*Federal\s+income\s+tax\s+withheld\b"], "box5_investment_interest_expense": [ r"\b5\s*[.\-:]?\s*Investment\s+interest\s+expense\b", r"\b5\s*[.\-:]?\s*Investment\s+expenses?\b", ], "box6_foreign_tax_paid": [r"\b6\s*[.\-:]?\s*Foreign\s+tax\s+paid\b"], "box7_foreign_country": [ r"\b7\s*[.\-:]?\s*Foreign\s+country\s+or\s+U\.?S\.?\s+(?:possession|territory)\b", ], "box8_tax_exempt_interest": [r"\b8\s*[.\-:]?\s*Tax-exempt\s+interest\b"], "box9_specified_private_activity": [ r"\b9\s*[.\-:]?\s*Specified\s+private\s+activity\s+bond\s+interest\b", ], "box10_market_discount": [r"\b10\s*[.\-:]?\s*Market\s+discount\b"], "box11_bond_premium": [r"\b11\s*[.\-:]?\s*Bond\s+premium\b"], "box12_bond_premium_treasury": [ r"\b12\s*[.\-:]?\s*Bond\s+premium\s+on\s+Treasury\s+obligations\b", ], "box13_bond_premium_tax_exempt": [ r"\b13\s*[.\-:]?\s*Bond\s+premium\s+on\s+tax-exempt\s+bond\b", ], "box14_cusip_no": [ r"\b14\s*[.\-:]?\s*(?:Tax-exempt\s+and\s+tax\s+credit\s+bond\s+)?CUSIP\s+no\.?", ], "box15_state": [r"\b15\s*[.\-:]?\s*State\b"], "box16_state_id_no": [r"\b16\s*[.\-:]?\s*State\s+identification\s+no\.?"], "box17_state_tax_withheld": [r"\b17\s*[.\-:]?\s*State\s+tax\s+withheld\b"], } _BOX_REGEXES = { key: [re.compile(p, re.IGNORECASE) for p in patterns] for key, patterns in _BOX_ANCHORS.items() } _ANY_BOX_LABEL_RE = re.compile( r"\b(?:box\s*)?(?:[1-9]|1[0-7])\s*[.\-:]?\s*" r"(?:interest\s+income|early\s+withdrawal|interest\s+on\s+u\.?s\.?|" r"federal\s+income\s+tax|investment\s+interest|investment\s+expenses?|" r"foreign\s+tax|foreign\s+country|tax-exempt\s+interest|specified\s+private|" r"market\s+discount|bond\s+premium|" r"tax-exempt\s+and\s+tax\s+credit|state\s+tax|" r"state\s+identification|state\b)", re.IGNORECASE, ) # --------------------------------------------------------------------- # TIN validation # --------------------------------------------------------------------- _INVALID_EIN_PREFIXES = { "00", "07", "08", "09", "17", "18", "19", "28", "29", "38", "39", "48", "49", "58", "59", "68", "69", "78", "79", "88", "89", "96", "97", "98", "99", } def _is_valid_ssn(tin: str) -> bool: digits = re.sub(r"\D", "", tin or "") if len(digits) != 9: return False area, group, serial = digits[:3], digits[3:5], digits[5:] if area in ("000", "666") or area.startswith("9"): return False if group == "00" or serial == "0000": return False return True def _is_valid_ein(tin: str) -> bool: digits = re.sub(r"\D", "", tin or "") if len(digits) != 9: return False return digits[:2] not in _INVALID_EIN_PREFIXES def _is_valid_tin(tin: str) -> bool: if not tin: return False if re.match(r"^\d{3}[-.]\d{2}[-.]\d{4}$", tin): return _is_valid_ssn(tin) if re.match(r"^\d{2}[-.]\d{7}$", tin): return _is_valid_ein(tin) return False # --------------------------------------------------------------------- # Text utilities # --------------------------------------------------------------------- def _normalise_text(value: str) -> str: value = (value or "").replace("\x00", " ").replace("\u00a0", " ") value = value.replace("\r\n", "\n").replace("\r", "\n") value = re.sub(r"[ \t]+", " ", value) value = re.sub(r"\n[ \t]+", "\n", value) value = re.sub(r"\n{3,}", "\n\n", value) return value.strip() def _cellify_html_tables(text: str) -> str: if not text: return text lowered = text.lower() if "", "\n", text, flags=re.IGNORECASE) text = re.sub(r"]*>", "\n", text, flags=re.IGNORECASE) text = re.sub(r"]*>", "\n", text, flags=re.IGNORECASE) text = re.sub(r"", "\n", text, flags=re.IGNORECASE) text = re.sub(r"]*>", "\n", text, flags=re.IGNORECASE) text = re.sub(r"<[^>]+>", "", text) text = re.sub(r"\n[ \t]*\n+", "\n", text) return text def _clean_line(value: str) -> str: value = re.sub(r"\s+", " ", value or "").strip() value = value.strip(" |:;,_") value = re.sub(r"^[•*]+", "", value).strip() return value _STRUCTURAL_LABEL_RE = re.compile( r"(?:PAYER'?S?|RECIPIENT'?S?)\s+(?:name|TIN|RTN|account|" r"federal|street\s+address)", re.IGNORECASE, ) _BARCODE_NOISE_RE = re.compile(r"[01]{40,}") def _has_structural_labels(text: str) -> bool: return bool(text and _STRUCTURAL_LABEL_RE.search(text)) def _has_barcode_noise(text: str) -> bool: return bool(text and _BARCODE_NOISE_RE.search(text)) def _join_first_page_text(pages: list[dict]) -> str: if not pages: return "" first = pages[0] or {} primary = _normalise_text(first.get("text") or "") alternate = _normalise_text(first.get("text_alt") or "") if not primary: return alternate if not alternate: return primary p_labels = _has_structural_labels(primary) a_labels = _has_structural_labels(alternate) if p_labels and not a_labels: return primary if a_labels and not p_labels: return alternate p_noise = _has_barcode_noise(primary) a_noise = _has_barcode_noise(alternate) if p_noise and not a_noise: return alternate if a_noise and not p_noise: return primary if len(alternate) > len(primary) * 1.20: return alternate return primary def _clean_amount(raw: str) -> float | None: if not raw: return None raw = raw.replace("$", "").replace(",", "").strip() try: amount = float(raw) except (ValueError, TypeError): return None return int(amount) if amount.is_integer() else amount def _extract_tax_year(text: str) -> int | None: patterns = [ r"for\s+(?:the\s+)?calendar\s+year\b[\s\S]{0,200}?\b(20\d{2})\b", r"(?:tax\s+year|calendar\s+year|for\s+the\s+year)" r"\s*[:\-]?\s*(20\d{2})", r"FORM\s+1099[-\s]?INT\b.{0,80}?\b(20\d{2})\b", r"\b(20\d{2})\s+FORM\s+1099[-\s]?INT\b", ] for pattern in patterns: m = re.search(pattern, text, re.IGNORECASE | re.DOTALL) if m: y = int(m.group(1)) if 2000 <= y <= 2100: return y return None def _extract_tax_year_fallback(text: str) -> int | None: """Fallback: 4-digit year within ~80 chars after a phone number.""" m = re.search( r"\(?\d{3}\)?[-.\s]\d{3}[-.\s]\d{4}[\s\S]{0,80}?\b((?:19|20)\d{2})\b", text, ) if m: y = int(m.group(1)) if 2000 <= y <= 2100: return y return None # --------------------------------------------------------------------- # TIN extraction # --------------------------------------------------------------------- def _normalise_tin(value: str) -> str: if not value: return "" original = value.strip() digits = re.sub(r"\D", "", original) if len(digits) != 9: return original if re.match(r"^\d{3}[-.\s]\d{2}[-.\s]\d{4}$", original): return f"{digits[:3]}-{digits[3:5]}-{digits[5:]}" if re.match(r"^\d{2}[-.\s]\d{7}$", original): return f"{digits[:2]}-{digits[2:]}" return digits def _tin_digits(value: str) -> str: return re.sub(r"\D", "", value or "") def _find_tin_after_label(text: str, label_pattern: str) -> str: m = re.search(label_pattern, text, re.IGNORECASE) if not m: return "" line_end = text.find("\n", m.end()) if line_end == -1: line_end = len(text) same_line = text[m.end():line_end] tm = _TIN_FORMATTED_RE.search(same_line) if tm: candidate = _normalise_tin(tm.group(0)) if _is_valid_tin(candidate): return candidate window = text[m.end():m.end() + 200] for tm in _TIN_FORMATTED_RE.finditer(window): candidate = _normalise_tin(tm.group(0)) if _is_valid_tin(candidate): return candidate # MASKED FIX #2 — also allow a masked SSN/EIN near the label. mssn = _MASKED_SSN_RE.search(window) if mssn: return f"***-**-{mssn.group(1)}" mein = _MASKED_EIN_RE.search(window) if mein: return f"**-{mein.group(1)}" return "" def _extract_tins(text: str) -> tuple[str, str]: """Return (payer_tin, recipient_tin). Supports full and masked TINs.""" # --- Payer --- payer_tin = _find_tin_after_label( text, r"PAYER'?S?\s+(?:TIN|taxpayer\s+identification\s+number)", ) if not payer_tin: for m in _TIN_FORMATTED_RE.finditer(text): candidate = _normalise_tin(m.group(0)) if re.match(r"^\d{2}-\d{7}$", candidate) and _is_valid_ein(candidate): payer_tin = candidate break # MASKED FIX #3 — masked payer EIN fallback (**-NNNNNNN). if not payer_tin: mein = _MASKED_EIN_RE.search(text) if mein: payer_tin = f"**-{mein.group(1)}" payer_digits = _tin_digits(payer_tin) # --- Recipient --- recipient_tin = _find_tin_after_label( text, r"RECIPIENT'?S?\s+(?:TIN|taxpayer\s+identification\s+number)", ) if not recipient_tin: for m in _TIN_FORMATTED_RE.finditer(text): candidate = _normalise_tin(m.group(0)) if ( re.match(r"^\d{3}-\d{2}-\d{4}$", candidate) and _is_valid_ssn(candidate) and _tin_digits(candidate) != payer_digits ): recipient_tin = candidate break # MASKED FIX #4 — masked recipient SSN fallback (***-**-NNNN). # This is the display form printed on most 1099-INTs, e.g. ***-**-2535. if not recipient_tin: mssn = _MASKED_SSN_RE.search(text) if mssn: last4 = mssn.group(1) # Ensure we don't pick up a masked payer EIN that happens to # look like an SSN (rare, but guard anyway). if _tin_digits(last4) != payer_digits: recipient_tin = f"***-**-{last4}" if recipient_tin and _tin_digits(recipient_tin) == payer_digits and payer_digits: recipient_tin = "" return payer_tin, recipient_tin # --------------------------------------------------------------------- # Account number # --------------------------------------------------------------------- def _extract_account_number(text: str) -> str: for pattern in [ r"RECIPIENT'?S?\s+account\s+number" r"(?:\s*\([^)]*\))?[\s:#]*([A-Za-z0-9\-]{4,30})", r"account\s+number(?:\s*\([^)]*\))?[\s:#]*([A-Za-z0-9\-]{4,30})", ]: m = re.search(pattern, text, re.IGNORECASE) if m: candidate = m.group(1).strip() if re.search(r"\d", candidate) and len(candidate) >= 4: return candidate m = re.search(r"account\s+number(?:\s*\([^)]*\))?", text, re.IGNORECASE) if m: after = text[m.end():m.end() + 300] for am in re.finditer(r"\b([A-Za-z0-9][A-Za-z0-9\-]{5,25})\b", after): candidate = am.group(1) digit_count = sum(c.isdigit() for c in candidate) has_letter = bool(re.search(r"[A-Za-z]", candidate)) if digit_count >= 6 or (digit_count >= 4 and has_letter): return candidate return "" def _extract_account_number_fallback(text: str) -> str: """Fallback: standalone 10-15 digit number (typical account number).""" for line in _lines(text): if re.fullmatch(r"\d{10,15}", line): return line return "" # --------------------------------------------------------------------- # Name extraction # --------------------------------------------------------------------- _NAME_CAPS_RE = re.compile(r"\b[A-Z][A-Z'.\-]{1,}(?:\s+[A-Z][A-Z'.\-]{1,}){1,4}\b") _NAME_TITLE_RE = re.compile(r"\b[A-Z][a-z]+(?:\s+[A-Z][a-z]+){1,4}\b") _FORM_WORD_RE = re.compile( r"\b(?:interest|income|early|withdrawal|penalty|savings|bonds|" r"treasury|obligations|federal|tax|withheld|investment|expense|" r"expenses|foreign|country|possession|territory|exempt|specified|" r"private|activity|bond|market|discount|premium|credit|cusip|state|" r"identification|payer|recipient|name|tin|street|address|city|town|" r"province|postal|telephone|account|form|copy|department|internal|" r"revenue|service|calendar|year|OMB|and|or|of|the|to|for|" r"important|information|enclosed|notice|mail|statement|please|see|" r"instructions|records|furnished|being|required|file|return|" r"penalty|sanction|imposed|determines|reported|negligence|" r"this|is|your|you|if|be|or|other|has|not)\b", re.IGNORECASE, ) _DATE_WORDS = { "rev", "january", "february", "march", "april", "may", "june", "july", "august", "september", "october", "november", "december", } _ADDRESS_SUFFIXES = { "st", "street", "ave", "avenue", "rd", "road", "ct", "court", "dr", "drive", "ln", "lane", "blvd", "boulevard", "way", "pl", "place", "ter", "terrace", "pkwy", "parkway", "cir", "circle", "hwy", "highway", "sq", "square", "ste", "suite", "fl", "floor", } _ADDRESS_LIKE_WORDS = {"PO", "P.O.", "P.O", "BOX", "RR", "HC", "RURAL", "ROUTE"} _US_STATES = { "AL", "AK", "AZ", "AR", "CA", "CO", "CT", "DE", "FL", "GA", "HI", "ID", "IL", "IN", "IA", "KS", "KY", "LA", "ME", "MD", "MA", "MI", "MN", "MS", "MO", "MT", "NE", "NV", "NH", "NJ", "NM", "NY", "NC", "ND", "OH", "OK", "OR", "PA", "RI", "SC", "SD", "TN", "TX", "UT", "VT", "VA", "WA", "WV", "WI", "WY", "DC", "PR", "VI", "GU", "AS", "MP", } _COMPANY_SUFFIXES = re.compile( r"\b(?:LLC|INC|CORP|CORPORATION|CO|COMPANY|LTD|LIMITED|LP|LLP|PLLC|" r"NA|N\.A\.|BANK|S\.A\.|S\.A|PLC)\b", re.IGNORECASE, ) _NOISE_RE = re.compile( r"^(?:" r"payer'?s?\s+(?:name|tin|federal|rtn)|" r"recipient'?s?\s+(?:name|tin|account)|" r"street\s+address|city\s+or\s+town|" r"state\s+or\s+province|foreign\s+postal|" r"account\s+number|2nd\s+TIN|" r"form\s+1099|copy\s+[a-z]|" r"corrected|void|" r"department\s+of\s+the\s+treasury|" r"internal\s+revenue\s+service|" r"for\s+calendar\s+year|" r"filing\s+requirement|" r"OMB\s+No\.?" r")\b", re.IGNORECASE, ) def _is_noise_line(line: str) -> bool: if not line: return True if _NOISE_RE.search(line): return True if len(line) > 120: return True if re.fullmatch(r"[\W\d_]+", line): return True if re.fullmatch(r"[01]{20,}", line): return True return False def _lines(text: str) -> list[str]: return [_clean_line(l) for l in text.splitlines() if _clean_line(l)] def _is_valid_name_candidate(cand: str, payer_name: str = "") -> bool: if not cand: return False words = cand.split() if not (2 <= len(words) <= 5): return False if _FORM_WORD_RE.fullmatch(cand): return False last = re.sub(r"[^A-Za-z]", "", words[-1]).lower() if last in _ADDRESS_SUFFIXES: return False if last.upper() in _US_STATES and len(words) >= 2: return False if any(w.upper() in _ADDRESS_LIKE_WORDS for w in words): return False if all(w.lower().rstrip(".") in _DATE_WORDS for w in words): return False if all(_FORM_WORD_RE.fullmatch(w) for w in words): return False if payer_name and cand.upper() in payer_name.upper(): return False return True def _extract_clean_name(value: str) -> str: if not value: return "" for pattern in (_NAME_CAPS_RE, _NAME_TITLE_RE): for m in pattern.finditer(value): candidate = m.group(0).strip() if _is_valid_name_candidate(candidate): return candidate return "" def _extract_name(text: str, party: str) -> str: if party == "payer": label = r"PAYER'?S?" other_label = r"RECIPIENT'?S?" else: label = r"RECIPIENT'?S?" other_label = r"PAYER'?S?" lines = _lines(text) for index, line in enumerate(lines): if not re.search(label, line, re.IGNORECASE): continue parts: list[str] = [] for j in range(index + 1, min(index + 15, len(lines))): candidate = lines[j] if re.search(other_label, candidate, re.IGNORECASE): break if _is_noise_line(candidate): continue if _EIN_RE.search(candidate) or _SSN_RE.search(candidate): continue if _ANY_BOX_LABEL_RE.search(candidate): continue name = _extract_clean_name(candidate) if name: parts.append(name) if len(parts) >= 2: break if parts: return " ".join(parts)[:100] return "" def _find_payer_name_by_suffix(text: str) -> str: for line in _lines(text): if _is_noise_line(line): continue if not _COMPANY_SUFFIXES.search(line): continue m = re.match( r"([A-Z][A-Za-z0-9&.'\-\s]{2,60}?" r"(?:LLC|INC|CORP|CORPORATION|CO|COMPANY|LTD|LIMITED|LP|LLP|" r"PLLC|N\.A\.?|NA|BANK|PLC))\b", line, ) if m: candidate = m.group(1).strip() if len(candidate) <= 100: return candidate return "" def _find_recipient_name_heuristic(text: str, payer_name: str) -> str: seen: set[str] = set() candidates: list[str] = [] for line in _lines(text): for m in _NAME_CAPS_RE.finditer(line): cand = m.group(0).strip() if not _is_valid_name_candidate(cand, payer_name): continue if len(cand.split()) > 3: continue upper = cand.upper() if upper in seen: continue seen.add(upper) candidates.append(cand) if len(candidates) >= 2: return " ".join(candidates) if candidates: return candidates[0] return "" # --------------------------------------------------------------------- # Address extraction # --------------------------------------------------------------------- _STREET_RE = re.compile( r"\b\d+\s+[A-Za-z][A-Za-z.'\-]*" r"(?:\s+[A-Za-z0-9][A-Za-z0-9.'\-]*){0,4}" r"\s+(?:St|Street|Ave|Avenue|Rd|Road|Ct|Court|Dr|Drive|" r"Ln|Lane|Blvd|Boulevard|Way|Pl|Place|Ter|Terrace|Pkwy|" r"Parkway|Cir|Circle|Hwy|Highway|Sq|Square)\b\.?", re.IGNORECASE, ) _CITY_STATE_ZIP_RE = re.compile( r"\b([A-Z][A-Za-z]+(?:\s+[A-Z][A-Za-z]+){0,3})\s*,?\s+" r"([A-Z]{2})\s+(\d{5}(?:-\d{4})?)\b" ) def _extract_us_address(value: str) -> str: if not value: return "" for city_m in _CITY_STATE_ZIP_RE.finditer(value): city = city_m.group(1).strip() state = city_m.group(2) zipcode = city_m.group(3) before = value[:city_m.start()] street_matches = list(_STREET_RE.finditer(before)) if street_matches: street = street_matches[-1].group(0).strip() rest = before[street_matches[-1].end():] ste_m = re.match( r"\s+(STE|SUITE|APT|UNIT|#)\s*\.?\s*([A-Za-z0-9\-]+)", rest, re.IGNORECASE, ) if ste_m: street = f"{street} {ste_m.group(1).upper()} {ste_m.group(2)}" return f"{street}, {city} {state} {zipcode}" m = _STREET_RE.search(value) if m: return m.group(0).strip() return "" def _extract_address(text: str, party: str) -> str: if party == "payer": label = r"PAYER'?S?\s+(?:name\s+and\s+address|name)" other = r"RECIPIENT'?S?\s+" else: label = r"RECIPIENT'?S?\s+(?:name\s+and\s+address|name)" other = r"PAYER'?S?\s+" lines = _lines(text) label_rx = re.compile(label, re.IGNORECASE) for index, line in enumerate(lines): m = label_rx.search(line) if not m: continue candidates: list[str] = [] remainder = _clean_line(line[m.end():]) remainder = re.sub(r"^[\s:|.\-]+", "", remainder) if remainder and not _is_noise_line(remainder): candidates.append(remainder) for next_line in lines[index + 1:index + 26]: if re.search(other, next_line, re.IGNORECASE): break if _is_noise_line(next_line): continue if _EIN_RE.fullmatch(next_line) or _SSN_RE.fullmatch(next_line): continue candidates.append(next_line) if not candidates: continue joined = "\n".join(candidates) us_addr = _extract_us_address(joined) if us_addr: return us_addr[:200] addr_lines = [c for c in candidates if re.search(r"\d", c)] if addr_lines: return ", ".join(addr_lines[:3])[:200] return ", ".join(candidates[:3])[:200] return "" def _find_address_by_name(text: str, name: str, exclude: str = "") -> str: if not name: return "" lines = _lines(text) targets = [name.strip().upper()] words = targets[0].split() if len(words) > 2: targets.append(" ".join(words[:2])) targets.append(words[0]) for target in targets: for i, line in enumerate(lines): if target not in line.upper(): continue block = "\n".join(lines[i: i + 4]) addr = _extract_us_address(block) if addr and addr != exclude: return addr[:200] return "" def _find_recipient_address_fallback(text: str, payer_address: str) -> str: lines = _lines(text) for i in range(len(lines)): block = "\n".join(lines[i: i + 4]) addr = _extract_us_address(block) if addr and addr != payer_address: return addr[:200] return "" def _strip_name_words_from_address(address: str, name: str) -> str: if not address or not name or "," not in address: return address name_words = { re.sub(r"[^A-Za-z]", "", w).upper() for w in name.split() if len(w) > 1 } parts = address.split(", ") if len(parts) != 2: return address city_part = parts[1].split() while city_part and re.sub(r"[^A-Za-z]", "", city_part[0]).upper() in name_words: city_part.pop(0) if city_part: parts[1] = " ".join(city_part) return ", ".join(parts) return address # --------------------------------------------------------------------- # Phone # --------------------------------------------------------------------- _PHONE_RE = re.compile( r"(?:\+?1[-.\s]?)?" r"\(?\d{3}\)?[-.\s]\d{3}[-.\s]?\d{4}\b" ) def _extract_phone(text: str) -> str: for pattern in [ r"(?:payer'?s?|recipient'?s?)\s*(?:telephone|phone)" r"(?:\s*(?:number|no\.?))?[^0-9]{0,30}" r"(\+?1?[-.\s]?\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4})", r"\b(?:tel|telephone|phone|ph)\b[^0-9]{0,20}" r"(\+?1?[-.\s]?\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4})", ]: m = re.search(pattern, text, re.IGNORECASE) if m: return m.group(1).strip() m = _PHONE_RE.search(text) if m: return m.group(0).strip() return "" # --------------------------------------------------------------------- # Amount / text / int box lookups # --------------------------------------------------------------------- def _find_bare_amount(snippet: str) -> float | None: for m in _MONEY_RE.finditer(snippet): raw = m.group("amount") if re.fullmatch(r"(?:19|20)\d{2}", raw): continue if re.fullmatch(r"\d{5}", raw): ctx = snippet[max(0, m.start() - 30):m.start()] if re.search(r"\b[A-Z]{2}\s*,?\s*$", ctx): continue after = snippet[m.end():m.end() + 60] if _BOX_NUMBER_FOLLOW_RE.match(after): continue amount = _clean_amount(raw) if amount is not None: return amount return None def _find_box_value(text: str, key: str) -> float | None: for rx in _BOX_REGEXES.get(key, []): for m in rx.finditer(text): snippet_end = min(len(text), m.end() + 300) snippet = text[m.end():snippet_end] nxt = _ANY_BOX_LABEL_RE.search(snippet) if nxt: snippet = snippet[:nxt.start()] dollar_m = _DOLLAR_AMOUNT_RE.search(snippet) if dollar_m: amount = _clean_amount(dollar_m.group(1)) if amount is not None: return amount if "$" in snippet: return None value = _find_bare_amount(snippet) if value is not None: return value return None def _extract_box1_fallback(text: str) -> float | None: m = re.search( r"\(?\d{3}\)?[-.\s]\d{3}[-.\s]\d{4}\s+\$?([\d,]+\.\d{2})\s+(?:19|20)\d{2}", text, ) if m: return _clean_amount(m.group(1)) return None def _find_box_text(text: str, key: str, max_chars: int = 60) -> str: for rx in _BOX_REGEXES.get(key, []): for m in rx.finditer(text): snippet = text[m.end():m.end() + max_chars * 4] lines_after = [l.strip() for l in snippet.split("\n") if l.strip()] for candidate in lines_after[:4]: if _is_noise_line(candidate): continue if _ANY_BOX_LABEL_RE.search(candidate): break if not re.search(r"[A-Za-z]{2,}", candidate): continue if key == "box7_foreign_country" and re.search(r"\d", candidate): continue if key == "box14_cusip_no": if not re.fullmatch(r"[A-Za-z0-9]{6,12}", candidate): continue if key == "box15_state": if not re.fullmatch(r"[A-Za-z]{2}", candidate): continue if key == "box16_state_id_no": if not re.search(r"\d", candidate): continue if re.fullmatch(r"\d{9,}", candidate): continue return candidate[:max_chars] return "" # --------------------------------------------------------------------- # Public API # --------------------------------------------------------------------- _MONEY_FIELDS = [ "box1_interest_income", "box2_early_withdrawal_penalty", "box3_us_savings_bonds_treasury", "box4_fed_tax_withheld", "box5_investment_interest_expense", "box6_foreign_tax_paid", "box8_tax_exempt_interest", "box9_specified_private_activity", "box10_market_discount", "box11_bond_premium", "box12_bond_premium_treasury", "box13_bond_premium_tax_exempt", "box17_state_tax_withheld", ] def parse_1099_int_2024(pages: list[dict], filename: str = "") -> dict[str, Any]: """Parse a Form 1099-INT (2024/2025) using page 1 only.""" raw_joined = _join_first_page_text(pages) if not raw_joined: raise ValueError( "The uploaded Form 1099-INT contained no readable text on page 1. " "If this is a scanned PDF, OCR must run in app.py." ) text = _cellify_html_tables(raw_joined) warnings: list[str] = [] payer_tin, recipient_tin = _extract_tins(text) payer_name = _extract_name(text, "payer") if not payer_name: payer_name = _find_payer_name_by_suffix(text) recipient_name = _extract_name(text, "recipient") if not recipient_name: recipient_name = _find_recipient_name_heuristic(text, payer_name) payer_address = _extract_address(text, "payer") if not payer_address and payer_name: payer_address = _find_address_by_name(text, payer_name) if payer_address and payer_name: payer_address = _strip_name_words_from_address(payer_address, payer_name) recipient_address = _extract_address(text, "recipient") if not recipient_address and recipient_name: recipient_address = _find_address_by_name( text, recipient_name, exclude=payer_address ) if not recipient_address: recipient_address = _find_recipient_address_fallback(text, payer_address) if recipient_address and recipient_name: recipient_address = _strip_name_words_from_address( recipient_address, recipient_name ) tax_year = _extract_tax_year(text) or _extract_tax_year_fallback(text) or 2024 account_number = _extract_account_number(text) or _extract_account_number_fallback(text) if not payer_tin: warnings.append("Could not locate the payer's TIN.") if not recipient_tin: warnings.append("Could not locate the recipient's TIN.") elif recipient_tin.startswith("*"): warnings.append( "Recipient TIN is masked on the form (last 4 digits only): " f"{recipient_tin}." ) if not payer_name: warnings.append("Could not confidently locate the payer's name.") if not recipient_name: warnings.append("Could not confidently locate the recipient's name.") result: dict[str, Any] = { "form_type": "1099_int_2024", "filename": filename, "tax_year": tax_year, "account_number": account_number, "payer_tin": payer_tin, "payer_name": payer_name, "payer_address": payer_address, "payer_phone": _extract_phone(text), "recipient_tin": recipient_tin, "recipient_name": recipient_name, "recipient_address": recipient_address, "box7_foreign_country": _find_box_text(text, "box7_foreign_country", 60), "box14_cusip_no": _find_box_text(text, "box14_cusip_no", 40), "box15_state": _find_box_text(text, "box15_state", 4), "box16_state_id_no": _find_box_text(text, "box16_state_id_no", 40), } for key in _MONEY_FIELDS: result[key] = _find_box_value(text, key) if result["box1_interest_income"] is None: fallback = _extract_box1_fallback(text) if fallback is not None: result["box1_interest_income"] = fallback if not any(result.get(k) is not None for k in _MONEY_FIELDS): warnings.append( "No monetary box values were detected. The PDF may be scanned " "at low resolution or use a non-standard layout." ) result["raw_text"] = text result["parse_warnings"] = warnings return result # --------------------------------------------------------------------- # Debug helper # --------------------------------------------------------------------- def debug_print_1099_int(result: dict) -> None: print("\n" + "=" * 78) print(f" 1099-INT PARSE RESULT file = {result.get('filename', '')!r}") print("=" * 78) groups = { "META": ["form_type", "filename", "tax_year", "account_number"], "PAYER": ["payer_tin", "payer_name", "payer_address", "payer_phone"], "RECIPIENT": ["recipient_tin", "recipient_name", "recipient_address"], "TEXT BOXES": ["box7_foreign_country", "box14_cusip_no", "box15_state", "box16_state_id_no"], "MONEY BOXES": _MONEY_FIELDS, "DIAG": ["parse_warnings"], } for title, keys in groups.items(): print(f"\n--- {title} " + "-" * (60 - len(title))) for k in keys: v = result.get(k, "") if isinstance(v, float) and v.is_integer(): v = int(v) print(f" {k:<35} : {v!r}") print("\n" + "=" * 78 + "\n")