""" 1099-G (2021) — Certain Government Payments. Handles TWO layouts: 1. LABELED forms — text like "1 Unemployment compensation". 2. PRE-PRINTED forms — text layer has only values; labels are graphics. Boxes: 1 Unemployment compensation 2 State or local income tax refunds, credits, or offsets 3 Box 2 amount is for tax year (a YEAR, not money) 4 Federal income tax withheld 5 RTAA payments 6 Taxable grants 7 Agriculture payments 8 Check if box 2 is trade or business income (checkbox) 9 Market gain 10a State | 10b State identification no. | 11 State income tax withheld `raw_text` is always returned so you can inspect what the extractor captured. """ from __future__ import annotations import re from typing import Any, Dict, List, Optional # --------------------------------------------------------------------------- # 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"(? Optional[float]: if not raw: return None s = re.sub(r"[^\d.\-]", "", str(raw).replace(",", "").replace("$", "").strip()) if s in ("", "-", ".", "-."): return None try: return float(s) except ValueError: return None 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 _left_side(line: str, col: int = 80) -> str: 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() 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_structured = bool(_EIN_RE.search(raw)) and bool( _MONEY_COMMA.search(raw) or _MONEY_PLAIN.search(raw) ) return has_structured and not has_labels def _find_payer_name(raw: str) -> str: 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 _split_into_blocks(raw: str, payer_name: str) -> List[str]: 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 _extract_payer_address(lines: List[str], idx: int) -> str: parts: List[str] = [] for line in lines[idx + 1: 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]: 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: 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 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) # --------------------------------------------------------------------------- # Identity extraction (shared by labeled branch) # --------------------------------------------------------------------------- def _extract_identity_block( raw: str, label_regex: str, stop_patterns: List[str], max_names: int = 2, ) -> Dict[str, str]: """ Scan forward from `label_regex`, returning { name?, address?, phone?, tin? }. Handles the OCR interleaving of left-column address text with right-column box labels / dollar values by stripping the right side before analysis. """ m = re.search(label_regex, raw, re.IGNORECASE) if not m: return {} block = raw[m.end():m.end() + 1500] name_lines: List[str] = [] street = "" city_state_zip = "" phone = "" tin = "" for line in block.split("\n"): if any(re.search(p, line, re.IGNORECASE) for p in stop_patterns): break clean = line # Strip right-side content (money, box labels, form numbers) clean = re.split(r"\$\s*[\d,.]+", clean)[0] clean = re.split( r"\s+\d+\s+(?:Unemployment|State(?:\s+(?:or|identification|income))?" r"|Box\s+2\s+amount|Federal\s+income|RTAA|Taxable\s+grants|" r"Agriculture\s+payments|Check\s+if|If\s+checked|Market\s+gain)", clean, )[0] clean = re.split(r"\s+Form\s+1099", clean)[0] clean = re.sub( r"\s+(?:Certain|Government|Payments|For\s+(?:Recipient|Payer)" r"|Copy\s+\w+)\s*$", "", clean, ) clean = clean.strip() if not clean: continue # Skip label continuations if re.match(r"^(?:or\s+(?:foreign|province)|address,\s+city)", clean, re.IGNORECASE): continue if re.search(label_regex, clean, re.IGNORECASE): continue # Phone pm = re.search(r"Phone:?\s*([\d\-\(\)\s\.]{7,})", clean, re.IGNORECASE) if pm and not phone: phone = pm.group(1).strip().rstrip(".") continue # TIN (payer EIN takes precedence over masked SSN) if not tin: tm = re.search(r"\b(\d{2}-\d{7})\b", clean) if tm: tin = tm.group(1) continue tm = _MASKED_SSN_RE.search(clean) if tm: tin = tm.group(1) continue # Street if not street and re.match(r"^\d+\s+[A-Z0-9]", clean): street = clean continue # City / State / ZIP if not city_state_zip: addr_m = re.match( r"^([A-Z][A-Za-z\s]+?,?\s+[A-Z]{2}\s+\d{5}(?:-\d{4})?)", clean, ) if addr_m: city_state_zip = addr_m.group(1).strip() continue # ALL-CAPS company/person name if ( len(name_lines) < max_names and re.match(r"^[A-Z][A-Z0-9 .,&'\-]{5,}$", clean) ): name_lines.append(clean) continue result: Dict[str, str] = {} if name_lines: result["name"] = " ".join(name_lines) if street or city_state_zip: result["address"] = ", ".join(filter(None, [street, city_state_zip])) if phone: result["phone"] = phone if tin: result["tin"] = tin return result def _extract_identity_labeled(raw: str, base: Dict[str, Any]) -> Dict[str, Any]: payer = _extract_identity_block( raw, r"PAYER'?S?\s+name", stop_patterns=[r"PAYER'?S?\s+TIN", r"RECIPIENT'?S?\s+name"], max_names=2, ) if payer.get("name"): base["payer_name"] = payer["name"] if payer.get("address"): base["payer_address"] = payer["address"] if payer.get("phone"): base["payer_phone"] = payer["phone"] if payer.get("tin"): base["payer_tin"] = payer["tin"] recip = _extract_identity_block( raw, r"RECIPIENT'?S?\s+name", stop_patterns=[r"Account\s+number", r"Form\s+1099"], max_names=1, ) if recip.get("name"): base["recipient_name"] = recip["name"] if recip.get("address"): base["recipient_address"] = recip["address"] if recip.get("tin"): base["recipient_tin"] = recip["tin"] return base # --------------------------------------------------------------------------- # Labeled-form value lookup (two-column aware) # --------------------------------------------------------------------------- def _find_box_money(text: str, box_key: str, max_scan: int = 400) -> Optional[float]: # Box 11 (state tax withheld) is a short field; keep scan short so we # don't accidentally pick up "$600" or "$0.00" from later copies. if box_key == "box11_state_tax_withheld": max_scan = 150 for rx in _BOX_REGEXES.get(box_key, []): for m in rx.finditer(text): # Determine if this label is the 1st, 2nd, … box on its line line_start = text.rfind("\n", 0, m.start()) + 1 prefix = text[line_start:m.start()] box_idx_on_line = len(list(_ANY_BOX_LABEL_RE.finditer(prefix))) # Slice the chunk and cut at the next box label that appears # on a DIFFERENT line (so we don't stop at "6 Taxable grants" # immediately after "5 RTAA payments"). chunk = text[m.end(): m.end() + max_scan] for nxt in _ANY_BOX_LABEL_RE.finditer(chunk): if "\n" in chunk[:nxt.start()]: chunk = chunk[:nxt.start()] break dollar_vals: List[float] = [] for dm in _MONEY_DOLLAR.finditer(chunk): v = _clean_money(dm.group(1)) if v is not None: dollar_vals.append(v) if len(dollar_vals) > box_idx_on_line: break if dollar_vals: idx = min(box_idx_on_line, len(dollar_vals) - 1) return dollar_vals[idx] val = _find_bare_amount(chunk) if val is not None: return val return None def _find_box_money_by_desc( text: str, desc_pattern: str, max_scan: int = 400 ) -> Optional[float]: 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 and "\n" in chunk[:nxt.start()]: chunk = chunk[:nxt.start()] val = _find_bare_amount(chunk) if val is not None: return val return None def _find_box_text(text: str, box_key: str, max_scan: int = 80) -> Optional[str]: 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 and "\n" in chunk[:nxt.start()]: chunk = chunk[:nxt.start()] # Box 10a: 2-letter state code, but not "10b" / "10a" if box_key == "box10a_state": if re.match(r"\s*\|\s*10b", chunk, re.IGNORECASE): continue state_m = re.search(r"\b([A-Z]{2})\b", chunk) if state_m and state_m.group(1) in _US_STATE_CODES: return state_m.group(1) continue # Box 10b: 4+ digit state ID, but not the label "State identification" if box_key == "box10b_state_id": if re.match(r"\s+State\s+identification", chunk, re.IGNORECASE): continue id_m = re.search(r"\b(\d{4,})\b", chunk) if id_m: return id_m.group(1) continue # Default token = re.search(r"[A-Za-z0-9][A-Za-z0-9'.\-]{0,40}", chunk) if token: return token.group(0).strip() return None # --------------------------------------------------------------------------- # Labeled branch # --------------------------------------------------------------------------- def _parse_labeled(raw: str, base: Dict[str, Any]) -> Dict[str, Any]: base["form_layout"] = "labeled" _extract_identity_labeled(raw, base) base["box1_unemployment_compensation"] = ( _find_box_money(raw, "box1_unemployment_compensation") or _find_box_money_by_desc(raw, r"Unemployment\s+compensation") ) base["box2_state_local_refunds"] = ( _find_box_money(raw, "box2_state_local_refunds") or _find_box_money_by_desc(raw, r"State\s+or\s+local\s+income\s+tax\s+refunds") ) base["box3_box2_tax_year"] = _find_box_money(raw, "box3_box2_tax_year") 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_rtaa_payments"] = _find_box_money(raw, "box5_rtaa_payments") base["box6_taxable_grants"] = _find_box_money(raw, "box6_taxable_grants") base["box7_agriculture_payments"] = _find_box_money(raw, "box7_agriculture_payments") base["box8_trade_or_business_income"] = bool( re.search(r"Box\s*2\s+is\s+trade\s+or\s+business\s+income\s*[X✓☒]", raw, re.IGNORECASE) ) base["box9_market_gain"] = _find_box_money(raw, "box9_market_gain") base["box10a_state"] = _find_box_text(raw, "box10a_state") base["box10b_state_id"] = _find_box_text(raw, "box10b_state_id") base["box11_state_tax_withheld"] = _find_box_money(raw, "box11_state_tax_withheld") return base # --------------------------------------------------------------------------- # Pre-printed branch # --------------------------------------------------------------------------- def _parse_preprinted( raw: str, base: Dict[str, Any], warnings: List[str] ) -> Dict[str, Any]: base["form_layout"] = "preprinted" payer_name = _find_payer_name(raw) if payer_name: base["payer_name"] = payer_name blocks = _split_into_blocks(raw, payer_name) block = blocks[0] if blocks else raw lines = block.split("\n") values: List[tuple[int, int, float]] = [] for i, line in enumerate(lines): for m in re.finditer(r"\$\s*([\d,]+\.\d{2})", line): val = _clean_money(m.group(1)) if val is not None: values.append((i, m.start(), val)) if not values: warnings.append("No dollar amounts found in the pre-printed layout.") else: state_tax = None remaining = list(values) has_state_anchor = bool(re.search( r"State\s+income\s+tax\s+withheld|Box\s*11\b|\b11\s+State", raw, re.IGNORECASE, )) if has_state_anchor and len(remaining) >= 1: state_tax = remaining[-1][2] remaining = remaining[:-1] elif len(values) >= 9: state_tax = remaining[-1][2] remaining = remaining[:-1] elif len(values) >= 2: if values[-1][0] - values[0][0] > 12: state_tax = remaining[-1][2] remaining = remaining[:-1] if state_tax is not None: base["box11_state_tax_withheld"] = state_tax reading_order = [ "box1_unemployment_compensation", "box2_state_local_refunds", "box3_box2_tax_year", "box4_fed_tax_withheld", "box5_rtaa_payments", "box6_taxable_grants", "box7_agriculture_payments", "box9_market_gain", ] for i, (_, _, val) in enumerate(remaining): if i < len(reading_order): base[reading_order[i]] = val base["box8_trade_or_business_income"] = bool(re.search( r"Box\s*2\s+is\s+trade\s+or\s+business\s+income\s*[X✓☒]" r"|[X✓☒]\s*Box\s*2\s+is\s+trade", raw, re.IGNORECASE, )) m = re.search(r"\b([A-Z]{2})\s+(\d{6,})\b", block) if m: base["box10a_state"] = m.group(1) base["box10b_state_id"] = m.group(2) m = _EIN_RE.search(block) if m: base["payer_tin"] = m.group(1) m = _MASKED_SSN_RE.search(block) or _SSN_RE.search(block) if m: base["recipient_tin"] = m.group(1) 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 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 if tin_row_idx >= 0: rn, rn_idx = _extract_recipient_name(lines, tin_row_idx) if rn: base["recipient_name"] = rn if rn_idx >= 0: addr = _extract_recipient_address(lines, rn_idx) if addr: base["recipient_address"] = addr m = _ACCOUNT_NO_RE.search(block) if m: base["account_number"] = m.group(1) m = _PHONE_RE.search(block) if m: base["payer_phone"] = m.group(0) if ( base.get("box1_unemployment_compensation") is None and base.get("box2_state_local_refunds") is None ): warnings.append("Neither Box 1 nor Box 2 could be located.") return base # --------------------------------------------------------------------------- # Public entry point # --------------------------------------------------------------------------- def parse_1099_g_2021( 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_g_2021", "form_code": "1099-G-2021", # ★ FIX #2: hardcode tax year for this parser. Auto-detection from # `raw[:500]` was picking up `2020` from interleaved instruction # text (page 2 says "refund is for 2020 taxes"). "tax_year": 2021, "filename": filename, "parse_warnings": warnings, "source_pages": [ {"page": p.get("page", 1), "source": p.get("source", "text")} for p in pages ], "void": False, "corrected": False, "second_tin_not": False, } # Only override if the form explicitly says "for tax year YYYY". m = re.search( r"(?:for\s+tax\s+year|tax\s+year)\s*[:\-]?\s*(20\d{2})", raw, re.IGNORECASE, ) 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) # --- Fallback TIN extraction (in case the labels were lost by OCR) --- if not base.get("payer_tin"): m = _EIN_RE.search(raw) if m: base["payer_tin"] = m.group(1) if not base.get("recipient_tin"): m = _MASKED_SSN_RE.search(raw) or _SSN_RE.search(raw) if m: base["recipient_tin"] = m.group(1) # --- Fallback account number (letters + digits, start-of-line, allow noise) if not base.get("account_number"): m = _ACCOUNT_NO_RE.search(raw) if m: base["account_number"] = m.group(1) if not base.get("account_number"): m = re.search( r"Account\s+number\s*[:\-]?\s*([A-Z0-9][A-Z0-9\-]{3,25})\b", raw, re.IGNORECASE, ) if m: base["account_number"] = m.group(1) if not base.get("account_number"): # "Account number (see instructions)" label, value on NEXT line m = re.search( r"Account\s+number[^\n]*\n\s*([A-Z]{2,5}\d{6,15})\b", raw, re.IGNORECASE, ) if m: base["account_number"] = m.group(1) # --- Checkboxes --- base["void"] = bool(re.search( r"VOID\s*[X✓☒]|[X✓☒]\s*VOID", raw, re.IGNORECASE, )) base["corrected"] = bool(re.search( r"CORRECTED\s*[X✓☒]|[X✓☒]\s*CORRECTED", raw, re.IGNORECASE, )) base["second_tin_not"] = bool(re.search( r"2nd\s+TIN\s+not\s*[X✓☒]|[X✓☒]\s*2nd\s+TIN", raw, re.IGNORECASE, )) base["raw_text"] = raw return base