Project

General

Profile

Download (90.4 KB)

Feature #92 » courier_bill_parser_dynamic_v5.py

Main Code to convert PDF to Json - Rajesh Gotur, 08/20/2026 11:00 AM

 
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")
(2-2/2)