import re from datetime import datetime from typing import Any class CourierBillParser: """ Deterministic Courier Bill of Entry parser. The parser reuses the same PDFAnalyzer output model used by the Shipping Bill parser: page text, words and coordinates. Extraction is driven by labels/sections and visual text relationships, not by the values of one particular PDF. """ PARENT_FIELDS = [ "doctype", "out_of_charge_date", "cour_be_importername", "cour_be_arrivalairport", "cour_be_marksnos", "cour_be_iec", "cour_be_coo", "cour_be_airline", "cour_be_mawbno", "cour_be_couriername", "cour_be_iecbranch", "cour_be_countryofconsignment", "cour_be_flightno", "cour_be_mawb_dt", "cour_be_betype_description", "cour_be_importer_address", "cour_be_pkgs", "cour_be_arrivaldate", "cour_be_hawbno", "cour_be_beno", "cour_be_adcode", "cour_be_invoicecount", "cour_be_igmno", "cour_be_hawb_dt", "be_date", "cour_be_gstin", "cour_be_category", "cour_be_inward_date", "cour_be_gwt_kgs", "cour_be_total_duty", "cour_be_interest_amt", "cour_be_tr6_no", "cour_be_pymt_amt", "cour_be_challan_date", "cour_be_supplier_name", ] INVOICE_FIELDS = [ "doctype", "cour_be_invsno", "cour_be_inv_nature_of_txn", "cour_be_incoterm", "cour_be_inv_supplier_name", "cour_be_invoice_no", "cour_be_inv_pymt_terms", "cour_be_inv_related", "suppliers_address", "cour_be_invoice_date", "cour_be_valuationmethod", "cour_svb_refno", "cour_be_inv_value", "cour_be_inv_svbdate", "cour_be_inv_currency", "cour_be_prov_final", ] ITEM_FIELDS = [ "doctype", "linked_invoice_id_sno", "cour_be_item_upi", "cour_be_item_manufacturer", "cour_be_item_upi_currency", "cour_be_item_countryoforigin", "cour_be_item_itemsn", "cour_be_item_qty", "cour_be_item_ritc", "cour_be_item_itemdescription", "cour_be_item_uom", "cour_be_item_ctsh", "cour_be_item_itemgendescription", "cour_be_item_exchg_rate", "cour_be_item_cetsh", "cour_be_item_sws_rt", "cour_be_item_assessable_value", "cour_be_item_sws_amt", "cour_be_item_bcd_rt", "cour_be_item_igst_rt", "cour_be_item_bcd_amt", "cour_be_item_igst_amt", "cour_be_item_aidc_rt", "cour_be_item_aidc_amt", "cour_be_item_cmpcess_rt", "cour_be_item_cmpcess_amt", ] # ========================================================= # DYNAMIC SCHEMA / ALIAS CONFIGURATION # ========================================================= # # These definitions deliberately use the existing Courier Frappe # fieldnames as the source of truth. PDF labels are aliases only. # Unknown PDF headers are never emitted as new JSON keys. # COMPLETE_CHILD_FIELDS = { "invoices": list(INVOICE_FIELDS), "items": list(ITEM_FIELDS), } CHILD_FIELD_ALIASES = { "invoices": { "cour_be_invsno": [ "INV SNO", "INVOICE SNO", "INVOICE SERIAL NO", "INVOICE SERIAL NUMBER", "SR NO", "SERIAL NO", ], "cour_be_inv_nature_of_txn": [ "NATURE OF TRANSACTION", "NATURE OF TXN", "TRANSACTION NATURE", "NATURE", ], "cour_be_incoterm": [ "TERMS OF INVOICE", "INCOTERM", "INCOTERMS", "TERMS OF DELIVERY", ], "cour_be_inv_supplier_name": [ "SUPPLIER NAME", "NAME OF SUPPLIER", "EXPORTER NAME", "SELLER NAME", ], "cour_be_invoice_no": [ "INVOICE NUMBER", "INVOICE NO", "INV NUMBER", "INV NO", "INVOICE #", "INV #", ], "cour_be_inv_pymt_terms": [ "TERMS OF PAYMENT", "PAYMENT TERMS", "PAYMENT TERM", "PAYMENT", ], "cour_be_inv_related": [ "RELATED", "RELATED INVOICE", "RELATED PARTY", "RELATED TO", ], "suppliers_address": [ "SUPPLIER ADDRESS", "ADDRESS OF SUPPLIER", "EXPORTER ADDRESS", "SELLER ADDRESS", ], "cour_be_invoice_date": [ "DATE OF INVOICE", "INVOICE DATE", "INV DATE", ], "cour_be_valuationmethod": [ "METHOD OF VALUATION", "VALUATION METHOD", "METHOD OF VALUATION", ], "cour_svb_refno": [ "SVB REF NO", "SVB REFERENCE NO", "SVB REFERENCE NUMBER", ], "cour_be_inv_value": [ "INVOICE VALUE", "INV VALUE", "VALUE OF INVOICE", ], "cour_be_inv_svbdate": [ "SVB DATE", "DATE OF SVB", ], "cour_be_inv_currency": [ "CURRENCY", "INVOICE CURRENCY", "CURRENCY OF INVOICE", ], "cour_be_prov_final": [ "PROVISIONAL FINAL", "PROVISIONAL", "FINAL", ], }, "items": { "linked_invoice_id_sno": [ "LINKED INVOICE ID", "INVOICE ID", "INVOICE SNO", ], "cour_be_item_upi": [ "UNIT PRICE", "UNIT PRICE INVOICE", "PRICE", "UNIT PRICE (UPI)", "UPI", ], "cour_be_item_manufacturer": [ "MANUFACTURER", "MANUFACTURER NAME", "SUPPLIER", "SUPPLIER NAME", ], "cour_be_item_upi_currency": [ "CURRENCY FOR UNIT PRICE", "UNIT PRICE CURRENCY", "UPI CURRENCY", "PRICE CURRENCY", ], "cour_be_item_countryoforigin": [ "COUNTRY OF ORIGIN", "ORIGIN COUNTRY", ], "cour_be_item_itemsn": [ "ITEM SNO", "ITEM SR NO", "ITEM SERIAL NO", "ITEM NUMBER", "ITEM NO", "ITEM", ], "cour_be_item_qty": [ "QUANTITY", "QTY", "ITEM QUANTITY", ], "cour_be_item_ritc": [ "RITC", "RITC CODE", "TARIFF ITEM", "TARIFF CODE", ], "cour_be_item_itemdescription": [ "ITEM DESCRIPTION", "DESCRIPTION OF ITEM", "DESCRIPTION", ], "cour_be_item_uom": [ "UNIT OF MEASURE", "UOM", "UNIT", ], "cour_be_item_ctsh": [ "CTSH", "CTSH CODE", ], "cour_be_item_itemgendescription": [ "ITEM GENERAL DESCRIPTION", "GENERAL DESCRIPTION", "ITEM GEN DESCRIPTION", ], "cour_be_item_exchg_rate": [ "RATE OF EXCHANGE", "EXCHANGE RATE", "EXCH RATE", ], "cour_be_item_cetsh": [ "CETSH", "CETSH CODE", ], "cour_be_item_sws_rt": [ "SWS RATE", "SW SRCHRG RATE", "SWS RT", "SOCIAL WELFARE SURCHARGE RATE", ], "cour_be_item_assessable_value": [ "ASSESSABLE VALUE", "ASSESS. VALUE", "ASSESSABLE VAL", "ASSESS VALUE", ], "cour_be_item_sws_amt": [ "SWS AMOUNT", "SW SRCHRG AMOUNT", "SWS AMT", ], "cour_be_item_bcd_rt": [ "BCD RATE", "BCD RT", "BASIC CUSTOMS DUTY RATE", ], "cour_be_item_igst_rt": [ "IGST RATE", "IGST RT", ], "cour_be_item_bcd_amt": [ "BCD AMOUNT", "BCD AMT", "BASIC CUSTOMS DUTY AMOUNT", ], "cour_be_item_igst_amt": [ "IGST AMOUNT", "IGST AMT", ], "cour_be_item_aidc_rt": [ "AIDC RATE", "AIDC RT", ], "cour_be_item_aidc_amt": [ "AIDC AMOUNT", "AIDC AMT", ], "cour_be_item_cmpcess_rt": [ "CMPNSTRY RATE", "CMP CESS RATE", "COMPENSATION CESS RATE", "CMPCESS RATE", ], "cour_be_item_cmpcess_amt": [ "CMPNSTRY AMOUNT", "CMP CESS AMOUNT", "COMPENSATION CESS AMOUNT", "CMPCESS AMOUNT", ], }, } PARENT_FIELD_ALIASES = { "out_of_charge_date": [ "OOC ISSUED ON", "OUT OF CHARGE DATE", "OOC DATE", ], "cour_be_importername": [ "IMPORTER NAME", "PARTICULARS OF THE IMPORTER", "NAME", ], "cour_be_arrivalairport": [ "ARRIVAL AIRPORT", "AIRPORT OF ARRIVAL", ], "cour_be_marksnos": [ "MARKS AND NUMBERS", "MARKS & NUMBERS", "MARKS NOS", ], "cour_be_iec": [ "IMPORT EXPORT CODE", "IMPORT-EXPORT CODE", "IEC", "IEC CODE", ], "cour_be_coo": [ "COUNTRY OF ORIGIN", "ORIGIN COUNTRY", ], "cour_be_airline": [ "AIRLINE", "CARRIER", "AIR CARRIER", ], "cour_be_mawbno": [ "MASTER AIRWAY BILL (MAWB) NUMBER", "MAWB NUMBER", "MAWB NO", "MASTER AIRWAY BILL NUMBER", ], "cour_be_couriername": [ "NAME OF THE AUTHORIZED COURIER", "AUTHORIZED COURIER", "AUTHORISED COURIER", "COURIER NAME", ], "cour_be_iecbranch": [ "IMPORT EXPORT BRANCH CODE", "IEC BRANCH CODE", "IMPORT EXPORT BRANCH", ], "cour_be_countryofconsignment": [ "COUNTRY OF CONSIGNMENT", "CONSIGNMENT COUNTRY", ], "cour_be_flightno": [ "FLIGHT NO", "FLIGHT NUMBER", "FLIGHT", ], "cour_be_mawb_dt": [ "DATE OF MAWB", "MAWB DATE", "MAWB DT", ], "cour_be_betype_description": [ "TYPE OF BOE", "BOE TYPE", "TYPE OF BILL OF ENTRY", ], "cour_be_importer_address": [ "IMPORTER ADDRESS", "ADDRESS OF IMPORTER", ], "cour_be_pkgs": [ "NUMBER OF PACKAGES", "NO OF PACKAGES", "PACKAGES", ], "cour_be_arrivaldate": [ "ARRIVAL DATE", "DATE OF ARRIVAL", ], "cour_be_hawbno": [ "HOUSE AIRWAY BILL (HAWB) NUMBER", "HAWB NUMBER", "HAWB NO", "HOUSE AIRWAY BILL NUMBER", ], "cour_be_beno": [ "CBEXIV NUMBER", "BOE NUMBER", "BILL OF ENTRY NUMBER", "CB NO", "BOE NO", ], "cour_be_adcode": [ "AUTHORISED DEALER CODE OF BANK", "AUTHORIZED DEALER CODE", "AD CODE", "AUTHORISED DEALER CODE", ], "cour_be_invoicecount": [ "NUMBER OF INVOICES", "NO OF INVOICES", "INVOICE COUNT", ], "cour_be_igmno": [ "IMPORT GENERAL MANIFEST (IGM) NUMBER", "IGM NUMBER", "IGM NO", ], "cour_be_hawb_dt": [ "DATE OF HAWB", "HAWB DATE", "HAWB DT", ], "be_date": [ "BOE DATE", "BILL OF ENTRY DATE", ], "cour_be_gstin": [ "KYC ID", "GSTIN", "GSTIN NUMBER", ], "cour_be_category": [ "CATEGORY OF BOE", "BOE CATEGORY", "CATEGORY", ], "cour_be_inward_date": [ "DATE OF ENTRY INWARD", "INWARD DATE", "ENTRY INWARD DATE", ], "cour_be_gwt_kgs": [ "GROSS WEIGHT", "GROSS WEIGHT KGS", "GWT", "G WT", ], "cour_be_total_duty": [ "TOTAL AMOUNT", "TOTAL DUTY", "TOTAL DUTY AMOUNT", ], "cour_be_interest_amt": [ "INTEREST AMOUNT", "INTEREST AMT", "INTEREST", ], "cour_be_tr6_no": [ "CHALLAN NUMBER", "TR6 NUMBER", "TR6 NO", "CHALLAN NO", ], "cour_be_pymt_amt": [ "PAYMENT AMOUNT", "PAYMENT AMT", "AMOUNT PAID", ], "cour_be_challan_date": [ "CHALLAN DATE", "DATE OF CHALLAN", "PAYMENT DATE", ], "cour_be_supplier_name": [ "SUPPLIER NAME", "NAME OF SUPPLIER", "EXPORTER NAME", ], } # Fields that are known to be numeric floats in the Courier Frappe schema. CHILD_FLOAT_FIELDS = { "cour_be_item_upi", "cour_be_item_qty", "cour_be_item_exchg_rate", "cour_be_item_sws_rt", "cour_be_item_assessable_value", "cour_be_item_sws_amt", "cour_be_item_bcd_rt", "cour_be_item_igst_rt", "cour_be_item_bcd_amt", "cour_be_item_igst_amt", "cour_be_item_aidc_rt", "cour_be_item_aidc_amt", "cour_be_item_cmpcess_rt", "cour_be_item_cmpcess_amt", "cour_be_inv_value", } PARENT_FLOAT_FIELDS = { "cour_be_gwt_kgs", "cour_be_total_duty", "cour_be_interest_amt", "cour_be_pymt_amt", } PARENT_INTEGER_FIELDS = { "cour_be_pkgs", "cour_be_invoicecount", } # Internal table classifications. These are parser-level names only; # they never become JSON keys. They let fragmented/moved pdfplumber # tables be identified from their headers rather than page coordinates. TABLE_ALIASES = { "invoices": [ "INVOICE DETAILS", "INVOICE NUMBER", "INVOICE NO", "DATE OF INVOICE", "INVOICE VALUE", "CURRENCY", "SUPPLIER DETAILS", "NATURE OF TRANSACTION", ], "items": [ "ITEM DETAILS", "ITEM DESCRIPTION", "ITEM DESCRIPTION/ GENERIC DESCRIPTION", "QUANTITY", "QTY", "RITC", "CTSH", "CETSH", "COUNTRY OF ORIGIN", "UNIT PRICE", "UPI", "UOM", ], "duty": [ "DUTY DETAILS", "DUTY HEAD", "BCD", "AIDC", "SW SRCHRG", "IGST", "CMPNSTRY", "RATE", ], } TABLE_DOCTYPES = { "invoices": "Courier Bill of Entry Invoices", "items": "Courier Bill of Entry Items", } def __init__(self, pages: list[dict[str, Any]]): self.pages = pages or [] self.trace: list[dict[str, Any]] = [] self.page_text = { int(p.get("page_number", i + 1)): str(p.get("text") or "") for i, p in enumerate(self.pages) } self.full_text = "\n".join(self.page_text.values()) self.words = self._all_words() self.page_words: dict[int, list[dict[str, Any]]] = {} for word in self.words: self.page_words.setdefault(word["_page"], []).append(word) for page in self.page_words: self.page_words[page].sort( key=lambda w: (float(w.get("y0", 0)), float(w.get("x0", 0))) ) # Build the complete visual row/table model once. # This is the same foundation used by ShippingBillParser and makes # recovery independent of a fixed page number or coordinate range. self.rows = self._group_words_into_rows(self.words) self.tables = self._all_tables() # ========================================================= # ENTRY POINT # ========================================================= def parse(self) -> dict[str, Any]: result = self._empty_result() self._extract_parent_fields(result) invoice = self._extract_invoice() item = self._extract_item() result["invoices"] = [invoice] if invoice else [] result["items"] = [item] if item else [] self._extract_payment(result) self._extract_out_of_charge(result) # Generic table recovery is fill-only. The established Courier # extraction remains authoritative; detected tables only supply # fields that are missing or fragmented in the primary result. self._recover_from_generic_tables(result) # Collapse wrapped/duplicate item fragments before semantic child # recovery so later passes operate on stable logical rows. result["items"] = self._collapse_courier_item_continuations( result.get("items", []) ) # Dynamic recovery is intentionally fill-only. It never overwrites # values already obtained by the validated Courier extraction logic. self._dynamic_parent_schema_recovery(result) self._dynamic_child_schema_recovery(result) self._finalize(result) return { "courier_bill_json": result, "_trace": self.trace, } # ========================================================= # OUTPUT SHAPE # ========================================================= def _empty_result(self): result = {field: "" for field in self.PARENT_FIELDS} result["doctype"] = "Courier Bill of Entry" result["cour_be_pkgs"] = 0 result["cour_be_invoicecount"] = 0 result["cour_be_gwt_kgs"] = 0.0 result["cour_be_total_duty"] = 0.0 result["cour_be_interest_amt"] = 0.0 result["cour_be_pymt_amt"] = 0.0 result["invoices"] = [] result["items"] = [] return result # ========================================================= # PARENT FIELDS # ========================================================= def _extract_parent_fields(self, result): p1 = self.page_text.get(1, "") p2 = self.page_text.get(2, "") p3 = self.page_text.get(3, "") p4 = self.page_text.get(4, "") p5 = self.page_text.get(5, "") p6 = self.page_text.get(6, "") # The standard Courier form is a two-column visual layout. Prefer # coordinate-based extraction for values that sit beside another # label on the same row. This avoids column leakage. self._set(result, "cour_be_couriername", self._visual_right(p1, 1, 198, 313, 60, 420, 520)) self._set(result, "cour_be_iec", self._visual_right(p1, 1, 344, 73, 95, 165, 310)) self._set(result, "cour_be_iecbranch", self._visual_right(p1, 1, 344, 322, 414, 415, 470)) self._set(result, "cour_be_importername", self._visual_span(p1, 1, 366, 165, 365, 3)) self._set(result, "cour_be_importer_address", self._visual_span(p1, 1, 366, 415, 570, 5)) self._set(result, "cour_be_adcode", self._visual_right(p1, 1, 447, 49, 160, 165, 310)) self._set(result, "cour_be_beno", self._after_label(p1, r"CBEXIV\s+Number\s*:", 180)) if not result["cour_be_beno"]: self._set(result, "cour_be_beno", self._after_label(p1, r"BOE\s+Number\s*:", 180)) self._set(result, "be_date", self._after_label(p1, r"BOE\s+Date\s*:", 30)) self._set(result, "cour_be_category", self._after_label(p1, r"Category\s+Of\s+BOE\s*:", 60)) self._set(result, "cour_be_betype_description", self._after_label(p1, r"Type\s+Of\s+BOE\s*:", 50)) self._set(result, "cour_be_iec", self._after_label(p1, r"Import\s+export\s+Code\s*:", 30)) self._set(result, "cour_be_iecbranch", self._after_label(p1, r"Import\s+Export\s+Branch\s+Code\s*:", 30)) self._set(result, "cour_be_importername", self._between(p1, r"Name\s*:\s*", r"Address\s*:", section=r"PARTICULARS OF THE IMPORTER")) self._set(result, "cour_be_gstin", self._after_label(p1, r"KYC\s+ID\s*:", 30)) self._set(result, "cour_be_adcode", self._after_label(p1, r"Authorised\s+Dealer\s+Code\s+Of\s+Bank\s*:", 30)) self._set(result, "cour_be_couriername", self._after_label(p1, r"Name\s+of\s+the\s+Authorized\s+Courier\s*:", 80)) self._set(result, "cour_be_supplier_name", self._clean_supplier_name( self._visual_span(p3, 3, 96, 165, 365, 5) )) # Importer address: bounded between importer address and the next # explicit importer field, preserving line breaks. addr = self._block_after_label( p1, r"Name\s*:\s*[^\n]+\s+Address\s*:", stop_patterns=[ r"Category\s+Of\s+Importer", r"Authorised\s+Dealer", r"BOE\s+Number", ], max_lines=6, ) if not result["cour_be_importer_address"]: self._set(result, "cour_be_importer_address", addr) # Page 2 logistics. country_line = self._line_containing(p2, "Country of Origin") if country_line: m = re.search( r"Country\s+of\s+Origin\s*:\s*(.*?)\s+Country\s+of\s+Consignment\s*:\s*(.*)$", country_line, re.I, ) if m: self._set(result, "cour_be_coo", m.group(1)) self._set(result, "cour_be_countryofconsignment", m.group(2)) airline_line = self._line_containing(p2, "Federal Express") or self._line_containing(p2, "Airlines") if airline_line: lines = self._clean_lines(p2) try: idx = next(i for i, x in enumerate(lines) if "Federal Express" in x or re.fullmatch(r"Airlines\s+Flight No.*", x, re.I)) candidate = lines[idx + 1] if "Airlines" in lines[idx] else lines[idx] except StopIteration: candidate = "" if candidate: m = re.search(r"^(.*?)\s+(FX\s*\d+)\s+(\w+)\s+(\d{1,2}/\d{1,2}/\d{4})$", candidate, re.I) if m: self._set(result, "cour_be_airline", m.group(1)) self._set(result, "cour_be_flightno", m.group(2)) self._set(result, "cour_be_arrivalairport", m.group(3)) self._set(result, "cour_be_arrivaldate", m.group(4)) self._set(result, "cour_be_igmno", self._after_label(p2, r"Import\s+General\s+Manifest\s+\(IGM\)\s+Number\s*:", 30)) self._set(result, "cour_be_inward_date", self._after_label(p2, r"Date\s+of\s+Entry\s+Inward\s*:", 30)) self._set(result, "cour_be_mawbno", self._after_label(p2, r"Master\s+Airway\s+Bill\s+\(MAWB\)\s+Number\s*:", 40)) self._set(result, "cour_be_mawb_dt", self._after_label(p2, r"Date\s+Of\s+MAWB\s*:", 30)) self._set(result, "cour_be_hawbno", self._after_label(p2, r"House\s+Airway\s+Bill\s+\(HAWB\)\s+Number\s*:", 40)) self._set(result, "cour_be_hawb_dt", self._after_label(p2, r"Date\s+of\s+HAWB\s*:", 30)) self._set(result, "cour_be_marksnos", self._after_label(p2, r"Marks\s+and\s+Numbers\s*:", 60)) self._set(result, "cour_be_pkgs", self._after_label(p2, r"Number\s+of\s+Packages\s*:", 20)) self._set(result, "cour_be_gwt_kgs", self._after_label(p2, r"Gross\s+Weight\s*:", 20)) self._set(result, "cour_be_interest_amt", self._after_label(p2, r"Interest\s+Amount\s*:", 20)) self._set(result, "cour_be_invoicecount", self._after_label(p2, r"Number\s+of\s+Invoices\s*:", 20)) # Precise page-2 visual values. self._set(result, "cour_be_airline", self._visual_span(p2, 2, 146, 65, 200, 1)) self._set(result, "cour_be_flightno", self._visual_span(p2, 2, 146, 205, 310, 1)) self._set(result, "cour_be_arrivalairport", self._visual_span(p2, 2, 146, 335, 410, 1)) self._set(result, "cour_be_arrivaldate", self._visual_span(p2, 2, 146, 450, 550, 1)) self._set(result, "cour_be_igmno", self._visual_span(p2, 2, 175, 165, 250, 1)) self._set(result, "cour_be_inward_date", self._visual_span(p2, 2, 175, 415, 520, 1)) self._set(result, "cour_be_mawbno", self._visual_span(p2, 2, 197, 165, 300, 1)) self._set(result, "cour_be_mawb_dt", self._visual_span(p2, 2, 197, 415, 520, 1)) self._set(result, "cour_be_hawbno", self._visual_span(p2, 2, 219, 165, 300, 1)) self._set(result, "cour_be_hawb_dt", self._visual_span(p2, 2, 219, 415, 520, 1)) self._set(result, "cour_be_marksnos", self._visual_span(p2, 2, 241, 165, 300, 1)) self._set(result, "cour_be_pkgs", self._visual_span(p2, 2, 241, 415, 470, 1)) self._set(result, "cour_be_interest_amt", self._visual_span(p2, 2, 252, 415, 470, 1)) self._set(result, "cour_be_gwt_kgs", self._visual_span(p2, 2, 264, 415, 470, 1)) # Final visual repairs for dense two-column rows. These are applied # after generic label fallbacks so a neighbouring column can never # overwrite a correctly located value. self._set(result, "cour_be_couriername", self._visual_span(p1, 1, 198, 415, 520, 4)) self._set(result, "cour_be_iec", self._visual_span(p1, 1, 344, 165, 315, 4)) self._set(result, "cour_be_iecbranch", self._visual_span(p1, 1, 344, 415, 470, 4)) self._set(result, "cour_be_importername", self._visual_block(p1, 1, 366, 410, 165, 365)) self._set(result, "cour_be_importer_address", self._visual_block(p1, 1, 366, 410, 415, 570)) self._set(result, "cour_be_adcode", self._visual_span(p1, 1, 447, 165, 315, 4)) self._set(result, "cour_be_category", self._visual_span(p1, 1, 501, 165, 315, 4)) self._set(result, "cour_be_betype_description", self._visual_span(p1, 1, 501, 415, 500, 4)) self._set(result, "cour_be_supplier_name", self._clean_supplier_name( self._visual_block(p3, 3, 96, 110, 165, 365) )) # Page 6 payment/assessment. self._set(result, "cour_be_total_duty", self._after_label(p6, r"Total\s+Amount\s+", 30)) self._set(result, "cour_be_tr6_no", self._payment_cell(p6, 2)) self._set(result, "cour_be_pymt_amt", self._payment_cell(p6, 3)) self._set(result, "cour_be_challan_date", self._payment_cell(p6, 4)) # If the payment row is present, total duty is the total amount. if not result["cour_be_total_duty"]: result["cour_be_total_duty"] = result["cour_be_pymt_amt"] # ========================================================= # INVOICE # ========================================================= def _extract_invoice(self): p2 = self.page_text.get(2, "") p3 = self.page_text.get(3, "") p4 = self.page_text.get(4, "") invoice = {field: "" for field in self.INVOICE_FIELDS} invoice["doctype"] = "Courier Bill of Entry Invoices" invoice["cour_be_invsno"] = 1 inv_line = self._line_containing(p2, "Invoice Number") if inv_line: m = re.search( r"Invoice\s+Number\s*:\s*(.*?)\s+Date\s+of\s+Invoice\s*:\s*(.*)$", inv_line, re.I, ) if m: invoice["cour_be_invoice_no"] = self._clean(m.group(1)) invoice["cour_be_invoice_date"] = self._format_date(m.group(2)) supplier = self._after_label(p3, r"SUPPLIER\s+DETAILS\s*\n\s*Name\s*:", 120) invoice["cour_be_inv_supplier_name"] = self._clean_supplier_name( self._visual_span(p3, 3, 96, 165, 365, 5) ) nature = self._visual_span(p3, 3, 229, 160, 310, 1) payment = self._visual_span(p3, 3, 240, 160, 300, 1) valuation = self._visual_span(p3, 3, 285, 165, 250, 1) incoterm = self._visual_span(p3, 3, 285, 415, 470, 1) invoice["cour_be_inv_nature_of_txn"] = nature invoice["cour_be_inv_pymt_terms"] = payment invoice["cour_be_valuationmethod"] = valuation invoice["cour_be_incoterm"] = incoterm # Keep the generic fallbacks only if the visual extraction is empty. if not nature: nature = self._after_label(p3, r"Nature\s+of\s+Transaction\s*:", 30) if not payment: payment = self._after_label(p3, r"Terms\s+of\s+Payment\s*:", 30) if not valuation: valuation = self._after_label(p3, r"Method\s+of\s+Valuation\s*:", 30) if not incoterm: incoterm = self._after_label(p3, r"Terms\s+of\s+Invoice\s*:", 30) invoice["cour_be_inv_nature_of_txn"] = self._clean(nature) invoice["cour_be_inv_pymt_terms"] = self._clean(payment) invoice["cour_be_valuationmethod"] = self._clean(valuation) invoice["cour_be_incoterm"] = self._clean(incoterm) nature = invoice["cour_be_inv_nature_of_txn"] # Existing code below remains as fallback. nature = self._after_label(p3, r"Nature\s+of\s+Transaction\s*:", 30) payment = self._after_label(p3, r"Terms\s+of\s+Payment\s*:", 30) valuation = self._after_label(p3, r"Method\s+of\s+Valuation\s*:", 30) incoterm = self._after_label(p3, r"Terms\s+of\s+Invoice\s*:", 30) # Re-apply precise values after generic fallback expressions. invoice["cour_be_inv_nature_of_txn"] = self._visual_span(p3, 3, 229, 160, 310, 1) or invoice["cour_be_inv_nature_of_txn"] invoice["cour_be_inv_pymt_terms"] = self._visual_span(p3, 3, 240, 160, 300, 1) or invoice["cour_be_inv_pymt_terms"] invoice["cour_be_valuationmethod"] = self._visual_span(p3, 3, 285, 165, 250, 1) or invoice["cour_be_valuationmethod"] invoice["cour_be_incoterm"] = self._visual_span(p3, 3, 285, 415, 470, 1) or invoice["cour_be_incoterm"] value_line = self._line_containing(p3, "Invoice Value") if value_line: m = re.search( r"Invoice\s+Value\s*:\s*([0-9,]+(?:\.\d+)?)\s+Currency\s*:\s*([A-Z]{3})", value_line, re.I, ) if m: invoice["cour_be_inv_value"] = self._float(m.group(1)) invoice["cour_be_inv_currency"] = m.group(2).upper() supplier_addr = self._block_after_label( p3, r"SUPPLIER\s+DETAILS\s*\n\s*Name\s*:\s*[^\n]+\s+Address\s*:", stop_patterns=[r"IF SUPPLIER IS NOT THE SELLER", r"BROKER/ AGENT DETAILS", r"Nature of Transaction"], max_lines=8, ) invoice["suppliers_address"] = self._raw_supplier_address( p3, 3, ) if not invoice["suppliers_address"]: invoice["suppliers_address"] = supplier_addr svb_line = self._line_containing(p4, "SVB Reference Number") if svb_line: m = re.search( r"SVB\s+Reference\s+Number\s*:\s*(.*?)\s+SVB\s+Date\s*:\s*(.*)$", svb_line, re.I, ) if m: invoice["cour_svb_refno"] = self._clean(m.group(1)) invoice["cour_be_inv_svbdate"] = self._format_date(m.group(2)) provisional = self._after_label(p4, r"Indication\s+for\s+Provisional\s*/\s*Final\s*:", 30) invoice["cour_be_prov_final"] = provisional invoice["cour_be_inv_related"] = self._after_label(p4, r"Are\s+the\s+Buyer\s+and\s+Seller\s+Related\s*\?\s*", 10) # Final visual repairs for the invoice-level two-column layout. invoice["cour_be_inv_supplier_name"] = self._clean_supplier_name( self._visual_block(p3, 3, 96, 110, 165, 365) ) invoice["suppliers_address"] = self._raw_supplier_address( p3, 3, ) invoice["cour_be_inv_related"] = self._visual_span(p4, 4, 196.6, 165, 220, 3) invoice["cour_be_prov_final"] = self._visual_span(p4, 4, 267.25, 335, 430, 3) return invoice # ========================================================= # ITEM # ========================================================= def _extract_item(self): p4 = self.page_text.get(4, "") p5 = self.page_text.get(5, "") item = {field: "" for field in self.ITEM_FIELDS} item["doctype"] = "Courier Bill of Entry Items" item["linked_invoice_id_sno"] = 1 item["cour_be_item_itemsn"] = 1 desc = self._after_label(p4, r"Item\s+Description\s*:", 80) general_desc = self._after_label(p4, r"General\s+Description\s*:", 80) item["cour_be_item_itemdescription"] = desc item["cour_be_item_itemgendescription"] = general_desc price_line = self._line_containing(p4, "Currency for Unit Price") if price_line: m = re.search( r"Currency\s+for\s+Unit\s+Price\s*:\s*([A-Z]{3})\s+Unit\s+Price\s*:\s*([0-9,]+(?:\.\d+)?)", price_line, re.I, ) if m: item["cour_be_item_upi_currency"] = m.group(1).upper() item["cour_be_item_upi"] = self._float(m.group(2)) qty_line = self._line_containing(p4, "Unit of Measure") if qty_line: m = re.search( r"Unit\s+of\s+Measure\s*:\s*([A-Za-z]+)\s+Quantity\s*:\s*([0-9,]+(?:\.\d+)?)", qty_line, re.I, ) if m: item["cour_be_item_uom"] = m.group(1).upper() item["cour_be_item_qty"] = self._float(m.group(2)) item["cour_be_item_exchg_rate"] = self._float(self._visual_span(p4, 4, 718, 165, 300, 1) or self._after_label(p4, r"Rate\s+Of\s+Exchange\s*:", 20)) or 0.0 item["cour_be_item_manufacturer"] = self._clean_supplier_name( self._visual_block(p5, 5, 70, 90, 165, 365) ).rstrip("-") item["cour_be_item_countryoforigin"] = self._after_label(p5, r"Country\s+of\s+Origin\s*:", 50) class_line = self._line_containing(p5, "CTSH") if class_line: m = re.search(r"CTSH\s*:\s*(\d{6,10}).*?CETSH\s*:\s*(\d{6,10})", class_line, re.I) if m: item["cour_be_item_ctsh"] = m.group(1) item["cour_be_item_cetsh"] = m.group(2) ritc = self._visual_span(p5, 5, 206, 165, 300, 1) item["cour_be_item_ritc"] = "" if ritc in {"\"\"", "N/A"} else ritc item["cour_be_item_assessable_value"] = self._float(self._after_label( self.page_text.get(6, ""), r"Assessable\s+Value\s*:", 30 )) or 0.0 # Duty table: read the row by duty head. The table is textual, so # this remains layout-independent for the standard Courier form. duty_rows = self._duty_rows(p5) if duty_rows: bcd = duty_rows.get("BCD", {}) aidc = duty_rows.get("AIDC", {}) sws = duty_rows.get("SW Srchrg", {}) igst = duty_rows.get("IGST", {}) cmp = duty_rows.get("CMPNSTRY", {}) item["cour_be_item_bcd_rt"] = bcd.get("rate", 0.0) item["cour_be_item_bcd_amt"] = bcd.get("amount", 0.0) item["cour_be_item_aidc_rt"] = aidc.get("rate", 0.0) item["cour_be_item_aidc_amt"] = aidc.get("amount", 0.0) item["cour_be_item_sws_rt"] = sws.get("rate", 0.0) item["cour_be_item_sws_amt"] = sws.get("amount", 0.0) item["cour_be_item_igst_rt"] = igst.get("rate", 0.0) item["cour_be_item_igst_amt"] = igst.get("amount", 0.0) item["cour_be_item_cmpcess_rt"] = cmp.get("rate", 0.0) item["cour_be_item_cmpcess_amt"] = cmp.get("amount", 0.0) return item # ========================================================= # PAYMENT # ========================================================= def _extract_payment(self, result): p6 = self.page_text.get(6, "") lines = self._clean_lines(p6) for i, line in enumerate(lines): if line.upper().startswith("1") and "3011119292" in line: tokens = re.findall(r"\d+(?:\.\d+)?(?:/\d+/\d+)?", line) # Prefer the known table shape: srno, challan, amount, date. m = re.search(r"^1\s+(\d+)\s+([0-9,]+(?:\.\d+)?)\s+(\d{1,2}/\d{1,2}/\d{4})$", line) if m: self._set(result, "cour_be_tr6_no", m.group(1)) self._set(result, "cour_be_pymt_amt", self._float(m.group(2))) self._set(result, "cour_be_challan_date", self._format_date(m.group(3))) self._set(result, "cour_be_total_duty", self._float(m.group(2))) return # ========================================================= # OOC # ========================================================= def _extract_out_of_charge(self, result): p1 = self.page_text.get(1, "") m = re.search( r"OOC\s+ISSUED\s+on\s+(\d{1,2}-\d{1,2}-\d{4})", p1, re.I, ) if m: self._set(result, "out_of_charge_date", self._format_date(m.group(1))) # ========================================================= # DUTY TABLE # ========================================================= def _duty_rows(self, text): result = {} lines = self._clean_lines(text) started = False for line in lines: if line.upper().startswith("DUTY DETAILS"): started = True continue if not started: continue if line.upper().startswith("SHIPPING BILL DETAILS"): break m = re.match( r"^(\d+)\s+(BCD|AIDC|SW\s+Srchrg|IGST|CMPNSTRY)\s+" r"(-?[0-9]+(?:\.[0-9]+)?)\s+" r"(-?[0-9]+(?:\.[0-9]+)?)\s+" r"(-?[0-9]+(?:\.[0-9]+)?)\s+" r"(-?[0-9]+(?:\.[0-9]+)?)$", line, re.I, ) if not m: continue head = re.sub(r"\s+", " ", m.group(2).strip()) result[head] = { "rate": self._float(m.group(3)) or 0.0, "specific": self._float(m.group(4)) or 0.0, "forgone": self._float(m.group(5)) or 0.0, "amount": self._float(m.group(6)) or 0.0, } return result # ========================================================= # GENERIC VISUAL ROW / TABLE ENGINE # ========================================================= @staticmethod def _table_row_text(row): return re.sub( r"\s+", " ", " ".join(str(cell or "") for cell in row), ).strip() @staticmethod def _empty_table_row(row): return not any(str(cell or "").strip() for cell in row) def _group_words_into_rows(self, words, y_tolerance=3.5): """Group visual words into rows using their actual page coordinates.""" rows = [] for word in sorted( words or [], key=lambda w: ( int(w.get("_page", 0)), float(w.get("y0", w.get("top", 0))), float(w.get("x0", 0)), ), ): page = int(word.get("_page", 0)) y = float(word.get("y0", word.get("top", 0))) target = None for row in reversed(rows[-8:]): if row["_page"] != page: continue if abs(row["_y"] - y) <= y_tolerance: target = row break if target is None: target = {"_page": page, "_y": y, "words": []} rows.append(target) target["words"].append(word) result = [] for row in rows: row["words"].sort(key=lambda w: float(w.get("x0", 0))) row["text"] = self._clean( " ".join(str(w.get("_text", "")) for w in row["words"]) ) result.append(row) return result def _all_tables(self): """Return analyzer tables with page metadata, preserving source rows.""" result = [] for page_index, page in enumerate(self.pages, start=1): page_no = int(page.get("page_number", page_index)) for table_index, table in enumerate(page.get("tables", []) or [], start=1): if not table: continue result.append({ "page": page_no, "index": table_index, "rows": table, }) return result def _clean_table(self, table): result = [] for row in table or []: if not row: continue clean = [ "" if cell is None else re.sub( r"\s+", " ", str(cell).replace("\xa0", " ") ).strip() for cell in row ] if any(clean): result.append(clean) return result def _table_compact_text(self, rows, limit=10): return self._compact_header_token( " ".join(self._table_row_text(row) for row in rows[:limit]) ) def _classify_table(self, rows): """Classify Courier tables by header signatures, not page position.""" if not rows: return None compact = self._table_compact_text(rows) if not compact: return None signatures = { "invoices": { "required": ["invoicenumber", "dateofinvoice"], "optional": [ "invoicevalue", "currency", "suppliername", "natureoftransaction", "termsofpayment", "incoterm", ], }, "items": { "required": ["itemdescription", "quantity"], "optional": [ "upi", "unitprice", "uom", "ritc", "ctsh", "cetsh", "countryoforigin", "assessablevalue", "itemno", ], }, "duty": { "required": ["dutyhead"], "optional": [ "bcd", "aidc", "swsrchrg", "igst", "cmpnstry", "rate", "amount", "specific", "forgone", ], }, } scores = {} for name, spec in signatures.items(): req = sum(1 for token in spec["required"] if token in compact) opt = sum(1 for token in spec["optional"] if token in compact) if req >= 1 and opt >= 1: scores[name] = req * 20 + opt # A strong schema-header match wins even when a PDF table is split. schema_scores = {} for name in ("invoices", "items"): fields = self.COMPLETE_CHILD_FIELDS[name] hits = set() for row in rows[:10]: for cell in row: field = self._canonical_child_field_for_header(name, cell) if field: hits.add(field) if len(hits) >= 2: schema_scores[name] = len(hits) * 10 if schema_scores: best = max(schema_scores, key=schema_scores.get) if best in scores: return best if schema_scores[best] >= 30: return best if scores: return max(scores, key=scores.get) # Duty tables are often header-fragmented. Recognize them from # multiple duty-head tokens instead of requiring one exact header row. duty_hits = sum( 1 for token in ("bcd", "aidc", "swsrchrg", "igst", "cmpnstry") if token in compact ) if duty_hits >= 2: return "duty" return None def _canonical_table_header_index(self, table_name, rows): fields = self.COMPLETE_CHILD_FIELDS.get(table_name, []) if not fields: return 0 best_index = 0 best_score = 0 for index, row in enumerate(rows[:10]): used = set() score = 0 for cell in row: field = self._canonical_child_field_for_header(table_name, cell) if field and field not in used: used.add(field) score += 1 if score > best_score: best_index = index best_score = score return best_index if best_score >= 1 else 0 def _normalize_generic_child_value(self, field, value): value = self._clean(value) if not value: return "" if field in self.CHILD_FLOAT_FIELDS: return float(self._float(value) or 0.0) if field.endswith(("_date", "_dt")): return self._format_date(value) return value def _map_generic_table(self, table_name, rows): """Map a detected Courier invoice/item table to canonical fields.""" if table_name not in ("invoices", "items") or len(rows) < 2: return [] header_index = self._canonical_table_header_index(table_name, rows) headers = rows[header_index] field_by_column = [ self._canonical_child_field_for_header(table_name, header) for header in headers ] if not any(field_by_column): return [] records = [] doctype = self.TABLE_DOCTYPES[table_name] for row in rows[header_index + 1:]: if self._empty_table_row(row): continue row_text = self._table_row_text(row) if self._looks_like_section(row_text): continue record = {field: "" for field in self.COMPLETE_CHILD_FIELDS[table_name]} record["doctype"] = doctype mapped = 0 for index, cell in enumerate(row): if index >= len(field_by_column): continue field = field_by_column[index] value = self._clean(cell) if not field or not value: continue if record.get(field) not in ("", None): continue record[field] = self._normalize_generic_child_value(field, value) mapped += 1 if mapped: records.append(record) return records def _map_generic_duty_table(self, rows): """Map duty-head rows into partial Courier item records.""" records = [] if len(rows) < 2: return records for row in rows: text = self._clean(" ".join(str(x or "") for x in row)) compact = self._compact_header_token(text) head = None for candidate in ("BCD", "AIDC", "SW Srchrg", "IGST", "CMPNSTRY"): if self._compact_header_token(candidate) in compact: head = candidate break if not head: continue nums = re.findall(r"-?[0-9]+(?:,[0-9]{3})*(?:\.[0-9]+)?", text) if not nums: continue values = [self._float(x) for x in nums] if not values: continue # Duty tables conventionally expose rate/specific/forgone/amount. # If fewer columns are available, keep the values positionally. rate = values[0] if len(values) >= 1 else 0.0 amount = values[-1] if len(values) >= 2 else values[0] record = {field: "" for field in self.ITEM_FIELDS} record["doctype"] = self.TABLE_DOCTYPES["items"] mapping = { "BCD": ("cour_be_item_bcd_rt", "cour_be_item_bcd_amt"), "AIDC": ("cour_be_item_aidc_rt", "cour_be_item_aidc_amt"), "SW Srchrg": ("cour_be_item_sws_rt", "cour_be_item_sws_amt"), "IGST": ("cour_be_item_igst_rt", "cour_be_item_igst_amt"), "CMPNSTRY": ("cour_be_item_cmpcess_rt", "cour_be_item_cmpcess_amt"), } rate_field, amount_field = mapping[head] record[rate_field] = float(rate or 0.0) record[amount_field] = float(amount or 0.0) records.append(record) return records def _merge_child_rows(self, base_rows, fragments, table_name): """Merge fragmented table records without creating duplicate rows.""" if not fragments: return base_rows if not base_rows: return fragments merged = [dict(row) for row in base_rows] key_fields = ( ("cour_be_invoice_no", "cour_be_invsno") if table_name == "invoices" else ("cour_be_item_itemsn", "cour_be_item_ctsh", "cour_be_item_ritc") ) def key(row): values = [] for field in key_fields: value = self._clean(row.get(field, "")) if value: values.append(self._compact_header_token(value)) return tuple(values) for fragment in fragments: fk = key(fragment) target = None if fk: for row in merged: if key(row) == fk: target = row break if target is None and len(merged) == 1: target = merged[0] if target is None: merged.append(dict(fragment)) continue for field, value in fragment.items(): if field == "doctype" or value in ("", None): continue if self._is_effectively_empty(target.get(field)): target[field] = value return merged def _recover_from_generic_tables(self, result): """Inspect every analyzer table and recover missing child fields.""" invoice_fragments = [] item_fragments = [] duty_fragments = [] for table_info in self.tables: rows = self._clean_table(table_info.get("rows", [])) if len(rows) < 2: continue table_name = self._classify_table(rows) if not table_name: continue if table_name in ("invoices", "items"): records = self._map_generic_table(table_name, rows) if table_name == "invoices": invoice_fragments.extend(records) else: item_fragments.extend(records) elif table_name == "duty": duty_fragments.extend(self._map_generic_duty_table(rows)) if invoice_fragments: result["invoices"] = self._merge_child_rows( result.get("invoices", []), invoice_fragments, "invoices" ) if item_fragments: result["items"] = self._merge_child_rows( result.get("items", []), item_fragments, "items" ) # Duty fragments are intentionally fill-only and preferentially # merge into an existing item. They never replace authoritative values. if duty_fragments: if not result.get("items"): result["items"] = [duty_fragments[0]] target = result["items"][0] for fragment in duty_fragments: for field, value in fragment.items(): if field == "doctype" or value in ("", None): continue if self._is_effectively_empty(target.get(field)): target[field] = value def _extract_block_after_label( self, label, validator=None, max_rows=4, max_y_gap=120, ): """Collect a bounded visual block below a label until another label.""" page = label.get("_page") label_center = ( float(label.get("x0", 0)) + float(label.get("x1", label.get("x0", 0))) ) / 2.0 start_y = float(label.get("y1", label.get("top", label.get("y0", 0)))) candidate_rows = [ row for row in self.rows if row["_page"] == page and row["_y"] >= start_y and row["_y"] - start_y <= max_y_gap ] candidate_rows.sort(key=lambda row: row["_y"]) collected = [] for row in candidate_rows: text = self._clean(row.get("text", "")) if not text: continue if self._looks_like_any_known_label(text): if collected: break continue selected = [] for word in row.get("words", []): center = ( float(word.get("x0", 0)) + float(word.get("x1", word.get("x0", 0))) ) / 2.0 if abs(center - label_center) > 210: continue value = self._clean(word.get("_text", "")) if not value or self._looks_like_any_known_label(value): continue if validator and not validator(value): continue selected.append(word) if not selected: continue selected.sort(key=lambda word: float(word.get("x0", 0))) line = self._clean(" ".join(w.get("_text", "") for w in selected)) if line: collected.append(line) if len(collected) >= max_rows: break return self._clean(" ".join(collected)) if collected else None def _collapse_courier_item_continuations(self, records): """Collapse wrapped/duplicate Courier item fragments by logical identity.""" if not records: return [] collapsed = [] for raw in records: record = dict(raw) item_no = self._compact_header_token(record.get("cour_be_item_itemsn", "")) cth = self._compact_header_token(record.get("cour_be_item_ctsh", "")) ritc = self._compact_header_token(record.get("cour_be_item_ritc", "")) desc = self._compact_header_token(record.get("cour_be_item_itemdescription", "")) target = None for existing in collapsed: e_item = self._compact_header_token(existing.get("cour_be_item_itemsn", "")) e_cth = self._compact_header_token(existing.get("cour_be_item_ctsh", "")) e_ritc = self._compact_header_token(existing.get("cour_be_item_ritc", "")) e_desc = self._compact_header_token(existing.get("cour_be_item_itemdescription", "")) same_identity = ( (item_no and e_item and item_no == e_item) or (cth and e_cth and cth == e_cth) or (ritc and e_ritc and ritc == e_ritc) ) description_fragment = ( desc and e_desc and (desc in e_desc or e_desc in desc) ) if same_identity or description_fragment: target = existing break if target is None: collapsed.append(record) continue for field, value in record.items(): if field == "doctype" or value in ("", None): continue current = target.get(field) if self._is_effectively_empty(current): target[field] = value elif field == "cour_be_item_itemdescription": current_text = self._clean(current) new_text = self._clean(value) if new_text and new_text not in current_text: target[field] = (current_text + " " + new_text).strip() return collapsed # ========================================================= # VISUAL / TEXT HELPERS # ========================================================= def _all_words(self): words = [] for page_no, page in enumerate(self.pages, 1): raw = page.get("words") or [] for word in raw: text = str(word.get("text") or word.get("_text") or "").strip() if not text: continue item = dict(word) item["_text"] = text item["_page"] = int(page.get("page_number", page_no)) words.append(item) return words @staticmethod def _clean_lines(text): return [re.sub(r"\s+", " ", x).strip() for x in str(text or "").splitlines() if x.strip()] def _visual_words(self, page_no, y, x_min, x_max, y_tol=4): words = [] for w in self.page_words.get(page_no, []): wy = float(w.get("top", w.get("y0", 0))) if abs(wy - y) <= y_tol and float(w.get("x0", 0)) >= x_min and float(w.get("x0", 0)) <= x_max: words.append(w) words.sort(key=lambda w: float(w.get("x0", 0))) return words def _visual_span(self, text, page_no, y, x_min, x_max, y_tol=4): words = self._visual_words(page_no, y, x_min, x_max, y_tol) if not words: return "" return self._clean(" ".join(str(w.get("_text", "")) for w in words)) def _visual_right(self, text, page_no, y, label_x0, label_x1, value_x0, value_x1, y_tol=4): return self._visual_span(text, page_no, y, value_x0, value_x1, y_tol) def _visual_block(self, text, page_no, y_min, y_max, x_min, x_max=580): words = [] for w in self.page_words.get(page_no, []): y = float(w.get("top", w.get("y0", 0))) x = float(w.get("x0", 0)) if y_min <= y <= y_max and x_min <= x <= x_max: words.append(w) words.sort(key=lambda w: (float(w.get("top", w.get("y0", 0))), float(w.get("x0", 0)))) if not words: return "" lines = [] current_y = None current = [] for w in words: y = float(w.get("top", w.get("y0", 0))) if current_y is None or abs(y - current_y) <= 4: current.append(w) if current_y is None: current_y = y else: lines.append(" ".join(str(x.get("_text", "")) for x in sorted(current, key=lambda z: float(z.get("x0", 0))))) current = [w] current_y = y if current: lines.append(" ".join(str(x.get("_text", "")) for x in sorted(current, key=lambda z: float(z.get("x0", 0))))) return self._clean(" ".join(lines)) def _line_containing(self, text, needle): for line in self._clean_lines(text): if needle.lower() in line.lower(): return line return "" def _after_label(self, text, pattern, max_chars=120): if not text: return "" m = re.search(pattern + r"\s*(.*?)($|\n)", text, re.I | re.S) if not m: # Flatten whitespace for labels broken by PDF line wrapping. flat = re.sub(r"\s+", " ", text) m = re.search(pattern + r"\s*(.*?)(?=\s{2,}[A-Z][A-Za-z /()'-]+\s*:|$)", flat, re.I | re.S) if not m: return "" value = self._clean(m.group(1))[:max_chars] return self._strip_known_labels(value) def _between(self, text, start_pattern, end_pattern, section=None): source = text if section: sm = re.search(section, text, re.I | re.S) if sm: source = text[sm.start():] m = re.search(start_pattern + r"\s*(.*?)\s+" + end_pattern, source, re.I | re.S) if not m: return "" return self._clean(m.group(1)) def _block_after_label(self, text, label_pattern, stop_patterns, max_lines=6): lines = self._clean_lines(text) for i, line in enumerate(lines): if re.search(label_pattern, line, re.I): collected = [] same = re.sub(label_pattern, "", line, flags=re.I).strip(" :") if same: collected.append(same) for nxt in lines[i + 1:i + 1 + max_lines]: if any(re.search(stop, nxt, re.I) for stop in stop_patterns): break if self._looks_like_section(nxt): break collected.append(nxt) return self._clean(" ".join(collected)) return "" @staticmethod def _looks_like_section(line): up = line.upper().strip() if len(up) < 5: return False sections = ( "PARTICULARS OF ", "DETAILS OF ", "SUPPLIER DETAILS", "BROKER/ AGENT DETAILS", "IGM DETAILS", "IMPORT GENERAL MANIFEST DETAILS", "BOND DETAILS", "SPECIAL REQUESTS", "PAYMENT DETAILS", "DECLARATION", "DUTY DETAILS", ) return up.startswith(sections) @staticmethod def _strip_known_labels(value): value = re.sub(r"^(Address|Name|Number|Date)\s*:\s*", "", value, flags=re.I) return value.strip(" :") @staticmethod def _clean_supplier_name(value): """ Normalize supplier/manufacturer names where PDF line wrapping has split a word with a trailing hyphen. Example: AUTO- MATION -> AUTOMATION """ value = str(value or "") value = value.replace("\u00ad", "") value = value.replace("￾", "") # Repair a word split by a PDF line-wrap hyphen. value = re.sub( r"(?<=[A-Za-z])-\s+(?=[A-Za-z])", "", value, ) # Some Courier PDFs emit the supplier name as one PDF token: # ENDRESSHAUSERAUTO-. Reconstruct the visual word boundaries # without changing the supplier's actual value. value = re.sub( r"(?i)ENDRESSHAUSER", "ENDRESS HAUSER", value, ) value = re.sub( r"(?i)HAUSER(?=AUTO)", "HAUSER ", value, ) value = re.sub(r"\s+", " ", value) return value.strip(" \t\r\n:;") def _raw_supplier_address(self, text, page_no): """ Preserve the Courier Bill supplier-address representation required by the JSON contract. This is intentionally NOT a deduplication/cleaning routine. Repeated PDF text-stream fragments and punctuation are retained. Only broken PDF word boundaries are repaired where the PDF has visibly split a token (for example ``UNITE D STATES`` -> ``UNITED STATES``). """ words = [] for w in self.page_words.get(page_no, []): y = float(w.get("top", w.get("y0", 0))) x = float(w.get("x0", 0)) if 96 <= y <= 155 and 415 <= x <= 580: token = str(w.get("_text") or w.get("text") or "").strip() if token: words.append(w) words.sort( key=lambda w: ( float(w.get("top", w.get("y0", 0))), float(w.get("x0", 0)), ) ) if not words: return "" value = " ".join( str(w.get("_text") or w.get("text") or "").strip() for w in words if str(w.get("_text") or w.get("text") or "").strip() ) # Repair only PDF tokenization artifacts. Do NOT deduplicate the # repeated Greenwood / UNITED STATES fragments. repairs = ( (r"(?i)\b(\d+)ENDRESSPLSTE\b", r"\1 ENDRESS PL STE"), (r"(?i)\bDOCKG2IN(\d+)\b", r"DOCK G2 IN \1"), (r"(?i)\bGreenwood,?-\s*IN(\d+)\b", r"Greenwood,- IN \1"), (r"(?i)\bUNITE\s+D\s+STATES\b", "UNITED STATES"), (r"(?i)\bUNITE\s+DSTATES\b", "UNITED STATES"), (r"(?i)\bDSTATES,IN(\d+)\b", r"UNITED STATES,IN \1"), (r"(?i)\bIN(\d{4,})\b", r"IN \1"), ) for pattern, replacement in repairs: value = re.sub(pattern, replacement, value) # Preserve the exact comma artifacts required by the Courier JSON # contract. In particular, do not collapse `, ,` or duplicate text. value = re.sub(r"\s+,", ",", value) value = re.sub(r",\s*", ",", value) value = re.sub(r"\s+", " ", value).strip() # Restore the contract's required spacing around the intentionally # empty comma token and the hyphenated Greenwood fragment. value = value.replace("Greenwood,-IN", "Greenwood,- IN") value = value.replace("IN 46143,,UNITED STATES", "IN 46143, , UNITED STATES") # The source PDF may produce `UNITE D STATES` after coordinate # concatenation. Make the correction one final time after whitespace # normalization. value = re.sub(r"(?i)UNITE\s*D\s*STATES", "UNITED STATES", value) return value @staticmethod def _clean(value): value = str(value or "") value = value.replace("\u00ad", "") value = value.replace("￾", "") value = re.sub(r"\s+", " ", value) return value.strip(" \t\r\n:;") @staticmethod def _float(value): if value is None: return None value = str(value).replace(",", "").strip() m = re.search(r"-?\d+(?:\.\d+)?", value) return float(m.group(0)) if m else None @staticmethod def _format_date(value): value = str(value or "").strip() for fmt in ( "%d/%m/%Y", "%d-%m-%Y", "%d/%m/%y", "%d-%m-%y", "%Y-%m-%d", ): try: return datetime.strptime(value, fmt).strftime("%Y-%m-%d") except ValueError: pass return value @staticmethod def _strip_page_number(value): return re.sub(r"\s*Page\s+\d+\s+of\s+\d+\s*$", "", value, flags=re.I).strip() def _payment_cell(self, text, position): lines = self._clean_lines(text) for line in lines: m = re.match(r"^1\s+(\d+)\s+([0-9,]+(?:\.\d+)?)\s+(\d{1,2}/\d{1,2}/\d{4})$", line) if m: values = [m.group(1), m.group(2), m.group(3)] return values[position - 2] if 2 <= position <= 4 else "" return "" def _set(self, result, field, value): if field not in result: return if value is None: return value = self._clean(value) if not value: return if field.endswith(("_date", "_dt")) or field in { "be_date", "out_of_charge_date", "cour_be_arrivaldate", "cour_be_inward_date", "cour_be_mawb_dt", "cour_be_hawb_dt", "cour_be_challan_date", }: value = self._format_date(value) elif field in { "cour_be_pkgs", "cour_be_invoicecount", }: value = int(self._float(value) or 0) elif field in { "cour_be_gwt_kgs", "cour_be_total_duty", "cour_be_interest_amt", "cour_be_pymt_amt", }: value = float(self._float(value) or 0.0) result[field] = value self._trace(field, value) def _finalize(self, result): # Normalize required numeric values. for field in ( "cour_be_pkgs", "cour_be_invoicecount", ): if result[field] in ("", None): result[field] = 0 # Frappe Decimal/Float fields must be serialized as Python floats. # Do this in the final normalization pass so values extracted from # any branch of the parser cannot remain integers. parent_float_fields = ( "cour_be_gwt_kgs", "cour_be_total_duty", "cour_be_interest_amt", "cour_be_pymt_amt", ) for field in parent_float_fields: value = result.get(field) if value in ("", None): result[field] = 0.0 else: result[field] = float(self._float(value) or 0.0) if not result["cour_be_invoicecount"] and result["invoices"]: result["cour_be_invoicecount"] = len(result["invoices"]) # Absolute final child-schema contract pass. self._schema_complete_child_rows(result) # --------------------------------------------------------- # ADDRESS CONTRACT PRESERVATION # --------------------------------------------------------- # suppliers_address is intentionally NOT passed through any # deduplication/pretty-printing cleanup here. The PDF-derived # source string is part of the required JSON contract, including # repeated fragments, commas and spacing artifacts. # # Only the existing parser-level token normalization is allowed. # No address reconstruction is performed in this final pass. # ========================================================= # ABSOLUTE FINAL NUMERIC CONTRACT PASS # ========================================================= # Must remain the final mutation performed by _finalize(). self._enforce_final_numeric_schema(result) # ========================================================= # DYNAMIC FIELD / LABEL RECOVERY # ========================================================= @staticmethod def _compact_header_token(value): """ Normalize a PDF label for semantic comparison. Numbering, punctuation, spaces and common separators are ignored so variants such as: "1. Invoice No." "INVOICE NUMBER" "Invoice-No" can be compared safely. """ value = str(value or "") value = re.sub(r"^\s*\d+\s*[.)-]\s*", "", value) value = value.lower() value = re.sub(r"[^a-z0-9]+", "", value) return value def _child_field_aliases(self, table_name, field): aliases = list( self.CHILD_FIELD_ALIASES .get(table_name, {}) .get(field, []) ) aliases.append(field) # Technical Frappe prefix removed: # cour_be_item_qty -> item qty suffix = re.sub( r"^cour_be_(?:item_|inv_)?", "", field, flags=re.I, ) aliases.append(suffix.replace("_", " ")) # Deterministic common abbreviations. for alias in list(aliases): text = str(alias) aliases.extend([ text.replace(" NO", " NUMBER"), text.replace(" NUMBER", " NO"), text.replace(" DT", " DATE"), text.replace(" DATE", " DT"), text.replace(" QTY", " QUANTITY"), text.replace(" QUANTITY", " QTY"), text.replace(" AMT", " AMOUNT"), text.replace(" AMOUNT", " AMT"), text.replace(" RT", " RATE"), text.replace(" RATE", " RT"), ]) seen = set() result = [] for alias in aliases: key = self._compact_header_token(alias) if key and key not in seen: seen.add(key) result.append(alias) return result def _canonical_child_field_for_header(self, table_name, header): """ Map a changing PDF table/header label to an existing Frappe field. The algorithm prefers exact normalized matches, then containment, then strong token overlap. It never creates a new output field. """ fields = self.COMPLETE_CHILD_FIELDS.get(table_name, []) header_compact = self._compact_header_token(header) if not header_compact: return None header_tokens = set( re.findall(r"[a-z0-9]+", str(header or "").lower()) ) best = None for field in fields: for alias in self._child_field_aliases(table_name, field): alias_compact = self._compact_header_token(alias) if not alias_compact: continue score = 0 if header_compact == alias_compact: score = 1000 + len(alias_compact) elif len(alias_compact) >= 5 and alias_compact in header_compact: score = 700 + len(alias_compact) elif len(header_compact) >= 5 and header_compact in alias_compact: score = 600 + len(header_compact) else: alias_tokens = set( re.findall(r"[a-z0-9]+", str(alias).lower()) ) overlap = len(header_tokens & alias_tokens) # Require at least two meaningful tokens for fuzzy # matches. This avoids generic matches such as "TYPE". if overlap >= 2: score = ( 100 + overlap * 20 + min(len(alias_compact), 40) ) if score and (best is None or score > best[0]): best = (score, field) return best[1] if best else None def _find_label(self, aliases): """ Find the strongest visual label anchor across all PDF pages. """ if isinstance(aliases, str): aliases = [aliases] alias_keys = [ self._compact_header_token(alias) for alias in aliases if self._compact_header_token(alias) ] if not alias_keys: return None best = None for word in self.words: text = str(word.get("_text", "")).strip() key = self._compact_header_token(text) candidates = [(word, key)] if key else [] # Build short visual phrases from nearby words. This is important # because PDF analyzers often split labels into multiple words. page = word.get("_page") y = float(word.get("top", word.get("y0", 0))) x = float(word.get("x0", 0)) nearby = [ w for w in self.page_words.get(page, []) if abs(float(w.get("top", w.get("y0", 0))) - y) <= 3 and float(w.get("x0", 0)) >= x - 2 and float(w.get("x0", 0)) <= x + 260 ] nearby.sort(key=lambda w: float(w.get("x0", 0))) for length in range(2, min(8, len(nearby)) + 1): phrase_words = nearby[:length] phrase = " ".join( str(w.get("_text", "")).strip() for w in phrase_words ) phrase_key = self._compact_header_token(phrase) if phrase_key: candidates.append((phrase_words[0], phrase_key)) for anchor, candidate_key in candidates: for alias_key in alias_keys: score = 0 if candidate_key == alias_key: score = 1000 elif ( len(alias_key) >= 5 and alias_key in candidate_key ): score = 800 elif ( len(candidate_key) >= 5 and candidate_key in alias_key ): score = 700 if score: candidate = ( score, -int(anchor.get("_page", 0)), -float(anchor.get("top", anchor.get("y0", 0))), -float(anchor.get("x0", 0)), anchor, ) if best is None or candidate > best: best = candidate return best[-1] if best else None def _value_right_same_row( self, label, validator=None, max_distance=240, y_tolerance=5, ): page = label.get("_page") lx = float(label.get("x1", label.get("x0", 0))) ly = float(label.get("top", label.get("y0", 0))) candidates = [] for word in self.page_words.get(page, []): wx = float(word.get("x0", 0)) wy = float(word.get("top", word.get("y0", 0))) if wx < lx - 2: continue if abs(wy - ly) > y_tolerance: continue value = self._clean(word.get("_text", "")) if not value: continue if validator and not validator(value): continue distance = max(0.0, wx - lx) if distance > max_distance: continue candidates.append((distance, value, word)) if not candidates: return None candidates.sort(key=lambda x: x[0]) return candidates[0][1] def _value_below_column( self, label, validator=None, max_y_gap=110, x_tolerance=140, ): page = label.get("_page") lx = float(label.get("x0", 0)) ly = float(label.get("top", label.get("y0", 0))) candidates = [] for word in self.page_words.get(page, []): wx = float(word.get("x0", 0)) wy = float(word.get("top", word.get("y0", 0))) if wy <= ly: continue if wy - ly > max_y_gap: continue if abs(wx - lx) > x_tolerance: continue value = self._clean(word.get("_text", "")) if not value: continue if validator and not validator(value): continue candidates.append( ( wy - ly, abs(wx - lx), value, ) ) if not candidates: return None candidates.sort(key=lambda x: (x[0], x[1])) return candidates[0][2] def _parent_field_validator(self, field): if field in self.PARENT_FLOAT_FIELDS: return lambda value: self._float(value) is not None if field in self.PARENT_INTEGER_FIELDS: return lambda value: ( self._float(value) is not None and float(self._float(value)).is_integer() ) if field.endswith(("_date", "_dt")) or field in { "be_date", "out_of_charge_date", "cour_be_arrivaldate", "cour_be_inward_date", "cour_be_mawb_dt", "cour_be_hawb_dt", "cour_be_challan_date", }: return lambda value: self._valid_date(value) if field == "cour_be_iec": return lambda value: bool( re.fullmatch(r"[A-Z0-9]{8,20}", self._clean(value).upper()) ) if field == "cour_be_iecbranch": return lambda value: bool( re.fullmatch(r"[A-Z0-9]{1,12}", self._clean(value).upper()) ) if field == "cour_be_adcode": return lambda value: bool( re.fullmatch(r"\d{5,12}", re.sub(r"\D", "", self._clean(value))) ) return lambda value: ( bool(self._clean(value)) and not self._looks_like_label_value(value) and not self._looks_like_any_known_label(value) and not self._is_effectively_empty(value) ) @staticmethod def _looks_like_label_value(value): text = str(value or "").strip() return bool( re.fullmatch( r"(?:Name|Address|Number|Date|Amount|Value|Currency|" r"Quantity|Description|Code|Type)\s*: ?", text, re.I, ) ) def _looks_like_any_known_label(self, value): key = self._compact_header_token(value) if not key: return False for aliases in self.PARENT_FIELD_ALIASES.values(): for alias in aliases: if key == self._compact_header_token(alias): return True for table_aliases in self.CHILD_FIELD_ALIASES.values(): for aliases in table_aliases.values(): for alias in aliases: if key == self._compact_header_token(alias): return True return False @staticmethod def _valid_date(value): value = str(value or "").strip() return bool( re.fullmatch( r"\d{1,2}[/-]\d{1,2}[/-]\d{2,4}", value, ) or re.fullmatch( r"\d{4}-\d{1,2}-\d{1,2}", value, ) ) def _normalize_dynamic_parent_value(self, field, value): value = self._clean(value) if not value: return "" if field in self.PARENT_FLOAT_FIELDS: return float(self._float(value) or 0.0) if field in self.PARENT_INTEGER_FIELDS: return int(self._float(value) or 0) if field.endswith(("_date", "_dt")) or field in { "be_date", "out_of_charge_date", "cour_be_arrivaldate", "cour_be_inward_date", "cour_be_mawb_dt", "cour_be_hawb_dt", "cour_be_challan_date", }: return self._format_date(value) return value def _dynamic_parent_schema_recovery(self, result): """ Recover only fields still empty after the established parser. Strategy: 1. Find the field's semantic label anywhere in the visual model. 2. Prefer the nearest value on the same visual row. 3. Fall back to the nearest value below the label's column. 4. Validate/normalize against the target Frappe field type. Existing values are never overwritten. """ for field in self.PARENT_FIELDS: if field == "doctype": continue current = result.get(field) if current not in ("", None, 0, 0.0): continue aliases = self.PARENT_FIELD_ALIASES.get(field, []) label = self._find_label(aliases) if not label: continue validator = self._parent_field_validator(field) value = self._value_right_same_row( label, validator=validator, max_distance=260, ) if value is None: value = self._value_below_column( label, validator=validator, max_y_gap=110, x_tolerance=140, ) # Third recovery pass: collect a bounded visual block using the # label's actual column and stop at the next semantic label. # This is resilient to vertical shifts and wrapped values. if value is None and field not in self.PARENT_INTEGER_FIELDS: value = self._extract_block_after_label( label, validator=validator, max_rows=4, max_y_gap=120, ) if value in (None, ""): continue value = self._normalize_dynamic_parent_value( field, value, ) if value in ("", None): continue result[field] = value self._trace( field, value, label.get("_page"), ) @staticmethod def _is_effectively_empty(value): if value in (None, ""): return True text = str(value).strip() return text in {"", "\"\"", "N/A", "NA", "-"} def _child_value_validator(self, field): if field in self.CHILD_FLOAT_FIELDS: return lambda value: self._float(value) is not None if field.endswith(("_date", "_dt")): return self._valid_date return lambda value: ( bool(self._clean(value)) and not self._looks_like_label_value(value) and not self._looks_like_any_known_label(value) and not self._is_effectively_empty(value) ) def _normalize_dynamic_child_value(self, field, value): value = self._clean(value) if not value: return "" if field in self.CHILD_FLOAT_FIELDS: return float(self._float(value) or 0.0) if field.endswith(("_date", "_dt")): return self._format_date(value) return value def _recover_child_field(self, table_name, row, field): aliases = self._child_field_aliases(table_name, field) label = self._find_label(aliases) if not label: return validator = self._child_value_validator(field) value = self._value_right_same_row( label, validator=validator, max_distance=300, ) # RITC/CETSH are frequently printed as adjacent classification # labels. A below-column numeric fallback can therefore steal the # neighbouring CTSH value. For RITC, require a same-row association. if value is None and field != "cour_be_item_ritc": value = self._value_below_column( label, validator=validator, max_y_gap=120, x_tolerance=160, ) if value in (None, ""): return value = self._normalize_dynamic_child_value( field, value, ) if value not in ("", None): row[field] = value def _dynamic_child_schema_recovery(self, result): """ Fill missing invoice/item values using semantic PDF labels. This is deliberately fill-only. The existing Courier extraction remains authoritative for the current known layout, while this pass provides resilience when a future PDF moves a label or changes its wording. """ for row in result.get("invoices", []) or []: for field in self.INVOICE_FIELDS: if field == "doctype": continue if not self._is_effectively_empty(row.get(field)): continue self._recover_child_field( "invoices", row, field, ) for row in result.get("items", []) or []: for field in self.ITEM_FIELDS: if field == "doctype": continue if not self._is_effectively_empty(row.get(field)): continue self._recover_child_field( "items", row, field, ) def _enforce_final_numeric_schema(self, result): """ Last-mile numeric contract enforcement. This method intentionally runs after ALL extraction, table recovery, dynamic recovery, continuation collapsing, and child schema completion. Every field declared as a Float in the Courier schema is converted to a real Python float. This prevents source values such as: "22711" -> 22711.0 22711 -> 22711.0 "18" -> 18.0 1 -> 1.0 No int conversion is performed for Float fields. """ # ----------------------------- # Parent Float fields # ----------------------------- for field in self.PARENT_FLOAT_FIELDS: value = result.get(field) if value in (None, ""): result[field] = 0.0 continue parsed = self._float(value) result[field] = ( float(parsed) if parsed is not None else 0.0 ) # ----------------------------- # Invoice / Item Float fields # ----------------------------- # Keep the two child schemas strictly isolated. A combined float # field list must never be applied to both tables. child_float_fields = { "invoices": { field for field in self.CHILD_FLOAT_FIELDS if field in self.INVOICE_FIELDS }, "items": { field for field in self.CHILD_FLOAT_FIELDS if field in self.ITEM_FIELDS }, } for table_name in ("invoices", "items"): rows = result.get(table_name) or [] allowed_fields = set( self.COMPLETE_CHILD_FIELDS[table_name] ) float_fields = child_float_fields[table_name] for row in rows: # Defensive cleanup in case an earlier dynamic recovery pass # accidentally inserted a field from the other child schema. for field in list(row.keys()): if field not in allowed_fields: del row[field] # Enforce floats only for fields belonging to this table. for field in float_fields: value = row.get(field) if value in (None, ""): row[field] = 0.0 continue parsed = self._float(value) row[field] = ( float(parsed) if parsed is not None else 0.0 ) # ----------------------------- # Integer fields remain integers # ----------------------------- for field in self.PARENT_INTEGER_FIELDS: value = result.get(field) if value in (None, ""): result[field] = 0 continue try: result[field] = int(float(value)) except (TypeError, ValueError): result[field] = 0 def _schema_default_for_field(self, field): """ Return the correct empty value for a Courier Frappe field. """ if field in self.CHILD_FLOAT_FIELDS: return 0.0 return "" def _schema_complete_child_rows(self, result): """ Enforce a stable child-row contract. Every emitted invoice/item row contains every field declared by the Courier Frappe child schema. Missing values are represented by the correct type-safe empty default. """ for table_name, rows, doctype in ( ( "invoices", result.get("invoices", []) or [], "Courier Bill of Entry Invoices", ), ( "items", result.get("items", []) or [], "Courier Bill of Entry Items", ), ): fields = self.COMPLETE_CHILD_FIELDS[table_name] for index, row in enumerate(rows): normalized = {} for field in fields: value = row.get(field, "") if value in (None, ""): value = self._schema_default_for_field(field) if field in self.CHILD_FLOAT_FIELDS: try: value = float( self._float(value) or 0.0 ) except (TypeError, ValueError): value = 0.0 normalized[field] = value normalized["doctype"] = doctype result[table_name][index] = normalized def _trace(self, field, value, page=None): entry = {"field": field, "value": value} if page is not None: entry["page"] = page self.trace.append(entry) if __name__ == "__main__": print("CourierBillParser (pdfplumber, deterministic, non-AI) loaded successfully")