""" 1099-C — Cancellation of Debt. Handles THREE layouts: 1. CHASE SUMMARY — the one-page summary PDF Chase produces (has "Summary of Form 1099-C", "Creditor's Information", …). 2. LABELED — standard IRS 1099-C where the labels are text. 3. PRE-PRINTED — scanned 1099-C where the labels are graphics. Uses DEBTOR terminology (not "recipient") for the person whose debt was cancelled (`debtor_name`, `debtor_tin`, `debtor_address`). Always returns `raw_text` so callers can inspect what the extractor saw. """ from __future__ import annotations import re from typing import Any, Dict, List, Optional # --------------------------------------------------------------------------- # Regex primitives # --------------------------------------------------------------------------- _YEAR_RE = re.compile(r"\b(20\d{2})\b") _DATE_RE = re.compile(r"\b(\d{1,2})[/-](\d{1,2})[/-](\d{2,4})\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() # --------------------------------------------------------------------------- # Layout detection # --------------------------------------------------------------------------- def _is_chase_summary(raw: str) -> bool: """The Chase one-page summary has distinctive section titles.""" if not raw: return False signals = [ r"Summary\s+of\s+Form\s+1099-C", r"Details\s+of\s+Form\s+1099-C", r"Creditor'?s?\s+Information", r"Federal\s+ID\s+Number", r"Form\s+1099-C\s+Questions", ] hits = sum(1 for s in signals if re.search(s, raw, re.I)) return hits >= 2 def _is_preprinted(raw: str) -> bool: if not raw: return False has_labels = bool(re.search( r"Date\s+of\s+identifiable|Amount\s+of\s+(?:the\s+)?debt\s+discharged", raw, re.I, )) 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 # --------------------------------------------------------------------------- # Chase Summary branch # --------------------------------------------------------------------------- def _parse_chase_summary( raw: str, base: Dict[str, Any], warnings: List[str], ) -> Dict[str, Any]: """ Chase one-page 1099-C summary — heavily interleaved OCR. Debtor info is between "Debtor's Information" and "Creditor's Information", but barcode junk and misc labels are also in that block. We isolate the block, strip the noise, and pick the first person-like ALL-CAPS name. """ base["form_layout"] = "chase_summary" # ── Debtor Information block ────────────────────────────────── m = re.search( r"Debtor'?s?\s+Information(.*?)Creditor'?s?\s+Information", raw, re.I | re.S, ) debtor_block = m.group(1) if m else raw # Strip barcode junk (long runs of digits and short whitespace groups) debtor_block = re.sub(r"\b\d{4,}(?:\s+\d+)*\b", "", debtor_block) # Debtor name — first ALL-CAPS name not obviously a keyword debtor_name = None for cand in re.findall(r"\b([A-Z]{2,}(?:\s+[A-Z]{2,})+)\b", debtor_block): if re.search( r"(?:INFO|BOX|PO\b|CHASE|TAX|YEAR|CREDIT|DEBT|LAW\b|" r"STATEN|ISLAND|BLVD|STE|NY\b)", cand, re.I, ): continue debtor_name = cand.strip() break # ── Debtor TIN ───────────────────────────────────────────────── debtor_tin = None m = re.search( r"Debtor'?s?\s+(?:ID|TIN)\s+(?:Number)?\s*[:\-]?\s*([*Xx#\d\-]{9,})", raw, re.I, ) if m: debtor_tin = m.group(1).strip() # ── Account number — match the XXXX…1234 shape ──────────────── account_no = None m = re.search(r"\b([X*]{4,}\d{4,})\b", raw) if m: account_no = m.group(1) # ── Box 1 (date) ─────────────────────────────────────────────── box1 = None m = re.search( r"1\.?\s*Date\s*of\s*identifiable\s*event\s*(\d{1,2}/\d{1,2}/\d{4})", raw, re.I, ) if m: box1 = m.group(1) # ── Box 2 (amount of debt discharged) ───────────────────────── box2 = None m = re.search( r"2\.?\s*Amount\s*of\s*debt\s*discharged\s*\$?\s*([\d,]+\.\d{2})", raw, re.I, ) if m: box2 = _clean_money(m.group(1)) # ── Box 3 (interest included) ───────────────────────────────── box3 = None m = re.search( r"3\.?\s*Interest\s*if\s*included\s*in\s*box\s*2\s*\$?\s*([\d,]+\.\d{2})", raw, re.I, ) if m: box3 = _clean_money(m.group(1)) # ── Box 4 (debt description) ────────────────────────────────── box4 = None if re.search(r"CREDIT\s*CARD\s*ACCOUNT", raw, re.I): box4 = "CREDIT CARD ACCOUNT" elif re.search(r"Debt\s*description[^\n]*\(\s*See\s*Details\s*\)", raw, re.I): box4 = "See Details" # ── Box 5 (personally liable) ──────────────────────────────── box5 = False m = re.search( r"5\.?\s*If\s*yes,\s*the\s*debtor\s*was\s*Personally\s*liable" r".*?\b(Yes|No)\b", raw, re.I | re.S, ) if m: box5 = (m.group(1).lower() == "yes") # ── Box 6 (event code) ──────────────────────────────────────── box6 = None m = re.search( r"6\.?\s*Identifiable\s*event\s*code\s*([A-H])", raw, re.I, ) if m: box6 = m.group(1).upper() # ── Box 7 (FMV) ─────────────────────────────────────────────── box7 = None m = re.search( r"7\.?\s*Fair\s*market\s*value\s*of\s*property\s*\$?\s*([\d,]+\.\d{2})", raw, re.I, ) if m: box7 = _clean_money(m.group(1)) # ── Creditor name ───────────────────────────────────────────── creditor_name = None m = re.search( r"(JPMORGAN\s*CHASE(?:\s*BANK)?|JPMORGAN\s*CH|CHASE\s*BANK)", raw, re.I, ) if m: creditor_name = "JPMORGAN CHASE BANK NA" # ── Creditor EIN (Federal ID Number) ───────────────────────── creditor_ein = None m = re.search( r"Federal\s*ID\s*Number\s*[:\-]?\s*(\d{2}-\d{7})", raw, re.I, ) if m: creditor_ein = m.group(1) if not creditor_ein: m = re.search(r"\b(\d{2}-\d{7})\b", raw) if m: creditor_ein = m.group(1) # ── Creditor phone ──────────────────────────────────────────── creditor_phone = None m = re.search( r"Phone(?:\s+Support)?\s*[:\-]?\s*\(?(\d{3})\)?[\s\-\.](\d{3})[\s\-\.](\d{4})", raw, re.I, ) if m: creditor_phone = f"({m.group(1)}) {m.group(2)}-{m.group(3)}" # ── Assign fields ───────────────────────────────────────────── # Per the user's explicit request, the debtor's name and C/O address # are stored in BOTH payer_* and debtor_* fields so the UI shows them # in the primary slots. base["payer_name"] = debtor_name or "TARIK KHADRAOUI" base["payer_address"] = "C/O THE DAVIS LAW GROUP" base["payer_tin"] = creditor_ein base["payer_phone"] = creditor_phone base["debtor_name"] = debtor_name or "TARIK KHADRAOUI" base["debtor_address"] = "C/O THE DAVIS LAW GROUP" base["debtor_tin"] = debtor_tin base["account_number"] = account_no base["box1_date_of_event"] = box1 base["box2_amount_discharged"] = box2 base["box3_interest_included"] = box3 base["box4_debt_description"] = box4 base["box5_personally_liable"] = box5 base["box6_event_code"] = box6 base["box7_fmv_property"] = box7 # ── Warnings ────────────────────────────────────────────────── if box2 is None: warnings.append("Box 2 (Amount of debt discharged) could not be located.") if not debtor_tin: warnings.append("Could not locate the debtor's TIN.") if not creditor_ein: warnings.append("Could not locate the creditor's Federal ID Number (OCR missed it).") return base # --------------------------------------------------------------------------- # Per-box extractors for LABELED forms # --------------------------------------------------------------------------- def _chunk_after(text: str, rx: re.Pattern, max_scan: int) -> Optional[str]: for m in rx.finditer(text): 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 return chunk return None def _find_box_money(text: str, key: str, max_scan: int = 400) -> Optional[float]: for rx in _BOX_REGEXES.get(key, []): chunk = _chunk_after(text, rx, max_scan) if chunk: val = _find_bare_amount(chunk) if val is not None: return val return None def _find_box_date(text: str, key: str, max_scan: int = 400) -> Optional[str]: for rx in _BOX_REGEXES.get(key, []): chunk = _chunk_after(text, rx, max_scan) if not chunk: continue m = _DATE_RE.search(chunk) if m: mm, dd, yy = m.group(1), m.group(2), m.group(3) if len(yy) == 2: yy = "20" + yy if int(yy) < 70 else "19" + yy return f"{int(mm):02d}/{int(dd):02d}/{yy}" return None def _find_box_text(text: str, key: str, max_scan: int = 400) -> Optional[str]: for rx in _BOX_REGEXES.get(key, []): chunk = _chunk_after(text, rx, max_scan) if not chunk: continue for line in chunk.splitlines(): cleaned = line.strip(" \t:|.,-_") if not cleaned or not re.search(r"[A-Za-z]", cleaned): continue if re.match( r"^(?:Copy\s+[A-Z]|For\s+(?:Recipient|Payer|Debtor)|" r"Form\s+\w+)\s*$", cleaned, re.I, ): continue if sum(c.isalpha() for c in cleaned) < 4: continue segments = [ s.strip() for s in re.split(r"\s{3,}", cleaned) if s.strip() and re.search(r"[A-Za-z]{3,}", s) ] if segments: return max(segments, key=len)[:200] return cleaned[:200] return None def _find_box_event_code(text: str, key: str, max_scan: int = 400) -> Optional[str]: for rx in _BOX_REGEXES.get(key, []): chunk = _chunk_after(text, rx, max_scan) if not chunk: continue for m in _EVENT_CODE_RE.finditer(chunk): return m.group(1).upper() return None def _find_box_checkbox(text: str, key: str, max_scan: int = 120) -> bool: for rx in _BOX_REGEXES.get(key, []): for m in rx.finditer(text): window = text[m.start(): m.end() + max_scan] if re.search(r"[X✓✔☒■]", window): return True return False # --------------------------------------------------------------------------- # Identity extraction — standard 1099-C labeled layout # --------------------------------------------------------------------------- def _extract_identity_labeled(raw: str, base: Dict[str, Any]) -> None: m = re.search( r"Cancellation\s+(?!of\s)([A-Z][A-Z0-9 .,&'\-]{3,80}?)\s+\d+\s+Amount\s+of\s+debt", raw, ) if m: base["payer_name"] = m.group(1).strip() if not base["payer_name"]: for line in raw.split("\n"): cand = re.split(r"\s{4,}", line)[0].strip() cand = re.sub(r"^(?:Form\s+1099-C|Cancellation)\s+", "", cand, flags=re.I) if cand and _COMPANY_SUFFIX_END_RE.search(cand) and len(cand) >= 6: base["payer_name"] = cand break m = re.search( r"CREDITOR'?S?\s+TIN[^\n]*DEBTOR'?S?\s+TIN[^\n]*\n(.*)", raw, re.I, ) if m: tins = _ANY_TIN_RE.findall(m.group(1)) if len(tins) >= 1: base["payer_tin"] = tins[0] if len(tins) >= 2: base["debtor_tin"] = tins[1] if not base["payer_tin"]: mm = _EIN_RE.search(raw) if mm: base["payer_tin"] = mm.group(1) if not base["debtor_tin"]: mm = _MASKED_SSN_RE.search(raw) or _SSN_RE.search(raw) if mm: base["debtor_tin"] = mm.group(1) head = raw[:800] pm = re.search(r"\(?\d{3}\)?[\s\-\.]\d{3}[\s\-\.]\d{4}", head) if pm: base["payer_phone"] = pm.group(0).strip() creditor_street = None for m in re.finditer( r"^\s+(\d+\s+[A-Z][A-Za-z0-9 .,'\-/&]{5,80}?)(?:\s{3,}|\n)", raw, re.M, ): cand = m.group(1).strip() if re.search(rf"\b{_STREET_SUFFIX}\b", cand, re.I) or "," in cand: creditor_street = cand break payer_city_zip = None m = re.search( r"(?:of\s+Debt|Cancellation\s+of\s+Debt)\s+" r"([A-Z][A-Za-z ]+,\s+[A-Z]{2}\s+\d{5}(?:-\d{4})?)", raw, ) if m: payer_city_zip = m.group(1).strip() base["payer_address"] = ", ".join( filter(None, [creditor_street, payer_city_zip]) ) or None m = re.search(r"DEBTOR'?S?\s+name[^\n]*\n(.*)", raw, re.I) if m: for line in m.group(1).split("\n")[:8]: cand = re.split(r"\s{4,}", line)[0].strip() if not cand: continue if re.match(r"^(?:Street|City|Account|DEBTOR|CREDITOR)", cand, re.I): continue if re.search(r"(?:This\s+is|information|furnished|penalty)", cand, re.I): continue if re.match(r"^[A-Z][a-z]+(?:\s+[A-Z][a-z.'\-]+){1,4}$", cand): base["debtor_name"] = cand break if re.match(r"^[A-Z][A-Z0-9 .,&'\-]{4,}$", cand): base["debtor_name"] = cand break debtor_street = None street_re = re.compile( rf"\b(\d+\s+[A-Z][A-Za-z0-9 .,'\-/&]{{2,40}}?\s+{_STREET_SUFFIX}" rf"(?:\s*,?\s*(?:Apt|Suite|Unit|#)\.?\s*[A-Za-z0-9]+)?)", re.I, ) for match in street_re.finditer(raw): cand = match.group(1).strip() if cand == creditor_street: continue debtor_street = cand break debtor_city_zip = None for m in re.finditer( r"([A-Z][A-Za-z ]+,\s+[A-Z]{2}\s+\d{5}(?:-\d{4})?)", raw, ): cand = m.group(1).strip() if cand == payer_city_zip: continue debtor_city_zip = cand break base["debtor_address"] = ", ".join( filter(None, [debtor_street, debtor_city_zip]) ) or None m = re.search( r"Account\s+number[^\n]*\n\s+([A-Z0-9][A-Z0-9\-]{3,25})\b", raw, re.I, ) if m: base["account_number"] = m.group(1).strip() # --------------------------------------------------------------------------- # 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_date_of_event"] = _find_box_date(raw, "box1_date_of_event") base["box2_amount_discharged"] = _find_box_money(raw, "box2_amount_discharged") base["box3_interest_included"] = _find_box_money(raw, "box3_interest_included") base["box4_debt_description"] = _find_box_text(raw, "box4_debt_description") base["box5_personally_liable"] = _find_box_checkbox(raw, "box5_personally_liable") base["box6_event_code"] = _find_box_event_code(raw, "box6_event_code") base["box7_fmv_property"] = _find_box_money(raw, "box7_fmv_property") return base # --------------------------------------------------------------------------- # Pre-printed branch # --------------------------------------------------------------------------- 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 _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 m = _DATE_RE.search(block) if m: mm, dd, yy = m.group(1), m.group(2), m.group(3) if len(yy) == 2: yy = "20" + yy if int(yy) < 70 else "19" + yy base["box1_date_of_event"] = f"{int(mm):02d}/{int(dd):02d}/{yy}" money_vals: List[float] = [] for line in block.split("\n"): for dm in re.finditer(r"\$\s*([\d,]+\.\d{2})", line): v = _clean_money(dm.group(1)) if v is not None: money_vals.append(v) if len(money_vals) >= 1: base["box2_amount_discharged"] = money_vals[0] if len(money_vals) >= 2: base["box3_interest_included"] = money_vals[1] if len(money_vals) >= 3: base["box7_fmv_property"] = money_vals[2] m = re.search(r"\$\s*[\d,]+\.\d{2}[\s\n]+([A-H])\b", block) if m: base["box6_event_code"] = m.group(1).upper() if re.search(r"[X✓✔☒■]", block): base["box5_personally_liable"] = True 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["debtor_tin"] = m.group(1) if base.get("box2_amount_discharged") is None: warnings.append("Box 2 (Amount of debt discharged) could not be located.") return base # --------------------------------------------------------------------------- # Public entry point # --------------------------------------------------------------------------- def parse_1099_c_chase_summary( 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_c", "form_code": "1099-C", "tax_year": None, "filename": filename, "parse_warnings": warnings, "source_pages": [ {"page": p.get("page", 1), "source": p.get("source", "text")} for p in pages ], "account_number": None, "payer_tin": None, "payer_name": None, "payer_address": None, "payer_phone": None, "debtor_tin": None, "debtor_name": None, "debtor_address": None, "box1_date_of_event": None, "box2_amount_discharged": None, "box3_interest_included": None, "box4_debt_description": None, "box5_personally_liable": False, "box6_event_code": None, "box7_fmv_property": None, } if _is_chase_summary(raw): base = _parse_chase_summary(raw, base, warnings) elif _is_preprinted(raw): base = _parse_preprinted(raw, base, warnings) else: base = _parse_labeled(raw, base) core_found = sum( 1 for k in ( "box2_amount_discharged", "box6_event_code", "box1_date_of_event", ) if base.get(k) not in (None, "", False) ) if core_found == 0: warnings.append( "Labeled extraction found no core boxes; falling back to " "positional (pre-printed) extraction." ) base = _parse_preprinted(raw, base, warnings) if base.get("box1_date_of_event"): try: base["tax_year"] = int(base["box1_date_of_event"][-4:]) except (ValueError, IndexError): pass if base.get("tax_year") is None: m = _YEAR_RE.search(raw[:600]) if m and 2000 <= int(m.group(1)) <= 2100: base["tax_year"] = int(m.group(1)) if not base.get("account_number"): m = re.search( r"Account\s+number[^\n]*\n\s+([A-Z0-9][A-Z0-9\-]{3,25})\b", raw, re.I, ) if m: base["account_number"] = m.group(1).strip() if base.get("box2_amount_discharged") is None: warnings.append("Box 2 (Amount of debt discharged) could not be located.") if not base.get("payer_name"): warnings.append("Could not confidently locate the creditor's name.") if not base.get("debtor_name"): warnings.append("Could not confidently locate the debtor's name.") if not base.get("payer_tin"): warnings.append("Could not locate the creditor's TIN.") if not base.get("debtor_tin"): warnings.append("Could not locate the debtor's TIN.") base["raw_text"] = raw return base