""" 1099-R (2025) — Distributions From Pensions, Annuities, Retirement or Profit-Sharing Plans, IRAs, Insurance Contracts, etc. Handles TWO layouts: 1. LABELED forms — text like "Box 1 Gross distribution" (issuer-generated). 2. PRE-PRINTED forms — the text layer contains ONLY the filled-in values; box labels are graphics. Parsed positionally (right column = boxes, left column = identity info). `raw_text` is always returned so you can inspect what the extractor actually captured. """ from __future__ import annotations import re from typing import Any, Dict, List, Optional, Tuple # --------------------------------------------------------------------------- # Regex primitives # --------------------------------------------------------------------------- _YEAR_RE = re.compile(r"\b(?:19|20)\d{2}\b") _MONEY_DOLLAR = re.compile(r"\$\s*([\d,]+(?:\.\d{1,2})?)") _MONEY_COMMA = re.compile(r"(? str: """ Return the left-column portion of a pre-printed 1099-R line. The right column (monetary values / checkbox marks) always starts after a run of 2+ spaces that is *preceded by real content* and followed by a digit, $, or X. The `(?<=\\S)` lookbehind is critical: without it, lines like " 4 CHASE METROTECH CENTER" match on the LEADING whitespace (5 spaces + digit '4') and lose their street data. With it, we only cut where there is actual content before the gap. """ if not line: return "" m = re.search(r"(?<=\S)\s{2,}(?=[\d$X])", line) if m: return line[:m.start()].rstrip() return line[:col].rstrip() # --------------------------------------------------------------------------- # Box anchors for LABELED forms # --------------------------------------------------------------------------- _BOX_ANCHORS: Dict[str, List[str]] = { "box1_gross_distribution": [ r"\bBox\s*1\b(?!\s*[0-9a-z])", r"\b1\s*[.\-:]\s*Gross\s+distribution\b", ], "box2a_taxable_amount": [ r"\bBox\s*2a\b", r"\b2a\s*[.\-:]\s*Taxable\s+amount\b", ], "box3_capital_gain": [ r"\bBox\s*3\b(?!\s*[0-9a-z])", r"\b3\s*[.\-:]\s*Capital\s+gain\b", ], "box4_fed_tax_withheld": [ r"\bBox\s*4\b(?!\s*[0-9a-z])", r"\b4\s*[.\-:]\s*Federal\s+income\s+tax\s+withheld\b", ], "box5_employee_contributions": [ r"\bBox\s*5\b(?!\s*[0-9a-z])", r"\b5\s*[.\-:]\s*Employee\s+contributions\b", ], "box6_net_unrealized_appreciation": [ r"\bBox\s*6\b(?!\s*[0-9a-z])", r"\b6\s*[.\-:]\s*Net\s+unrealized\b", ], "box7_distribution_code": [ r"\bBox\s*7\b(?!\s*[0-9a-z])", r"\b7\s*[.\-:]\s*Distribution\s+code", ], "box8_other": [r"\bBox\s*8\b(?!\s*[0-9a-z])"], "box9a_percentage": [r"\bBox\s*9a\b"], "box9b_total_employee_contributions": [r"\bBox\s*9b\b"], "box10_irr_within_5_years": [r"\bBox\s*10\b"], "box11_first_year_designated_roth": [r"\bBox\s*11\b"], "box14_state_tax_withheld": [r"\bBox\s*14\b"], "box15_state_payer_no": [r"\bBox\s*15\b"], "box16_state_distribution": [r"\bBox\s*16\b"], "box17_local_tax_withheld": [r"\bBox\s*17\b"], "box18_name_of_locality": [r"\bBox\s*18\b"], "box19_local_distribution": [r"\bBox\s*19\b"], } _BOX_REGEXES = { k: [re.compile(p, re.IGNORECASE) for p in v] for k, v in _BOX_ANCHORS.items() } _ANY_BOX_LABEL_RE = re.compile( r"\bBox\s*(?:1|2[abc]?|3|4|5|6|7|8|9[ab]?|1[0-9])\b", re.IGNORECASE, ) _DESCRIPTION_RE = re.compile( r"\bthis\s+box\s+shows\b" r"|\bshows\s+(?:your|the|a|an)\b" r"|\bblank\s+because\b" r"|\bnot\s+pertinent\b", re.IGNORECASE, ) # --------------------------------------------------------------------------- # Small helpers # --------------------------------------------------------------------------- def _clean_money(raw: Optional[str]) -> Optional[float]: if raw is None: return None s = str(raw).strip() if not s: return None s = re.sub(r"[^\d.\-]", "", s.replace(",", "").replace("$", "")) if s in ("", "-", ".", "-."): return None try: return float(s) except ValueError: return None def _is_description_snippet(snippet: str, window: int = 250) -> bool: return bool(_DESCRIPTION_RE.search(snippet[:window])) def _find_bare_amount(snippet: str) -> Optional[float]: for regex in (_MONEY_DOLLAR, _MONEY_COMMA, _MONEY_PLAIN): for m in regex.finditer(snippet): clean = m.group(1).replace(",", "").replace("$", "").strip() if _YEAR_RE.fullmatch(clean): continue if "." not in clean and len(clean) > 7: continue val = _clean_money(m.group(1)) if val is not None: return val return None def _is_preprinted(raw: str) -> bool: if not raw: return False has_labels = bool(re.search(r"\bBox\s+[1-9]", raw, re.IGNORECASE)) has_structure = bool(_EIN_RE.search(raw)) and bool( _MONEY_COMMA.search(raw) or _MONEY_PLAIN.search(raw) ) return has_structure and not has_labels # --------------------------------------------------------------------------- # LABELED-form helpers # --------------------------------------------------------------------------- def _find_box_money(text, box_key, max_scan=400): for rx in _BOX_REGEXES.get(box_key, []): for m in rx.finditer(text): chunk = text[m.end(): m.end() + max_scan] nxt = _ANY_BOX_LABEL_RE.search(chunk) if nxt: chunk = chunk[:nxt.start()] if _is_description_snippet(chunk): continue val = _find_bare_amount(chunk) if val is not None: return val return None def _find_box_money_by_desc(text, desc_pattern, max_scan=400): for m in re.finditer(desc_pattern, text, re.IGNORECASE): chunk = text[m.end(): m.end() + max_scan] nxt = _ANY_BOX_LABEL_RE.search(chunk) if nxt: chunk = chunk[:nxt.start()] if _is_description_snippet(chunk): continue val = _find_bare_amount(chunk) if val is not None: return val return None def _find_box_code(text, box_key, max_scan=30): for rx in _BOX_REGEXES.get(box_key, []): for m in rx.finditer(text): chunk = text[m.end(): m.end() + max_scan] nxt = _ANY_BOX_LABEL_RE.search(chunk) if nxt: chunk = chunk[:nxt.start()] if _is_description_snippet(chunk): continue code_m = re.search(r"\b([A-Z0-9]{1,3})\b", chunk) if code_m and _VALID_DISTRIBUTION_CODE_RE.fullmatch(code_m.group(1)): return code_m.group(1) return None def _find_box_text(text, box_key, max_scan=80): for rx in _BOX_REGEXES.get(box_key, []): for m in rx.finditer(text): chunk = text[m.end(): m.end() + max_scan] nxt = _ANY_BOX_LABEL_RE.search(chunk) if nxt: chunk = chunk[:nxt.start()] if _is_description_snippet(chunk): continue token = re.search(r"[A-Za-z0-9][A-Za-z0-9'.\-]{0,40}", chunk) if token: return token.group(0).strip() return None def _parse_labeled(raw: str, base: Dict[str, Any]) -> Dict[str, Any]: base["box1_gross_distribution"] = ( _find_box_money(raw, "box1_gross_distribution") or _find_box_money_by_desc(raw, r"gross\s+distribution") ) base["box2a_taxable_amount"] = ( _find_box_money(raw, "box2a_taxable_amount") or _find_box_money_by_desc(raw, r"taxable\s+amount") ) base["box2b_taxable_amount_not_determined"] = bool( re.search(r"2b[^\n]{0,60}not\s+determined", raw, re.IGNORECASE) ) base["box2c_total_distribution"] = bool( re.search(r"2c[^\n]{0,60}total\s+distribution", raw, re.IGNORECASE) ) base["box3_capital_gain"] = _find_box_money(raw, "box3_capital_gain") base["box4_fed_tax_withheld"] = ( _find_box_money(raw, "box4_fed_tax_withheld") or _find_box_money_by_desc(raw, r"federal\s+income\s+tax\s+withheld") ) base["box5_employee_contributions"] = ( _find_box_money(raw, "box5_employee_contributions") or _find_box_money_by_desc(raw, r"employee\s+contributions") ) base["box6_net_unrealized_appreciation"] = ( _find_box_money(raw, "box6_net_unrealized_appreciation") or _find_box_money_by_desc(raw, r"net\s+unrealized\s+appreciation") ) base["box7_distribution_code"] = _find_box_code(raw, "box7_distribution_code") base["box8_other"] = _find_box_money(raw, "box8_other") base["box9a_percentage"] = _find_box_money(raw, "box9a_percentage") base["box9b_total_employee_contributions"] = ( _find_box_money(raw, "box9b_total_employee_contributions") or _find_box_money_by_desc(raw, r"remaining\s+contributions") ) base["box10_irr_within_5_years"] = _find_box_money(raw, "box10_irr_within_5_years") base["box11_first_year_designated_roth"] = _find_box_code( raw, "box11_first_year_designated_roth" ) base["box12_fatca"] = bool(re.search(r"12[^\n]{0,60}FATCA", raw, re.IGNORECASE)) m = re.search( r"13\s+date\s+of\s+payment[^\d]{0,20}(\d{1,2}[/-]\d{1,2}[/-]\d{2,4})", raw, re.IGNORECASE, ) if m: base["box13_date_of_payment"] = m.group(1) base["box14_state_tax_withheld"] = ( _find_box_money(raw, "box14_state_tax_withheld") or _find_box_money_by_desc(raw, r"state\s+income\s+tax\s+withheld") ) base["box15_state_payer_no"] = _find_box_text(raw, "box15_state_payer_no") base["box16_state_distribution"] = ( _find_box_money(raw, "box16_state_distribution") or _find_box_money_by_desc(raw, r"taxable\s+by\s+the\s+state") ) base["box17_local_tax_withheld"] = _find_box_money(raw, "box17_local_tax_withheld") base["box18_name_of_locality"] = _find_box_text(raw, "box18_name_of_locality") base["box19_local_distribution"] = _find_box_money(raw, "box19_local_distribution") return base # --------------------------------------------------------------------------- # PRE-PRINTED-form helpers (positional) # --------------------------------------------------------------------------- def _split_into_blocks(raw: str, payer_name: str) -> List[str]: """Split raw text by payer name → one string per 1099-R on the page.""" if not payer_name: return [raw] positions = [] start = 0 while True: idx = raw.find(payer_name, start) if idx < 0: break positions.append(idx) start = idx + len(payer_name) if len(positions) < 2: return [raw] blocks = [] for i, pos in enumerate(positions): end = positions[i + 1] if i + 1 < len(positions) else len(raw) blocks.append(raw[pos:end]) return blocks def _find_payer_name(raw: str) -> str: """ Find the payer (company) name. Uses _left_side() to drop right-column money, then accepts any line that ends with a company suffix — including compound ones such as "HLDG, INC" — so we don't get cut off after HLDG. """ for line in raw.split("\n"): clean = _left_side(line).strip() if not clean or len(clean) < 5: continue if not clean[0].isupper(): continue if re.match(r"^\d", clean): continue if _COMPANY_SUFFIX_END_RE.search(clean): return clean return "" def _find_right_column_values( lines: List[str], right_col_min: int = 40, ) -> Tuple[List[Tuple[int, float]], List[Tuple[int, str]], List[Tuple[int, str]]]: """Scan lines for right-column values.""" money_vals: List[Tuple[int, float]] = [] checkbox_marks: List[Tuple[int, str]] = [] code_marks: List[Tuple[int, str]] = [] for i, line in enumerate(lines): if not line: continue # --- "7 X" on one line --- codex = re.search(r"\b([1-9A-Z])\s{3,}X\b", line[right_col_min:]) if codex: code_marks.append((i, codex.group(1))) checkbox_marks.append((i, "X")) line = line[:right_col_min + codex.start()] + " " * len(line) else: # --- standalone X after col_min --- xm = re.search(r"\bX\b", line[right_col_min:]) if xm: checkbox_marks.append((i, "X")) line = line[:right_col_min + xm.start()] + " " * len(line) # --- money --- for m in re.finditer(r"\d{1,3}(?:,\d{3})*\.\d{2}", line): if m.start() >= right_col_min: val = _clean_money(m.group(0)) if val is not None: money_vals.append((i, val)) break return money_vals, checkbox_marks, code_marks def _extract_payer_address(lines: List[str], payer_name_idx: int) -> str: parts: List[str] = [] for line in lines[payer_name_idx + 1: payer_name_idx + 8]: clean = _left_side(line).strip() if not clean: if parts: break continue if re.match(r"^\d+\s+[A-Z0-9]", clean) or re.match( r"^[A-Z][A-Z\s.]+,\s+[A-Z]{2}\s+\d{5}", clean ): parts.append(clean) elif parts: break return ", ".join(parts) def _extract_recipient_name( lines: List[str], tin_idx: int ) -> Tuple[str, int]: """Return ONLY the first recipient-name line + its index.""" for i in range(tin_idx + 1, min(tin_idx + 12, len(lines))): clean = _left_side(lines[i]).strip() if not clean: continue if re.match(r"^\d", clean): break if re.match(r"^[A-Z][A-Z0-9 .,&'\-]+$", clean) and len(clean) >= 3: return clean, i return "", -1 def _extract_recipient_address(lines: List[str], start_idx: int) -> str: """ Skip name-continuation lines (e.g. custodian "JPMS LLC CUST."), then read the street + city/state/zip. Uses _left_side() to drop the trailing `7 X` right-column junk. """ # --- 1. skip name-continuation lines --- i = start_idx + 1 while i < len(lines) and i < start_idx + 8: clean = _left_side(lines[i]).strip() if not clean: i += 1 continue is_name_like = ( re.match(r"^[A-Z][A-Z0-9 .,&'\-]+$", clean) and not re.match(r"^\d", clean) and not re.match(r"^[A-Z][A-Z\s.]+[A-Z]{2}\s+\d{5}", clean) ) if is_name_like: i += 1 continue break # --- 2. read the address lines --- parts: List[str] = [] while i < len(lines) and i < start_idx + 12: clean = _left_side(lines[i]).strip() if not clean: if parts: break i += 1 continue if re.match(r"^\d+\s+[A-Z0-9]", clean): parts.append(clean) elif re.match(r"^[A-Z][A-Z\s.]+[A-Z]{2}\s+\d{5}(?:-\d{4})?", clean): parts.append(clean) break elif parts: break i += 1 return ", ".join(parts) def _parse_preprinted( raw: str, base: Dict[str, Any], warnings: List[str] ) -> Dict[str, Any]: base["form_layout"] = "preprinted" # --- 1. Payer name --- payer_name = _find_payer_name(raw) if payer_name: base["payer_name"] = payer_name # --- 2. Split into blocks; use the first --- blocks = _split_into_blocks(raw, payer_name) block = blocks[0] if blocks else raw lines = block.split("\n") # --- 3. Right-column values --- money_vals, checkbox_marks, code_marks = _find_right_column_values(lines) if len(money_vals) >= 1: base["box1_gross_distribution"] = money_vals[0][1] if len(money_vals) >= 2: base["box2a_taxable_amount"] = money_vals[1][1] if len(checkbox_marks) >= 1: base["box2b_taxable_amount_not_determined"] = True if len(checkbox_marks) >= 2: base["box2c_total_distribution"] = True if code_marks: base["box7_distribution_code"] = code_marks[0][1] # --- 4. TINs --- m = _EIN_RE.search(block) if m: base["payer_tin"] = m.group(1) m = _MASKED_SSN_RE.search(block) if m: base["recipient_tin"] = m.group(1) else: m = _SSN_RE.search(block) if m and m.group(1) != base.get("payer_tin"): base["recipient_tin"] = m.group(1) # --- 5. Payer address --- payer_name_idx = next( (i for i, l in enumerate(lines) if payer_name and payer_name in l), 0 ) payer_addr = _extract_payer_address(lines, payer_name_idx) if payer_addr: base["payer_address"] = payer_addr # --- 6. TIN row index --- tin_row_idx = -1 for i, line in enumerate(lines): if base.get("payer_tin") and base["payer_tin"] in line: tin_row_idx = i break if base.get("recipient_tin") and base["recipient_tin"] in line: tin_row_idx = i break # --- 7. Recipient name (first line only) --- recipient_name = "" recipient_name_idx = -1 if tin_row_idx >= 0: recipient_name, recipient_name_idx = _extract_recipient_name( lines, tin_row_idx ) if recipient_name: base["recipient_name"] = recipient_name # --- 8. Recipient address --- if recipient_name and recipient_name_idx >= 0: addr = _extract_recipient_address(lines, recipient_name_idx) if addr: base["recipient_address"] = addr # --- 9. Box 15 --- m = _STATE_PAYER_NO_RE.search(raw) if m: base["box15_state_payer_no"] = m.group(1) # --- 10. Account number --- m = _ACCOUNT_NO_RE.search(raw) if m: base["account_number"] = m.group(1) # --- 11. Payer phone --- m = _PHONE_RE.search(block) if m: base["payer_phone"] = m.group(0) # --- 12. Warnings --- if base.get("box1_gross_distribution") is None: warnings.append("Box 1 (Gross distribution) could not be located.") if base.get("box2a_taxable_amount") is None: warnings.append("Box 2a (Taxable amount) could not be located.") if not base.get("payer_name"): warnings.append("Could not confidently locate the payer's name.") if not base.get("recipient_name"): warnings.append("Could not confidently locate the recipient's name.") if not base.get("payer_tin"): warnings.append("Could not locate the payer's TIN.") if not base.get("recipient_tin"): warnings.append("Could not locate the recipient's TIN.") return base # --------------------------------------------------------------------------- # Public entry point # --------------------------------------------------------------------------- def parse_1099_r_2025( pages: List[Dict[str, Any]], filename: str = "", ) -> Dict[str, Any]: warnings: List[str] = [] parts: List[str] = [] for p in pages: parts.append(p.get("text") or "") parts.append(p.get("text_alt") or "") raw = "\n".join(parts).replace("\u2019", "'").replace("\u2014", "-") base: Dict[str, Any] = { "form_type": "1099_r_2025", "form_code": "1099-R-2025", "tax_year": 2025, "filename": filename, "parse_warnings": warnings, "source_pages": [ {"page": p.get("page", 1), "source": p.get("source", "text")} for p in pages ], } m = re.search(r"\b(20\d{2})\b", raw) if m and 2000 <= int(m.group(1)) <= 2100: base["tax_year"] = int(m.group(1)) if _is_preprinted(raw): base = _parse_preprinted(raw, base, warnings) else: base = _parse_labeled(raw, base) m = _EIN_RE.search(raw) if m: base.setdefault("payer_tin", m.group(1)) m = _MASKED_SSN_RE.search(raw) or _SSN_RE.search(raw) if m: base.setdefault("recipient_tin", m.group(1)) base["raw_text"] = raw return base # --------------------------------------------------------------------------- # Debug helper # --------------------------------------------------------------------------- def debug_print_1099_r(result: Dict[str, Any]) -> None: print("\n" + "=" * 78) print(f" 1099-R PARSE RESULT file = {result.get('filename', '')!r}") print("=" * 78) groups = { "META": ["form_type", "form_code", "tax_year", "filename", "form_layout"], "IDENTITY": ["payer_tin", "payer_name", "payer_address", "payer_phone", "recipient_tin", "recipient_name", "recipient_address"], "BOXES": ["box1_gross_distribution", "box2a_taxable_amount", "box2b_taxable_amount_not_determined", "box2c_total_distribution", "box3_capital_gain", "box4_fed_tax_withheld", "box5_employee_contributions", "box6_net_unrealized_appreciation", "box7_distribution_code", "box8_other", "box9a_percentage", "box9b_total_employee_contributions", "box10_irr_within_5_years", "box11_first_year_designated_roth", "box12_fatca", "box13_date_of_payment", "box14_state_tax_withheld", "box15_state_payer_no", "box16_state_distribution", "box17_local_tax_withheld", "box18_name_of_locality", "box19_local_distribution"], "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:<42} : {v!r}") print("\n" + "=" * 78 + "\n")