"""Shipping Bill PDF -> Frappe ``Shipping Bill`` JSON parser. Input is the ``pages`` list produced by ``PDFAnalyzer`` (text, words with coordinates, tables). Output is the Frappe document JSON. Design ------ ICEGATE Shipping Bills are fixed-layout forms. The pipeline is: 1. Page selection Only pages carrying the ICEGATE Shipping Bill header are kept, and only those belonging to the first Shipping Bill number found. Scanned/image pages, invoices, packing lists and courier waybills bundled in the same PDF are ignored. 2. Clean layout model (word geometry, not pdfplumber tables) The ICEGATE PDFs contain three kinds of text noise that corrupt table extraction, because pdfplumber assigns every glyph to whichever cell its centre falls in: * the diagonal "ASSESSED COPY" / "LET EXPORT COPY" watermark, which is real text at ~50-58pt (caused ``D`` in AD Code, ``O Maharashtra``, ``S DTY`` ...); * rotated side-bar captions ("B DECLARANT DETAILS", "INVOICE (1/1)", "OTHER ADDITIONAL INFORMATION") printed letter by letter; * label text overflowing into the value cell ("FINALDESTINATIO|N" -> ``NGERMANY``). Words are therefore filtered by font size relative to the page body text and by position (left of the form's content edge), grouped into visual rows, and every section is read from that clean model. 3. Section readers Grid sections (label row followed by a value row) assign each value word to the column whose cell it overlaps most. Key/value sections match the printed label on a character stream, so a label glued to or overlapping its value is split correctly. Repeating blocks (items, Single Window rows) are read with a sequential state machine across page breaks. 4. Normalisation Values are converted to the Frappe field contract (numbers, ISO dates, HH:MM:SS times) and every parent field / child row is schema-complete. No values from sample documents are hardcoded; everything is read from the input PDF. """ from __future__ import annotations import re from collections import Counter from collections.abc import Iterable from dataclasses import dataclass, field from datetime import datetime from typing import Any, ClassVar # --------------------------------------------------------------------------- # Regular expressions shared by the layout model and the readers # --------------------------------------------------------------------------- _DATE_PATTERNS = ( re.compile(r"\d{1,2}-[A-Za-z]{3}-\d{2,4}"), re.compile(r"\d{1,2}/\d{1,2}/\d{2,4}"), re.compile(r"\d{1,2}\.\d{1,2}\.\d{4}"), re.compile(r"\d{4}-\d{1,2}-\d{1,2}"), ) _TIME_RE = re.compile(r"(\d{1,2}):(\d{2})(?::(\d{2}))?") _NUMBER_RE = re.compile(r"-?\d+(?:\.\d+)?") # A printed label that starts with a field number: "12.PORT", "1INVSN", # "3.HS", "13CESAMT", "4.", "19.". _NUMBERED_LABEL_RE = re.compile(r"^\d{1,2}(?:\.|(?=[A-Za-z]))") # A whole word that is a numbered field label ("12.PORT", "1INVSN", # "8VALUE(F/C)") and not a code that merely starts with digits ("29AAACW1685J1ZW"). _FIELD_LABEL_WORD_RE = re.compile(r"^\d{1,2}(?:\.\s*)?[A-Za-z][A-Za-z.()/'&-]*$") _PURE_NUMBER_PREFIX_RE = re.compile(r"^\d{1,2}\.?$") # Two labels printed without a space: "NO.4.HAWB", "3.INSURANC4.DISCOU", # "3FREIGHT4.INSURANCE5DISCOUNT6.COMMISON", "14.DBKCLMD15.IGSTSTAT". _GLUED_LABEL_RE = re.compile(r"(?<=[A-Za-z.)])(?=\d{1,2}\.?[A-Za-z]{2,})") _PART_TITLES = { "partishippingbillsummary": "I", "partiiinvoicedetails": "II", "partiiiitemdetails": "III", "partivexportschemedetails": "IV", "partvdeclarations": "V", } def _compact(value: Any) -> str: return re.sub(r"[^a-z0-9]+", "", str(value or "").lower()) def _label_key(text: str) -> str: """Compact label text without its printed field number ("12.PORT OF" -> "portof").""" return re.sub(r"^\d{1,2}", "", _compact(text)) def _clean_text(value: Any) -> str: if value is None: return "" text = str(value).replace("\x00", "").replace("\r", " ").replace("\n", " ") return re.sub(r"\s+", " ", text).strip() def _is_date(value: Any) -> bool: text = str(value or "").strip() return any(p.fullmatch(text) for p in _DATE_PATTERNS) def _is_time(value: Any) -> bool: return bool(_TIME_RE.fullmatch(str(value or "").strip())) def _to_number(value: Any) -> float | None: if value is None: return None text = str(value).replace(",", "").strip() return float(text) if _NUMBER_RE.fullmatch(text) else None # --------------------------------------------------------------------------- # Layout model # --------------------------------------------------------------------------- @dataclass class _Word: text: str x0: float x1: float top: float bottom: float size: float = 0.0 @property def key(self) -> str: return _compact(self.text) @dataclass class _Row: page: int top: float words: list[_Word] = field(default_factory=list) @property def text(self) -> str: return " ".join(w.text for w in self.words) @property def key(self) -> str: return _compact(self.text) def between(self, x_min: float = float("-inf"), x_max: float = float("inf")) -> list[_Word]: return [w for w in self.words if x_min <= w.x0 < x_max] @dataclass class _Label: key: str text: str x0: float x1: float @dataclass class _Page: number: int width: float rows: list[_Row] part: str | None = None header_rows: list[_Row] = field(default_factory=list) body_rows: list[_Row] = field(default_factory=list) header: dict[str, str] = field(default_factory=dict) icegate_banner: bool = False class _LayoutBuilder: """Build clean visual rows from PDFAnalyzer page data.""" ROW_TOLERANCE = 2.5 MIN_SIZE_RATIO = 0.85 MAX_SIZE_RATIO = 1.6 SIDEBAR_MARGIN = 3.0 def __init__(self, known_label_keys: Iterable[str]): self.known_label_keys = sorted({k for k in known_label_keys if len(k) >= 3}, key=len, reverse=True) def build(self, raw_page: dict[str, Any], index: int) -> _Page: number = int(raw_page.get("page_number") or index + 1) words = self._words(raw_page) banner = "indiancustomsedisystem" in _compact(" ".join(w.text for w in words)) words = self._drop_noise(words) words = [part for w in words for part in self._split_glued(w)] rows = self._group_rows(words, number) page = _Page( number=number, width=float(raw_page.get("width") or 612.0), rows=rows, icegate_banner=banner, ) self._assign_bands(page) return page # -- word extraction ---------------------------------------------------- def _words(self, raw_page: dict[str, Any]) -> list[_Word]: words: list[_Word] = [] for raw in raw_page.get("words") or []: text = _clean_text(raw.get("text")) if not text: continue x0 = float(raw.get("x0") or 0.0) top = float(raw.get("top") or 0.0) words.append( _Word( text=text, x0=x0, x1=float(raw.get("x1") or x0), top=top, bottom=float(raw.get("bottom") or top), size=float(raw.get("size") or 0.0), ) ) if words: return words return self._words_from_layout_text(str(raw_page.get("text") or "")) @staticmethod def _words_from_layout_text(text: str) -> list[_Word]: """Fallback for analyzers that return only layout text. PDFAnalyzer renders layout text at x_density=7.25 and y_density=13, so character columns/lines map back to approximate PDF coordinates. """ words = [] for line_no, line in enumerate(text.splitlines()): for match in re.finditer(r"\S+", line): x0 = match.start() * 7.25 words.append( _Word( text=match.group(), x0=x0, x1=match.end() * 7.25, top=line_no * 13.0, bottom=line_no * 13.0 + 8.0, ) ) return words # -- noise removal ------------------------------------------------------ def _drop_noise(self, words: list[_Word]) -> list[_Word]: sized = [round(w.size, 1) for w in words if w.size > 0] if sized: body = Counter(sized).most_common(1)[0][0] low, high = body * self.MIN_SIZE_RATIO, body * self.MAX_SIZE_RATIO # Watermark glyphs are several times the body size; rotated side-bar # captions report a reduced size. Words without size are kept. words = [w for w in words if w.size <= 0 or low <= w.size <= high] # Side-bar captions that survive the size filter sit in the narrow # column left of the form's content edge. The content edge is where # the numbered field labels ("12.PORT", "1INVSN") start. label_x = [w.x0 for w in words if _FIELD_LABEL_WORD_RE.match(w.text)] if label_x: content_left = min(label_x) page_right = max(w.x1 for w in words) if content_left < page_right * 0.3: words = [w for w in words if w.x1 > content_left - self.SIDEBAR_MARGIN] return words def _split_glued(self, word: _Word) -> list[_Word]: """Split labels printed without a separating space ("NO.4.HAWB").""" cuts = [m.start() for m in _GLUED_LABEL_RE.finditer(word.text)] cuts = [c for c in cuts if self._starts_known_label(word.text[c:])] if not cuts: return [word] pieces, start = [], 0 for cut in cuts + [len(word.text)]: pieces.append((start, cut)) start = cut width = (word.x1 - word.x0) / max(len(word.text), 1) return [ _Word( text=word.text[a:b], x0=word.x0 + a * width, x1=word.x0 + b * width, top=word.top, bottom=word.bottom, size=word.size, ) for a, b in pieces if word.text[a:b] ] def _starts_known_label(self, text: str) -> bool: key = _label_key(text) return any(key.startswith(k) or (len(key) >= 4 and k.startswith(key)) for k in self.known_label_keys) # -- rows and page bands -------------------------------------------------- def _group_rows(self, words: list[_Word], page_no: int) -> list[_Row]: rows: list[_Row] = [] for word in sorted(words, key=lambda w: (w.top, w.x0)): if rows and abs(rows[-1].top - word.top) <= self.ROW_TOLERANCE: rows[-1].words.append(word) else: rows.append(_Row(page=page_no, top=word.top, words=[word])) for row in rows: row.words.sort(key=lambda w: w.x0) return rows @staticmethod def _assign_bands(page: _Page) -> None: title_index = next((i for i, r in enumerate(page.rows) if r.key in _PART_TITLES), None) if title_index is None: page.body_rows = list(page.rows) return page.part = _PART_TITLES[page.rows[title_index].key] page.header_rows = page.rows[:title_index] body = [] for row in page.rows[title_index + 1 :]: if row.key.startswith("glossary") or "scanqrcode" in row.key: break body.append(row) page.body_rows = body # --------------------------------------------------------------------------- # Parser # --------------------------------------------------------------------------- class ShippingBillParser: """Deterministic, layout-aware ICEGATE Shipping Bill parser.""" # ------------------------------------------------------------------ # Frappe DocType contract # ------------------------------------------------------------------ PARENT_FIELDS: ClassVar[list[str]] = [ "doctype", "port_code", "state_of_origin", "inv", "cb_code", "gwt", "shipping_bill_no", "port_of_loading", "item", "buyer_name", "gwt_unit", "shipping_bill_date", "cntry_of_finaldstn", "cntry_of_discharge", "cont", "pkg", "iec", "port_of_finaldstn", "port_of_discharge", "iec_branch_code", "p_1a_mode", "p_1a_meis", "p_1a_dfrc", "p_1a_assess", "p_1a_dbk", "p_1a_re_exp", "p_1a_exam", "p_1a_rodtp", "p_1a_lut", "p_1a_job", "p_1a_licence", "p_1b_exporter_name", "p_1b_exporter_address", "p_1b_cb_name", "p_1b_dbk_ac_no", "p_1b_consignee_name", "p_1b_consignee_address", "p_1b_aeo", "p_1b_ifsc_no", "p_1b_type", "p_1b_rbi_wr_no", "p_1b_gstin", "p_1b_ad_code", "p_1b_rbi_wr_dt", "p_1b_forex_ac_no", "p_1c_fob_val", "p_1c_com", "p_1c_cess", "p_1c_freight", "p_1c_deductions", "p_1c_insurance", "p_1c_p_c", "p_1c_discount", "p_1c_duty", "p_1d_dbk", "p_1d_rodtep_amt", "p_1d_igst_amt", "p_1d_rosctle_amt", "p_1d_cess_amt", "p_1d_igst_val", "p_1j_subm_dt", "p_1i_subm_time", "p_1i_leo_no", "p_1i_assess_dt", "p_1i_assess_time", "p_1i_leo_dt", "p_1i_exmn_dt", "p_1i_exmn_time", "p_1i_leo_time", "p_1i_brc_realzn_dt", "p_1i_sez_unit_data", ] CHILD_TABLES: ClassVar[list[str]] = [ "manifest_details", "challan_details", "annex_details", "invoice_details", "aa__dfia_licence_details", "item_details", "drawback_and_rosl_claim", "job_details", "single_window_declaration", "single_window_declaration_constituents", "single_window_declaration_control", "supporting_documents", "ar4_details", "third_party_details", "item_manufacturerproducergrower_details", "rodtep_details", "container_details", ] COMPLETE_CHILD_FIELDS: ClassVar[dict[str, list[str]]] = { "manifest_details": [ "p_1e_mawb_no", "p_1e_noc", "p_1e_mawb_dt", "p_1e_cin_no", "p_1e_hawb_no", "p_1e_cin_dt", "p_1e_hawb_dt", "p_1e_cin_site_id", ], "challan_details": ["p_4h_challan_serial_no", "p_4h_challan_no", "p_4h_paymt_dt", "p_4h_challan_amount"], "annex_details": [ "p_1i_seal_typ", "p_1i_loose_pkts", "p_1i_nature_cargo", "p_1i_marks_numbers", "p_1i_no_of_pkgs", "p_1i_no_of_containers", ], "invoice_details": [ "inv_sn", "p_2a_inv_no", "p_2a_inv_dt", "p_2a_ad_code", "p_2a_invterm", "p_2b_exporter_name", "p_2b_buyer_addr", "p_2b_exporter_addr", "p_2b_buyer_name", "p_2c_invoice_value", "p_2c_freight", "p_2c_discount", "p_2c_deduct", "p_2c_exchng_rate_desc", "p_2c_exchng_rate", "p_2c_invoice_curr", "p_2c_frieght_curr", "p_2c_discount_curr", "p_2c_deduct_curr", "p_2c_fob_val", "p_2c_insurance", "p_2c_commison", "p_2c_p_c", "p_2c_fob_curr", "p_2c_insurance_curr", "p_2c_commison_curr", "p_2c_p_c_curr", ], "aa__dfia_licence_details": [ "p_4b_invsn", "p_4b_itmsn", "p_4b_licence_no", "p_4b_licence_dt", "p_4b_licence_value", "p_4b_licence_qty", ], "item_details": [ "p_3a_invsno", "p_3a_itemsn", "p_3a_cth", "p_3a_item_desc", "p_3a_qty", "p_3a_uqc", "p_3a_rate", "p_3a_value", "p_3a_fob", "p_3a_pmv", "p_3a_duty_amt", "p_3a_cess_rate", "p_3a_cess_amt", "p_3a_dbk_claimed", "p_3a_igststat", "p_3a_igst_val", "p_3a_igst_amt", "p_3a_schcod", "p_3a_scheme_desc", "p_3a_sqc_mst", "p_3a_sqc_uqc", "p_3a_state_of_origin", "p_3a_district_of_origin", "p_3a_pt_abroad", "p_3a_comp_cess", "p_3a_end_use", "p_3a_benefit_availd", "p_3a_reward_benefit", "p_3a_third_party_item", ], "drawback_and_rosl_claim": [ "p_3b_invsno", "p_3b_itemsn", "p_3b_dbksn", "p_3b_dbk_rate", "p_3b_dbk_amt", "p_3b_rosl_rate", "p_3b_rosl_amt", ], "job_details": ["p_3c_invsn", "p_3c_itmsn", "p_3c_job_no", "p_3c_job_date", "p_3c_job_amount"], "single_window_declaration": [ "p_4d_invsn", "p_4d_itmsn", "p_4d_info", "p_4d_qualifier", "p_4d_info_cd", "p_4d_info_text", "p_4d_info_msr", "p_4d_uqc", ], "single_window_declaration_constituents": ["p_4e_invsn", "p_4e_itmsn", "p_4e_constituent", "p_4e_percentage"], "single_window_declaration_control": ["p_4f_invsn", "p_4f_itmsn", "p_4f_control", "p_4f_result"], "supporting_documents": [ "p_4g_invsn", "p_4g_itmsn", "p_4g_doc_code", "p_4g_doc_no", "p_4g_doc_date", "p_4g_remarks", ], "ar4_details": [ "p_4j_invsn", "p_4j_itmsno", "p_4j_ar4_number", "p_4j_ar4_date", "p_4j_commissionerate", "p_4j_division", "p_4j_range", ], "third_party_details": [ "p_4k_invsn", "p_4k_itmsn", "p_4k_iec", "p_4k_exporter_name", "p_4k_address", "p_4k_gstn_id_type", ], "item_manufacturerproducergrower_details": [ "p_4l_invsn", "p_4l_itmsn", "p_4l_type", "p_4l_manufact_cd", "p_4l_source_state", "p_4l_trans_cy", "p_4l_address", ], "rodtep_details": ["p_4m_invsn", "p_4m_itmsn", "p_4m_quantity", "p_4m_uqc", "p_4m_no_of_units", "p_4m_value"], "container_details": ["p_4i_container", "p_4i_cont_seal", "p_4i_cont_dt", "p_4i_cont_sno"], } CHILD_DOCTYPES: ClassVar[dict[str, str]] = { "manifest_details": "Manifest Detail", "challan_details": "Challan Detail", "annex_details": "Annex Detail", "invoice_details": "Invoice Detail", "aa__dfia_licence_details": "AA or DFIA Licence Detail", "item_details": "Item Detail", "drawback_and_rosl_claim": "Drawback And Rosl Claim", "job_details": "Job Detail", "single_window_declaration": "Single Window Declaration", "single_window_declaration_constituents": "Single Window Declaration Constituent", "single_window_declaration_control": "Single Window Declaration Control", "supporting_documents": "Supporting Document", "ar4_details": "AR4 Detail", "third_party_details": "Third Party Detail", "item_manufacturerproducergrower_details": "Item Manufacturer-Producer-Grower Detail", "rodtep_details": "Rodtep Detail", "container_details": "Container Detail", } STATUS_FIELDS: ClassVar[list[str]] = [ "p_1a_assess", "p_1a_meis", "p_1a_dbk", "p_1a_re_exp", "p_1a_exam", "p_1a_rodtp", "p_1a_lut", "p_1a_job", "p_1a_licence", "p_1a_dfrc", ] NUMBER_FIELDS: ClassVar[set[str]] = { "p_1c_fob_val", "p_1c_discount", "p_1d_dbk", "cont", "p_1d_rosctle_amt", "p_1d_cess_amt", "item", "pkg", "inv", "p_1c_freight", "p_1d_igst_val", "p_1c_com", "p_1c_deductions", "p_1c_p_c", "p_1d_rodtep_amt", "p_1c_duty", "p_1c_cess", "p_1c_insurance", "p_1d_igst_amt", "gwt", } DATE_FIELDS: ClassVar[set[str]] = { "p_1i_leo_dt", "p_1i_assess_dt", "p_1i_brc_realzn_dt", "p_1i_exmn_dt", "shipping_bill_date", "p_1j_subm_dt", } TIME_FIELDS: ClassVar[set[str]] = {"p_1i_subm_time", "p_1i_assess_time", "p_1i_exmn_time", "p_1i_leo_time"} CHILD_NUMERIC_DEFAULT_FIELDS: ClassVar[set[str]] = { "p_3a_qty", "p_3a_rate", "p_3a_value", "p_3a_fob", "p_3a_pmv", "p_3a_duty_amt", "p_3a_cess_rate", "p_3a_cess_amt", "p_3a_igst_val", "p_3a_igst_amt", "p_3a_comp_cess", "p_1i_loose_pkts", "p_1i_no_of_pkgs", "p_1i_no_of_containers", "p_2c_invoice_value", "p_2c_freight", "p_2c_discount", "p_2c_deduct", "p_2c_fob_val", "p_2c_insurance", "p_2c_commison", "p_2c_p_c", } # Identifiers/codes that must stay strings even when they look numeric. CHILD_STRING_FIELDS: ClassVar[set[str]] = { "p_1e_mawb_no", "p_1e_noc", "p_1e_hawb_no", "p_1e_cin_no", "p_1e_cin_site_id", "p_1i_seal_typ", "p_1i_nature_cargo", "p_1i_marks_numbers", "p_2a_inv_no", "p_2a_ad_code", "p_2a_invterm", "p_2b_exporter_name", "p_2b_buyer_addr", "p_2b_exporter_addr", "p_2b_buyer_name", "p_2c_exchng_rate_desc", "p_3a_cth", "p_3a_item_desc", "p_3a_uqc", "p_3a_dbk_claimed", "p_3a_igststat", "p_3a_schcod", "p_3a_scheme_desc", "p_3a_sqc_mst", "p_3a_sqc_uqc", "p_3a_state_of_origin", "p_3a_district_of_origin", "p_3a_pt_abroad", "p_3a_end_use", "p_3a_benefit_availd", "p_3a_reward_benefit", "p_3a_third_party_item", "p_3b_dbksn", "p_4b_licence_no", "p_3c_job_no", "p_4d_info", "p_4d_qualifier", "p_4d_info_cd", "p_4d_info_text", "p_4d_info_msr", "p_4d_uqc", "p_4e_constituent", "p_4f_control", "p_4f_result", "p_4g_doc_code", "p_4g_doc_no", "p_4g_remarks", "p_4h_challan_no", "p_4i_container", "p_4i_cont_seal", "p_4j_ar4_number", "p_4j_commissionerate", "p_4j_division", "p_4j_range", "p_4k_iec", "p_4k_exporter_name", "p_4k_address", "p_4k_gstn_id_type", "p_4l_type", "p_4l_manufact_cd", "p_4l_source_state", "p_4l_trans_cy", "p_4l_address", "p_4m_uqc", } # ------------------------------------------------------------------ # Printed-label specifications (label key -> field) # Keys are compact label texts without the printed field number. # ------------------------------------------------------------------ STATUS_GRID: ClassVar[dict[str, str]] = { "mode": "p_1a_mode", "assess": "p_1a_assess", "exmn": "p_1a_exam", "jobbing": "p_1a_job", "meis": "p_1a_meis", "dbk": "p_1a_dbk", "rodtp": "p_1a_rodtp", "licence": "p_1a_licence", "dfrc": "p_1a_dfrc", "reexp": "p_1a_re_exp", "lut": "p_1a_lut", } VALUATION_GRID: ClassVar[dict[str, str]] = { "fobvalue": "p_1c_fob_val", "freight": "p_1c_freight", "insuranc": "p_1c_insurance", "insurance": "p_1c_insurance", "discou": "p_1c_discount", "discount": "p_1c_discount", "com": "p_1c_com", "commission": "p_1c_com", "deductions": "p_1c_deductions", "pc": "p_1c_p_c", "duty": "p_1c_duty", "cess": "p_1c_cess", "dbkclaim": "p_1d_dbk", "igstamt": "p_1d_igst_amt", "cessamt": "p_1d_cess_amt", "igstvalue": "p_1d_igst_val", "rodtepamt": "p_1d_rodtep_amt", "rosctlamt": "p_1d_rosctle_amt", } MANIFEST_GRID: ClassVar[dict[str, str]] = { "mawbno": "p_1e_mawb_no", "mawbdt": "p_1e_mawb_dt", "hawbno": "p_1e_hawb_no", "hawbdt": "p_1e_hawb_dt", "noc": "p_1e_noc", } CIN_GRID: ClassVar[dict[str, str]] = {"cinno": "p_1e_cin_no", "cindt": "p_1e_cin_dt", "cinsiteid": "p_1e_cin_site_id"} INVOICE_SUMMARY_GRID: ClassVar[dict[str, str]] = { "sno": "inv_sn", "invno": "p_2a_inv_no", "invamt": "p_2c_invoice_value", "currenc": "p_2c_invoice_curr", } EQUIPMENT_GRID: ClassVar[dict[str, str]] = { "container": "p_4i_container", "seal": "p_4i_cont_seal", "date": "p_4i_cont_dt", "sno": "p_4i_cont_sno", } CHALLAN_GRID: ClassVar[dict[str, str]] = { "srno": "p_4h_challan_serial_no", "challanno": "p_4h_challan_no", "paymtdt": "p_4h_paymt_dt", "amount": "p_4h_challan_amount", } ANNEX_GRID: ClassVar[dict[str, str]] = { "sealtype": "p_1i_seal_typ", "natureofcargo": "p_1i_nature_cargo", "noofpackets": "p_1i_no_of_pkgs", "noofcontainers": "p_1i_no_of_containers", "loosepackets": "p_1i_loose_pkts", } # Part I key/value labels: field -> (aliases, side) PART_ONE_VALUES: ClassVar[dict[str, tuple[tuple[str, ...], str]]] = { "port_of_loading": (("PORT OF LOADING",), "left"), "state_of_origin": (("STATE OF ORIGIN",), "left"), "port_of_discharge": (("PORT OF DISCHARGE",), "left"), "cntry_of_finaldstn": (("COUNTRY OF FINAL DESTINATION",), "right"), "port_of_finaldstn": (("PORT OF FINAL DESTINATION",), "right"), "cntry_of_discharge": (("COUNTRY OF DISCHARGE",), "right"), "p_1b_ad_code": (("AD CODE",), "left"), "_rbi_waiver": (("RBI WAIVER NO & DT", "RBI WAIVER NO"), "left"), "p_1b_cb_name": (("CB NAME",), "left"), "p_1b_aeo": (("AEO",), "left"), "_gstin_type": (("GSTIN / TYPE", "GSTIN"), "right"), "p_1b_forex_ac_no": (("FOREX BANK A/C NO",), "right"), "p_1b_dbk_ac_no": (("DBK BANK A/C NO",), "right"), "p_1b_ifsc_no": (("IFSC NO",), "right"), } INVOICE_REF_GRID: ClassVar[dict[str, str]] = { "sno": "inv_sn", "invoicenodt": "_inv_no_dt", "ponodt": "_po_no_dt", "locnodt": "_loc_no_dt", "contractnodt": "_contract_no_dt", "adcode": "p_2a_ad_code", "invterm": "p_2a_invterm", } INVOICE_VALUE_GRID: ClassVar[dict[str, str]] = { "invoicevalue": "p_2c_invoice_value", "fobvalue": "p_2c_fob_val", "freight": "p_2c_freight", "insurance": "p_2c_insurance", "discount": "p_2c_discount", "commison": "p_2c_commison", "commission": "p_2c_commison", "deduct": "p_2c_deduct", "pc": "p_2c_p_c", "exchangerate": "p_2c_exchng_rate_desc", } INVOICE_CURRENCY_FIELDS: ClassVar[dict[str, str]] = { "p_2c_invoice_value": "p_2c_invoice_curr", "p_2c_fob_val": "p_2c_fob_curr", "p_2c_freight": "p_2c_frieght_curr", "p_2c_insurance": "p_2c_insurance_curr", "p_2c_discount": "p_2c_discount_curr", "p_2c_commison": "p_2c_commison_curr", "p_2c_deduct": "p_2c_deduct_curr", "p_2c_p_c": "p_2c_p_c_curr", } INVOICE_ITEM_GRID: ClassVar[dict[str, str]] = { "itemsno": "p_3a_itemsn", "hscd": "p_3a_cth", "description": "p_3a_item_desc", "quantity": "p_3a_qty", "uqc": "p_3a_uqc", "rate": "p_3a_rate", "valuefc": "p_3a_value", } ITEM_MAIN_GRID: ClassVar[dict[str, str]] = { "invsn": "p_3a_invsno", "itemsn": "p_3a_itemsn", "hscd": "p_3a_cth", "description": "p_3a_item_desc", "quantity": "p_3a_qty", "uqc": "p_3a_uqc", "rate": "p_3a_rate", "valuefc": "p_3a_value", "fobinr": "p_3a_fob", "fob": "p_3a_fob", "pmv": "p_3a_pmv", } ITEM_SUB_GRIDS: ClassVar[list[dict[str, str]]] = [ { "dutyamt": "p_3a_duty_amt", "cessrt": "p_3a_cess_rate", "cesamt": "p_3a_cess_amt", "dbkclmd": "p_3a_dbk_claimed", "igststat": "p_3a_igststat", "igstvalue": "p_3a_igst_val", "igstamount": "p_3a_igst_amt", "schcod": "p_3a_schcod", }, { "schemedescription": "p_3a_scheme_desc", "sqcmsr": "p_3a_sqc_mst", "sqcuqc": "p_3a_sqc_uqc", "stateoforigin": "p_3a_state_of_origin", "districtoforigin": "p_3a_district_of_origin", }, { "ptabroad": "p_3a_pt_abroad", "compcess": "p_3a_comp_cess", "enduse": "p_3a_end_use", "ftabenefitavailed": "p_3a_benefit_availd", "rewardbenefit": "p_3a_reward_benefit", "thirdpartyitem": "p_3a_third_party_item", }, ] _SERIALS: ClassVar[dict[str, str]] = {"invsn": "invsn", "invsno": "invsn", "itmsn": "itmsn", "itmsno": "itmsn", "itemsn": "itmsn", "itemsno": "itmsn"} # Part IV sections: title key -> (child table, label key -> field) EXPORT_SCHEME_SECTIONS: ClassVar[dict[str, tuple[str | None, dict[str, str]]]] = { "drawbackroslclaim": ("drawback_and_rosl_claim", { "invsno": "p_3b_invsno", "itemsno": "p_3b_itemsn", "dbksno": "p_3b_dbksn", "rate": "p_3b_dbk_rate", "dbkamt": "p_3b_dbk_amt", "rosctlamt": "p_3b_rosl_amt", }), "aadfialicencedetails": ("aa__dfia_licence_details", { "invsno": "p_4b_invsn", "itemsno": "p_4b_itmsn", "licenceno": "p_4b_licence_no", "expqty": "p_4b_licence_qty", "fobvalue": "p_4b_licence_value", }), "jobbingdetails": ("job_details", {"beno": "p_3c_job_no", "bedate": "p_3c_job_date"}), "singlewindowdeclaration": ("single_window_declaration", { "invsn": "p_4d_invsn", "itmsn": "p_4d_itmsn", "info": "p_4d_info", "qualifier": "p_4d_qualifier", "infocd": "p_4d_info_cd", "infotext": "p_4d_info_text", "infomsr": "p_4d_info_msr", "uqc": "p_4d_uqc", }), "singlewindowdeclarationconstituents": ("single_window_declaration_constituents", { "invsn": "p_4e_invsn", "itmsno": "p_4e_itmsn", "itmsn": "p_4e_itmsn", "name": "p_4e_constituent", "percentage": "p_4e_percentage", }), "singlewindowdeclarationcontrol": ("single_window_declaration_control", { "invsn": "p_4f_invsn", "itmsno": "p_4f_itmsn", "itmsn": "p_4f_itmsn", "controltype": "p_4f_control", "rescd": "_res_cd", "restext": "_res_text", }), "supportingdocuments": ("supporting_documents", { "invsn": "p_4g_invsn", "itmsno": "p_4g_itmsn", "itmsn": "p_4g_itmsn", "doctypcd": "p_4g_doc_code", "irn": "p_4g_doc_no", "icegateid": "_icegate_id", "issdt": "p_4g_doc_date", }), "invoicedetails": (None, {}), "containerdetails": ("container_details", { "sno": "p_4i_cont_sno", "container": "p_4i_container", "seal": "p_4i_cont_seal", "date": "p_4i_cont_dt", }), "ar4details": ("ar4_details", { "invsn": "p_4j_invsn", "itmsn": "p_4j_itmsno", "ar4number": "p_4j_ar4_number", "ar4date": "p_4j_ar4_date", "commissionerate": "p_4j_commissionerate", "division": "p_4j_division", "range": "p_4j_range", }), "thirdpartydetails": ("third_party_details", { "invsn": "p_4k_invsn", "itmsn": "p_4k_itmsn", "iec": "p_4k_iec", "exportername": "p_4k_exporter_name", "address": "p_4k_address", "gstnidandtype": "p_4k_gstn_id_type", }), "itemmanufacturerproducergrowerdetails": ("item_manufacturerproducergrower_details", { "invsn": "p_4l_invsn", "itmsn": "p_4l_itmsn", "type": "p_4l_type", "manufactcd": "p_4l_manufact_cd", "sourcestate": "p_4l_source_state", "transcy": "p_4l_trans_cy", "address": "p_4l_address", }), "rodtepdetails": ("rodtep_details", { "invsn": "p_4m_invsn", "itmsn": "p_4m_itmsn", "quantity": "p_4m_quantity", "uqc": "p_4m_uqc", "noofunits": "p_4m_no_of_units", "value": "p_4m_value", }), "reexportdetails": (None, {}), } # Tolerances in PDF points. VALUE_ROW_MAX_GAP = 16.0 LABEL_GAP = 5.5 # ================================================================== # Public API # ================================================================== def __init__(self, pages: list[dict[str, Any]]): self.trace: list[dict[str, Any]] = [] builder = _LayoutBuilder(self._known_label_keys()) built = [builder.build(page or {}, i) for i, page in enumerate(pages or [])] self.pages: list[_Page] = self._select_shipping_bill_pages(built) self.split_x: float = self._column_split() # Cross-section evidence gathered while reading, used as fallbacks. self._invoice_summary: list[dict[str, Any]] = [] self._invoice_items: list[dict[str, Any]] = [] self._part_one_containers: list[dict[str, Any]] = [] def parse(self) -> dict[str, Any]: result = self._empty_result() if not self.pages: self._note("document", "No Shipping Bill page with extractable text was found.") else: self._read_header(result) self._read_part_one(result) self._read_invoices(result) self._read_items(result) self._read_export_schemes(result) self._reconcile(result) self._finalize(result) return {"shipping_bill_json": result, "_trace": self.trace} # ================================================================== # Page selection # ================================================================== def _select_shipping_bill_pages(self, pages: list[_Page]) -> list[_Page]: for page in pages: page.header = self._page_header_identity(page) identified = [p for p in pages if p.header.get("shipping_bill_no")] if identified: primary = identified[0].header["shipping_bill_no"] selected = [p for p in identified if p.header["shipping_bill_no"] == primary] others = sorted({p.header["shipping_bill_no"] for p in identified} - {primary}) self.trace.append({ "stage": "page_selection", "shipping_bill_no": primary, "selected_pages": [p.number for p in selected], "skipped_pages": [p.number for p in pages if p not in selected], "reason": "ICEGATE Shipping Bill header matched", }) if others: self._note("page_selection", f"PDF also contains Shipping Bill(s) {', '.join(others)}; only {primary} was parsed.") return selected fallback = [p for p in pages if p.part] or [p for p in pages if p.rows] self._note("page_selection", "ICEGATE header not found; parsing all pages with extractable text.") return fallback def _page_header_identity(self, page: _Page) -> dict[str, str]: """Port code, SB number and SB date from the ICEGATE page header. A page is only accepted as a Shipping Bill page when it carries the ICEGATE signature (system banner or a PART title) and the header value row yields all three identifiers; look-alike pages (invoices, packing lists, forms reusing the same captions) are rejected. """ if not self._has_icegate_signature(page): return {} rows = page.header_rows or page.rows[:15] for i, row in enumerate(rows): if "portcodesbnosbdate" not in row.key or i + 1 >= len(rows): continue left = next(w.x0 for w in row.words if w.key == "port") - 3 values = [w.text for w in rows[i + 1].words if w.x0 >= left] identity = {} for value in values: if "port_code" not in identity and re.fullmatch(r"[A-Z]{2}[A-Z0-9]{3,6}", value) and re.search(r"[A-Z]{4}", value): identity["port_code"] = value elif "shipping_bill_no" not in identity and re.fullmatch(r"\d{5,10}", value): identity["shipping_bill_no"] = value elif "shipping_bill_date" not in identity and _is_date(value): identity["shipping_bill_date"] = value if len(identity) == 3: return identity return {} return {} @staticmethod def _has_icegate_signature(page: _Page) -> bool: return page.icegate_banner or page.part is not None def _column_split(self) -> float: """X coordinate where the right-hand column of two-column blocks starts.""" for row in self._part_rows("I"): for word in row.words: if _label_key(word.text) in {"consignee", "country"} and word.x0 > 150: return word.x0 - 1.5 width = self.pages[0].width if self.pages else 612.0 return width / 2 # ================================================================== # Layout helpers # ================================================================== def _part_rows(self, part: str) -> list[_Row]: return [row for page in self.pages if page.part == part for row in page.body_rows] @classmethod def _known_label_keys(cls) -> set[str]: keys: set[str] = set() for grid in ( cls.STATUS_GRID, cls.VALUATION_GRID, cls.MANIFEST_GRID, cls.CIN_GRID, cls.INVOICE_SUMMARY_GRID, cls.EQUIPMENT_GRID, cls.CHALLAN_GRID, cls.ANNEX_GRID, cls.INVOICE_REF_GRID, cls.INVOICE_VALUE_GRID, cls.INVOICE_ITEM_GRID, cls.ITEM_MAIN_GRID, *cls.ITEM_SUB_GRIDS, ): keys.update(grid) for _table, grid in cls.EXPORT_SCHEME_SECTIONS.values(): keys.update(grid) for aliases, _side in cls.PART_ONE_VALUES.values(): keys.update(_compact(a) for a in aliases) keys.update({"country", "consignee", "exporters", "buyers", "event", "leono", "leodate", "brc", "sez"}) return keys @staticmethod def _starts_numbered_label(word: _Word) -> bool: return bool(_NUMBERED_LABEL_RE.match(word.text) and re.search(r"[A-Za-z]", word.text)) def _label_groups(self, words: list[_Word]) -> list[_Label]: """Group header words into printed labels ("19." "SCHEME" "DESCRIPTION").""" groups: list[list[_Word]] = [] for i, word in enumerate(words): nxt = words[i + 1] if i + 1 < len(words) else None number_token = bool(_PURE_NUMBER_PREFIX_RE.match(word.text) and nxt and re.match(r"[A-Za-z]", nxt.text)) new_group = ( not groups or self._starts_numbered_label(word) or number_token or word.x0 - groups[-1][-1].x1 > self.LABEL_GAP ) if groups and _PURE_NUMBER_PREFIX_RE.match(groups[-1][-1].text) and len(groups[-1]) == 1: new_group = False # "19." continues with "SCHEME" if new_group: groups.append([word]) else: groups[-1].append(word) labels = [] for group in groups: text = " ".join(w.text for w in group) labels.append(_Label(key=_label_key(text), text=text, x0=group[0].x0, x1=group[-1].x1)) return labels @staticmethod def _is_label_token(word: _Word, nxt: _Word | None) -> bool: """Printed field-number label: "12.PORT", "4.", "1INVSN", "13CESAMT".""" text = word.text if re.match(r"^\d{1,2}\.\s*[A-Za-z]", text) or re.match(r"^\d{1,2}[A-Za-z]{2,}", text): return True return bool(re.fullmatch(r"\d{1,2}\.", text) and nxt is not None and re.match(r"[A-Za-z]{2}", nxt.text)) def _is_label_row(self, row: _Row) -> bool: """A header row has several numbered labels and they form a large share of it. The share test keeps data rows such as "3M TAPE 5KG ..." or "1 1 CHR SQC" from being mistaken for headers. """ words = row.words numbered = sum( 1 for i, w in enumerate(words) if self._is_label_token(w, words[i + 1] if i + 1 < len(words) else None) ) return numbered >= 2 and numbered >= 0.2 * len(words) @staticmethod def _match_spec(key: str, spec: dict[str, str]) -> str | None: if key in spec: return spec[key] best = None for spec_key, target in spec.items(): prefix_match = len(spec_key) >= 4 and len(key) >= 4 and (key.startswith(spec_key) or spec_key.startswith(key)) if prefix_match and (best is None or len(spec_key) > best[0]): best = (len(spec_key), target) return best[1] if best else None def _grid_labels(self, row: _Row, spec: dict[str, str], x_min: float = float("-inf"), x_max: float = float("inf")) -> list[_Label]: labels = self._label_groups(row.between(x_min, x_max)) return [label for label in labels if self._match_spec(label.key, spec)] @staticmethod def _assign_to_labels(labels: list[_Label], words: list[_Word]) -> dict[int, list[_Word]]: """Assign value words to label columns by horizontal cell overlap. A label's cell is approximated as the span between its neighbours' inner edges. ICEGATE centres most values but left-aligns some (e.g. invoice number), and labels sometimes overflow their cells, so the largest overlap wins and ties fall back to the value's left edge. """ assigned: dict[int, list[_Word]] = {} if not labels: return assigned ordered = sorted(range(len(labels)), key=lambda i: labels[i].x0) for word in words: scores = [] for pos, idx in enumerate(ordered): lo = labels[ordered[pos - 1]].x1 if pos > 0 else float("-inf") hi = labels[ordered[pos + 1]].x0 if pos + 1 < len(ordered) else float("inf") overlap = min(word.x1, hi) - max(word.x0, lo) starts_inside = lo <= word.x0 < labels[idx].x1 + 0.5 scores.append((round(overlap, 1), starts_inside, idx)) best = max(scores, key=lambda s: (s[0], s[1])) if best[0] <= 0 and not best[1]: continue assigned.setdefault(best[2], []).append(word) return assigned def _read_grid( self, rows: list[_Row], header_index: int, spec: dict[str, str], x_min: float = float("-inf"), x_max: float = float("inf"), max_rows: int = 1, ) -> list[dict[str, str]]: """Read value rows under a label row; returns one ``{field: text}`` per row.""" header = rows[header_index] labels = self._grid_labels(header, spec, x_min, x_max) if not labels: return [] records = [] for row in self._value_rows(rows, header_index, max_rows): words = row.between(x_min, x_max) if not words: continue record: dict[str, str] = {} for idx, assigned in self._assign_to_labels(labels, words).items(): target = self._match_spec(labels[idx].key, spec) record[target] = " ".join(w.text for w in sorted(assigned, key=lambda w: w.x0)) if record: records.append(record) return records def _value_rows(self, rows: list[_Row], header_index: int, max_rows: int) -> list[_Row]: header = rows[header_index] found = [] previous = header for row in rows[header_index + 1 :]: if row.page != header.page or row.top - previous.top > self.VALUE_ROW_MAX_GAP: break if self._is_label_row(row) or self._is_title_row(row): break found.append(row) previous = row if len(found) >= max_rows: break return found @staticmethod def _is_title_row(row: _Row) -> bool: key = row.key return key in _PART_TITLES or key[1:] in ShippingBillParser.EXPORT_SCHEME_SECTIONS def _find_row(self, rows: list[_Row], required_keys: Iterable[str], start: int = 0) -> int | None: required = list(required_keys) for i in range(start, len(rows)): keys = {label.key for label in self._label_groups(rows[i].words)} if all(any(k == r or (len(r) >= 4 and k.startswith(r)) for k in keys) for r in required): return i return None # -- key/value label matching ------------------------------------------ def _match_label(self, words: list[_Word], start: int, alias: str) -> tuple[int, int] | None: """Match a printed label on the character stream starting at ``words[start]``. Returns ``(word_index, chars_consumed_in_that_word)`` of the label end. The label may end inside a word (label text glued to its value) and the printed field number prefix ("3." / "12.") is skipped. """ target = _compact(alias) pos, j, leading = 0, start, True while j < len(words) and pos < len(target): chunk = _compact(words[j].text) if leading: stripped = re.sub(r"^\d{1,2}", "", chunk) if not stripped and chunk.isdigit(): j += 1 continue chunk = stripped leading = False need = target[pos:] if not chunk: j += 1 continue if need.startswith(chunk): pos += len(chunk) j += 1 continue if chunk.startswith(need): return j, len(need) return None return (j, 0) if pos == len(target) else None @staticmethod def _remainder(word: _Word, consumed: int) -> _Word | None: """Part of ``word`` after its first ``consumed`` alphanumeric characters.""" seen = 0 for offset, char in enumerate(word.text): if seen == consumed: rest = word.text[offset:].lstrip(" :.-") if not rest: return None width = (word.x1 - word.x0) / max(len(word.text), 1) return _Word(rest, word.x0 + offset * width, word.x1, word.top, word.bottom, word.size) if char.isalnum(): seen += 1 return None def _label_value(self, words: list[_Word], aliases: Iterable[str]) -> list[_Word] | None: """Words printed after one of ``aliases`` on the same row, up to the next label.""" for alias in aliases: for start in range(len(words)): first = words[start] if start and not (self._starts_numbered_label(first) or _PURE_NUMBER_PREFIX_RE.match(first.text)): continue match = self._match_label(words, start, alias) if not match: continue end, consumed = match value: list[_Word] = [] if consumed: rest = self._remainder(words[end], consumed) if rest: value.append(rest) end += 1 for k, word in enumerate(words[end:], end): if self._is_next_label(word, words[k + 1] if k + 1 < len(words) else None): break value.append(word) return value return None @staticmethod def _is_next_label(word: _Word, nxt: _Word | None) -> bool: """Start of the next printed label on a key/value row ("10.DBK", "11. IFSC"). ICEGATE label numbers are always followed by a dot and a word, so values such as "07X-XXXXXX-001", "31.03.2027GR" or "NO.4675546" never match. """ if re.fullmatch(r"\d{1,2}\.", word.text): return nxt is not None and re.match(r"[A-Za-z]{2}", nxt.text) is not None return bool(re.match(r"^\d{1,2}\.\s*[A-Za-z]", word.text)) @staticmethod def _line_text(words: list[_Word]) -> str: """Text of a wrapped text-block line; stray punctuation glyphs are dropped.""" text = " ".join(w.text for w in words) return text if re.search(r"[A-Za-z0-9]", text) else "" def _side(self, row: _Row, side: str) -> list[_Word]: return row.between(x_max=self.split_x) if side == "left" else row.between(x_min=self.split_x) # ================================================================== # Header block (top-right of every page) # ================================================================== def _read_header(self, result: dict[str, Any]) -> None: page = self.pages[0] result["port_code"] = page.header.get("port_code", "") result["shipping_bill_no"] = page.header.get("shipping_bill_no", "") result["shipping_bill_date"] = page.header.get("shipping_bill_date", "") rows = page.header_rows port_row = next((r for r in rows if "portcodesbnosbdate" in r.key), None) x_min = next((w.x0 for w in port_row.words if w.key == "port"), 0.0) - 3 if port_row else 0.0 block = [r.between(x_min=x_min) for r in rows] for i, words in enumerate(block): texts = [w.text for w in words] key = _compact(" ".join(texts)) if key.startswith("iecbr"): values = texts[1:] result["iec"] = values[0] if values else "" result["iec_branch_code"] = values[1] if len(values) > 1 else "" elif key.startswith("gstintype"): result["p_1b_gstin"] = " ".join(texts[1:]) elif key.startswith("cbcode"): result["cb_code"] = " ".join(texts[2:]) elif key.startswith("typeinvitemcont") and i + 1 < len(block): labels = [_Label(w.key, w.text, w.x0, w.x1) for w in words if w.key in {"inv", "item", "cont"}] values = [w for w in block[i + 1] if _to_number(w.text) is not None] for idx, assigned in self._assign_to_labels(labels, values).items(): result[labels[idx].key] = assigned[0].text elif key.startswith("pkg"): self._read_package_row(result, texts) @staticmethod def _read_package_row(result: dict[str, Any], texts: list[str]) -> None: """``PKG G.WT ``; unit/weight may be blank.""" tokens = [t for t in texts if re.search(r"[A-Za-z0-9]", t)] gwt_at = next((i for i, t in enumerate(tokens) if _compact(t) == "gwt"), len(tokens)) pkg = [t for t in tokens[1:gwt_at] if _to_number(t) is not None] result["pkg"] = pkg[0] if pkg else 0 for token in tokens[gwt_at + 1 :]: if _to_number(token) is not None: result["gwt"] = token elif re.fullmatch(r"[A-Za-z]{1,6}", token): result["gwt_unit"] = token.upper() # ================================================================== # Part I - Shipping Bill Summary # ================================================================== def _read_part_one(self, result: dict[str, Any]) -> None: rows = self._part_rows("I") if not rows: self._note("part_i", "PART - I - SHIPPING BILL SUMMARY not found.") return self._read_status(rows, result) self._read_part_one_values(rows, result) self._read_parties(rows, result) self._read_valuation(rows, result) self._read_manifest(rows, result) self._read_equipment_and_challans(rows, result) self._read_annex(rows, result) self._read_events(rows, result) def _read_status(self, rows: list[_Row], result: dict[str, Any]) -> None: index = self._find_row(rows, ("mode", "assess", "exmn")) if index is None: return for record in self._read_grid(rows, index, self.STATUS_GRID)[:1]: for target, value in record.items(): value = value.upper() if target == "p_1a_mode" or value in {"Y", "N"}: result[target] = value for status in self.STATUS_FIELDS: if not result.get(status): result[status] = "N" def _read_part_one_values(self, rows: list[_Row], result: dict[str, Any]) -> None: for target, (aliases, side) in self.PART_ONE_VALUES.items(): for row in rows: value = self._label_value(self._side(row, side), aliases) if value is None: continue text = " ".join(w.text for w in value) if target == "_rbi_waiver": number, date = self._split_number_and_date(text) result["p_1b_rbi_wr_no"], result["p_1b_rbi_wr_dt"] = number, date elif target == "_gstin_type": if text: result["p_1b_gstin"] = text else: result[target] = text break def _read_parties(self, rows: list[_Row], result: dict[str, Any]) -> None: label_index = next((i for i, r in enumerate(rows) if "exportersnameaddress" in r.key and "consigneenameaddress" in r.key), None) if label_index is None: return end = next( (i for i in range(label_index + 1, len(rows)) if self._label_value(self._side(rows[i], "left"), ("AD CODE",)) is not None), min(label_index + 7, len(rows)), ) exporter_lines, consignee_lines = [], [] for row in rows[label_index + 1 : end]: left = self._side(row, "left") type_at = next((i for i, w in enumerate(left) if _label_key(w.text) == "type" and i > 0), None) if type_at is not None: result["p_1b_type"] = " ".join(w.text for w in left[type_at + 1 :]) left = left[:type_at] if self._line_text(left): exporter_lines.append(self._line_text(left)) right = self._side(row, "right") if self._line_text(right): consignee_lines.append(self._line_text(right)) if exporter_lines: result["p_1b_exporter_name"] = exporter_lines[0] result["p_1b_exporter_address"] = " ".join(exporter_lines[1:]) if consignee_lines: result["p_1b_consignee_name"] = consignee_lines[0] result["p_1b_consignee_address"] = " ".join(consignee_lines[1:]) result["buyer_name"] = consignee_lines[0] def _read_valuation(self, rows: list[_Row], result: dict[str, Any]) -> None: for required in (("fobvalue", "freight"), ("deductions", "duty")): index = self._find_row(rows, required) if index is None: continue for record in self._read_grid(rows, index, self.VALUATION_GRID)[:1]: for target, value in record.items(): if _to_number(value) is not None: result[target] = value def _read_manifest(self, rows: list[_Row], result: dict[str, Any]) -> None: index = self._find_row(rows, ("mawbno", "hawbno")) if index is None: return cin_index = self._find_row(rows, ("cinno",), index + 1) records = [] for record in self._read_grid(rows, index, self.MANIFEST_GRID, x_max=self.split_x, max_rows=10): records.append(self._repair_manifest_columns(record)) if cin_index is not None: cin = self._read_grid(rows, cin_index, self.CIN_GRID, x_max=self.split_x) if cin: if not records: records.append({}) records[0].update(cin[0]) result["manifest_details"] = [r for r in records if r] # F. Invoice summary shares the manifest band on the right-hand side. summary = self._read_grid(rows, index, self.INVOICE_SUMMARY_GRID, x_min=self.split_x, max_rows=20) self._invoice_summary = [r for r in summary if r.get("p_2a_inv_no")] @staticmethod def _repair_manifest_columns(record: dict[str, str]) -> dict[str, str]: """MAWB/HAWB numbers are printed centred in cells narrower than their overflowing labels, so a number may overlap the date column. A date column must hold a date; otherwise the value belongs to its number column. """ for number_field, date_field in (("p_1e_mawb_no", "p_1e_mawb_dt"), ("p_1e_hawb_no", "p_1e_hawb_dt")): value = record.get(date_field, "") if not value or _is_date(value): continue dates = [t for t in value.split() if _is_date(t)] others = " ".join(t for t in value.split() if not _is_date(t)) if not record.get(number_field): record[number_field] = others record[date_field] = dates[0] if dates else "" return record def _read_equipment_and_challans(self, rows: list[_Row], result: dict[str, Any]) -> None: index = self._find_row(rows, ("container", "seal")) if index is None: return self._part_one_containers = [ r for r in self._read_grid(rows, index, self.EQUIPMENT_GRID, x_max=self.split_x, max_rows=20) if r.get("p_4i_container") ] challans = self._read_grid(rows, index, self.CHALLAN_GRID, x_min=self.split_x, max_rows=20) result["challan_details"] = [r for r in challans if r.get("p_4h_challan_no")] def _read_annex(self, rows: list[_Row], result: dict[str, Any]) -> None: index = self._find_row(rows, ("sealtype", "natureofcargo")) if index is None: return record = next(iter(self._read_grid(rows, index, self.ANNEX_GRID)), {}) marks = self._read_marks(rows, index + 1) if marks: record["p_1i_marks_numbers"] = marks if record: result["annex_details"] = [record] def _read_marks(self, rows: list[_Row], start: int) -> str: for i in range(start, len(rows)): value = self._label_value(rows[i].words, ("MARKS & NUMBERS", "MARKS AND NUMBERS")) if value is None: continue value_x = value[0].x0 if value else max(w.x1 for w in rows[i].words) parts = [" ".join(w.text for w in value)] for row in rows[i + 1 :]: if self._is_label_row(row) or row.page != rows[i].page: break continuation = row.between(x_min=value_x - 5) if len(continuation) != len(row.words): break parts.append(" ".join(w.text for w in continuation)) text = " ".join(p for p in parts if p).strip() return "" if text in {".", "-"} else text return "" def _read_events(self, rows: list[_Row], result: dict[str, Any]) -> None: index = self._find_row(rows, ("event", "date", "time")) if index is None: return events = { "submission": ("p_1j_subm_dt", "p_1i_subm_time"), "assessment": ("p_1i_assess_dt", "p_1i_assess_time"), "examination": ("p_1i_exmn_dt", "p_1i_exmn_time"), "leo": ("p_1i_leo_dt", "p_1i_leo_time"), } right_labels = { "p_1i_leo_no": ("LEO NO",), "_leo_date": ("LEO DATE",), "p_1i_brc_realzn_dt": ("BRC REALISATION DATE", "BRC REALIZATION DATE"), } leo_date = "" for row in rows[index:]: left = self._side(row, "left") if left: event = _label_key(left[0].text) sez = self._label_value(left, ("SEZ UNIT DETAILS", "SEZ UNIT")) if sez is not None: result["p_1i_sez_unit_data"] = " ".join(w.text for w in sez) elif event in events: date_field, time_field = events[event] for word in left[1:]: if _is_date(word.text) and not result.get(date_field): result[date_field] = word.text elif _is_time(word.text) and not result.get(time_field): result[time_field] = word.text right = self._side(row, "right") for target, aliases in right_labels.items(): value = self._label_value(right, aliases) if value is None: continue text = " ".join(w.text for w in value) if target == "_leo_date": leo_date = text elif target == "p_1i_brc_realzn_dt": result[target] = text if _is_date(text) else "" else: result[target] = text if leo_date and _is_date(leo_date): result["p_1i_leo_dt"] = leo_date # ================================================================== # Part II - Invoice Details # ================================================================== def _read_invoices(self, result: dict[str, Any]) -> None: rows = self._part_rows("II") invoices: list[dict[str, Any]] = [] current: dict[str, Any] | None = None i = 0 while i < len(rows): row = rows[i] keys = {label.key for label in self._label_groups(row.words)} if {"sno"} <= keys and any(k.startswith("invoiceno") for k in keys): current = self._invoice_reference(rows, i) invoices.append(current) elif current is not None and "exportersnameaddress" in row.key and "buyersnameaddress" in row.key: i = self._invoice_parties(rows, i, current) continue elif current is not None and any(k.startswith("invoicevalue") for k in keys): self._invoice_values(rows, i, current) elif any(k.startswith("hscd") for k in keys) and any(k.startswith("description") for k in keys): for item in self._read_item_rows(rows, i, self.INVOICE_ITEM_GRID): item["p_3a_invsno"] = (current or {}).get("inv_sn") or 1 self._invoice_items.append(item) i += 1 if not invoices: invoices = list(self._invoice_summary) if invoices: self._note("invoice_details", "PART - II not found; invoice rows taken from Part I invoice summary.") result["invoice_details"] = invoices def _invoice_reference(self, rows: list[_Row], index: int) -> dict[str, Any]: invoice: dict[str, Any] = {} for record in self._read_grid(rows, index, self.INVOICE_REF_GRID)[:1]: number, date = self._split_number_and_date(record.pop("_inv_no_dt", "")) invoice.update({k: v for k, v in record.items() if not k.startswith("_")}) invoice["p_2a_inv_no"], invoice["p_2a_inv_dt"] = number, date return invoice def _invoice_parties(self, rows: list[_Row], index: int, invoice: dict[str, Any]) -> int: exporter, buyer = [], [] i = index + 1 while i < len(rows): row = rows[i] if self._is_label_row(row) or any(k in row.key for k in ("thirdpartynameaddress", "buyeraeostatus", "invoicevalue")): break left, right = self._line_text(self._side(row, "left")), self._line_text(self._side(row, "right")) if left: exporter.append(left) if right: buyer.append(right) i += 1 if exporter: invoice["p_2b_exporter_name"] = exporter[0] invoice["p_2b_exporter_addr"] = " ".join(exporter[1:]) if buyer: invoice["p_2b_buyer_name"] = buyer[0] invoice["p_2b_buyer_addr"] = " ".join(buyer[1:]) return i def _invoice_values(self, rows: list[_Row], index: int, invoice: dict[str, Any]) -> None: records = self._read_grid(rows, index, self.INVOICE_VALUE_GRID, max_rows=2) if not records: return values = records[0] for target, value in values.items(): if target == "p_2c_exchng_rate_desc": invoice[target] = value numbers = _NUMBER_RE.findall(value.replace(",", "")) if numbers: invoice["p_2c_exchng_rate"] = numbers[-1] else: invoice[target] = value if len(records) > 1: for target, currency in records[1].items(): currency_field = self.INVOICE_CURRENCY_FIELDS.get(target) if currency_field and re.fullmatch(r"[A-Z]{3}", currency): invoice[currency_field] = currency # ================================================================== # Part III - Item Details # ================================================================== def _read_items(self, result: dict[str, Any]) -> None: rows = self._part_rows("III") items: list[dict[str, Any]] = [] current: dict[str, Any] | None = None for i, row in enumerate(rows): keys = {label.key for label in self._label_groups(row.words)} if any(k.startswith("hscd") for k in keys) and any(k.startswith("description") for k in keys): current = None for item in self._read_item_rows(rows, i, self.ITEM_MAIN_GRID): items.append(item) current = item elif current is not None: for spec in self.ITEM_SUB_GRIDS: if sum(1 for k in keys if self._match_spec(k, spec)) >= 3: for record in self._read_grid(rows, i, spec, max_rows=3)[:1]: for target, value in record.items(): current.setdefault(target, value) break invoice_lines = {self._item_key(i): i.pop("_desc_lines", []) for i in self._invoice_items} for item in self._invoice_items: item["p_3a_item_desc"] = self._join_wrapped_lines(invoice_lines[self._item_key(item)]) for item in items: reference = invoice_lines.get(self._item_key(item), ()) item["p_3a_item_desc"] = self._join_wrapped_lines(item.pop("_desc_lines", []), reference) if not items and self._invoice_items: items = self._invoice_items self._note("item_details", "PART - III not found; items taken from PART - II item list.") result["item_details"] = items @staticmethod def _item_key(item: dict[str, Any]) -> tuple[str, str]: invoice = _clean_text(item.get("p_3a_invsno")) or "1" return invoice, _clean_text(item.get("p_3a_itemsn")) def _read_item_rows(self, rows: list[_Row], header_index: int, spec: dict[str, str]) -> list[dict[str, Any]]: """Read an item block: one data row per item plus wrapped description lines. An item row is recognised by an HS code printed in the HS column; the description column spans from the HS value to the Quantity column, so its wrapped lines are captured regardless of label centring. A long number inside a wrapped description ("SR NO 24350229") is therefore never mistaken for a new item. """ header = rows[header_index] labels = self._grid_labels(header, spec) by_field = {self._match_spec(label.key, spec): label for label in labels} qty_label = by_field.get("p_3a_qty") hs_label = by_field.get("p_3a_cth") if not labels or qty_label is None or hs_label is None: return [] desc_right = qty_label.x0 - 1 numeric_labels = [label for label in labels if label.x0 >= qty_label.x0 - 1] key_labels = [label for label in labels if label.x1 <= hs_label.x0 + 1] items: list[dict[str, Any]] = [] item: dict[str, Any] | None = None for row in rows[header_index + 1 :]: if row.page != header.page or self._is_label_row(row) or self._is_title_row(row): break hs_word = next( (w for w in row.words if re.fullmatch(r"\d{4,10}", w.text) and hs_label.x0 - 20 <= w.x0 < hs_label.x1), None, ) if hs_word is not None: item = {"p_3a_cth": hs_word.text, "_desc_lines": []} items.append(item) serials = [w for w in row.words if w.x1 <= hs_word.x0] for idx, assigned in self._assign_to_labels(key_labels, serials).items(): item[self._match_spec(key_labels[idx].key, spec)] = assigned[0].text values = [w for w in row.words if w.x0 >= desc_right] for idx, assigned in self._assign_to_labels(numeric_labels, values).items(): item[self._match_spec(numeric_labels[idx].key, spec)] = " ".join(w.text for w in assigned) line = [w.text for w in row.words if hs_word.x1 < w.x0 and w.x1 <= desc_right + 1] elif item is not None: line = [w.text for w in row.words if w.x0 >= hs_label.x1 and w.x1 <= desc_right + 1] else: continue if line: item["_desc_lines"].append(" ".join(line)) return items @staticmethod def _join_wrapped_lines(lines: list[str], reference_lines: Iterable[str] = ()) -> str: """Join visually wrapped lines of one text cell. Lines normally wrap at a space. When a line ends with "-" the wrap may have happened inside a hyphenated token ("4000-|5000WAS") or after a hyphen that is followed by a space ("PN-| 2044384"). ICEGATE prints the same description in Part II and Part III with different wrap points, so the other rendering (``reference_lines``) decides; without evidence of a space the hyphenated token is kept intact. """ references = [re.sub(r"\s+", " ", r) for r in reference_lines] text = "" for line in (raw.strip() for raw in lines): if not line: continue if not text: text = line elif text.endswith("-"): spaced = f"{text.split()[-1]} {line.split()[0]}" text = f"{text} {line}" if any(spaced in r for r in references) else f"{text}{line}" else: text = f"{text} {line}" return text # ================================================================== # Part IV - Export Scheme Details # ================================================================== def _read_export_schemes(self, result: dict[str, Any]) -> None: rows = self._part_rows("IV") collected: dict[str, list[dict[str, Any]]] = {} table: str | None = None spec: dict[str, str] = {} headers: list[list[_Label]] = [] pending: list[_Row] = [] def flush() -> None: if table and headers and pending: collected.setdefault(table, []).extend(self._section_records(pending, headers, spec)) pending.clear() for row in rows: title = row.key[1:] if row.key[1:] in self.EXPORT_SCHEME_SECTIONS else None if title: flush() table, spec = self.EXPORT_SCHEME_SECTIONS[title] headers = [] elif self._is_label_row(row): if pending: flush() headers = [] headers.append(self._label_groups(row.words)) else: pending.append(row) flush() for name, records in collected.items(): result[name] = records if not result.get("container_details") and self._part_one_containers: result["container_details"] = self._part_one_containers def _section_records(self, rows: list[_Row], headers: list[list[_Label]], spec: dict[str, str]) -> list[dict[str, Any]]: """Map Part IV data rows; multi-line headers produce multi-line records.""" records: list[dict[str, Any]] = [] lines_per_record = max(len(headers), 1) for offset in range(0, len(rows), lines_per_record): record: dict[str, Any] = {} for line, row in enumerate(rows[offset : offset + lines_per_record]): labels = headers[line] for idx, assigned in self._assign_to_labels(labels, row.words).items(): target = self._match_spec(labels[idx].key, spec) if target: record[target] = " ".join(w.text for w in sorted(assigned, key=lambda w: w.x0)) if not record: continue serial = next((f for f in record if f.endswith(("_invsn", "_invsno"))), None) first_label_field = self._match_spec(headers[0][0].key, spec) if headers[0] else None if records and first_label_field and first_label_field not in record: # Wrapped text line of the previous record. for key, value in record.items(): records[-1][key] = " ".join(filter(None, [str(records[-1].get(key, "")), value])) continue if serial is None and not any(v for k, v in record.items() if not k.startswith("_")): continue records.append(record) for record in records: if "_res_text" in record or "_res_cd" in record: record["p_4f_result"] = record.pop("_res_text", "") or record.pop("_res_cd", "") for key in [k for k in record if k.startswith("_")]: record.pop(key) return records # ================================================================== # Cross-section reconciliation # ================================================================== def _reconcile(self, result: dict[str, Any]) -> None: invoices = result.get("invoice_details") or [] for position, invoice in enumerate(invoices, 1): invoice["inv_sn"] = invoice.get("inv_sn") or position if not invoice.get("p_2a_ad_code") and result.get("p_1b_ad_code"): invoice["p_2a_ad_code"] = result["p_1b_ad_code"] items = result.get("item_details") or [] for position, item in enumerate(items, 1): item["p_3a_invsno"] = item.get("p_3a_invsno") or 1 item["p_3a_itemsn"] = item.get("p_3a_itemsn") or position # Header counts are authoritative; parsed rows are the fallback. if _to_number(result.get("inv")) in (None, 0) and invoices: result["inv"] = len(invoices) if _to_number(result.get("item")) in (None, 0) and items: result["item"] = len(items) declared_items = _to_number(result.get("item")) or 0 if items and declared_items and int(declared_items) != len(items): self._note("item_details", f"Header declares {int(declared_items)} item(s) but {len(items)} item block(s) were read.") if not result.get("p_1j_subm_dt") and result.get("shipping_bill_date"): result["p_1j_subm_dt"] = result["shipping_bill_date"] # ================================================================== # Schema / normalisation # ================================================================== def _empty_result(self) -> dict[str, Any]: result: dict[str, Any] = {f: (0 if f in self.NUMBER_FIELDS else "") for f in self.PARENT_FIELDS} result["doctype"] = "Shipping Bill" for table in self.CHILD_TABLES: result[table] = [] return result def _finalize(self, result: dict[str, Any]) -> None: result["doctype"] = "Shipping Bill" for name in self.PARENT_FIELDS: if name != "doctype": result[name] = self._normalize_parent(name, result.get(name)) for table in self.CHILD_TABLES: rows = [] for record in result.get(table) or []: clean = {"doctype": self.CHILD_DOCTYPES[table]} for name in self.COMPLETE_CHILD_FIELDS[table]: clean[name] = self._normalize_child(name, record.get(name)) if self._has_identity(table, clean): rows.append(clean) result[table] = rows for key in [k for k in result if k not in self.PARENT_FIELDS and k not in self.CHILD_TABLES]: result.pop(key) def _has_identity(self, table: str, record: dict[str, Any]) -> bool: identity = { "manifest_details": ("p_1e_mawb_no", "p_1e_hawb_no", "p_1e_cin_no"), "annex_details": ("p_1i_seal_typ", "p_1i_nature_cargo", "p_1i_marks_numbers", "p_1i_no_of_pkgs"), "invoice_details": ("p_2a_inv_no",), "item_details": ("p_3a_cth", "p_3a_item_desc"), "single_window_declaration": ("p_4d_info", "p_4d_qualifier"), }.get(table) if identity: return any(record.get(f) not in (None, "", 0) for f in identity) serials = {f for f in self.COMPLETE_CHILD_FIELDS[table] if re.search(r"_(invsn|invsno|itmsn|itmsno|itemsn|sno|serial_no)$", f)} return any(record.get(f) not in (None, "", 0) for f in self.COMPLETE_CHILD_FIELDS[table] if f not in serials) def _normalize_parent(self, name: str, value: Any) -> Any: if name in self.NUMBER_FIELDS: return self._number_or_zero(value) if name in self.DATE_FIELDS: return self._format_date(value) if name in self.TIME_FIELDS: return self._format_time(value) text = _clean_text(value) if name == "p_1b_gstin": match = re.search(r"\b(\d{2}[A-Z]{5}\d{4}[A-Z][0-9A-Z]{3})\b", text.upper()) return match.group(1) if match else text return text def _normalize_child(self, name: str, value: Any) -> Any: text = _clean_text(value) numeric = name in self.CHILD_NUMERIC_DEFAULT_FIELDS or ( name not in self.CHILD_STRING_FIELDS and not name.endswith("_curr") and re.search(r"(qty|value|amount|amt|rate|percentage|no_of|fob|pmv|freight|insurance|discount|deduct|commison|p_c|quantity|units)$", name) ) if not text: return 0 if numeric else "" if name.endswith(("_dt", "_date")): return self._format_date(text) if name in self.CHILD_STRING_FIELDS or name.endswith("_curr"): return text if name == "p_3a_comp_cess" or numeric: match = _NUMBER_RE.search(text.replace(",", "")) return self._as_number(float(match.group())) if match else (0 if numeric else text) number = _to_number(text) return self._as_number(number) if number is not None else text def _number_or_zero(self, value: Any) -> int | float: number = _to_number(value) return self._as_number(number) if number is not None else 0 @staticmethod def _as_number(number: float) -> int | float: return int(number) if float(number).is_integer() else number @staticmethod def _format_date(value: Any) -> str: text = _clean_text(value).strip(".,;:").upper() if not text: return "" for fmt in ("%d-%b-%y", "%d-%b-%Y", "%d/%m/%Y", "%d/%m/%y", "%d-%m-%Y", "%d-%m-%y", "%Y-%m-%d", "%d.%m.%Y"): try: return datetime.strptime(text, fmt).strftime("%Y-%m-%d") except ValueError: continue return text @staticmethod def _format_time(value: Any) -> str: match = _TIME_RE.search(str(value or "")) if not match: return _clean_text(value) return f"{int(match.group(1)):02d}:{match.group(2)}:{match.group(3) or '00'}" @staticmethod def _split_number_and_date(text: str) -> tuple[str, str]: tokens = _clean_text(text).split() date = next((t for t in reversed(tokens) if _is_date(t)), "") number = " ".join(t for t in tokens if t != date) return number, date def _note(self, stage: str, message: str) -> None: self.trace.append({"stage": stage, "message": message}) if __name__ == "__main__": import json import sys from shipping_bill_extractor.services.pdf_analyzer import PDFAnalyzer if len(sys.argv) < 2: print("Usage: python shipping_bill_parser.py ") raise SystemExit(1) analysis = PDFAnalyzer(sys.argv[1]).analyze() print(json.dumps(ShippingBillParser(analysis["pages"]).parse(), indent=2, default=str))