|
"""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 <count> G.WT <unit> <weight>``; 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 <shipping_bill.pdf>")
|
|
raise SystemExit(1)
|
|
analysis = PDFAnalyzer(sys.argv[1]).analyze()
|
|
print(json.dumps(ShippingBillParser(analysis["pages"]).parse(), indent=2, default=str))
|