#!/usr/bin/env python3 """ Shipping Bill Bulk Uploader =========================== Reads every Shipping Bill PDF in a folder, converts each one to JSON using the Shipping Bill Extractor framework logic (pdfplumber only, no OCR, no AI), and creates a "Shipping Bill" document in Frappe / ERPNext through the REST API. When the run finishes, an Excel status report is saved to the user's Downloads folder (and opened automatically). Validation pipeline for each PDF (one after the other). Every step is recorded as PASS / FAIL in the Excel report: 0. Unique file in folder identical PDFs (same content, any file name) are detected before processing and reported as "Duplicate in Folder" - never uploaded twice 1. PDF readable PDFAnalyzer: text, words, tables per page 2. JSON extracted ShippingBillParser: finds the Customs Shipping Bill pages (image/attachment pages skipped) 3. JSON valid vs DocType mandatory fields, date/time/number formats, field lengths and Select options checked against the live Frappe DocType before anything is sent 4. Unique SB in folder two different PDFs of the same Shipping Bill (same SB No + Port Code) -> only the first is used 5. Not already in Frappe same SB No + Port Code, or the exact same PDF already attached to a Shipping Bill 6. Created in Frappe POST /api/resource/Shipping Bill 7. Verified in Frappe document read back and compared; also checks that only ONE Shipping Bill exists for the SB 8. PDF attached + verified source PDF attached to the Shipping Bill and the File record checked (size / content hash) Usage ----- python shipping_bill_uploader.py # uses CONFIG below python shipping_bill_uploader.py --folder "D:\\SB" # override PDF folder python shipping_bill_uploader.py --dry-run # extract + validate only, # nothing is sent to Frappe python shipping_bill_uploader.py --no-open # don't open the report Requirements ------------ pip install pdfplumber requests openpyxl Python 3.9+ """ from __future__ import annotations import argparse import hashlib import html import json import logging import os import re import subprocess import sys import time import traceback from dataclasses import dataclass, field as dataclass_field from datetime import datetime from enum import Enum from pathlib import Path from typing import Any, ClassVar, Optional from urllib.parse import quote try: import pdfplumber import requests from openpyxl import Workbook from openpyxl.chart import DoughnutChart, Reference from openpyxl.chart.label import DataLabelList from openpyxl.chart.series import DataPoint from openpyxl.formatting.rule import DataBarRule from openpyxl.styles import Alignment, Border, Font, PatternFill, Side from openpyxl.utils import get_column_letter from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry except ImportError as _import_error: # pragma: no cover - environment guard sys.stderr.write( f"Missing dependency: {_import_error.name}\n" "Install the required packages with:\n" " pip install pdfplumber requests openpyxl\n" ) sys.exit(2) # ============================================================================= # CONFIGURATION - this is the ONLY section you need to edit # ============================================================================= # Every value can also be supplied as an environment variable with the same # name (environment variables win). That keeps secrets out of the file when # you want to, e.g.: set FRAPPE_API_SECRET=xxxx (Windows) CONFIG = { # ---- Frappe site & API credentials ------------------------------------- # API key / secret: Frappe > User > (API user) > Settings > API Access # > "Generate Keys" "FRAPPE_URL": "https://eapx.canofy.tech/", "FRAPPE_API_KEY": "194afd33933ed9d", "FRAPPE_API_SECRET": "33aededf52fd002", # ---- Input --------------------------------------------------------------- # Folder that contains the Shipping Bill PDFs. "PDF_FOLDER": r"C:\Users\DELL\OneDrive\Desktop\shipping_bill", # Also pick up PDFs inside sub-folders. "INCLUDE_SUBFOLDERS": False, # ---- Frappe DocType ------------------------------------------------------ "DOCTYPE": "Shipping Bill", # Leave empty to use the DocType's default naming series (SB-.YY.-.#####). "NAMING_SERIES": "", # ---- Duplicate handling -------------------------------------------------- # What to do when a Shipping Bill with the same values in # DUPLICATE_MATCH_FIELDS already exists in Frappe: # "skip" -> do not create it again, mark it "Skipped - Duplicate" # "update" -> overwrite the existing document with the PDF data # "create" -> always create a new document (no duplicate check) "DUPLICATE_ACTION": "skip", "DUPLICATE_MATCH_FIELDS": ["shipping_bill_no", "port_code"], # ---- PDF attachment -------------------------------------------------------- # Attach the source PDF to the created Shipping Bill (and verify it). # The attached PDF's content hash is also used to detect a PDF that was # already uploaded earlier, even if it was renamed. "ATTACH_PDF_TO_SHIPPING_BILL": True, "ATTACH_AS_PRIVATE_FILE": True, # ---- Validation ---------------------------------------------------------- # Validate every JSON against the live DocType definition (mandatory # fields, types, lengths, Select options) before sending it to Frappe. "VALIDATE_AGAINST_DOCTYPE": True, # A PDF is only pushed to Frappe when these fields were extracted. "REQUIRED_FIELDS": { "shipping_bill_no": "Shipping Bill No", "shipping_bill_date": "Shipping Bill Date", "port_code": "Port Code", }, # ---- Output -------------------------------------------------------------- # Excel report folder. Empty = the user's Downloads folder. "REPORT_FOLDER": "", # Open the Excel report automatically when the run is finished. "OPEN_REPORT_WHEN_DONE": True, # Save the extracted JSON of every PDF (useful for checking / debugging). "SAVE_JSON_FILES": True, # Empty = "/_extracted_json" "JSON_OUTPUT_FOLDER": "", # Empty = "logs" folder next to this script. "LOG_FOLDER": "", # ---- Network ------------------------------------------------------------- "REQUEST_TIMEOUT_SECONDS": 60, "MAX_RETRIES": 3, # retries for read requests (GET) only "VERIFY_SSL": True, # Stop the run early when Frappe is unreachable this many times in a row # (prevents waiting on hundreds of time-outs). Remaining PDFs are listed # in the report as "Not Processed". "MAX_CONSECUTIVE_CONNECTION_FAILURES": 5, } # ============================================================================= # END OF CONFIGURATION # ============================================================================= SCRIPT_NAME = "Shipping Bill Bulk Uploader" SCRIPT_VERSION = "2.0.0" PLACEHOLDER_VALUES = { "https://your-frappe-site.com", "your-api-key", "your-api-secret", } logger = logging.getLogger("shipping_bill_uploader") # ============================================================================= # EXTRACTION ENGINE # ----------------------------------------------------------------------------- # PDFAnalyzer and ShippingBillParser below are taken unchanged from the # shipping_bill_extractor framework (services/pdf_analyzer.py and # services/shipping_bill_parser.py), so the extraction rules, page selection # (mixed text/image PDFs start at the Customs Shipping Bill header page) and # the JSON produced are identical to the framework. # Do not edit these two classes here; change the framework and re-sync. # ============================================================================= class PDFAnalyzer: """ Deterministic PDF analyzer for Shipping Bill extraction. Uses pdfplumber only. Each page preserves: - text : layout-preserved text - words : individual words with coordinates - tables : tables detected using multiple strategies - lines : PDF drawing lines useful for table detection No AI/OCR is used. """ def __init__(self, file_path): self.file_path = file_path def analyze(self): start_time = time.perf_counter() self._validate_file() pages_data = [] text_pages = 0 image_pages = 0 total_words = 0 total_tables = 0 with pdfplumber.open(self.file_path) as pdf: total_pages = len(pdf.pages) if total_pages == 0: raise ValueError("The PDF contains no pages.") for page_number, page in enumerate(pdf.pages, start=1): page_data = self._analyze_page(page, page_number) pages_data.append(page_data) if page_data["text"].strip(): text_pages += 1 else: image_pages += 1 total_words += page_data["word_count"] total_tables += page_data["table_count"] elapsed = time.perf_counter() - start_time if text_pages > 0 and image_pages == 0: pdf_type = "Text-based PDF" elif text_pages == 0 and image_pages > 0: pdf_type = "Image-based / Scanned PDF" else: pdf_type = "Mixed PDF" return { "pdf_type": pdf_type, "total_pages": total_pages, "pages_processed": total_pages, "text_pages": text_pages, "image_pages": image_pages, "total_words": total_words, "total_tables": total_tables, "processing_time": round(elapsed, 2), "pages": pages_data, } def _validate_file(self): if not self.file_path: raise FileNotFoundError("PDF file path was not provided.") if not os.path.exists(self.file_path): raise FileNotFoundError(f"PDF file not found: {self.file_path}") if not os.path.isfile(self.file_path): raise ValueError(f"PDF path is not a file: {self.file_path}") if not self.file_path.lower().endswith(".pdf"): raise ValueError("Only PDF files are supported.") def _analyze_page(self, page, page_number): text = self._extract_text(page) words = self._extract_words(page) tables = self._extract_tables(page) lines = self._extract_lines(page) return { "page_number": page_number, "width": self._safe_float(page.width), "height": self._safe_float(page.height), "text": text, "words": words, "word_count": len(words), "tables": tables, "table_count": len(tables), "lines": lines, "line_count": len(lines), } def _extract_text(self, page): """Return visually arranged text without changing its content.""" try: return ( page.extract_text( layout=True, x_tolerance=2, y_tolerance=3, x_density=7.25, y_density=13, ) or "" ) except Exception: try: return page.extract_text(layout=True) or "" except Exception: return "" def _extract_words(self, page): """ Return normalized pdfplumber words. Coordinates are retained because the parser can use them for fields whose visual position is meaningful. No page-specific coordinates are hardcoded here. """ settings = { "x_tolerance": 2, "y_tolerance": 3, "keep_blank_chars": False, "use_text_flow": False, } try: raw_words = page.extract_words( extra_attrs=["fontname", "size"], **settings, ) except Exception: try: raw_words = page.extract_words(**settings) except Exception: raw_words = [] words = [] for word in raw_words or []: text = str(word.get("text", "")).strip() if not text: continue words.append( { "text": text, "x0": self._safe_float(word.get("x0")), "x1": self._safe_float(word.get("x1")), "top": self._safe_float(word.get("top")), "bottom": self._safe_float(word.get("bottom")), "doctop": self._safe_float(word.get("doctop")), "width": self._safe_float(word.get("width")), "height": self._safe_float(word.get("height")), "fontname": str(word.get("fontname", "")), "size": self._safe_float(word.get("size")), } ) # Deterministic visual order. This is intentionally separate from # pdfplumber's text-flow order because Shipping Bills are layouts. words.sort(key=lambda w: (round(w["top"], 2), w["x0"])) return words def _extract_tables(self, page): """ Try several pdfplumber table strategies and retain unique tables. Shipping Bills vary between bordered and whitespace-aligned tables, so one table strategy cannot be assumed to work for every PDF. """ candidates = [] strategies = [ {}, { "vertical_strategy": "lines", "horizontal_strategy": "lines", "snap_tolerance": 3, "join_tolerance": 3, "intersection_tolerance": 5, "edge_min_length": 3, }, { "vertical_strategy": "text", "horizontal_strategy": "text", "text_x_tolerance": 2, "text_y_tolerance": 3, "intersection_tolerance": 5, "snap_tolerance": 3, "join_tolerance": 3, }, ] for settings in strategies: try: extracted = page.extract_tables(table_settings=settings) candidates.extend(self._clean_tables(extracted)) except Exception: continue # Default/line detection is normally sufficient. Text strategy # is retained as a fallback for unruled tables. return self._deduplicate_tables(candidates) def _clean_tables(self, tables): cleaned = [] for table in tables or []: if not table: continue rows = [] for row in table: if row is None: continue cleaned_row = [] for cell in row: if cell is None: value = "" else: value = str(cell).replace("\r\n", "\n").replace("\r", "\n") normalized_lines = [] for line in value.split("\n"): line = " ".join(line.split()) if line: normalized_lines.append(line) value = "\n".join(normalized_lines).strip() cleaned_row.append(value) if any(cell != "" for cell in cleaned_row): rows.append(cleaned_row) if rows: cleaned.append(rows) return cleaned def _deduplicate_tables(self, tables): unique = [] seen = set() for table in tables: # Normalize row lengths so harmless trailing empty cells do not # create different signatures. normalized = tuple(tuple(row) for row in table) signature = repr(normalized) if signature in seen: continue seen.add(signature) unique.append(table) return unique def _extract_lines(self, page): try: page_lines = page.lines or [] except Exception: page_lines = [] lines = [] for line in page_lines: lines.append( { "x0": self._safe_float(line.get("x0")), "x1": self._safe_float(line.get("x1")), "top": self._safe_float(line.get("top")), "bottom": self._safe_float(line.get("bottom")), "width": self._safe_float(line.get("width")), "height": self._safe_float(line.get("height")), } ) return lines @staticmethod def _safe_float(value): try: return float(value) if value is not None else 0.0 except (TypeError, ValueError): return 0.0 class ShippingBillParser: """Deterministic, layout-aware Shipping Bill parser.""" 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']} EMIT_EMPTY_CHILD_SCHEMA_ROWS: ClassVar[bool] = True 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", } LABELS: ClassVar[dict[str, list[str]]] = {'port_code': ['PORT CODE'], 'shipping_bill_no': ['SB NO', 'SB NO.'], 'shipping_bill_date': ['SB DATE'], 'iec': ['IEC', 'IEC/BR'], 'cb_code': ['CB CODE'], 'inv': ['INV'], 'item': ['ITEM'], 'cont': ['CONT'], 'pkg': ['PKG'], 'gwt': ['G.WT', 'GWT', 'GROSS WEIGHT'], 'gwt_unit': ['G.WT UNIT', 'UNIT'], 'port_of_loading': ['PORT OF LOADING'], 'state_of_origin': ['STATE OF ORIGIN'], 'port_of_finaldstn': ['PORT OF FINAL DESTINATION'], 'port_of_discharge': ['PORT OF DISCHARGE'], 'cntry_of_finaldstn': ['COUNTRY OF FINAL DESTINATION', 'COUNTRY OF FINALDESTINATION', 'COUNTRY OF FINALDESTINATIO'], 'cntry_of_discharge': ['COUNTRY OF DISCHARGE'], 'p_1b_exporter_name': ["EXPORTER'S NAME & ADDRESS"], 'p_1b_consignee_name': ['CONSIGNEE NAME & ADDRESS'], 'p_1b_cb_name': ['CB NAME'], 'p_1b_type': ['TYPE'], 'p_1b_gstin': ['GSTIN', 'GSTIN / TYPE'], 'p_1b_ad_code': ['AD CODE'], 'p_1b_forex_ac_no': ['FOREX BANK A/C NO', 'FOREX BANK A/C NO.'], 'p_1b_dbk_ac_no': ['DBK BANK A/C NO', 'DBK BANK A/C NO.'], 'p_1b_aeo': ['AEO'], 'p_1b_ifsc_no': ['IFSC NO', 'IFSC NO.'], 'p_1b_rbi_wr_no': ['RBI WAIVER NO', 'RBI WAIVER NO.'], 'p_1b_rbi_wr_dt': ['RBI WAIVER DT', 'RBI WAIVER DATE'], 'p_1c_fob_val': ['FOB VALUE'], 'p_1c_cess': ['CESS', '9.CESS'], 'p_1c_duty': ['DUTY', '8.DUTY'], 'p_1c_freight': ['FREIGHT'], 'p_1c_discount': ['DISCOUNT'], 'p_1c_deductions': ['DEDUCTIONS', 'DEDUCT'], 'p_1c_insurance': ['INSURANCE'], 'p_1c_com': ['COMMISSION', 'COM'], 'p_1c_p_c': ['P & C', 'P.C', 'P C'], 'p_1d_dbk': ['DBK'], 'p_1d_rodtep_amt': ['RODTEP'], 'p_1d_rosctle_amt': ['ROSCTL', 'ROSL'], 'p_1j_subm_dt': ['SUBMISSION'], 'p_1i_assess_dt': ['ASSESSMENT'], 'p_1i_assess_time': ['ASSESSMENT'], 'p_1i_exmn_dt': ['EXAMINATION'], 'p_1i_exmn_time': ['EXAMINATION'], 'p_1i_subm_time': ['SUBMISSION'], 'p_1i_leo_time': ['LEO TIME', 'LEO'], 'p_1i_sez_unit_data': ['SEZ UNIT DETAILS', 'SEZ UNIT'], 'p_1i_leo_no': ['LEO NO'], 'p_1i_leo_dt': ['LEO DATE'], 'p_1i_exmn_dt': ['EXAMINATION'], 'p_1i_brc_realzn_dt': ['BRC REALISATION DATE', 'BRC REALIZATION DATE']} TABLE_ALIASES: ClassVar[dict[str, list[str]]] = {'manifest_details': ['MANIFEST', 'MAWB', 'CIN'], 'annex_details': ['ANNEX', 'SEAL TYPE', 'NATURE OF CARGO'], 'invoice_details': ['INVOICE', 'INV NO', 'INVOICE NO'], 'item_details': ['ITEM DETAILS', 'HS CODE', 'CTH', 'ITEM DESCRIPTION'], 'single_window_declaration': ['SINGLE WINDOW', 'QUALIFIER', 'INFO CODE', 'INFO'], 'container_details': ['CONTAINER', 'CONTAINER NO'], 'challan_details': ['CHALLAN'], 'aa__dfia_licence_details': ['AA', 'DFIA', 'LICENCE'], 'drawback_and_rosl_claim': ['DRAWBACK', 'ROSL', 'RODTEP'], 'job_details': ['JOB DETAILS'], 'supporting_documents': ['SUPPORTING DOCUMENT'], 'ar4_details': ['AR4'], 'third_party_details': ['THIRD PARTY'], 'item_manufacturerproducergrower_details': ['MANUFACTURER', 'PRODUCER', 'GROWER'], 'rodtep_details': ['RODTEP'], 'single_window_declaration_constituents': ['CONSTITUENT'], 'single_window_declaration_control': ['CONTROL', 'CONTROL RESULT'], 'reexport_details': ['REEXPORT DETAILS', 'BE SITE ID', 'BE NUMBER']} SECTION_HEADINGS: ClassVar[dict[str, list[str]]] = { "challan_details": ["CHALLAN DETAILS", "1.SR.NO 2.CHALLAN NO 3.PAYMT DT 4.AMOUNT"], "aa__dfia_licence_details": ["AA / DFIA LICENCE DETAILS"], "drawback_and_rosl_claim": ["DRAWBACK & ROSL CLAIM"], "job_details": ["JOBBING DETAILS"], "single_window_declaration_constituents": ["SINGLE WINDOW DECLARATION - CONSTITUENTS"], "single_window_declaration_control": ["SINGLE WINDOW DECLARATION - CONTROL"], "supporting_documents": ["SUPPORTING DOCUMENTS"], "ar4_details": ["AR4 DETAILS"], "third_party_details": ["THIRD PARTY DETAILS"], "item_manufacturerproducergrower_details": ["ITEM MANUFACTURER/PRODUCER/GROWER DETAILS"], "rodtep_details": ["RODTEP DETAILS"], "container_details": ["CONTAINER DETAILS"], } PARENT_FIELD_ALIASES: ClassVar[dict[str, list[str]]] = {'state_of_origin': ['STATE OF ORIGIN', 'ORIGIN STATE'], 'buyer_name': ['BUYER NAME', 'BUYER', 'BUYER NAME & ADDRESS'], 'p_1a_mode': ['MODE', 'MODE OF TRANSPORT'], 'p_1a_meis': ['MEIS'], 'p_1a_dfrc': ['DFRC'], 'p_1a_assess': ['ASSESS', 'ASSESSMENT STATUS'], 'p_1a_dbk': ['DBK'], 'p_1a_re_exp': ['RE-EXP', 'RE EXP', 'REEXP', 'RE-EXPORT'], 'p_1a_exam': ['EXMN', 'EXAM', 'EXAMINATION STATUS'], 'p_1a_rodtp': ['RODTP', 'RODTEP'], 'p_1a_lut': ['LUT'], 'p_1a_job': ['JOBBING', 'JOB'], 'p_1a_licence': ['LICENCE', 'LICENSE'], 'p_1b_exporter_name': ["EXPORTER'S NAME & ADDRESS", 'EXPORTER NAME & ADDRESS'], 'p_1b_cb_name': ['CB NAME', 'CUSTOMS BROKER', 'CUSTOMS BROKER NAME'], 'p_1b_dbk_ac_no': ['DBK BANK A/C NO', 'DBK BANK A/C NO.'], 'p_1b_aeo': ['AEO', 'AEO NO'], 'p_1b_ifsc_no': ['IFSC NO', 'IFSC NO.'], 'p_1b_type': ['TYPE'], 'p_1b_rbi_wr_no': ['RBI WAIVER NO', 'RBI WAIVER NO.'], 'p_1b_gstin': ['GSTIN', 'GSTIN / TYPE', 'GSTN', 'GSTN ID'], 'p_1b_ad_code': ['AD CODE', 'ADCODE'], 'p_1b_rbi_wr_dt': ['RBI WAIVER DT', 'RBI WAIVER DATE'], 'p_1b_forex_ac_no': ['FOREX BANK A/C NO', 'FOREX BANK A/C NO.'], 'p_1c_fob_val': ['FOB VALUE'], 'p_1c_com': ['COM', 'COMMISSION', 'COMMISSION VALUE'], 'p_1c_cess': ['CESS', 'CESS AMOUNT'], 'p_1c_freight': ['FREIGHT'], 'p_1c_deductions': ['DEDUCTIONS', 'DEDUCT'], 'p_1c_insurance': ['INSURANCE'], 'p_1c_p_c': ['P/C', 'P & C', 'P.C', 'P C'], 'p_1c_discount': ['DISCOUNT'], 'p_1c_duty': ['DUTY', 'DUTY AMOUNT'], 'p_1d_dbk': ['DBK CLAIM', 'DBK'], 'p_1d_rodtep_amt': ['RODTEP AMT', 'RODTEP AMOUNT'], 'p_1d_igst_amt': ['IGST AMT', 'IGST AMOUNT'], 'p_1d_rosctle_amt': ['ROSCTL AMT', 'ROSCTL AMOUNT', 'ROSL AMT'], 'p_1d_cess_amt': ['CESS AMT', 'CESS AMOUNT'], 'p_1d_igst_val': ['IGST VALUE', 'IGST VAL'], 'p_1j_subm_dt': ['SUBMISSION', 'SUBMISSION DATE'], 'p_1i_subm_time': ['SUBMISSION TIME'], 'p_1i_leo_no': ['LEO NO', 'LEO NUMBER'], 'p_1i_assess_dt': ['ASSESSMENT DATE', 'ASSESS DT'], 'p_1i_assess_time': ['ASSESSMENT TIME', 'ASSESS TIME'], 'p_1i_leo_dt': ['LEO DATE', 'LEO DT'], 'p_1i_exmn_dt': ['EXAMINATION DATE', 'EXAMINATION DT', 'EXMN DATE'], 'p_1i_exmn_time': ['EXAMINATION TIME', 'EXMN TIME'], 'p_1i_leo_time': ['LEO TIME'], 'p_1i_brc_realzn_dt': ['BRC REALISATION DATE', 'BRC REALIZATION DATE', 'BRC REALZN DT'], 'p_1i_sez_unit_data': ['SEZ UNIT DETAILS', 'SEZ UNIT DATA', 'SEZ UNIT']} CHILD_FIELD_ALIASES: ClassVar[dict[str, dict[str, list[str]]]] = {'manifest_details': {'p_1e_mawb_no': ['MAWB NO', 'MAWB NUMBER', 'MASTER AWB', 'MASTER AWB NO'], 'p_1e_noc': ['N.O.C.', 'NOC', 'NO C', 'NOC NO'], 'p_1e_mawb_dt': ['MAWB DT', 'MAWB DATE', 'MAWB DT'], 'p_1e_cin_no': ['CIN NO', 'CIN NUMBER'], 'p_1e_hawb_no': ['HAWB NO', 'HAWB NUMBER', 'HOUSE AWB'], 'p_1e_cin_dt': ['CIN DT', 'CIN DATE'], 'p_1e_hawb_dt': ['HAWB DT', 'HAWB DATE'], 'p_1e_cin_site_id': ['CIN SITE ID', 'SITE ID', 'CIN SITE']}, 'challan_details': {'p_4h_challan_serial_no': ['CHALLAN SERIAL NO', 'CHALLAN SR NO', 'SERIAL NO', 'SERIAL NUMBER'], 'p_4h_challan_no': ['CHALLAN NO', 'CHALLAN NUMBER'], 'p_4h_paymt_dt': ['PAYMENT DT', 'PAYMENT DATE', 'PAYMT DT', 'PAYMT DATE'], 'p_4h_challan_amount': ['CHALLAN AMOUNT', 'AMOUNT', 'PAYMENT AMOUNT']}, 'annex_details': {'p_1i_seal_typ': ['SEAL TYPE', 'SEAL TYP', 'SEAL'], 'p_1i_loose_pkts': ['LOOSE PKTS', 'LOOSE PACKETS', 'LOOSE PACKAGES'], 'p_1i_nature_cargo': ['NATURE OF CARGO', 'NATURE CARGO', 'CARGO NATURE'], 'p_1i_marks_numbers': ['MARKS NUMBERS', 'MARKS & NUMBERS', 'MARKS AND NUMBERS', 'MARKS'], 'p_1i_no_of_pkgs': ['NO OF PKGS', 'NO OF PACKAGES', 'NO OF PACKETS', 'PACKAGES', 'PACKETS'], 'p_1i_no_of_containers': ['NO OF CONTAINERS', 'NUMBER OF CONTAINERS', 'CONTAINERS']}, 'invoice_details': {'inv_sn': ['INV SN', 'INV SNO', 'INVOICE SN', 'INVOICE SERIAL NO'], 'p_2a_inv_no': ['INV NO', 'INVOICE NO', 'INVOICE NUMBER'], 'p_2a_inv_dt': ['INV DT', 'INVOICE DATE', 'INV DATE'], 'p_2a_ad_code': ['AD CODE', 'ADCODE'], 'p_2a_invterm': ['INVTERM', 'INV TERM', 'INVOICE TERM', 'INCO TERM', 'INCOTERM'], 'p_2b_exporter_name': ['EXPORTER NAME', "EXPORTER'S NAME"], 'p_2b_buyer_addr': ['BUYER ADDRESS', 'BUYER ADDR'], 'p_2b_exporter_addr': ['EXPORTER ADDRESS', 'EXPORTER ADDR'], 'p_2b_buyer_name': ['BUYER NAME', 'BUYER', 'CONSIGNEE NAME'], 'p_2c_invoice_value': ['INVOICE VALUE', 'INVOICE AMOUNT', 'INV VALUE'], 'p_2c_freight': ['FREIGHT'], 'p_2c_discount': ['DISCOUNT'], 'p_2c_deduct': ['DEDUCT', 'DEDUCTIONS'], 'p_2c_exchng_rate_desc': ['EXCHANGE RATE', 'EXCHNG RATE', 'EXCHANGE RATE DESC', 'EXCHNG RATE DESC'], 'p_2c_exchng_rate': ['EXCHANGE RATE VALUE', 'EXCHNG RATE VALUE'], 'p_2c_invoice_curr': ['INVOICE CURR', 'INVOICE CURRENCY', 'INV CURRENCY', 'CURRENCY'], 'p_2c_frieght_curr': ['FREIGHT CURR', 'FREIGHT CURRENCY'], 'p_2c_discount_curr': ['DISCOUNT CURR', 'DISCOUNT CURRENCY'], 'p_2c_deduct_curr': ['DEDUCT CURR', 'DEDUCT CURRENCY'], 'p_2c_fob_val': ['FOB VALUE', 'FOB VAL'], 'p_2c_insurance': ['INSURANCE'], 'p_2c_commison': ['COMMISSION', 'COMMISON', 'COMM'], 'p_2c_p_c': ['P/C', 'P & C', 'P.C'], 'p_2c_fob_curr': ['FOB CURR', 'FOB CURRENCY'], 'p_2c_insurance_curr': ['INSURANCE CURR', 'INSURANCE CURRENCY'], 'p_2c_commison_curr': ['COMMISSION CURR', 'COMMISON CURR', 'COMMISSION CURRENCY'], 'p_2c_p_c_curr': ['P/C CURR', 'P & C CURR', 'P C CURR']}, 'aa__dfia_licence_details': {'p_4b_invsn': ['INVSN', 'INV SN', 'INVOICE SN'], 'p_4b_itmsn': ['ITMSN', 'ITEM SN', 'ITEM SNO'], 'p_4b_licence_no': ['LICENCE NO', 'LICENSE NO', 'LICENCE NUMBER', 'LICENSE NUMBER'], 'p_4b_licence_dt': ['LICENCE DT', 'LICENCE DATE', 'LICENSE DATE'], 'p_4b_licence_value': ['LICENCE VALUE', 'LICENSE VALUE'], 'p_4b_licence_qty': ['LICENCE QTY', 'LICENCE QUANTITY', 'LICENSE QTY', 'LICENSE QUANTITY']}, 'item_details': {'p_3a_invsno': ['INVSNO', 'INV SNO', 'INV SN', 'INVOICE SNO'], 'p_3a_itemsn': ['ITEMSN', 'ITEM SNO', 'ITEM SN', 'ITEM SERIAL NO'], 'p_3a_cth': ['CTH', 'HS CODE', 'HS CD', 'HSN', 'TARIFF CODE'], 'p_3a_item_desc': ['ITEM DESCRIPTION', 'DESCRIPTION', 'ITEM DESC', 'DESCRIPTION OF GOODS'], 'p_3a_qty': ['QTY', 'QUANTITY'], 'p_3a_uqc': ['UQC', 'UNIT'], 'p_3a_rate': ['RATE'], 'p_3a_value': ['VALUE', 'ITEM VALUE'], 'p_3a_fob': ['FOB'], 'p_3a_pmv': ['PMV'], 'p_3a_duty_amt': ['DUTY AMT', 'DUTY AMOUNT'], 'p_3a_cess_rate': ['CESS RATE'], 'p_3a_cess_amt': ['CESS AMT', 'CESS AMOUNT'], 'p_3a_dbk_claimed': ['DBK CLAIMED', 'DBK'], 'p_3a_igststat': ['IGST STAT', 'IGST STATUS', 'IGST STAT.'], 'p_3a_igst_val': ['IGST VALUE', 'IGST VAL'], 'p_3a_igst_amt': ['IGST AMT', 'IGST AMOUNT'], 'p_3a_schcod': ['SCHCOD', 'SCHEME CODE', 'SCH CODE'], 'p_3a_scheme_desc': ['SCHEME DESC', 'SCHEME DESCRIPTION'], 'p_3a_sqc_mst': ['SQC MST', 'SQC MASTER', 'SQC'], 'p_3a_sqc_uqc': ['SQC UQC'], 'p_3a_state_of_origin': ['STATE OF ORIGIN'], 'p_3a_district_of_origin': ['DISTRICT OF ORIGIN', 'DISTRICT'], 'p_3a_pt_abroad': ['PT ABROAD', 'PORT ABROAD'], 'p_3a_comp_cess': ['COMP CESS', 'COMPENSATION CESS'], 'p_3a_end_use': ['END USE', 'END-USE'], 'p_3a_benefit_availd': ['BENEFIT AVAILD', 'BENEFIT AVAILABLE', 'BENEFIT AVAILED'], 'p_3a_reward_benefit': ['REWARD BENEFIT', 'REWARD'], 'p_3a_third_party_item': ['THIRD PARTY ITEM', 'THIRD PARTY']}, 'drawback_and_rosl_claim': {'p_3b_invsno': ['INVSNO', 'INV SNO', 'INV SN'], 'p_3b_itemsn': ['ITEMSN', 'ITEM SNO', 'ITEM SN'], 'p_3b_dbksn': ['DBKSN', 'DBK SNO', 'DBK SN'], 'p_3b_dbk_rate': ['DBK RATE'], 'p_3b_dbk_amt': ['DBK AMT', 'DBK AMOUNT'], 'p_3b_rosl_rate': ['ROSL RATE', 'ROSCTL RATE'], 'p_3b_rosl_amt': ['ROSL AMT', 'ROSL AMOUNT', 'ROSCTL AMT', 'ROSCTL AMOUNT']}, 'job_details': {'p_3c_invsn': ['INVSN', 'INV SN'], 'p_3c_itmsn': ['ITMSN', 'ITEM SN'], 'p_3c_job_no': ['JOB NO', 'JOB NUMBER'], 'p_3c_job_date': ['JOB DATE', 'JOB DT'], 'p_3c_job_amount': ['JOB AMOUNT', 'JOB VALUE', 'AMOUNT']}, 'single_window_declaration': {'p_4d_invsn': ['INVSN', 'INV SN', 'INVOICE SN'], 'p_4d_itmsn': ['ITMSN', 'ITEM SN', 'ITEM SNO'], 'p_4d_info': ['INFO'], 'p_4d_qualifier': ['QUALIFIER'], 'p_4d_info_cd': ['INFO CODE', 'INFO CD', 'CODE'], 'p_4d_info_text': ['INFO TEXT', 'TEXT'], 'p_4d_info_msr': ['INFO MSR', 'MEASURE', 'MEASUREMENT'], 'p_4d_uqc': ['UQC', 'UNIT']}, 'single_window_declaration_constituents': {'p_4e_invsn': ['INVSN', 'INV SN'], 'p_4e_itmsn': ['ITMSN', 'ITEM SN'], 'p_4e_constituent': ['CONSTITUENT', 'CONSTITUENTS', 'CONSTITUENT NAME'], 'p_4e_percentage': ['PERCENTAGE', 'PERCENT', '%']}, 'single_window_declaration_control': {'p_4f_invsn': ['INVSN', 'INV SN'], 'p_4f_itmsn': ['ITMSN', 'ITEM SN'], 'p_4f_control': ['CONTROL', 'CONTROL TYPE'], 'p_4f_result': ['RESULT', 'CONTROL RESULT']}, 'supporting_documents': {'p_4g_invsn': ['INVSN', 'INV SN'], 'p_4g_itmsn': ['ITMSN', 'ITEM SN'], 'p_4g_doc_code': ['DOC CODE', 'DOCUMENT CODE'], 'p_4g_doc_no': ['DOC NO', 'DOCUMENT NO', 'DOCUMENT NUMBER'], 'p_4g_doc_date': ['DOC DATE', 'DOCUMENT DATE'], 'p_4g_remarks': ['REMARKS', 'REMARK']}, 'ar4_details': {'p_4j_invsn': ['INVSN', 'INV SN'], 'p_4j_itmsno': ['ITMSNO', 'ITEM SNO', 'ITEM SN'], 'p_4j_ar4_number': ['AR4 NUMBER', 'AR4 NO', 'AR4 NO.'], 'p_4j_ar4_date': ['AR4 DATE', 'AR4 DT'], 'p_4j_commissionerate': ['COMMISSIONERATE', 'COMMISSIONERATE NAME'], 'p_4j_division': ['DIVISION'], 'p_4j_range': ['RANGE']}, 'third_party_details': {'p_4k_invsn': ['INVSN', 'INV SN'], 'p_4k_itmsn': ['ITMSN', 'ITEM SN'], 'p_4k_iec': ['IEC', 'IEC NO'], 'p_4k_exporter_name': ['EXPORTER NAME', 'THIRD PARTY EXPORTER'], 'p_4k_address': ['ADDRESS', 'EXPORTER ADDRESS'], 'p_4k_gstn_id_type': ['GSTN ID TYPE', 'GSTIN ID TYPE', 'GSTN TYPE']}, 'item_manufacturerproducergrower_details': {'p_4l_invsn': ['INVSN', 'INV SN'], 'p_4l_itmsn': ['ITMSN', 'ITEM SN'], 'p_4l_type': ['TYPE'], 'p_4l_manufact_cd': ['MANUFACT CD', 'MANUFACTURER CODE', 'MANUFACTURE CODE'], 'p_4l_source_state': ['SOURCE STATE', 'STATE OF SOURCE'], 'p_4l_trans_cy': ['TRANS CY', 'TRANSACTION COUNTRY', 'TRANS COUNTRY'], 'p_4l_address': ['ADDRESS', 'MANUFACTURER ADDRESS']}, 'rodtep_details': {'p_4m_invsn': ['INVSN', 'INV SN'], 'p_4m_itmsn': ['ITMSN', 'ITEM SN'], 'p_4m_quantity': ['QUANTITY', 'QTY'], 'p_4m_uqc': ['UQC', 'UNIT'], 'p_4m_no_of_units': ['NO OF UNITS', 'NUMBER OF UNITS', 'UNITS'], 'p_4m_value': ['VALUE', 'RODTEP VALUE']}, 'container_details': {'p_4i_container': ['CONTAINER NO', 'CONTAINER NUMBER', 'CONTAINER'], 'p_4i_cont_seal': ['CONTAINER SEAL', 'CONT SEAL', 'SEAL NO', 'SEAL NUMBER'], 'p_4i_cont_dt': ['CONTAINER DT', 'CONTAINER DATE', 'CONT DT', 'CONT DATE'], 'p_4i_cont_sno': ['CONTAINER SNO', 'CONTAINER S NO', 'CONTAINER SR NO', 'CONT SNO', 'CONT S NO']}} KNOWN_LABELS: ClassVar[set[str]] = {'PORT CODE', 'SB NO', 'SB NO.', 'SB DATE', 'IEC', 'IEC/BR', 'CB CODE', 'INV', 'ITEM', 'CONT', 'PKG', 'G.WT', 'GWT', 'GROSS WEIGHT', 'G.WT UNIT', 'UNIT', 'PORT OF LOADING', 'STATE OF ORIGIN', 'PORT OF FINAL DESTINATION', 'PORT OF DISCHARGE', 'COUNTRY OF FINAL DESTINATION', 'COUNTRY OF FINALDESTINATION', 'COUNTRY OF FINALDESTINATIO', 'COUNTRY OF DISCHARGE', "EXPORTER'S NAME & ADDRESS", "1.EXPORTER'S NAME & ADDRESS", 'EXPORTER NAME & ADDRESS', 'CONSIGNEE NAME & ADDRESS', '7.CONSIGNEE NAME & ADDRESS', 'CB NAME', 'TYPE', 'GSTIN', 'GSTIN / TYPE', 'AD CODE', 'FOREX BANK A/C NO', 'FOREX BANK A/C NO.', 'FOB VALUE', 'FREIGHT', 'DISCOUNT', 'DEDUCTIONS', 'DEDUCT', 'INSURANCE', 'COMMISSION', 'COM', 'P & C', 'P.C', 'P C', 'DBK', 'RODTEP', 'ROSCTL', 'ROSL', 'SUBMISSION', 'LEO NO', 'LEO DATE', 'LEO TIME', 'EXAMINATION', 'BRC REALISATION DATE', 'BRC REALIZATION DATE', 'MODE', 'ASSESS', 'RE-EXP', 'RE EXP', 'REEXP', 'EXMN', 'EXAM', 'LUT', 'JOB', 'LICENCE', 'LICENSE', 'DFRC'} # ================================================================ # document-page selection # ================================================================ def _select_shipping_bill_pages(self, pages: list[dict[str, Any]]) -> list[dict[str, Any]]: """Select the actual Shipping Bill page range before field extraction. A PDF can contain unrelated invoices, packing lists, scans, or other attachments before/after the Customs Shipping Bill. The parser must not expose those pages to the field extractors because they can contain labels that look like valid Shipping Bill fields. The first Shipping Bill page is identified from the document header, not from a fixed physical page number. Once found, the embedded ``Page X Of N`` numbering is used when available to select the complete Customs document. If that numbering is unavailable, contiguous Shipping Bill pages are retained using the same structural signals. """ if not pages: return [] def page_text(page: dict[str, Any]) -> str: return self._clean(page.get("text") or "") def compact(text: str) -> str: return re.sub(r"[^a-z0-9]+", "", text.lower()) start_index = None best_score = -1 for index, page in enumerate(pages): text = page_text(page) c = compact(text) if not c: # A blank/image-only cover cannot be the Shipping Bill start. continue score = 0 if "indiancustomsedisystem" in c: score += 6 if "centralboardofindirecttaxesandcustoms" in c: score += 4 if "partishippingbillsummary" in c: score += 8 if "portcode" in c and "sbno" in c and "sbdate" in c: score += 6 if "iecbr" in c: score += 3 if "cbcode" in c: score += 3 # A page that contains the strong Customs header is preferred over # any later page that only happens to contain a Shipping Bill label. if score > best_score: best_score = score start_index = index # Require the actual first-page identity. This keeps an unrelated # invoice/supporting page from becoming the parser's starting point. if start_index is None or best_score < 14: return pages selected_end = len(pages) # Read the embedded Customs page count. Depending on the PDF text # extraction order it may appear as ``Page 5 Of 6`` or ``Page Of * 6``. total_pages = None page_number_pattern = re.compile(r"\bPage\s+(\d+)\s+Of\s+(\d+)\b", re.I) for page in pages[start_index:]: text = page_text(page) match = page_number_pattern.search(text) if match: total_pages = int(match.group(2)) break if total_pages and total_pages > 0: selected_end = min(len(pages), start_index + total_pages) else: # Fallback for PDFs where page numbering is not text-extractable. # Continue while pages retain strong Shipping Bill identity/section # signals; stop at the first unrelated page after the document has # started. selected_end = start_index + 1 section_markers = ( "partiishippingbillsummary", "partiiinvoicedetails", "partiiiitemdetails", "partivexportschemedetails", "partvdeclarations", "singlewindowdeclaration", "manifestdetails", "invoicedetails", "itemdetails", ) for index in range(start_index + 1, len(pages)): c = compact(page_text(pages[index])) if ( "indiancustomsedisystem" in c or any(marker in c for marker in section_markers) or "shippingbillsummary" in c ): selected_end = index + 1 continue break selected = pages[start_index:selected_end] self.trace.append({ "stage": "page_selection", "start_page": int(pages[start_index].get("page_number", start_index + 1)), "selected_page_count": len(selected), "source_page_count": len(pages), "reason": "Shipping Bill header detected; embedded Customs page count used when available", }) return selected # ================================================================ # public entry point # ================================================================ def __init__(self, pages: list[dict[str, Any]]): self.trace: list[dict[str, Any]] = [] self.pages = self._select_shipping_bill_pages(pages or []) self.page_text = {int(p.get("page_number", i + 1)): str(p.get("text") or "") for i, p in enumerate(self.pages)} self.full_text = "\n".join(self.page_text.values()) self.words = self._all_words() self.page_words: dict[int, list[dict[str, Any]]] = {} for w in self.words: self.page_words.setdefault(w["_page"], []).append(w) for p in self.page_words: self.page_words[p].sort(key=lambda w: (w["y0"], w["x0"])) self.rows = self._build_rows(self.words) self.page_rows: dict[int, list[dict[str, Any]]] = {} for r in self.rows: self.page_rows.setdefault(r["_page"], []).append(r) self.tables = self._all_tables() self._label_cache = {} self._candidate_cache = {} def parse(self) -> dict[str, Any]: result = self._parse_base() self._final_reconciliation(result) self._finalize_parent_fields(result) self._finalize_children(result) self._schema_complete(result) return {"shipping_bill_json": result, "_trace": self.trace} def _parse_base(self) -> dict[str, Any]: result = self._empty_result() # Pass 1: parent/header/party semantic extraction. self._extract_parent_semantic(result) # Pass 2: explicit invoice/item sections. self._extract_invoices(result) self._extract_items(result) # Pass 3: known Shipping Bill structural sections. self._extract_known_sections(result) # Pass 4: table-first generic recovery. self._recover_from_generic_tables(result) # Pass 5: remaining semantic fields. self._recover_missing_parent_fields(result) self._recover_missing_child_fields(result) # Pass 6: generic table recovery again, enriching partial rows. self._recover_from_generic_tables(result, second_pass=True) # Pass 7: deterministic child reconciliation. self._reconcile_child_rows(result) # Pass 8: schema/type contract. self._finalize_parent_fields(result) self._finalize_children(result) self._schema_complete(result) return result # ================================================================ # document model # ================================================================ def _all_words(self) -> list[dict[str, Any]]: out = [] for page_no, page in enumerate(self.pages, 1): for raw in page.get("words", []) or []: text = self._clean(raw.get("text", "")) if not text: continue w = dict(raw) w["_page"] = page_no w["_text"] = text w["_norm"] = self._norm(text) w["x0"] = self._num(w.get("x0")) or 0.0 w["x1"] = self._num(w.get("x1")) or w["x0"] w["y0"] = self._num(w.get("top", w.get("y0"))) or 0.0 w["y1"] = self._num(w.get("bottom", w.get("y1"))) or w["y0"] out.append(w) # Text-only fallback remains available when coordinates are absent. if not out: for page_no, page in enumerate(self.pages, 1): for line_no, line in enumerate(str(page.get("text") or "").splitlines()): for m in re.finditer(r"\S+", line): out.append({"_page": page_no, "_text": m.group(), "_norm": self._norm(m.group()), "x0": float(m.start()), "x1": float(m.end()), "y0": float(line_no * 10), "y1": float(line_no * 10 + 8)}) return out def _build_rows(self, words: list[dict[str, Any]], tolerance: float = 4.0): rows = [] for w in sorted(words, key=lambda x: (x["_page"], x["y0"], x["x0"])): target = None for r in reversed(rows[-12:]): if r["_page"] != w["_page"]: continue if abs(r["_y"] - w["y0"]) <= tolerance: target = r break if target is None: target = {"_page": w["_page"], "_y": w["y0"], "words": []} rows.append(target) target["words"].append(w) for r in rows: r["words"].sort(key=lambda x: x["x0"]) r["text"] = " ".join(x["_text"] for x in r["words"]) return rows def _all_tables(self): out = [] for page_no, page in enumerate(self.pages, 1): for idx, table in enumerate(page.get("tables", []) or [], 1): rows = [[self._clean(c) for c in (row or [])] for row in table or []] if rows: out.append({"page": page_no, "index": idx, "rows": rows}) return out def _norm(self, value): return re.sub(r"[^a-z0-9]+", " ", str(value or "").lower()).strip() def _compact(self, value): return re.sub(r"[^a-z0-9]+", "", str(value or "").lower()) def _find_labels(self, aliases): aliases = list(aliases or []) key = tuple(sorted(self._compact(a) for a in aliases)) if key in self._label_cache: return self._label_cache[key] hits = [] for row in self.rows: words = row["words"] for i in range(len(words)): for alias in aliases: parts = self._norm(alias).split() if not parts or i + len(parts) > len(words): continue candidate = " ".join(w["_text"] for w in words[i:i+len(parts)]) if self._compact(candidate) == self._compact(alias): ws = words[i:i+len(parts)] hits.append({"page": row["_page"], "_page": row["_page"], "x0": ws[0]["x0"], "x1": ws[-1]["x1"], "y0": min(w["y0"] for w in ws), "y1": max(w["y1"] for w in ws), "_y": row["_y"], "text": candidate, "row": row, "alias": alias}) self._label_cache[key] = hits return hits def _best_label(self, aliases, page=None, after_y=None): hits = self._find_labels(aliases) if page is not None: hits = [h for h in hits if h["_page"] == page] if after_y is not None: hits = [h for h in hits if h["y0"] >= after_y] return hits[0] if hits else None def _all_known_aliases(self): out = [] for groups in (self.PARENT_FIELD_ALIASES, self.CHILD_FIELD_ALIASES): if isinstance(groups, dict): for v in groups.values(): if isinstance(v, dict): for aliases in v.values(): out.extend(aliases) else: out.extend(v) out.extend(sum((v for v in self.LABELS.values()), [])) out.extend(sum((v for v in self.TABLE_ALIASES.values()), [])) return out def _label_word_spans(self, words): spans = set() aliases = self._all_known_aliases() for i in range(len(words)): for alias in aliases: parts = self._norm(alias).split() if parts and i + len(parts) <= len(words): candidate = " ".join(w["_text"] for w in words[i:i+len(parts)]) if self._compact(candidate) == self._compact(alias): spans.update(range(i, i + len(parts))) return spans def _candidate_right(self, label, validator=None, max_dx=420): if not label: return None row = label["row"] spans = self._label_word_spans(row["words"]) candidates = [] for i, w in enumerate(row["words"]): if i in spans or w["x0"] < label["x1"] - 1: continue dx = w["x0"] - label["x1"] if dx < 0 or dx > max_dx: continue if self._looks_like_any_known_label(w["_text"]): continue if validator and not validator(w["_text"]): continue score = dx candidates.append((score, w["_text"], w)) if not candidates: return None return min(candidates, key=lambda x:x[0])[1] def _candidate_below(self, label, validator=None, max_dy=120, x_tol=220): if not label: return None best = None for row in self.page_rows.get(label["_page"], []): dy = row["_y"] - label["y1"] if dy <= 0 or dy > max_dy: continue for w in row["words"]: cx = (w["x0"] + w["x1"])/2 if abs(cx - label["x0"]) > x_tol: continue if self._looks_like_any_known_label(w["_text"]): continue if validator and not validator(w["_text"]): continue score = dy + abs(cx-label["x0"])*0.02 if best is None or score < best[0]: best=(score,w["_text"]) return best[1] if best else None def _semantic_value(self, aliases, validator=None, max_dx=420, max_dy=120): for label in self._find_labels(aliases): value = self._candidate_right(label, validator, max_dx) if value is not None: return value, label value = self._candidate_below(label, validator, max_dy) if value is not None: return value, label return None, None # ================================================================ # parent extraction # ================================================================ def _extract_parent_semantic(self, result): self._extract_header(result) self._extract_statuses(result) self._extract_locations(result) self._extract_parties(result) self._extract_financials(result) self._extract_dates_times(result) def _extract_header(self, result): # Header values are taken from the repeated labelled header block, not # from arbitrary nearby occurrences on later pages. head = self.full_text[:5000] m = re.search(r"Port\s+Code\s+SB\s+No\s+SB\s+Date\s+(\w{3,8}\d{1,3})\s+(\d{5,10})\s+(\d{1,2}-[A-Za-z]{3}-\d{2,4})", head, re.I) if m: result["port_code"] = m.group(1).upper() result["shipping_bill_no"] = m.group(2) result["shipping_bill_date"] = self._format_date(m.group(3)) m = re.search(r"IEC\s*/?Br\s+([0-9]{8,15})\s+(\d{1,4})", head, re.I) if m: result["iec"], result["iec_branch_code"] = m.group(1), m.group(2) m = re.search(r"CB\s+CODE\s+([A-Z0-9]{10,20})", head, re.I) if m: result["cb_code"] = m.group(1).upper() m = re.search(r"GSTIN\s*/?TYPE\s+([0-9A-Z]{15})\s+([A-Z]+)", head, re.I) if m: result["p_1b_gstin"] = m.group(1).upper() # Count row: Nos / INV / ITEM / CONT and PKG / G.WT / unit / weight. for t in self.tables: rows=t["rows"] for i,r in enumerate(rows): c=self._compact(self._row_text(r)) if "type" in c and "invitemcont" in c and i+1=3: result["inv"] = self._integer(nums[0]); result["item"] = self._integer(nums[1]); result["cont"] = self._integer(nums[2]) if "pkg" in c and "gwt" in c and i+1=len(rows): continue value=rows[i+1] headers=[self._compact(c) for c in row] values=[self._clean(c) for c in value] mapping={} for j,h in enumerate(headers): if not h: continue token=None if "mode" in h: token="p_1a_mode" elif "assess" in h: token="p_1a_assess" elif "exmn" in h or "exam" in h: token="p_1a_exam" elif "jobbing" in h: token="p_1a_job" elif "meis" in h: token="p_1a_meis" elif re.search(r"(^|\d)6dbk",h) or h.endswith("dbk"): token="p_1a_dbk" elif "rodtp" in h: token="p_1a_rodtp" elif "licence" in h or "license" in h: token="p_1a_licence" elif "dfrc" in h: token="p_1a_dfrc" elif "reexp" in h or "re exp" in h: token="p_1a_re_exp" elif h.endswith("lut") or "lut" in h: token="p_1a_lut" if token and j=5: break return (lines[0],"\n".join(lines[1:])) if lines else (None,None) def _extract_financials(self,result): for t in self.tables: rows=t["rows"] for i,row in enumerate(rows): compact=self._compact(self._row_text(row)) if "1fobvalue" in compact and "2freight" in compact and "3insuranc" in compact: vals=rows[i+1] if i+1 2/8). page_no = next((t["page"] for t in self.tables if row in t["rows"]), None) if page_no is not None: label_word = next((w for w in self.page_words.get(page_no, []) if self._compact(w["_text"]) == "4leo"), None) if label_word: word_leo_value = next((self._clean(w["_text"]) for w in self.page_words.get(page_no, []) if abs(w["y0"] - label_word["y0"]) <= 2.0 and w["x0"] > label_word["x1"] and re.fullmatch(r"\d+\s*/\s*\d+", self._clean(w["_text"]))), "") if word_leo_value: leo_value = word_leo_value if leo_value: result["p_1i_leo_no"] = leo_value for rr in rows[i+1:i+8]: event=""; date=""; tm="" # Use the event-table column boundaries instead of scanning # the complete row. The right side contains independent # LEO/BRC fields and must never populate Assessment. header=row date_start=next((k for k,x in enumerate(header) if self._compact(x)=="2date"), None) time_start=next((k for k,x in enumerate(header) if self._compact(x)=="3time"), None) right_start=next((k for k,x in enumerate(header) if self._compact(x)=="4leono"), len(rr)) if date_start is None: date_start=0 if time_start is None: time_start=date_start for j,v in enumerate(rr[:right_start]): v=self._clean(v) if not v: continue cv=self._compact(v) if "submission" in cv and j < date_start: event="submission" elif "assessment" in cv and j < date_start: event="assessment" elif "examination" in cv and j < date_start: event="examination" elif cv in {"9leo","leo"} and j < date_start: event="leo" for v in rr[date_start:time_start]: v=self._clean(v) if self._valid_date(v) and not date: date=v for v in rr[time_start:right_start]: v=self._clean(v) if self._valid_time(v) and not tm: tm=v # Right-hand LEO date and BRC date are independent labels. for j,v in enumerate(rr): cv=self._compact(v) if "6leodate" in cv: for x in rr[j+1:]: x=self._clean(x) if self._valid_date(x): result["p_1i_leo_dt"]=self._format_date(x); break if "brcrealisationdate" in cv or "brcrealizationdate" in cv: for x in rr[j+1:]: x=self._clean(x) if self._valid_date(x): result["p_1i_brc_realzn_dt"]=self._format_date(x); break if event=="submission": if date: result["p_1j_subm_dt"]=self._format_date(date) if tm: result["p_1i_subm_time"]=self._format_time(tm) elif event=="assessment": if date: result["p_1i_assess_dt"]=self._format_date(date) if tm: result["p_1i_assess_time"]=self._format_time(tm) elif event=="examination": if date: result["p_1i_exmn_dt"]=self._format_date(date) if tm: result["p_1i_exmn_time"]=self._format_time(tm) elif event=="leo": if date: result["p_1i_leo_dt"]=self._format_date(date) if tm: result["p_1i_leo_time"]=self._format_time(tm) break if self._is_missing("p_1j_subm_dt",result.get("p_1j_subm_dt")) and result.get("shipping_bill_date"): result["p_1j_subm_dt"]=result["shipping_bill_date"] # ================================================================ # explicit child sections # ================================================================ def _extract_invoices(self,result): rows=self._section_rows("PART - II - INVOICE DETAILS") if not rows: return header_i=next((i for i,r in enumerate(rows) if "invoiceno" in self._compact(self._row_text(r)) and "invterm" in self._compact(self._row_text(r))),None) if header_i is None: header_i=next((i for i,r in enumerate(rows) if "invoice" in self._compact(self._row_text(r)) and "sno" in self._compact(self._row_text(r))),None) inv={"doctype":"Invoice Detail"} if header_i is not None and header_i+1=len(row): continue hc=self._compact(h) v=self._clean(row[j]) if not v or not re.fullmatch(r"[A-Z]{3}",v): continue f=None for key,field in field_by_header.items(): if key in hc: f=field; break if f: inv[f]=v self._inherit_invoice_currencies(inv) def _map_invoice_value_grid(self,inv,headers,values): for j,hraw in enumerate(headers): h=self._compact(hraw) if not h or j>=len(values): continue raw=self._clean(values[j]) if not raw: continue f=None if "invoicevalue" in h: f="p_2c_invoice_value" elif "fobvalue" in h: f="p_2c_fob_val" elif "freight" in h: f="p_2c_freight" elif "insurance" in h: f="p_2c_insurance" elif "discount" in h: f="p_2c_discount" elif "commison" in h or "commission" in h: f="p_2c_commison" elif "deduct" in h: f="p_2c_deduct" elif "pc" in h or "packing" in h: f="p_2c_p_c" elif "exchange" in h: f="p_2c_exchng_rate_desc" if f: if f=="p_2c_exchng_rate_desc": inv[f]=raw m=re.search(r"(?:INR\s*)?([0-9]+(?:\.[0-9]+)?)\s*$",raw,re.I) if m: inv["p_2c_exchng_rate"]=self._float(m.group(1)) else: inv[f]=self._float(raw) if self._float(raw) is not None else raw # Currency cells follow the numeric row in the standard layout. currencies=[v for v in [self._clean(x) for x in values] if re.fullmatch(r"[A-Z]{3}",v or "")] if currencies: inv["p_2c_invoice_curr"]=currencies[0] inv["p_2c_fob_curr"]=currencies[1] if len(currencies)>1 else currencies[0] if len(currencies)>2: inv["p_2c_frieght_curr"]=currencies[2] if len(currencies)>3: inv["p_2c_discount_curr"]=currencies[3] def _extract_items(self,result): # Prefer PART III because it contains the full 29-field item row. rows=self._section_rows("PART - III - ITEM DETAILS") if not rows: return hi=next((i for i,r in enumerate(rows) if "itemsn" in self._compact(self._row_text(r)) and "hscd" in self._compact(self._row_text(r))),None) if hi is None: return data=rows[hi+1:] first=next((r for r in data if any(self._valid_hscode(x) for x in r)),None) if not first: return item={"doctype":"Item Detail"}; hm=self._header_map(rows[hi],"item_details"); self._map_row_to_record(item,"item_details",hm,first) # The description is a merged cell containing the wrapped PDF text. # Keep that cell intact instead of joining neighboring numeric cells. if len(first)>4 and self._clean(first[4]): item["p_3a_item_desc"]=self._clean(first[4]) for c in first: if self._valid_hscode(c): item["p_3a_cth"]=self._clean(c); break # Recover second half from the same page/table. for r in data: c=self._compact(self._row_text(r)) if "11dutyamt" in c and data.index(r)+1=len(rows): return header=rows[header_idx] hm=self._header_map(header,table_name) if len(hm)<2: return recs=[] for rr in rows[header_idx+1:]: if self._is_major_section_row(rr): break if any(self._compact(h) in self._compact(self._row_text(rr)) for h in headings): break if not any(self._clean(x) for x in rr): continue rec={"doctype":self.CHILD_DOCTYPES.get(table_name,"")} self._map_row_to_record(rec,table_name,hm,rr) if self._child_identity(table_name,rec): recs.append(rec) if recs: result[table_name]=recs return def _recover_from_generic_tables(self,result,second_pass=False): for t in self.tables: rows=t["rows"] for i,row in enumerate(rows): table_name=self._classify_table(rows[max(0,i-1):i+8]) if not table_name or table_name in {"manifest_details","annex_details","invoice_details","item_details","single_window_declaration"}: continue header=self._header_map(row,table_name) if len(header)<2: continue for rr in rows[i+1:i+12]: if self._is_major_section_row(rr): break rec={"doctype":self.CHILD_DOCTYPES.get(table_name,"")} self._map_row_to_record(rec,table_name,header,rr) if not self._child_identity(table_name,rec): continue self._merge_child(result,table_name,rec) def _classify_table(self,rows): text=" ".join(self._row_text(r) for r in rows) c=self._compact(text) scores=[] for name,aliases in self.TABLE_ALIASES.items(): score=sum(1 for a in aliases if self._compact(a) in c) if score: scores.append((score,name)) return max(scores)[1] if scores else None def _header_map(self,row,table_name=None): result={} groups=self.CHILD_FIELD_ALIASES.get(table_name,{}) if table_name else self.CHILD_FIELD_ALIASES for idx,c in enumerate(row): raw=self._clean(c) if not raw: continue # Remove printed serial-number prefixes and vertical OCR noise. stripped=re.sub(r"^[^0-9A-Za-z]*\d+\s*\.?\s*", "", raw) cc=self._compact(stripped) candidates=[] for field,aliases in groups.items(): for a in aliases: aa=self._compact(a) if not aa: continue if cc==aa: candidates.append((100000+len(aa),field)); break if not candidates: for field,aliases in groups.items(): for a in aliases: aa=self._compact(a) if aa and (aa in cc or cc in aa): candidates.append((len(aa),field)); break if candidates: result[idx]=max(candidates)[1] return result def _map_row_to_record(self,rec,table_name,header_map,row): for idx,field in header_map.items(): if idx>=len(row): continue raw=self._clean(row[idx]) if not raw: continue value=self._normalize_child(field,raw) if self._is_missing(field,rec.get(field)) or self._stronger_child_value(field,value,rec.get(field)): rec[field]=value self._trace(f"{table_name}.{field}",value) # Some extracted tables have merged cells; recover common scalar fields by position/regex. return rec # ================================================================ # missing-field recovery / reconciliation # ================================================================ def _recover_missing_parent_fields(self,result): for field,aliases in self.PARENT_FIELD_ALIASES.items(): if field in {"p_1b_type","p_1b_cb_name","p_1b_forex_ac_no","p_1b_ad_code","p_1b_gstin","p_1b_ifsc_no","p_1b_rbi_wr_no","p_1b_aeo","p_1i_sez_unit_data"}: continue if not self._is_missing(field,result.get(field)): continue val,_=self._semantic_value(aliases,self._validator_for(field)) if val is not None: result[field]=self._normalize_parent(field,val) # Labels from LABELS cover basic header fields not duplicated above. for field,aliases in self.LABELS.items(): if field in {"p_1i_sez_unit_data","p_1b_type","p_1b_aeo","p_1b_ifsc_no","p_1b_rbi_wr_no"}: continue if field not in result or not self._is_missing(field,result.get(field)): continue val,_=self._semantic_value(aliases,self._validator_for(field)) if val is not None: result[field]=self._normalize_parent(field,val) def _recover_missing_child_fields(self,result): # Child fields are recovered only inside their owning section/table. # A global label search is intentionally avoided because Shipping Bill # PDFs repeat labels such as VALUE, DATE, UQC, INFO and AMOUNT across # unrelated sections. return def _reconcile_child_rows(self,result): for table in self.CHILD_TABLES: records=result.get(table) or [] if not records: result[table]=[]; continue merged=[] keys=self._identity_fields(table) for rec in records: rec={k:v for k,v in rec.items() if k in self.COMPLETE_CHILD_FIELDS.get(table,[]) or k=="doctype"} target=None for old in merged: if keys and any(self._compact(old.get(k)) and self._compact(old.get(k))==self._compact(rec.get(k)) for k in keys): target=old; break if target is None: merged.append(rec) else: for k,v in rec.items(): if self._is_missing(k,target.get(k)) and not self._is_missing(k,v): target[k]=v result[table]=merged def _final_reconciliation(self,result): self._reconcile_header_authority(result) self._reconcile_status_authority(result) self._reconcile_locations_authority(result) self._reconcile_party_authority(result) self._reconcile_financial_authority(result) self._reconcile_event_authority(result) self._reconcile_invoice_authority(result) self._reconcile_item_authority(result) self._reconcile_annex_authority(result) self._reconcile_manifest_authority(result) self._reconcile_single_window_authority(result) def _reconcile_header_authority(self,result): # Re-read only explicit top header rows; never accept a nearby body value. for t in self.tables: rows=t["rows"] for i,r in enumerate(rows): c=self._compact(self._row_text(r)) if "portcode" in c and "sbno" in c and "sbdate" in c and i+1len(str(old))) def _merge_child(self,result,table,rec): keys=self._identity_fields(table) for old in result.get(table,[]): if keys and all(self._clean(old.get(k))==self._clean(rec.get(k)) for k in keys if self._clean(rec.get(k))): for k,v in rec.items(): if self._is_missing(k,old.get(k)) and not self._is_missing(k,v): old[k]=v return result.setdefault(table,[]).append(rec) # ================================================================ # schema / normalization / validation # ================================================================ def _empty_result(self): result={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_parent_fields(self,result): result["doctype"]="Shipping Bill" for f in self.PARENT_FIELDS: if f not in result: result[f]=0 if f in self.NUMBER_FIELDS else "" result[f]=self._normalize_parent(f,result[f]) def _finalize_children(self,result): for table in self.CHILD_TABLES: out=[] for rec in result.get(table) or []: clean={"doctype":self.CHILD_DOCTYPES.get(table,rec.get("doctype",""))} for f in self.COMPLETE_CHILD_FIELDS.get(table,[]): clean[f]=self._normalize_child(f,rec.get(f)) if f in self.CHILD_NUMERIC_DEFAULT_FIELDS and self._is_missing(f,rec.get(f)): clean[f]=0 if self._child_identity(table,clean): out.append(clean) result[table]=out def _schema_complete(self,result): for f in self.PARENT_FIELDS: if f not in result: result[f]=0 if f in self.NUMBER_FIELDS else "" for table in self.CHILD_TABLES: result.setdefault(table,[]) return result def _normalize_parent(self,field,value): if value is None: return 0 if field in self.NUMBER_FIELDS else "" if field in self.DATE_FIELDS: return self._format_date(value) if field in self.TIME_FIELDS: return self._format_time(value) if field in self.NUMBER_FIELDS: n=self._float(value) return int(n) if n is not None and float(n).is_integer() else (n if n is not None else 0) if field=="p_1b_gstin": m=re.search(r"\b(\d{2}[A-Z]{5}\d{4}[A-Z]\d[A-Z0-9]{3})\b",str(value).upper()); return m.group(1) if m else self._clean(value) return self._clean(value) def _normalize_child(self,field,value): if field == "p_4d_info_msr": return "" if value is None or str(value).strip()=="" else self._clean(value) numeric_tokens=("qty","value","amount","rate","percentage","no_of","fob","pmv","freight","insurance","discount","deduct","commison","p_c","exchng_rate","cess_rate","cess","duty","sqc_mst") if value is None or str(value).strip()=="": return 0 if field in self.CHILD_NUMERIC_DEFAULT_FIELDS or (not field.endswith("_curr") and any(x in field.lower() for x in numeric_tokens)) else "" if field.endswith("_dt") or field.endswith("_date") or "date" in field.lower(): return self._format_date(value) if field in {"p_1e_mawb_no","p_1e_noc","p_1e_hawb_no","p_1e_cin_no","p_1e_cin_site_id","p_2a_ad_code","p_4d_info_cd","p_4d_info_text","p_4d_qualifier","p_4d_uqc","p_3a_cth","p_3a_sqc_mst"}: return self._clean(value) if field == "p_3a_comp_cess": m=re.search(r"-?\d+(?:\.\d+)?", self._clean(value)) return int(float(m.group(0))) if m and float(m.group(0)).is_integer() else (float(m.group(0)) if m else 0) if field == "p_3a_igststat": return re.sub(r"^[OPC]\s*\n?", "", self._clean(value), flags=re.I).strip() if field in {"p_1i_seal_typ","p_1i_nature_cargo","p_1i_marks_numbers","p_2a_inv_no","p_2a_invterm","p_2b_exporter_name","p_2b_buyer_name","p_2b_buyer_addr","p_2b_exporter_addr","p_2c_exchng_rate_desc","p_3a_item_desc","p_3a_scheme_desc","p_3a_dbk_claimed","p_3a_igststat","p_3a_schcod","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_4d_info_cd","p_4d_info_text","p_4d_qualifier","p_4d_uqc"}: return re.sub(r"^[A-Z]\s+(?=[YN]$)", "", self._clean(value)).upper() if field == "p_3a_benefit_availd" else self._clean(value) n=self._float(value) if n is not None: return int(n) if float(n).is_integer() else n return self._clean(value) def _validator_for(self,field): if field in self.NUMBER_FIELDS: return lambda x:self._float(x) is not None if field in self.DATE_FIELDS: return self._valid_date if field in self.TIME_FIELDS: return self._valid_time if field in {"port_code","port_of_loading","port_of_finaldstn","port_of_discharge"}: return lambda x:self._valid_location_candidate(field,x) if field in {"shipping_bill_no"}: return self._valid_sb_number if field=="cb_code": return self._valid_cb_code return lambda x: bool(self._clean(x)) def _first_date(self, values): for v in values: if self._valid_date(v): return v return "" def _first_time(self, values): for v in values: if self._valid_time(v): return v return "" def _valid_time(self,v): return bool(re.fullmatch(r"\d{1,2}:\d{2}(?::\d{2})?",str(v or "").strip())) def _valid_date(self,v): s=str(v or "").strip(); return any(re.fullmatch(p,s,re.I) for p in (r"\d{1,2}-[A-Za-z]{3}-\d{2,4}",r"\d{1,2}/\d{1,2}/\d{2,4}",r"\d{4}-\d{1,2}-\d{1,2}",r"\d{1,2}\.\d{1,2}\.\d{4}")) def _recover_gwt_from_text(self,result): m=re.search(r"\bPKG\s+(\d+(?:\.\d+)?)\s+G\.?WT\s+(?:KGS\s+)?(\d+(?:\.\d+)?)", self.full_text, re.I) if m: result["pkg"]=self._integer(m.group(1)); result["gwt"]=self._float(m.group(2)); result["gwt_unit"]="KGS" def _clean_location(self,v): s=self._clean(v) s=re.sub(r"\s+"," ",s).strip() # pdfplumber can fuse a vertical glyph from the adjacent label into # the country value (e.g. NSAUDI / NSINGAPORE). If removing one # leading alphabetic glyph produces a clean country phrase, prefer it. if re.match(r"^[A-Z][A-Z ]{5,}$",s) and s[1:] in {"SAUDI ARABIA","SINGAPORE","UNITED STATES"}: s=s[1:] return s def _valid_code(self,v): return bool(re.fullmatch(r"[A-Z0-9]{3,20}",str(v or "").strip().upper())) def _valid_clean_aeo(self,v): return bool(re.fullmatch(r"[A-Z0-9]{6,30}",str(v or "").strip(),re.I)) def _inherit_invoice_currencies(self,r): # Currency is a field-level value in the PDF. Do not propagate the # invoice currency into valuation fields whose currency cell is blank. return def _valid_port(self,v): return bool(re.fullmatch(r"[A-Z]{3,8}\d{1,3}",str(v or "").strip().upper())) def _valid_sb_number(self,v): return bool(re.fullmatch(r"\d{5,10}",str(v or "").strip())) def _valid_cb_code(self,v): return bool(re.fullmatch(r"[A-Z0-9]{10,20}",str(v or "").strip().upper())) def _valid_hscode(self,v): return bool(re.fullmatch(r"\d{6,10}",str(v or "").strip())) def _valid_location_candidate(self,field,v): s=self._clean(v); return bool(s) and len(s)<=120 and not re.search(r"\b(?:NUMBER|DATE|AMOUNT|CODE|TYPE)\b",s,re.I) def _clean_status(self,v): s=self._clean(v).upper().replace("0","O") return s if s in {"Y","N"} else s def _is_missing(self,field,value): return value is None or str(value).strip()=="" def _clean(self,value): if value is None: return "" s=str(value).replace("\x00", "").replace("\r", " ").replace("\n", " ").strip() s=re.sub(r"\s+", " ", s) s=re.sub(r"(?<=\w)-\s+(?=\w)", "-", s) return s.strip(" ;") def _clean_address(self,value): # PDF table cells/rows use newlines for visual wrapping. Preserve the # existing address-line filtering, but serialize the logical address # as one JSON string with spaces between wrapped lines. s=str(value or "").replace("\r", "") lines=[] for line in s.splitlines(): line=self._clean(line) line=re.sub(r"^[YOPC]\s+", "", line, flags=re.I).strip() if not line or line.upper() in {"Y","O","P","C","SA","L","PRIVATE","GOVERNMENT","PUBLIC"}: continue if re.fullmatch(r"\d+\.?TYPE", line, re.I) or line.lower() in {"2.type","type"}: continue if self._looks_like_stop_text(line): continue if line not in lines: lines.append(line) return " ".join(lines) def _clean_party_value(self,field,value): s=self._clean(value) s=re.sub(r"^[YOPC](?:\s+|\n+)","",s,flags=re.I) s=re.sub(r"\b(?:O|C|P|SA)\b$","",s).strip() return s def _looks_like_stop_text(self,value): c=self._compact(value) return any(x in c for x in ("glossary","scanqrcode","visiticegate","pageof","parti","partii","partiii","partiv","partv")) def _looks_like_any_known_label(self,value): c=self._compact(value) return any(c==self._compact(a) for a in self._all_known_aliases()) def _num(self,v): try:return float(v) except:return None def _float(self,v): if v is None:return None s=str(v).replace(",","").strip() m=re.fullmatch(r"-?\d+(?:\.\d+)?",s) return float(s) if m else None def _integer(self,v): n=self._float(v); return int(n) if n is not None else 0 def _format_date(self,v): if not v:return "" s=str(v).strip().strip(".,;:").upper() 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(s,fmt).strftime("%Y-%m-%d") except ValueError:pass return s def _format_time(self,v): m=re.search(r"\b(\d{1,2}):(\d{2})(?::(\d{2}))?\b",str(v or "")) return f"{int(m.group(1)):02d}:{m.group(2)}:{m.group(3) or '00'}" if m else self._clean(v) def _trace(self,field,value,page=None): e={"field":field,"value":value} if page is not None:e["page"]=page self.trace.append(e) # ================================================================ # table helpers # ================================================================ def _row_text(self,row): return " ".join(self._clean(c) for c in row if self._clean(c)) def _section_rows(self,title): target=self._compact(title) for t in self.tables: for i,r in enumerate(t["rows"]): if target in self._compact(self._row_text(r)): return t["rows"][i:] return [] def _is_major_section_row(self,row): c=self._compact(self._row_text(row)) if not c: return False if any(x in c for x in ("parti","partii","partiii","partiv","partv","glossary","scanqrcode","visiticegate")): return True for heads in self.SECTION_HEADINGS.values(): if any(self._compact(h) in c for h in heads): return True return any(x in c for x in ("singlewindowdeclarationconstituents","singlewindowdeclarationcontrol","supportingdocuments","containerdetails","ar4details","thirdpartydetails")) def _header_pairs(self,header,data): pairs={} for i,c in enumerate(header): cc=self._compact(c) if i>=len(data):continue v=self._clean(data[i]) if "portcode" in cc:pairs["port code"]=v elif "sbno" in cc:pairs["sb no"]=v elif "sbdate" in cc:pairs["sb date"]=v return pairs def _first_value_after_header(self,header,data): for v in data: v=self._clean(v) if v:return v return "" def _cell_for_header(self,header,data,needles): for i,h in enumerate(header): c=self._compact(h) if any(self._compact(n) in c or c in self._compact(n) for n in needles): return self._clean(data[i]) if i "Settings": def get(key: str) -> Any: env_value = os.getenv(key) return env_value if env_value not in (None, "") else config.get(key) def get_bool(key: str) -> bool: value = get(key) if isinstance(value, str): return value.strip().lower() in {"1", "true", "yes", "y", "on"} return bool(value) def get_int(key: str) -> int: try: return int(get(key)) except (TypeError, ValueError) as exc: raise ConfigError(f"{key} must be a whole number.") from exc def get_list(key: str) -> tuple[str, ...]: value = get(key) if isinstance(value, str): value = [part.strip() for part in value.split(",")] return tuple(part for part in (value or []) if part) script_dir = Path(__file__).resolve().parent pdf_folder = Path(args.folder or str(get("PDF_FOLDER") or "")).expanduser().resolve() report_folder = str(get("REPORT_FOLDER") or "").strip() json_folder = str(get("JSON_OUTPUT_FOLDER") or "").strip() log_folder = str(get("LOG_FOLDER") or "").strip() settings = cls( frappe_url=str(get("FRAPPE_URL") or "").strip().rstrip("/"), api_key=str(get("FRAPPE_API_KEY") or "").strip(), api_secret=str(get("FRAPPE_API_SECRET") or "").strip(), pdf_folder=pdf_folder, include_subfolders=get_bool("INCLUDE_SUBFOLDERS"), doctype=str(get("DOCTYPE") or "Shipping Bill").strip(), naming_series=str(get("NAMING_SERIES") or "").strip(), duplicate_action=str(get("DUPLICATE_ACTION") or "skip").strip().lower(), duplicate_match_fields=get_list("DUPLICATE_MATCH_FIELDS"), attach_pdf=get_bool("ATTACH_PDF_TO_SHIPPING_BILL"), attach_private=get_bool("ATTACH_AS_PRIVATE_FILE"), validate_against_doctype=get_bool("VALIDATE_AGAINST_DOCTYPE"), required_fields=dict(config.get("REQUIRED_FIELDS") or {}), report_folder=Path(report_folder).expanduser() if report_folder else get_downloads_folder(), open_report_when_done=get_bool("OPEN_REPORT_WHEN_DONE") and not args.no_open, save_json_files=get_bool("SAVE_JSON_FILES"), json_output_folder=(Path(json_folder).expanduser() if json_folder else pdf_folder / "_extracted_json"), log_folder=Path(log_folder).expanduser() if log_folder else script_dir / "logs", request_timeout=get_int("REQUEST_TIMEOUT_SECONDS"), max_retries=get_int("MAX_RETRIES"), verify_ssl=get_bool("VERIFY_SSL"), max_consecutive_connection_failures=get_int("MAX_CONSECUTIVE_CONNECTION_FAILURES"), dry_run=bool(args.dry_run), ) settings.validate() return settings def validate(self) -> None: problems = [] if not str(self.pdf_folder).strip() or not self.pdf_folder.is_dir(): problems.append(f"PDF_FOLDER does not exist or is not a folder: {self.pdf_folder}") if self.duplicate_action not in {"skip", "update", "create"}: problems.append("DUPLICATE_ACTION must be one of: skip, update, create") if not self.duplicate_match_fields: problems.append("DUPLICATE_MATCH_FIELDS cannot be empty") if not self.required_fields: problems.append("REQUIRED_FIELDS cannot be empty") if self.request_timeout <= 0: problems.append("REQUEST_TIMEOUT_SECONDS must be greater than 0") if not self.dry_run: if not self.frappe_url or self.frappe_url in PLACEHOLDER_VALUES: problems.append("FRAPPE_URL is not set") elif not re.match(r"^https?://", self.frappe_url, re.I): problems.append("FRAPPE_URL must start with http:// or https://") if not self.api_key or self.api_key in PLACEHOLDER_VALUES: problems.append("FRAPPE_API_KEY is not set") if not self.api_secret or self.api_secret in PLACEHOLDER_VALUES: problems.append("FRAPPE_API_SECRET is not set") if problems: raise ConfigError("Configuration problem(s):\n - " + "\n - ".join(problems)) def get_downloads_folder() -> Path: """Return the user's Downloads folder (honours a relocated folder on Windows).""" if os.name == "nt": try: import winreg with winreg.OpenKey( winreg.HKEY_CURRENT_USER, r"Software\Microsoft\Windows\CurrentVersion\Explorer\Shell Folders", ) as key: value, _ = winreg.QueryValueEx(key, "{374DE290-123F-4565-9164-39C4925E467B}") if value: return Path(os.path.expandvars(value)) except OSError: pass return Path.home() / "Downloads" # ============================================================================= # RESULT MODEL # ============================================================================= class Status(str, Enum): CREATED = "Created" UPDATED = "Updated" NEEDS_REVIEW = "Saved - Needs Review" ALREADY_IN_FRAPPE = "Skipped - Already in Frappe" DUPLICATE_FILE = "Skipped - Duplicate PDF File" DUPLICATE_SB = "Skipped - Same SB in Folder" VALIDATED = "Validated (Dry Run)" FAILED = "Failed" NOT_PROCESSED = "Not Processed" SUCCESS_STATUSES = {Status.CREATED, Status.UPDATED, Status.VALIDATED} DUPLICATE_STATUSES = {Status.ALREADY_IN_FRAPPE, Status.DUPLICATE_FILE, Status.DUPLICATE_SB} ATTENTION_STATUSES = {Status.FAILED, Status.NOT_PROCESSED, Status.NEEDS_REVIEW} class Check(str, Enum): """Validation steps, in pipeline order. Every PDF gets a state per step.""" UNIQUE_FILE = "Unique File" PDF_READ = "PDF Readable" EXTRACTED = "JSON Extracted" JSON_VALID = "JSON Valid" UNIQUE_SB = "Unique SB in Folder" NOT_IN_FRAPPE = "Not in Frappe" SAVED = "Saved in Frappe" VERIFIED = "Verified in Frappe" ATTACHED = "PDF Attached" CHECK_DESCRIPTIONS = { Check.UNIQUE_FILE: "The file is not an identical copy (same content, any name) of another PDF in the folder.", Check.PDF_READ: "The PDF opens and has a text layer (scanned / image-only PDFs are not supported).", Check.EXTRACTED: "Shipping Bill pages found and converted to JSON by the extractor.", Check.JSON_VALID: "Mandatory fields present and values match the Frappe DocType (types, dates, lengths, options).", Check.UNIQUE_SB: "No earlier PDF in this folder has the same SB No + Port Code.", Check.NOT_IN_FRAPPE: "No Shipping Bill with this SB No + Port Code (or this exact PDF) exists in Frappe yet.", Check.SAVED: "Shipping Bill created (or updated) in Frappe.", Check.VERIFIED: "Saved record read back from Frappe: key fields and child rows match, only one record exists.", Check.ATTACHED: "Source PDF attached to the Shipping Bill and the attachment confirmed (size and content).", } class CheckState(str, Enum): PASS = "\u2713" # check mark FAIL = "\u2717" # cross WARN = "!" NOT_RUN = "\u2014" # em dash class ProcessingError(Exception): """A per-PDF failure: which check failed, why, and how to fix it.""" def __init__(self, check: Check, message: str, action: str = "", connection_error: bool = False): super().__init__(message) self.check = check self.message = message self.action = action self.connection_error = connection_error @dataclass class ProcessResult: serial_no: int file_name: str file_path: str status: Status = Status.NOT_PROCESSED checks: dict[Check, CheckState] = dataclass_field( default_factory=lambda: {c: CheckState.NOT_RUN for c in Check}) shipping_bill_no: str = "" shipping_bill_date: str = "" port_code: str = "" exporter_name: str = "" frappe_doc_name: str = "" frappe_doc_url: str = "" failed_check: str = "" message: str = "" action_required: str = "" duplicate_of: str = "" warnings: list[str] = dataclass_field(default_factory=list) pdf_type: str = "" total_pages: int = 0 shipping_bill_pages: str = "" item_count: int = 0 file_size_kb: float = 0.0 duration_seconds: float = 0.0 processed_at: str = "" def mark(self, check: Check, state: CheckState) -> None: self.checks[check] = state @dataclass class RunSummary: started_at: datetime finished_at: Optional[datetime] = None interrupted: bool = False aborted_reason: str = "" frappe_user: str = "" schema_status: str = "" total_files: int = 0 unique_files: int = 0 results: list[ProcessResult] = dataclass_field(default_factory=list) def count(self, *statuses: Status) -> int: return sum(1 for r in self.results if r.status in statuses) @dataclass class PdfFile: path: Path size: int = 0 sha256: str = "" md5: str = "" # Frappe stores the md5 of a file as File.content_hash read_error: str = "" # ============================================================================= # EXTRACTION SERVICE (PDF -> Shipping Bill JSON) # ============================================================================= @dataclass class ExtractionOutput: shipping_bill_json: dict[str, Any] header_found: bool shipping_bill_pages: str class ShippingBillExtractor: """Runs the framework analyzer + parser and checks mandatory fields.""" def __init__(self, required_fields: dict[str, str]): self.required_fields = required_fields def extract(self, pdf_path: Path, result: ProcessResult) -> ExtractionOutput: # Step: PDF readable try: analysis = PDFAnalyzer(str(pdf_path)).analyze() except Exception as exc: # pdfminer raises many different exception types raise ProcessingError( Check.PDF_READ, f"The PDF could not be opened: {self._short(exc)}", "The file is corrupted, password-protected or not a real PDF. " "Download it again from ICEGATE.", ) from exc if not isinstance(analysis, dict) or not analysis.get("pages"): raise ProcessingError(Check.PDF_READ, "The PDF has no readable pages.", "Check that the PDF opens correctly.") result.pdf_type = str(analysis.get("pdf_type") or "") result.total_pages = int(analysis.get("total_pages") or 0) if int(analysis.get("text_pages") or 0) == 0: raise ProcessingError( Check.PDF_READ, "Scanned / image-only PDF: no page has selectable text (OCR is not supported).", "Use the original digitally generated Shipping Bill PDF from ICEGATE, not a scan.", ) result.mark(Check.PDF_READ, CheckState.PASS) # Step: JSON extracted (framework parser, unchanged) try: parsed = ShippingBillParser(analysis["pages"]).parse() except Exception as exc: raise ProcessingError( Check.EXTRACTED, f"Shipping Bill data could not be extracted: {self._short(exc)}", "The PDF layout differs from the standard ICEGATE Shipping Bill. Check it manually.", ) from exc data = parsed.get("shipping_bill_json") if isinstance(parsed, dict) else None if not isinstance(data, dict): raise ProcessingError(Check.EXTRACTED, "The extractor did not return Shipping Bill JSON.", "Check the PDF manually.") trace = parsed.get("_trace") if isinstance(parsed.get("_trace"), list) else [] selection = next((t for t in trace if t.get("stage") == "page_selection"), None) sb_pages = "" if selection: start = int(selection.get("start_page") or 1) end = start + int(selection.get("selected_page_count") or 1) - 1 sb_pages = f"{start}-{end}" if end > start else str(start) missing = [label for fieldname, label in self.required_fields.items() if str(data.get(fieldname) or "").strip() == ""] if missing and selection is None: raise ProcessingError( Check.EXTRACTED, "This is not a Customs Shipping Bill: the ICEGATE header (Port Code / SB No / " f"SB Date) was not found. Missing: {', '.join(missing)}.", "Remove this file from the folder or replace it with the ICEGATE Shipping Bill PDF.", ) result.mark(Check.EXTRACTED, CheckState.PASS) return ExtractionOutput(shipping_bill_json=data, header_found=selection is not None, shipping_bill_pages=sb_pages) def validate_required(self, data: dict[str, Any]) -> list[str]: errors = [] for fieldname, label in self.required_fields.items(): if str(data.get(fieldname) or "").strip() == "": errors.append(f"{label} ({fieldname}) could not be extracted from the PDF") sb_date = str(data.get("shipping_bill_date") or "") if sb_date and not _is_valid_date(sb_date): errors.append(f"Shipping Bill Date '{sb_date}' is not a valid date") return errors @staticmethod def _short(exc: Exception) -> str: text = str(exc).strip() or type(exc).__name__ return text if len(text) <= 300 else text[:297] + "..." def _is_valid_date(value: Any) -> bool: try: datetime.strptime(str(value), "%Y-%m-%d") return True except ValueError: return False def build_frappe_payload(shipping_bill_json: dict[str, Any], settings: Settings, for_update: bool = False) -> dict[str, Any]: """Turn the framework JSON into a Frappe REST payload. The extracted JSON itself is not changed; only transport details apply: * blank values are left out of a new document so Frappe applies the DocType defaults (on update they are sent as null to clear old values) * blank values inside child rows are always left out * the configured DocType / naming series are set """ def clean_row(row: dict[str, Any]) -> dict[str, Any]: return {k: v for k, v in row.items() if v is not None and v != ""} payload: dict[str, Any] = {} for key, value in shipping_bill_json.items(): if isinstance(value, list): payload[key] = [clean_row(row) for row in value if isinstance(row, dict)] elif value is None or value == "": if for_update: payload[key] = None else: payload[key] = value payload["doctype"] = settings.doctype if settings.naming_series and not for_update: payload["naming_series"] = settings.naming_series return payload # ============================================================================= # DOCTYPE SCHEMA VALIDATION (JSON vs live Frappe DocType) # ============================================================================= class DocTypeSchema: """Field definitions of the Shipping Bill DocType and its child tables, loaded from Frappe and used to validate each JSON before it is sent.""" LAYOUT_TYPES = {"Section Break", "Column Break", "Tab Break", "HTML", "Button", "Heading", "Fold", "Image"} TABLE_TYPES = {"Table", "Table MultiSelect"} INT_TYPES = {"Int", "Check"} FLOAT_TYPES = {"Float", "Currency", "Percent"} LENGTH_LIMITED_TYPES = {"Data", "Link", "Dynamic Link", "Select", "Read Only", "Autocomplete"} SYSTEM_KEYS = {"doctype", "name", "naming_series", "parent", "parentfield", "parenttype", "idx", "owner", "creation", "modified", "modified_by", "docstatus"} def __init__(self, doctype: str, meta_docs: list[dict[str, Any]]): by_name = {d.get("name"): d for d in meta_docs if isinstance(d, dict)} if doctype not in by_name: raise ValueError(f"DocType '{doctype}' definition was not returned by Frappe") self.doctype = doctype self.fields = self._fields(by_name[doctype]) self.tables = {f: df.get("options") for f, df in self.fields.items() if df.get("fieldtype") in self.TABLE_TYPES} self.child_fields = {child: self._fields(by_name[child]) for child in self.tables.values() if child in by_name} def describe(self) -> str: data_fields = sum(1 for df in self.fields.values() if df.get("fieldtype") not in self.TABLE_TYPES) return (f"Validated against the live DocType '{self.doctype}' " f"({data_fields} fields, {len(self.tables)} child tables)") def _fields(self, meta: dict[str, Any]) -> dict[str, dict[str, Any]]: return {df["fieldname"]: df for df in meta.get("fields") or [] if df.get("fieldname") and df.get("fieldtype") not in self.LAYOUT_TYPES} def validate(self, payload: dict[str, Any]) -> tuple[list[str], list[str]]: errors: list[str] = [] warnings: list[str] = [] unknown = sorted(k for k in payload if k not in self.fields and k not in self.SYSTEM_KEYS) if unknown: warnings.append(f"{len(unknown)} extracted field(s) are not in DocType '{self.doctype}' " f"and are ignored by Frappe: {', '.join(unknown)}") for fieldname, df in self.fields.items(): label = f"{df.get('label') or fieldname} ({fieldname})" value = payload.get(fieldname) if df.get("fieldtype") in self.TABLE_TYPES: rows = value or [] if df.get("reqd") and not rows: errors.append(f"Mandatory table {label} has no rows") child_doctype = df.get("options") child = self.child_fields.get(child_doctype, {}) unknown_child: set[str] = set() for idx, row in enumerate(rows, start=1): unknown_child |= {k for k in row if k not in child and k not in self.SYSTEM_KEYS} for cname, cdf in child.items(): clabel = (f"{df.get('label') or fieldname} row {idx} > " f"{cdf.get('label') or cname} ({cname})") self._check_value(errors, clabel, cdf, row.get(cname)) if unknown_child and child: warnings.append(f"{df.get('label') or fieldname}: field(s) not in child DocType " f"'{child_doctype}' are ignored: {', '.join(sorted(unknown_child))}") else: self._check_value(errors, label, df, value) return errors, warnings def _check_value(self, errors: list[str], label: str, df: dict[str, Any], value: Any) -> None: fieldtype = df.get("fieldtype") if value is None or (isinstance(value, str) and value.strip() == ""): if df.get("reqd") and not df.get("default") and fieldtype != "Check": errors.append(f"Mandatory field {label} is empty") return text = str(value) if fieldtype in self.INT_TYPES: try: if float(text) != int(float(text)): raise ValueError except ValueError: errors.append(f"{label} must be a whole number, got '{text}'") elif fieldtype in self.FLOAT_TYPES: try: float(text) except ValueError: errors.append(f"{label} must be a number, got '{text}'") elif fieldtype == "Date": if not _is_valid_date(text): errors.append(f"{label} is not a valid date: '{text}'") elif fieldtype == "Time": if not re.fullmatch(r"\d{1,2}:\d{2}(:\d{2}(\.\d+)?)?", text): errors.append(f"{label} is not a valid time: '{text}'") elif fieldtype == "Select": options = str(df.get("options") or "").split("\n") if any(o.strip() for o in options) and text not in options: allowed = ", ".join(o for o in options if o) errors.append(f"{label} value '{text}' is not an allowed option ({allowed})") if fieldtype in self.LENGTH_LIMITED_TYPES: max_len = int(df.get("length") or 140) if len(text) > max_len: errors.append(f"{label} is {len(text)} characters long; the DocType allows {max_len}") # ============================================================================= # FRAPPE REST CLIENT # ============================================================================= class FrappeError(Exception): def __init__(self, message: str, status_code: Optional[int] = None, connection_error: bool = False): super().__init__(message) self.message = message self.status_code = status_code self.connection_error = connection_error class FrappeClient: """Small, explicit wrapper around the Frappe REST API (token auth).""" STATUS_HINTS = { 400: "Bad request", 401: "Authentication failed", 403: "Permission denied", 404: "Not found", 409: "Duplicate entry", 413: "File/request too large", 417: "Validation error", 429: "Too many requests", } def __init__(self, settings: Settings): self.base_url = settings.frappe_url self.timeout = settings.request_timeout self.verify_ssl = settings.verify_ssl self.session = requests.Session() self.session.headers.update({ "Authorization": f"token {settings.api_key}:{settings.api_secret}", "Accept": "application/json", "User-Agent": f"{SCRIPT_NAME}/{SCRIPT_VERSION}", }) # Retries only for idempotent GET requests. POST/PUT are never retried # automatically, so a time-out can never create the same bill twice; # a re-run detects the bill through the duplicate checks instead. retry = Retry( total=settings.max_retries, connect=settings.max_retries, read=settings.max_retries, backoff_factor=1.5, status_forcelist=(429, 500, 502, 503, 504), allowed_methods=frozenset({"GET"}), raise_on_status=False, ) adapter = HTTPAdapter(max_retries=retry) self.session.mount("https://", adapter) self.session.mount("http://", adapter) # ---- connection / metadata -------------------------------------------------- def get_logged_user(self) -> str: data = self._request("GET", "/api/method/frappe.auth.get_logged_user") return str(data.get("message") or "") def check_doctype_access(self, doctype: str) -> None: self._request("GET", self._resource_path(doctype), params={"fields": json.dumps(["name"]), "limit_page_length": 1}) def get_doctype_meta(self, doctype: str) -> list[dict[str, Any]]: data = self._request("GET", "/api/method/frappe.desk.form.load.getdoctype", params={"doctype": doctype}) docs = data.get("docs") if not isinstance(docs, list): raise FrappeError("The DocType definition response did not contain 'docs'.") return docs # ---- documents ----------------------------------------------------------------- def find_names(self, doctype: str, filters: dict[str, Any], limit: int = 20) -> list[str]: data = self._request( "GET", self._resource_path(doctype), params={ "filters": json.dumps([[k, "=", v] for k, v in filters.items()]), "fields": json.dumps(["name"]), "order_by": "creation asc", "limit_page_length": limit, }, ) return [str(row.get("name")) for row in data.get("data") or [] if row.get("name")] def get_doc(self, doctype: str, name: str) -> dict[str, Any]: data = self._request("GET", self._resource_path(doctype, name)) doc = data.get("data") if not isinstance(doc, dict): raise FrappeError(f"Document {name} could not be read back from Frappe.") return doc def insert(self, doctype: str, doc: dict[str, Any]) -> str: data = self._request("POST", self._resource_path(doctype), json_body=doc) return self._doc_name(data) def update(self, doctype: str, name: str, doc: dict[str, Any]) -> str: data = self._request("PUT", self._resource_path(doctype, name), json_body=doc) return self._doc_name(data) or name # ---- attachments ----------------------------------------------------------------- def find_files_by_hash(self, content_hash: str, doctype: str) -> list[dict[str, Any]]: return self._list_files([["content_hash", "=", content_hash], ["attached_to_doctype", "=", doctype]]) def list_attachments(self, doctype: str, name: str) -> list[dict[str, Any]]: return self._list_files([["attached_to_doctype", "=", doctype], ["attached_to_name", "=", name]]) def upload_attachment(self, pdf: PdfFile, doctype: str, name: str, private: bool) -> dict[str, Any]: with pdf.path.open("rb") as fh: data = self._request( "POST", "/api/method/upload_file", files={"file": (pdf.path.name, fh, "application/pdf")}, form={"doctype": doctype, "docname": name, "is_private": "1" if private else "0", "folder": "Home/Attachments"}, ) message = data.get("message") if not isinstance(message, dict) or not message.get("name"): raise FrappeError("Frappe accepted the upload but did not return the File record.") return message def doc_url(self, doctype: str, name: str) -> str: slug = doctype.strip().lower().replace(" ", "-") return f"{self.base_url}/app/{quote(slug)}/{quote(name, safe='')}" def close(self) -> None: self.session.close() # ---- internals ----------------------------------------------------------------- def _list_files(self, filters: list[list[Any]]) -> list[dict[str, Any]]: data = self._request( "GET", self._resource_path("File"), params={ "filters": json.dumps(filters), "fields": json.dumps(["name", "file_name", "file_size", "content_hash", "attached_to_name", "file_url"]), "limit_page_length": 50, }, ) return [row for row in data.get("data") or [] if isinstance(row, dict)] @staticmethod def _resource_path(doctype: str, name: Optional[str] = None) -> str: path = f"/api/resource/{quote(doctype, safe='')}" return f"{path}/{quote(name, safe='')}" if name else path @staticmethod def _doc_name(data: dict[str, Any]) -> str: doc = data.get("data") if isinstance(data, dict) else None if not isinstance(doc, dict) or not doc.get("name"): raise FrappeError("Frappe accepted the request but did not return a document name.") return str(doc["name"]) def _request(self, method: str, path: str, params: Optional[dict] = None, json_body: Optional[dict] = None, files: Optional[dict] = None, form: Optional[dict] = None) -> dict[str, Any]: url = f"{self.base_url}{path}" headers = {} body: Any = form if json_body is not None: headers["Content-Type"] = "application/json" body = json.dumps(json_body, default=str) try: response = self.session.request( method, url, params=params, data=body, files=files, headers=headers, timeout=self.timeout, verify=self.verify_ssl, ) except requests.exceptions.SSLError as exc: raise FrappeError(f"SSL certificate error while connecting to {self.base_url}: {exc}", connection_error=True) from exc except requests.exceptions.Timeout as exc: raise FrappeError(f"Frappe did not respond within {self.timeout} seconds.", connection_error=True) from exc except requests.exceptions.ConnectionError as exc: raise FrappeError(f"Could not connect to {self.base_url} (site down, wrong URL " f"or no internet).", connection_error=True) from exc except requests.exceptions.RequestException as exc: raise FrappeError(f"HTTP request failed: {exc}", connection_error=True) from exc if response.status_code >= 400: raise FrappeError(self._describe_error(response), status_code=response.status_code, connection_error=response.status_code in {502, 503, 504}) try: return response.json() except ValueError as exc: raise FrappeError( f"Frappe returned a non-JSON response (HTTP {response.status_code}). " "Check that FRAPPE_URL points to the Frappe site root.", status_code=response.status_code, ) from exc @classmethod def _describe_error(cls, response: requests.Response) -> str: status = response.status_code try: body = response.json() except ValueError: body = {} if not isinstance(body, dict): body = {} messages: list[str] = [] server_messages = body.get("_server_messages") if server_messages: try: for raw in json.loads(server_messages): try: item = json.loads(raw) messages.append(str(item.get("message", raw)) if isinstance(item, dict) else str(item)) except (ValueError, TypeError): messages.append(str(raw)) except (ValueError, TypeError): messages.append(str(server_messages)) if not messages and body.get("exception"): messages.append(re.sub(r"^frappe\.exceptions\.", "", str(body["exception"]))) if not messages and body.get("exc_type"): messages.append(str(body["exc_type"])) if not messages and body.get("message"): messages.append(str(body["message"])) if not messages: messages.append(response.text[:300] if response.text else (response.reason or "")) detail = " | ".join(cls._clean_text(m) for m in messages if m) hint = cls.STATUS_HINTS.get(status, "Frappe server error" if status >= 500 else "Request failed") return f"HTTP {status} {hint}: {detail}".strip().rstrip(":") @staticmethod def _clean_text(text: str) -> str: text = re.sub(r"<[^>]+>", " ", html.unescape(str(text))) text = re.sub(r"\s+", " ", text).strip() return text if len(text) <= 500 else text[:497] + "..." def frappe_action_hint(error: FrappeError) -> str: status = error.status_code if error.connection_error: return ("Frappe was not reachable. Run the script again later - Shipping Bills that " "were already created are detected and skipped automatically.") if status in {401, 403}: return ("Check FRAPPE_API_KEY / FRAPPE_API_SECRET and that the API user has Read, Create " "and Write permission on Shipping Bill and File.") if status == 404: return "Check FRAPPE_URL and the DOCTYPE name in the configuration." if status == 409: return "A document with the same name already exists in Frappe." if status == 413: return "The PDF is larger than the site's maximum upload size (System Settings > Max File Size)." if status == 417: return "Frappe rejected the data (see the reason). Correct the value or the DocType rule and run again." if status and status >= 500: return "Frappe raised a server error. Check the Error Log in Frappe, then run again." return "See the reason, fix it and run the script again." # ============================================================================= # BATCH PROCESSOR (validation pipeline) # ============================================================================= class ShippingBillBatchUploader: def __init__(self, settings: Settings): self.settings = settings self.extractor = ShippingBillExtractor(settings.required_fields) self.client: Optional[FrappeClient] = None if settings.dry_run else FrappeClient(settings) self.schema: Optional[DocTypeSchema] = None self._consecutive_connection_failures = 0 # ---- entry point ----------------------------------------------------------------- def run(self) -> RunSummary: summary = RunSummary(started_at=datetime.now()) files = self._discover_and_fingerprint() readable = [f for f in files if not f.read_error] summary.total_files = len(files) summary.unique_files = len({f.sha256 for f in readable}) copies = len(readable) - summary.unique_files logger.info("Folder check: %d PDF file(s) found, %d unique, %d identical cop%s.", len(files), summary.unique_files, copies, "y" if copies == 1 else "ies") if self.client: summary.frappe_user, summary.schema_status = self._preflight_checks() else: summary.schema_status = "Mandatory fields only (dry run - DocType not checked, Frappe not contacted)" first_by_hash: dict[str, ProcessResult] = {} first_by_sb: dict[tuple[str, ...], ProcessResult] = {} try: for index, pdf in enumerate(files, start=1): if summary.aborted_reason: summary.results.append(self._not_processed(index, pdf, summary.aborted_reason)) continue result = self._process(index, pdf, first_by_hash, first_by_sb) summary.results.append(result) self._log_result(result, len(files)) if self._consecutive_connection_failures >= self.settings.max_consecutive_connection_failures: summary.aborted_reason = (f"Frappe unreachable {self._consecutive_connection_failures} " f"times in a row") logger.error("Stopping: %s. Remaining PDFs are marked 'Not Processed'.", summary.aborted_reason) except KeyboardInterrupt: summary.interrupted = True logger.warning("Interrupted by user. Writing the report for the PDFs processed so far.") done = {r.file_path for r in summary.results} for index, pdf in enumerate(files, start=1): if str(pdf.path) not in done: summary.results.append(self._not_processed(index, pdf, "Run stopped by user")) finally: summary.finished_at = datetime.now() if self.client: self.client.close() return summary # ---- folder scan + Frappe pre-flight ------------------------------------------------ def _discover_and_fingerprint(self) -> list[PdfFile]: """List the PDFs and fingerprint each one (SHA-256 for identical-copy detection in the folder, MD5 = Frappe File.content_hash).""" pattern = "**/*" if self.settings.include_subfolders else "*" json_dir = self.settings.json_output_folder.resolve() paths = [] for path in self.settings.pdf_folder.glob(pattern): if not path.is_file() or path.suffix.lower() != ".pdf" or path.name.startswith("~$"): continue if json_dir in path.resolve().parents: continue paths.append(path) paths.sort(key=lambda p: str(p).lower()) files = [] for path in paths: pdf = PdfFile(path=path) try: sha, md5 = hashlib.sha256(), hashlib.md5() with path.open("rb") as fh: for chunk in iter(lambda: fh.read(1024 * 1024), b""): sha.update(chunk) md5.update(chunk) pdf.sha256, pdf.md5 = sha.hexdigest(), md5.hexdigest() pdf.size = path.stat().st_size except OSError as exc: pdf.read_error = str(exc) files.append(pdf) return files def _preflight_checks(self) -> tuple[str, str]: """Fail fast on wrong URL / credentials / permissions before touching PDFs.""" assert self.client is not None try: user = self.client.get_logged_user() self.client.check_doctype_access(self.settings.doctype) except FrappeError as exc: raise ConfigError( f"Cannot use Frappe at {self.settings.frappe_url}:\n {exc.message}\n" f" -> {frappe_action_hint(exc)}" ) from exc logger.info("Connected to %s as '%s' (DocType '%s' accessible).", self.settings.frappe_url, user, self.settings.doctype) if not self.settings.validate_against_doctype: return user, "Mandatory fields only (VALIDATE_AGAINST_DOCTYPE is switched off)" try: self.schema = DocTypeSchema(self.settings.doctype, self.client.get_doctype_meta(self.settings.doctype)) status = self.schema.describe() logger.info("%s.", status) except (FrappeError, ValueError, KeyError) as exc: reason = exc.message if isinstance(exc, FrappeError) else str(exc) status = f"Mandatory fields only - DocType definition could not be loaded ({reason})" logger.warning("%s", status) return user, status # ---- one PDF ------------------------------------------------------------------------ def _process(self, index: int, pdf: PdfFile, first_by_hash: dict[str, ProcessResult], first_by_sb: dict[tuple[str, ...], ProcessResult]) -> ProcessResult: result = ProcessResult(serial_no=index, file_name=pdf.path.name, file_path=str(pdf.path), file_size_kb=round(pdf.size / 1024, 1)) started = time.perf_counter() try: # Step 0: unique file in the folder if pdf.read_error: raise ProcessingError(Check.UNIQUE_FILE, f"The file could not be read: {pdf.read_error}", "Close the file if it is open in another program and run again.") original = first_by_hash.get(pdf.sha256) if original: result.mark(Check.UNIQUE_FILE, CheckState.FAIL) result.status = Status.DUPLICATE_FILE result.duplicate_of = original.file_name for attr in ("shipping_bill_no", "shipping_bill_date", "port_code", "exporter_name", "frappe_doc_name", "frappe_doc_url"): setattr(result, attr, getattr(original, attr)) result.message = (f"Identical copy of '{original.file_name}' (same file content). " "Skipped so the same PDF is not uploaded twice.") result.action_required = "No action needed. You can delete this extra copy from the folder." return result first_by_hash[pdf.sha256] = result result.mark(Check.UNIQUE_FILE, CheckState.PASS) # Steps 1-2: PDF readable, JSON extracted extraction = self.extractor.extract(pdf.path, result) data = extraction.shipping_bill_json self._fill_from_extraction(result, extraction) self._save_json(pdf.path, data) # Step 3: JSON valid (mandatory fields + live DocType definition) self._validate_json(result, data) # Step 4: unique Shipping Bill within the folder key = tuple(str(data.get(f) or "").strip().upper() for f in self.settings.duplicate_match_fields) first = first_by_sb.get(key) if first and self.settings.duplicate_action != "create": result.mark(Check.UNIQUE_SB, CheckState.FAIL) result.status = Status.DUPLICATE_SB result.duplicate_of = first.file_name result.frappe_doc_name, result.frappe_doc_url = first.frappe_doc_name, first.frappe_doc_url result.message = (f"Same Shipping Bill (SB No {result.shipping_bill_no}, Port " f"{result.port_code}) as '{first.file_name}', which was processed first " f"(result: {first.status.value}). Skipped to avoid a duplicate.") result.action_required = ("No action needed if both files are the same Shipping Bill. If this " "file is the corrected version, run it with DUPLICATE_ACTION = 'update'.") return result first_by_sb.setdefault(key, result) result.mark(Check.UNIQUE_SB, CheckState.PASS) if self.settings.dry_run: result.status = Status.VALIDATED result.message = "All offline checks passed (dry run - nothing sent to Frappe)." return result # Steps 5-8: Frappe self._upload(result, pdf, data) self._consecutive_connection_failures = 0 except ProcessingError as exc: result.mark(exc.check, CheckState.FAIL) result.status = Status.FAILED result.failed_check = exc.check.value result.message = exc.message result.action_required = exc.action if exc.check in {Check.NOT_IN_FRAPPE, Check.SAVED} and not exc.connection_error: self._consecutive_connection_failures = 0 except Exception as exc: # never let one PDF stop the whole batch result.status = Status.FAILED result.failed_check = "Unexpected error" result.message = f"{type(exc).__name__}: {exc}" result.action_required = "Unexpected error - see the log file for the full details." logger.debug("Traceback for %s:\n%s", pdf.path.name, traceback.format_exc()) finally: result.duration_seconds = round(time.perf_counter() - started, 2) result.processed_at = datetime.now().strftime("%d-%m-%Y %H:%M:%S") return result def _validate_json(self, result: ProcessResult, data: dict[str, Any]) -> None: errors = self.extractor.validate_required(data) if self.schema: schema_errors, warnings = self.schema.validate(build_frappe_payload(data, self.settings)) errors.extend(e for e in schema_errors if e not in errors) result.warnings.extend(warnings) if errors: shown = errors[:8] more = f" (+{len(errors) - 8} more)" if len(errors) > 8 else "" raise ProcessingError( Check.JSON_VALID, "JSON does not match the Shipping Bill DocType: " + "; ".join(shown) + more + ".", "Check these values in the PDF. If the PDF is correct, the DocType field definition " "(mandatory / type / length / options) needs to be adjusted.", ) result.mark(Check.JSON_VALID, CheckState.WARN if result.warnings else CheckState.PASS) def _upload(self, result: ProcessResult, pdf: PdfFile, data: dict[str, Any]) -> None: assert self.client is not None client, doctype, action = self.client, self.settings.doctype, self.settings.duplicate_action match = {f: data.get(f) for f in self.settings.duplicate_match_fields} # Step 5: not already in Frappe (same SB No + Port, or this exact PDF attached earlier) existing, same_pdf_attached = None, False if action != "create": try: names = client.find_names(doctype, match) if names: existing = names[0] elif self.settings.attach_pdf: files = client.find_files_by_hash(pdf.md5, doctype) if files and files[0].get("attached_to_name"): existing, same_pdf_attached = str(files[0]["attached_to_name"]), True except FrappeError as exc: self._raise_frappe(Check.NOT_IN_FRAPPE, exc) if existing and (action == "skip" or same_pdf_attached): result.mark(Check.NOT_IN_FRAPPE, CheckState.FAIL) result.status = Status.ALREADY_IN_FRAPPE result.duplicate_of = f"Frappe: {existing}" result.frappe_doc_name, result.frappe_doc_url = existing, client.doc_url(doctype, existing) reason = (f"This exact PDF is already attached to Shipping Bill {existing}" if same_pdf_attached else f"Shipping Bill {result.shipping_bill_no} (Port {result.port_code}) already " f"exists in Frappe as {existing}") result.message = f"{reason}. Not created again." result.action_required = ("No action needed. Set DUPLICATE_ACTION = 'update' if the existing " "record must be overwritten with this PDF.") return result.mark(Check.NOT_IN_FRAPPE, CheckState.WARN if existing else CheckState.PASS) # Step 6: saved in Frappe try: if existing: name = client.update(doctype, existing, build_frappe_payload(data, self.settings, for_update=True)) result.status = Status.UPDATED result.message = f"Existing Shipping Bill {name} updated with the PDF data." else: name = client.insert(doctype, build_frappe_payload(data, self.settings)) result.status = Status.CREATED result.message = f"Shipping Bill {name} created." except FrappeError as exc: self._raise_frappe(Check.SAVED, exc) result.mark(Check.SAVED, CheckState.PASS) result.frappe_doc_name, result.frappe_doc_url = name, client.doc_url(doctype, name) # Steps 7-8: verify + attach. The record is already saved, so a problem # here never "fails" the PDF - it is flagged as "Saved - Needs Review". problems = [] problem = self._verify_saved(name, build_frappe_payload(data, self.settings), match) result.mark(Check.VERIFIED, CheckState.FAIL if problem else CheckState.PASS) if problem: problems.append(f"verification - {problem}") if self.settings.attach_pdf: problem = self._attach_and_verify(pdf, name) result.mark(Check.ATTACHED, CheckState.FAIL if problem else CheckState.PASS) if problem: problems.append(f"attachment - {problem}") if problems: result.status = Status.NEEDS_REVIEW result.failed_check = (Check.VERIFIED if result.checks[Check.VERIFIED] == CheckState.FAIL else Check.ATTACHED).value result.message = f"Shipping Bill {name} was saved, but: " + "; ".join(problems) + "." result.action_required = (f"Open {name} in Frappe and check it. A missing PDF can be attached " "manually from the record's sidebar.") def _verify_saved(self, name: str, payload: dict[str, Any], match: dict[str, Any]) -> str: assert self.client is not None try: saved = self.client.get_doc(self.settings.doctype, name) except FrappeError as exc: return f"the saved document could not be read back ({exc.message})" issues = [] for fieldname in dict.fromkeys(list(self.settings.required_fields) + list(match)): expected, actual = payload.get(fieldname), saved.get(fieldname) if str(expected or "") != str(actual or ""): issues.append(f"{fieldname} is '{actual}' in Frappe, expected '{expected}'") tables = (list(self.schema.tables) if self.schema else [k for k, v in payload.items() if isinstance(v, list) and k in saved]) for table in tables: expected_rows, actual_rows = len(payload.get(table) or []), len(saved.get(table) or []) if expected_rows != actual_rows: issues.append(f"{table} has {actual_rows} row(s) in Frappe, expected {expected_rows}") if self.settings.duplicate_action != "create": try: names = self.client.find_names(self.settings.doctype, match) if len(names) > 1: issues.append(f"{len(names)} Shipping Bills now exist for the same SB No / Port: " f"{', '.join(names)}") except FrappeError as exc: issues.append(f"duplicate re-check failed ({exc.message})") return "; ".join(issues) def _attach_and_verify(self, pdf: PdfFile, name: str) -> str: assert self.client is not None doctype = self.settings.doctype try: if any(f.get("content_hash") == pdf.md5 for f in self.client.list_attachments(doctype, name)): return "" # the same PDF is already attached (e.g. on update) - nothing to do uploaded = self.client.upload_attachment(pdf, doctype, name, self.settings.attach_private) attached = self.client.list_attachments(doctype, name) except FrappeError as exc: return f"the PDF could not be attached ({exc.message})" record = next((f for f in attached if f.get("name") == uploaded.get("name")), None) if not record: return "the upload succeeded, but the file is not linked to the Shipping Bill" if record.get("content_hash") and record.get("content_hash") != pdf.md5: return "the attached file content does not match the source PDF" if record.get("file_size") and int(record.get("file_size")) != pdf.size: return (f"the attached file is {record.get('file_size')} bytes but the PDF is " f"{pdf.size} bytes") return "" # ---- helpers ------------------------------------------------------------------------ @staticmethod def _fill_from_extraction(result: ProcessResult, extraction: ExtractionOutput) -> None: data = extraction.shipping_bill_json result.shipping_bill_no = str(data.get("shipping_bill_no") or "") result.shipping_bill_date = str(data.get("shipping_bill_date") or "") result.port_code = str(data.get("port_code") or "") result.exporter_name = str(data.get("p_1b_exporter_name") or "") result.item_count = len(data.get("item_details") or []) result.shipping_bill_pages = extraction.shipping_bill_pages def _save_json(self, pdf_path: Path, data: dict[str, Any]) -> None: if not self.settings.save_json_files: return try: folder = self.settings.json_output_folder folder.mkdir(parents=True, exist_ok=True) with (folder / f"{pdf_path.stem}.json").open("w", encoding="utf-8") as fh: json.dump({"shipping_bill_json": data}, fh, indent=4, ensure_ascii=False, default=str) except OSError as exc: # the JSON copy is a convenience - never fail the PDF for it logger.warning("Could not save JSON for %s: %s", pdf_path.name, exc) def _raise_frappe(self, check: Check, exc: FrappeError) -> None: if exc.connection_error: self._consecutive_connection_failures += 1 raise ProcessingError(check, exc.message, frappe_action_hint(exc), connection_error=exc.connection_error) from exc @staticmethod def _not_processed(index: int, pdf: PdfFile, reason: str) -> ProcessResult: return ProcessResult( serial_no=index, file_name=pdf.path.name, file_path=str(pdf.path), status=Status.NOT_PROCESSED, message=f"Not processed: {reason}.", action_required="Run the script again - Shipping Bills already created are skipped automatically.", file_size_kb=round(pdf.size / 1024, 1), processed_at=datetime.now().strftime("%d-%m-%Y %H:%M:%S"), ) @staticmethod def _log_result(result: ProcessResult, total: int) -> None: width = len(str(total)) prefix = f"[{result.serial_no:>{width}}/{total}] {result.file_name}" if result.status == Status.FAILED: logger.error("%s -> FAILED at '%s': %s", prefix, result.failed_check, result.message) elif result.status == Status.NEEDS_REVIEW: logger.warning("%s -> NEEDS REVIEW: %s", prefix, result.message) elif result.status in DUPLICATE_STATUSES: logger.info("%s -> %s (duplicate of %s)", prefix, result.status.value.upper(), result.duplicate_of) else: doc = f" {result.frappe_doc_name}" if result.frappe_doc_name else "" logger.info("%s -> %s%s (SB %s, %.1fs)", prefix, result.status.value.upper(), doc, result.shipping_bill_no, result.duration_seconds) # ============================================================================= # EXCEL REPORT # ============================================================================= class Palette: NAVY = "1F3864" NAVY_2 = "2F5597" NAVY_LIGHT = "D9E1F2" MUTED = "7F7F7F" WHITE = "FFFFFF" ZEBRA = "F5F7FB" LINE = "D9D9D9" GREEN_BG, GREEN_FG, GREEN_SOLID = "E2EFDA", "375623", "70AD47" BLUE_BG, BLUE_FG, BLUE_SOLID = "DDEBF7", "1F4E78", "4472C4" AMBER_BG, AMBER_FG = "FFF2CC", "7F6000" ORANGE_BG, ORANGE_FG, ORANGE_SOLID = "FBE5D6", "C55A11", "ED7D31" RED_BG, RED_FG, RED_SOLID = "FCE4D6", "C00000", "E15759" GREY_BG, GREY_FG, GREY_SOLID = "EDEDED", "595959", "A5A5A5" # status -> (cell fill, font colour, chart colour, meaning, what the user should do) STATUS_STYLE: dict[Status, tuple[str, str, str, str, str]] = { Status.CREATED: (Palette.GREEN_BG, Palette.GREEN_FG, Palette.GREEN_SOLID, "New Shipping Bill created in Frappe and verified.", "Nothing."), Status.UPDATED: (Palette.GREEN_BG, Palette.GREEN_FG, "A9D18E", "Existing Shipping Bill overwritten with this PDF's data (DUPLICATE_ACTION = update).", "Nothing."), Status.VALIDATED: (Palette.BLUE_BG, Palette.BLUE_FG, Palette.BLUE_SOLID, "Passed every offline check. Dry run, so nothing was uploaded.", "Run again without --dry-run to upload."), Status.NEEDS_REVIEW: (Palette.ORANGE_BG, Palette.ORANGE_FG, Palette.ORANGE_SOLID, "Saved in Frappe, but the read-back check or the PDF attachment had a problem.", "Open the record in Frappe and check it (see 'Action Required')."), Status.ALREADY_IN_FRAPPE: (Palette.AMBER_BG, Palette.AMBER_FG, "FFC000", "This Shipping Bill (or this exact PDF) already exists in Frappe. Not created again.", "Nothing - this protects against duplicates."), Status.DUPLICATE_FILE: (Palette.AMBER_BG, Palette.AMBER_FG, "FFD966", "Identical copy of another PDF in the folder. Ignored.", "Optional: delete the extra copy from the folder."), Status.DUPLICATE_SB: (Palette.AMBER_BG, Palette.AMBER_FG, "F4B183", "Another PDF in the folder has the same SB No + Port Code. Only the first one was used.", "Check that both files are the same Shipping Bill."), Status.FAILED: (Palette.RED_BG, Palette.RED_FG, Palette.RED_SOLID, "Not uploaded. The reason and the fix are shown for each file.", "Fix the reason shown in 'Action Required' and run again."), Status.NOT_PROCESSED: (Palette.GREY_BG, Palette.GREY_FG, Palette.GREY_SOLID, "The run stopped before this PDF was reached.", "Run the script again."), } CHECK_STYLE = { CheckState.PASS: (Palette.GREEN_BG, Palette.GREEN_FG), CheckState.FAIL: (Palette.RED_BG, Palette.RED_FG), CheckState.WARN: (Palette.AMBER_BG, Palette.AMBER_FG), CheckState.NOT_RUN: (None, "BFBFBF"), } class ExcelReportWriter: """Builds the user-facing Excel report (Dashboard, details, actions, duplicates, guide).""" THIN = Side(style="thin", color=Palette.LINE) BORDER = Border(left=THIN, right=THIN, top=THIN, bottom=THIN) WRAP = Alignment(wrap_text=True, vertical="top") CENTER = Alignment(horizontal="center", vertical="center", wrap_text=True) FONT = "Calibri" DETAIL_COLUMNS: list[tuple[str, float, str]] = [ ("#", 5, "serial_no"), ("PDF File Name", 42, "file_name"), ("Result", 25, "status"), ("SB No", 12, "shipping_bill_no"), ("SB Date", 11, "shipping_bill_date"), ("Port", 9, "port_code"), ("Exporter", 30, "exporter_name"), ("Frappe Record", 15, "frappe_doc_name"), *[(f"{i}\n{check.value}", 9.5, f"check:{check.name}") for i, check in enumerate(Check)], ("Stopped At", 18, "failed_check"), ("Details", 58, "message"), ("How to Fix", 46, "action_required"), ("Duplicate Of", 30, "duplicate_of"), ("Warnings", 46, "warnings"), ("PDF Type", 14, "pdf_type"), ("Pages", 7, "total_pages"), ("SB Pages", 9, "shipping_bill_pages"), ("Items", 7, "item_count"), ("Size (KB)", 9, "file_size_kb"), ("Time (s)", 8, "duration_seconds"), ("Processed At", 18, "processed_at"), ("Full Path", 60, "file_path"), ] def __init__(self, settings: Settings): self.settings = settings # ---- public -------------------------------------------------------------------------- def write(self, summary: RunSummary) -> Path: wb = Workbook() self._dashboard(wb.active, summary) self._details(wb.create_sheet("PDF Details"), summary.results) self._action_required(wb.create_sheet("Action Required"), summary.results) self._duplicates(wb.create_sheet("Duplicates"), summary.results) self._guide(wb.create_sheet("Guide")) wb.active = 0 stamp = summary.started_at.strftime("%Y-%m-%d_%H-%M-%S") file_name = f"Shipping_Bill_Upload_Report_{stamp}.xlsx" for folder in (self.settings.report_folder, Path(__file__).resolve().parent, Path.cwd()): try: folder.mkdir(parents=True, exist_ok=True) path = folder / file_name wb.save(path) return path except OSError as exc: logger.warning("Could not save the report in %s (%s); trying another folder.", folder, exc) raise OSError("The Excel report could not be saved in any folder.") # ---- Dashboard --------------------------------------------------------------------------- def _dashboard(self, ws, s: RunSummary) -> None: ws.title = "Dashboard" ws.sheet_view.showGridLines = False ws.sheet_properties.tabColor = Palette.NAVY ws.column_dimensions["A"].width = 2 for col in range(2, 12): ws.column_dimensions[get_column_letter(col)].width = 13.5 ws.column_dimensions["L"].width = 2 total = len(s.results) dry = self.settings.dry_run attention = s.count(*ATTENTION_STATUSES) duplicates = s.count(*DUPLICATE_STATUSES) ok = total - attention finished = s.finished_at or datetime.now() # Title banner ws.row_dimensions[1].height = 8 self._band(ws, 2, "Shipping Bill Upload Report", Palette.NAVY, Font(name=self.FONT, size=20, bold=True, color=Palette.WHITE), height=40) mode = "Dry run (validation only)" if dry else "Live upload to Frappe" site = self.settings.frappe_url or "-" self._band(ws, 3, f"{mode} | Run on {s.started_at:%d-%m-%Y %H:%M} | {total} PDF(s) | " f"Frappe: {site}", Palette.NAVY, Font(name=self.FONT, size=10, color=Palette.NAVY_LIGHT), height=22) # Overall verdict ws.row_dimensions[4].height = 10 if s.aborted_reason or s.interrupted: reason = s.aborted_reason or "stopped by the user (Ctrl+C)" verdict = (f"! The run stopped early: {reason}. {attention} PDF(s) need attention - " "see 'Action Required'.", Palette.RED_BG, Palette.RED_FG) elif attention: verdict = (f"! {attention} of {total} PDF(s) need your attention - open the 'Action Required' " "sheet for the reason and the fix for each file.", Palette.RED_BG, Palette.RED_FG) elif total == 0: verdict = ("No PDF files were found in the folder.", Palette.GREY_BG, Palette.GREY_FG) else: extra = f" {duplicates} duplicate(s) were skipped safely." if duplicates else "" text = ("All PDFs passed validation and are ready to upload." if dry else "All PDFs were processed successfully. No action is needed.") verdict = (f"\u2713 {text}{extra}", Palette.GREEN_BG, Palette.GREEN_FG) self._band(ws, 5, verdict[0], verdict[1], Font(name=self.FONT, size=12, bold=True, color=verdict[2]), height=32) ws.row_dimensions[6].height = 12 # KPI tiles rate = f"{(ok / total):.0%}" if total else "-" if dry: second = ("VALIDATED", s.count(Status.VALIDATED), "ready to upload", Palette.BLUE_BG, Palette.BLUE_FG) else: second = ("CREATED / UPDATED", s.count(Status.CREATED, Status.UPDATED), "Shipping Bills saved", Palette.GREEN_BG, Palette.GREEN_FG) tiles = [ ("TOTAL PDFs", total, f"{s.unique_files} unique file(s)", Palette.NAVY_LIGHT, Palette.NAVY), second, ("DUPLICATES SKIPPED", duplicates, "not uploaded twice", Palette.AMBER_BG, Palette.AMBER_FG), ("NEED ATTENTION", attention, "failed / needs review", Palette.RED_BG if attention else Palette.GREEN_BG, Palette.RED_FG if attention else Palette.GREEN_FG), ("SUCCESS RATE", rate, "processed without errors", Palette.BLUE_BG, Palette.BLUE_FG), ] for i, (label, value, caption, bg, fg) in enumerate(tiles): c1 = 2 + i * 2 fill = PatternFill("solid", fgColor=bg) for row, val, font, height in ( (7, label, Font(name=self.FONT, size=9, bold=True, color=fg), 20), (8, value, Font(name=self.FONT, size=26, bold=True, color=fg), 44), (9, caption, Font(name=self.FONT, size=9, color=fg), 20), ): ws.merge_cells(start_row=row, start_column=c1, end_row=row, end_column=c1 + 1) cell = ws.cell(row=row, column=c1, value=val) cell.font, cell.alignment = font, Alignment(horizontal="center", vertical="center") for col in (c1, c1 + 1): ws.cell(row=row, column=col).fill = fill ws.row_dimensions[row].height = height # Result breakdown + chart row = 11 self._section(ws, row, "Result Breakdown") row += 1 self._table_header(ws, row, [(2, 3, "Result"), (4, 4, "PDFs"), (5, 5, "Share"), (6, 8, "What it means")]) first_data_row = row + 1 always = {Status.VALIDATED, Status.DUPLICATE_FILE, Status.DUPLICATE_SB, Status.FAILED} if dry else \ {Status.CREATED, Status.ALREADY_IN_FRAPPE, Status.DUPLICATE_FILE, Status.DUPLICATE_SB, Status.FAILED} shown = [st for st in Status if s.count(st) or st in always] chart_colors = [] for st in shown: row += 1 bg, fg, solid, meaning, _ = STATUS_STYLE[st] count = s.count(st) self._merged(ws, row, 2, 3, st.value, font=Font(name=self.FONT, bold=True, color=fg), fill=bg, align=Alignment(vertical="center", indent=1)) self._merged(ws, row, 4, 4, count, align=self.CENTER) self._merged(ws, row, 5, 5, (count / total) if total else 0, align=self.CENTER, number_format="0%") self._merged(ws, row, 6, 8, meaning, align=Alignment(wrap_text=True, vertical="center"), font=Font(name=self.FONT, size=9, color=Palette.GREY_FG)) ws.row_dimensions[row].height = 30 chart_colors.append(solid) last_data_row = row row += 1 self._merged(ws, row, 2, 3, "Total", font=Font(name=self.FONT, bold=True), align=Alignment(vertical="center", indent=1), fill=Palette.NAVY_LIGHT) self._merged(ws, row, 4, 4, total, font=Font(name=self.FONT, bold=True), align=self.CENTER, fill=Palette.NAVY_LIGHT) self._merged(ws, row, 5, 5, 1 if total else 0, font=Font(name=self.FONT, bold=True), align=self.CENTER, number_format="0%", fill=Palette.NAVY_LIGHT) self._merged(ws, row, 6, 8, "", fill=Palette.NAVY_LIGHT) ws.conditional_formatting.add( f"D{first_data_row}:D{last_data_row}", DataBarRule(start_type="num", start_value=0, end_type="max", color=Palette.BLUE_SOLID, showValue=True)) if total: table_points = 22 + 30 * len(shown) + 15 # header + rows + total row self._doughnut(ws, first_data_row, last_data_row, chart_colors, anchor=f"I{first_data_row - 1}", height_cm=max(table_points * 0.0353, 6.5)) row += 3 # Validation steps self._section(ws, row, "Validation Steps - how many PDFs passed each check") row += 1 self._table_header(ws, row, [(2, 2, "Step"), (3, 4, "Check"), (5, 7, "What is checked"), (8, 8, "\u2713 Passed"), (9, 9, "\u2717 Failed / Skipped"), (10, 10, "! Warning"), (11, 11, "\u2014 Not reached")]) for i, check in enumerate(Check): row += 1 states = [r.checks[check] for r in s.results] passed = states.count(CheckState.PASS) failed = states.count(CheckState.FAIL) warned = states.count(CheckState.WARN) not_run = states.count(CheckState.NOT_RUN) # A stop on a duplicate check is protection, not an error -> amber. duplicate_check = check in {Check.UNIQUE_FILE, Check.UNIQUE_SB, Check.NOT_IN_FRAPPE} fail_bg, fail_fg = ((Palette.AMBER_BG, Palette.AMBER_FG) if duplicate_check else (Palette.RED_BG, Palette.RED_FG)) not_applicable = (dry and check in {Check.NOT_IN_FRAPPE, Check.SAVED, Check.VERIFIED, Check.ATTACHED}) or \ (check == Check.ATTACHED and not self.settings.attach_pdf) self._merged(ws, row, 2, 2, i, align=self.CENTER, font=Font(name=self.FONT, bold=True, color=Palette.NAVY)) self._merged(ws, row, 3, 4, check.value, font=Font(name=self.FONT, bold=True), align=Alignment(vertical="center", indent=1)) self._merged(ws, row, 5, 7, CHECK_DESCRIPTIONS[check] + (" (not run in this mode)" if not_applicable else ""), align=Alignment(wrap_text=True, vertical="center"), font=Font(name=self.FONT, size=9, color=Palette.GREY_FG)) self._merged(ws, row, 8, 8, passed, align=self.CENTER, font=Font(name=self.FONT, bold=True, color=Palette.GREEN_FG if passed else Palette.MUTED)) self._merged(ws, row, 9, 9, failed, align=self.CENTER, font=Font(name=self.FONT, bold=True, color=fail_fg if failed else Palette.MUTED), fill=fail_bg if failed else None) self._merged(ws, row, 10, 10, warned, align=self.CENTER, font=Font(name=self.FONT, bold=True, color=Palette.AMBER_FG if warned else Palette.MUTED), fill=Palette.AMBER_BG if warned else None) self._merged(ws, row, 11, 11, not_run, align=self.CENTER, font=Font(name=self.FONT, color=Palette.MUTED)) ws.row_dimensions[row].height = 30 # Next steps row += 2 self._section(ws, row, "What To Do Next") for text in self._next_steps(s): row += 1 self._merged(ws, row, 2, 11, f"\u2022 {text}", border=False, align=Alignment(wrap_text=True, vertical="center")) ws.row_dimensions[row].height = 30 if len(text) > 120 else 18 # Run information row += 2 self._section(ws, row, "Run Information") duration = (finished - s.started_at).total_seconds() if s.aborted_reason: outcome = f"Stopped early - {s.aborted_reason}" elif s.interrupted: outcome = "Stopped by user (Ctrl+C)" else: outcome = "Completed" attach = ("Off" if not self.settings.attach_pdf else f"On ({'private' if self.settings.attach_private else 'public'} file)") info = [ ("Run result", outcome), ("Mode", mode), ("Started / Finished", f"{s.started_at:%d-%m-%Y %H:%M:%S} -> {finished:%d-%m-%Y %H:%M:%S}"), ("Duration", self._format_duration(duration)), ("PDF folder", str(self.settings.pdf_folder)), ("Files in folder", f"{s.total_files} PDF(s), {s.unique_files} unique"), ("Frappe site", site), ("Frappe user", s.frappe_user or "-"), ("DocType", self.settings.doctype), ("JSON validation", s.schema_status or "-"), ("Duplicate rule", f"{self.settings.duplicate_action} (match on " f"{' + '.join(self.settings.duplicate_match_fields)})"), ("PDF attachment", attach), ("JSON copies", str(self.settings.json_output_folder) if self.settings.save_json_files else "Off"), ("Generated by", f"{SCRIPT_NAME} v{SCRIPT_VERSION}"), ] for label, value in info: row += 1 self._merged(ws, row, 2, 3, label, font=Font(name=self.FONT, bold=True, color=Palette.NAVY), fill=Palette.ZEBRA, align=Alignment(vertical="center", indent=1)) self._merged(ws, row, 4, 11, value, align=Alignment(wrap_text=True, vertical="center", indent=1)) ws.row_dimensions[row].height = 30 if len(str(value)) > 95 else 18 ws.page_setup.orientation = "portrait" ws.page_setup.fitToWidth, ws.page_setup.fitToHeight = 1, 0 ws.sheet_properties.pageSetUpPr.fitToPage = True ws.print_area = f"A1:L{row + 1}" def _next_steps(self, s: RunSummary) -> list[str]: steps = [] failed, review = s.count(Status.FAILED), s.count(Status.NEEDS_REVIEW) not_processed = s.count(Status.NOT_PROCESSED) if failed: steps.append(f"{failed} PDF(s) failed. Open 'Action Required', fix the reason shown for each file, " "then run the script again. Shipping Bills that are already in Frappe are skipped " "automatically, so re-running is always safe.") if review: steps.append(f"{review} Shipping Bill(s) were saved but need a quick check in Frappe " "(read-back check or PDF attachment problem - see 'Action Required').") if not_processed: steps.append(f"{not_processed} PDF(s) were not processed because the run stopped early. " "Run the script again to finish them.") if s.count(Status.DUPLICATE_FILE): n = s.count(Status.DUPLICATE_FILE) steps.append(f"{n} identical PDF {'copy was' if n == 1 else 'copies were'} found in the folder " "and ignored. You can delete them (listed in 'Duplicates').") if s.count(Status.DUPLICATE_SB): steps.append(f"{s.count(Status.DUPLICATE_SB)} PDF(s) contain a Shipping Bill that appears more than " "once in the folder; only the first file was used (see 'Duplicates').") if s.count(Status.ALREADY_IN_FRAPPE): steps.append(f"{s.count(Status.ALREADY_IN_FRAPPE)} Shipping Bill(s) already existed in Frappe and " "were not created again.") with_warnings = sum(1 for r in s.results if r.warnings) if with_warnings: steps.append(f"{with_warnings} PDF(s) have warnings (for example extracted fields that do not exist " "in the DocType and are ignored). See the 'Warnings' column in 'PDF Details'.") if self.settings.dry_run and s.count(Status.VALIDATED): steps.append(f"This was a dry run. Run the script without --dry-run to upload the " f"{s.count(Status.VALIDATED)} validated PDF(s).") if not steps: steps.append("Nothing to do - every Shipping Bill is in Frappe.") return steps def _doughnut(self, ws, first_row: int, last_row: int, colors: list[str], anchor: str, height_cm: float) -> None: chart = DoughnutChart(holeSize=55) chart.title = "Results" labels = Reference(ws, min_col=2, min_row=first_row, max_row=last_row) data = Reference(ws, min_col=4, min_row=first_row, max_row=last_row) chart.add_data(data, titles_from_data=False) chart.set_categories(labels) series = chart.series[0] for idx, color in enumerate(colors): point = DataPoint(idx=idx) point.graphicalProperties.solidFill = color point.graphicalProperties.line.solidFill = Palette.WHITE series.dPt.append(point) series.dLbls = DataLabelList() series.dLbls.showVal = True for attr in ("showSerName", "showCatName", "showLegendKey", "showPercent"): setattr(series.dLbls, attr, False) chart.legend = None # the colour-coded table beside the chart acts as the legend chart.width, chart.height = 7.9, height_cm ws.add_chart(chart, anchor) # ---- PDF Details --------------------------------------------------------------------------- def _details(self, ws, results: list[ProcessResult]) -> None: ws.sheet_properties.tabColor = Palette.NAVY_2 ws.sheet_view.zoomScale = 90 groups = [("PDF", 1, 3), ("SHIPPING BILL", 4, 8), ("VALIDATION STEPS \u2713 passed \u2717 failed / stopped ! warning " "\u2014 not reached", 9, 8 + len(Check)), ("OUTCOME", 9 + len(Check), 13 + len(Check)), ("PDF INFO", 14 + len(Check), len(self.DETAIL_COLUMNS))] for title, c1, c2 in groups: ws.merge_cells(start_row=1, start_column=c1, end_row=1, end_column=c2) cell = ws.cell(row=1, column=c1, value=title) cell.font = Font(name=self.FONT, bold=True, size=10, color=Palette.WHITE) cell.alignment = Alignment(horizontal="center", vertical="center") for col in range(c1, c2 + 1): ws.cell(row=1, column=col).fill = PatternFill("solid", fgColor=Palette.NAVY_2) ws.cell(row=1, column=col).border = Border(left=Side(style="medium", color=Palette.WHITE) if col == c1 else None) ws.row_dimensions[1].height = 22 for col, (header, width, _) in enumerate(self.DETAIL_COLUMNS, start=1): ws.column_dimensions[get_column_letter(col)].width = width self._header_cell(ws.cell(row=2, column=col, value=header)) ws.row_dimensions[2].height = 46 ws.freeze_panes = "D3" if not results: ws.cell(row=3, column=2, value="No PDF files were found in the folder.") return for i, r in enumerate(results): row = 3 + i zebra = PatternFill("solid", fgColor=Palette.ZEBRA) if i % 2 else None is_duplicate = r.status in DUPLICATE_STATUSES for col, (_, width, attr) in enumerate(self.DETAIL_COLUMNS, start=1): cell = ws.cell(row=row, column=col) cell.border = self.BORDER cell.font = Font(name=self.FONT, size=10) cell.alignment = self.WRAP if zebra: cell.fill = zebra if attr.startswith("check:"): state = r.checks[Check[attr.split(":", 1)[1]]] bg, fg = CHECK_STYLE[state] if state == CheckState.FAIL and is_duplicate: # stopped on purpose, not an error bg, fg = Palette.AMBER_BG, Palette.AMBER_FG cell.value = state.value cell.font = Font(name=self.FONT, size=12, bold=True, color=fg) cell.alignment = Alignment(horizontal="center", vertical="top") if bg: cell.fill = PatternFill("solid", fgColor=bg) continue value = getattr(r, attr) if attr == "status": bg, fg, *_ = STATUS_STYLE[r.status] value = r.status.value cell.fill = PatternFill("solid", fgColor=bg) cell.font = Font(name=self.FONT, size=10, bold=True, color=fg) elif attr == "warnings": value = "\n".join(value) elif attr == "frappe_doc_name" and r.frappe_doc_url: cell.hyperlink = r.frappe_doc_url cell.font = Font(name=self.FONT, size=10, color="0563C1", underline="single") elif attr in {"serial_no", "total_pages", "item_count", "file_size_kb", "duration_seconds"}: cell.alignment = Alignment(horizontal="center", vertical="top") numeric_blank = attr in {"total_pages", "item_count", "file_size_kb", "duration_seconds"} \ and not value cell.value = None if value in ("", None) or numeric_blank else value ws.row_dimensions[row].height = self._row_height(r, [ (r.file_name, 42), (r.message, 58), (r.action_required, 46), ("\n".join(r.warnings), 46), (r.duplicate_of, 30), (r.exporter_name, 30)]) last = 2 + len(results) ws.auto_filter.ref = f"A2:{get_column_letter(len(self.DETAIL_COLUMNS))}{last}" ws.page_setup.orientation = "landscape" ws.print_title_rows = "1:2" # ---- Action Required / Duplicates ------------------------------------------------------------ def _action_required(self, ws, results: list[ProcessResult]) -> None: rows = [r for r in results if r.status in ATTENTION_STATUSES] ws.sheet_properties.tabColor = Palette.RED_FG if rows else Palette.GREEN_SOLID columns = [("#", 5), ("PDF File Name", 42), ("Result", 22), ("Stopped At", 18), ("What Went Wrong", 60), ("How to Fix", 50), ("SB No", 12), ("Frappe Record", 15), ("Full Path", 60)] self._list_sheet_header( ws, columns, "Action Required - PDFs that were not (fully) uploaded", "Fix the problem shown for each file and run the script again. Shipping Bills that are already " "in Frappe are skipped automatically, so re-running is always safe.", Palette.RED_BG, Palette.RED_FG) if not rows: self._empty_message(ws, len(columns), "\u2713 Nothing needs your attention. Every PDF was " "processed without errors.") return for i, r in enumerate(rows): values = [r.serial_no, r.file_name, r.status.value, r.failed_check or "-", r.message, r.action_required, r.shipping_bill_no, r.frappe_doc_name, r.file_path] self._list_row(ws, 5 + i, i, values, r, status_col=3, link_col=8) ws.row_dimensions[5 + i].height = self._row_height(r, [(r.message, 60), (r.action_required, 50), (r.file_name, 42)]) ws.auto_filter.ref = f"A4:{get_column_letter(len(columns))}{4 + len(rows)}" def _duplicates(self, ws, results: list[ProcessResult]) -> None: rows = [r for r in results if r.status in DUPLICATE_STATUSES] ws.sheet_properties.tabColor = "FFC000" columns = [("#", 5), ("PDF File Name", 42), ("Duplicate Type", 26), ("Duplicate Of", 42), ("SB No", 12), ("Port", 9), ("Frappe Record", 15), ("Explanation", 70)] self._list_sheet_header( ws, columns, "Duplicates - PDFs skipped so nothing is uploaded twice", "These files were NOT uploaded because the same PDF or the same Shipping Bill was already " "processed. This is expected protection, not an error.", Palette.AMBER_BG, Palette.AMBER_FG) if not rows: self._empty_message(ws, len(columns), "\u2713 No duplicates found - every PDF in the folder is a " "different Shipping Bill.") return labels = {Status.DUPLICATE_FILE: "Identical file in folder", Status.DUPLICATE_SB: "Same SB No + Port in folder", Status.ALREADY_IN_FRAPPE: "Already in Frappe"} for i, r in enumerate(rows): values = [r.serial_no, r.file_name, labels[r.status], r.duplicate_of, r.shipping_bill_no, r.port_code, r.frappe_doc_name, r.message] self._list_row(ws, 5 + i, i, values, r, status_col=3, link_col=7) ws.row_dimensions[5 + i].height = self._row_height(r, [(r.message, 70), (r.file_name, 42), (r.duplicate_of, 42)]) ws.auto_filter.ref = f"A4:{get_column_letter(len(columns))}{4 + len(rows)}" # ---- Guide -------------------------------------------------------------------------------------- def _guide(self, ws) -> None: ws.sheet_properties.tabColor = Palette.GREY_SOLID ws.sheet_view.showGridLines = False for col, width in zip("ABCDE", (2, 30, 62, 50, 2)): ws.column_dimensions[col].width = width ws.merge_cells("B2:D2") ws["B2"] = "How to Read This Report" ws["B2"].font = Font(name=self.FONT, size=18, bold=True, color=Palette.NAVY) ws.row_dimensions[2].height = 30 ws.merge_cells("B3:D3") ws["B3"] = ("Start with the Dashboard. Anything that needs you is listed in 'Action Required'. " "'PDF Details' shows every file with a tick or cross for each validation step.") ws["B3"].font = Font(name=self.FONT, size=10, color=Palette.GREY_FG) ws["B3"].alignment = Alignment(wrap_text=True) ws.row_dimensions[3].height = 30 row = 5 self._guide_table(ws, row, ("Result", "Meaning", "What you should do"), [(st.value, STATUS_STYLE[st][3], STATUS_STYLE[st][4]) for st in Status], styles=[STATUS_STYLE[st][:2] for st in Status]) row += len(Status) + 3 self._guide_table(ws, row, ("Validation step", "What is checked", "If it fails"), [(f"{i}. {c.value}", CHECK_DESCRIPTIONS[c], self._check_failure_hint(c)) for i, c in enumerate(Check)]) row += len(Check) + 3 symbols = [(CheckState.PASS.value + " Passed", "The check was successful.", ""), (CheckState.FAIL.value + " Failed / stopped", "Red = an error. Amber = stopped on purpose to prevent a duplicate.", ""), (CheckState.WARN.value + " Warning", "Passed, but something should be looked at " "(see the Warnings column).", ""), (CheckState.NOT_RUN.value + " Not reached", "The PDF stopped at an earlier step, or the " "step is not used in this mode.", "")] self._guide_table(ws, row, ("Symbol", "Meaning", ""), symbols) ws.page_setup.orientation = "portrait" ws.page_setup.fitToWidth, ws.page_setup.fitToHeight = 1, 0 ws.sheet_properties.pageSetUpPr.fitToPage = True @staticmethod def _check_failure_hint(check: Check) -> str: return { Check.UNIQUE_FILE: "Skipped as a duplicate copy. Delete the extra copy if you like.", Check.PDF_READ: "Get the original ICEGATE PDF (not a scan / not corrupted).", Check.EXTRACTED: "File is not a standard ICEGATE Shipping Bill. Check it manually.", Check.JSON_VALID: "Check the listed values in the PDF, or adjust the DocType field rule.", Check.UNIQUE_SB: "Skipped - another file in the folder is the same Shipping Bill.", Check.NOT_IN_FRAPPE: "Skipped - already in Frappe. Use DUPLICATE_ACTION = 'update' to overwrite.", Check.SAVED: "Frappe rejected the record. The reason from Frappe is shown.", Check.VERIFIED: "Record saved but differs from the PDF. Open it in Frappe and check.", Check.ATTACHED: "Record saved but the PDF is not attached. Attach it manually.", }[check] # ---- building blocks ------------------------------------------------------------------------------ def _guide_table(self, ws, row: int, headers: tuple[str, str, str], rows: list[tuple[str, str, str]], styles: Optional[list[tuple[str, str]]] = None) -> None: for col, header in enumerate(headers, start=2): self._header_cell(ws.cell(row=row, column=col, value=header), align_left=True) for i, values in enumerate(rows, start=1): for col, value in enumerate(values, start=2): cell = ws.cell(row=row + i, column=col, value=value) cell.border = self.BORDER cell.alignment = Alignment(wrap_text=True, vertical="center", indent=1 if col == 2 else 0) cell.font = Font(name=self.FONT, size=10, bold=col == 2) if col == 2 and styles: bg, fg = styles[i - 1] cell.fill = PatternFill("solid", fgColor=bg) cell.font = Font(name=self.FONT, size=10, bold=True, color=fg) ws.row_dimensions[row + i].height = 32 def _list_sheet_header(self, ws, columns: list[tuple[str, float]], title: str, note: str, bg: str, fg: str) -> None: ws.sheet_view.zoomScale = 90 last = get_column_letter(len(columns)) ws.merge_cells(f"A1:{last}1") ws["A1"] = title ws["A1"].font = Font(name=self.FONT, size=15, bold=True, color=fg) ws["A1"].alignment = Alignment(vertical="center", indent=1) ws.merge_cells(f"A2:{last}2") ws["A2"] = note ws["A2"].font = Font(name=self.FONT, size=10, color=fg) ws["A2"].alignment = Alignment(vertical="center", indent=1, wrap_text=True) for col in range(1, len(columns) + 1): ws.cell(row=1, column=col).fill = PatternFill("solid", fgColor=bg) ws.cell(row=2, column=col).fill = PatternFill("solid", fgColor=bg) ws.row_dimensions[1].height = 30 ws.row_dimensions[2].height = 22 for col, (header, width) in enumerate(columns, start=1): ws.column_dimensions[get_column_letter(col)].width = width self._header_cell(ws.cell(row=4, column=col, value=header)) ws.row_dimensions[4].height = 24 ws.freeze_panes = "C5" ws.page_setup.orientation = "landscape" ws.print_title_rows = "4:4" def _list_row(self, ws, row: int, index: int, values: list[Any], result: ProcessResult, status_col: int, link_col: int) -> None: zebra = PatternFill("solid", fgColor=Palette.ZEBRA) if index % 2 else None for col, value in enumerate(values, start=1): cell = ws.cell(row=row, column=col, value=value if value != "" else None) cell.border = self.BORDER cell.alignment = self.WRAP if col != 1 else Alignment(horizontal="center", vertical="top") cell.font = Font(name=self.FONT, size=10) if zebra: cell.fill = zebra if col == status_col: bg, fg, *_ = STATUS_STYLE[result.status] cell.fill = PatternFill("solid", fgColor=bg) cell.font = Font(name=self.FONT, size=10, bold=True, color=fg) if col == link_col and result.frappe_doc_url: cell.hyperlink = result.frappe_doc_url cell.font = Font(name=self.FONT, size=10, color="0563C1", underline="single") def _empty_message(self, ws, width: int, text: str) -> None: ws.merge_cells(start_row=5, start_column=1, end_row=5, end_column=width) cell = ws.cell(row=5, column=1, value=text) cell.font = Font(name=self.FONT, size=12, bold=True, color=Palette.GREEN_FG) cell.fill = PatternFill("solid", fgColor=Palette.GREEN_BG) cell.alignment = Alignment(vertical="center", indent=1) ws.row_dimensions[5].height = 30 def _band(self, ws, row: int, text: str, fill: str, font: Font, height: float) -> None: ws.merge_cells(start_row=row, start_column=2, end_row=row, end_column=11) cell = ws.cell(row=row, column=2, value=text) cell.font = font cell.alignment = Alignment(vertical="center", indent=1, wrap_text=True) for col in range(2, 12): ws.cell(row=row, column=col).fill = PatternFill("solid", fgColor=fill) ws.row_dimensions[row].height = height def _section(self, ws, row: int, title: str) -> None: cell = ws.cell(row=row, column=2, value=title) cell.font = Font(name=self.FONT, size=13, bold=True, color=Palette.NAVY) for col in range(2, 12): ws.cell(row=row, column=col).border = Border(bottom=Side(style="medium", color=Palette.NAVY)) ws.row_dimensions[row].height = 24 def _table_header(self, ws, row: int, spans: list[tuple[int, int, str]]) -> None: for c1, c2, title in spans: if c2 > c1: ws.merge_cells(start_row=row, start_column=c1, end_row=row, end_column=c2) for col in range(c1, c2 + 1): self._header_cell(ws.cell(row=row, column=col, value=title if col == c1 else None)) ws.row_dimensions[row].height = 22 def _merged(self, ws, row: int, c1: int, c2: int, value: Any, font: Optional[Font] = None, fill: Optional[str] = None, align: Optional[Alignment] = None, border: bool = True, number_format: Optional[str] = None) -> None: if c2 > c1: ws.merge_cells(start_row=row, start_column=c1, end_row=row, end_column=c2) for col in range(c1, c2 + 1): cell = ws.cell(row=row, column=col) if border: cell.border = self.BORDER if fill: cell.fill = PatternFill("solid", fgColor=fill) cell = ws.cell(row=row, column=c1, value=value) cell.font = font or Font(name=self.FONT, size=10) cell.alignment = align or Alignment(vertical="center") if number_format: cell.number_format = number_format def _header_cell(self, cell, align_left: bool = False) -> None: cell.fill = PatternFill("solid", fgColor=Palette.NAVY) cell.font = Font(name=self.FONT, bold=True, size=10, color=Palette.WHITE) cell.border = self.BORDER cell.alignment = (Alignment(vertical="center", indent=1, wrap_text=True) if align_left else self.CENTER) @staticmethod def _row_height(result: ProcessResult, texts: list[tuple[str, float]]) -> float: lines = 1 for text, width in texts: if not text: continue chars_per_line = max(int(width * 1.15), 10) text_lines = sum(max(1, -(-len(part) // chars_per_line)) for part in str(text).split("\n")) lines = max(lines, text_lines) return min(15 * lines + 4, 150) @staticmethod def _format_duration(seconds: float) -> str: seconds = int(round(seconds)) hours, rem = divmod(seconds, 3600) minutes, secs = divmod(rem, 60) if hours: return f"{hours}h {minutes}m {secs}s" if minutes: return f"{minutes}m {secs}s" return f"{secs}s" # ============================================================================= # ENTRY POINT # ============================================================================= def setup_logging(log_folder: Path, started_at: datetime) -> Optional[Path]: logger.setLevel(logging.DEBUG) logger.handlers.clear() formatter = logging.Formatter("%(asctime)s | %(levelname)-7s | %(message)s", "%Y-%m-%d %H:%M:%S") console = logging.StreamHandler(sys.stdout) console.setLevel(logging.INFO) console.setFormatter(formatter) logger.addHandler(console) log_path = None try: log_folder.mkdir(parents=True, exist_ok=True) log_path = log_folder / f"shipping_bill_upload_{started_at:%Y-%m-%d_%H-%M-%S}.log" file_handler = logging.FileHandler(log_path, encoding="utf-8") file_handler.setLevel(logging.DEBUG) file_handler.setFormatter(formatter) logger.addHandler(file_handler) except OSError as exc: logger.warning("File logging disabled (%s).", exc) # pdfminer is very noisy about harmless font/encoding issues. for noisy in ("pdfminer", "pdfplumber", "urllib3"): logging.getLogger(noisy).setLevel(logging.ERROR) return log_path def open_file(path: Path) -> None: try: if os.name == "nt": os.startfile(str(path)) # type: ignore[attr-defined] elif sys.platform == "darwin": subprocess.Popen(["open", str(path)]) else: subprocess.Popen(["xdg-open", str(path)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) except Exception as exc: # opening the report is a convenience only logger.warning("Could not open the report automatically: %s", exc) def parse_args(argv: Optional[list[str]] = None) -> argparse.Namespace: parser = argparse.ArgumentParser( description="Convert Shipping Bill PDFs to JSON and create Shipping Bills in Frappe.") parser.add_argument("--folder", help="PDF folder (overrides PDF_FOLDER in CONFIG)") parser.add_argument("--dry-run", action="store_true", help="Extract and validate only; nothing is sent to Frappe") parser.add_argument("--no-open", action="store_true", help="Do not open the Excel report when finished") return parser.parse_args(argv) def main(argv: Optional[list[str]] = None) -> int: args = parse_args(argv) started_at = datetime.now() try: settings = Settings.load(CONFIG, args) except ConfigError as exc: print(f"\nERROR: {exc}\n\nEdit the CONFIGURATION section at the top of this script.", file=sys.stderr) return 2 log_path = setup_logging(settings.log_folder, started_at) logger.info("%s v%s - %s", SCRIPT_NAME, SCRIPT_VERSION, "DRY RUN (nothing will be sent to Frappe)" if settings.dry_run else "LIVE UPLOAD") try: summary = ShippingBillBatchUploader(settings).run() except ConfigError as exc: logger.error("%s", exc) logger.error("No PDFs were processed.") return 2 report_path = ExcelReportWriter(settings).write(summary) total = len(summary.results) logger.info("-" * 72) logger.info("Finished: %d PDF(s) in %s", total, ExcelReportWriter._format_duration( ((summary.finished_at or datetime.now()) - summary.started_at).total_seconds())) for status in Status: count = summary.count(status) if count: logger.info(" %-30s %d", status.value, count) logger.info("Excel report : %s", report_path) if log_path: logger.info("Log file : %s", log_path) if settings.save_json_files and total: logger.info("JSON files : %s", settings.json_output_folder) logger.info("-" * 72) if settings.open_report_when_done: open_file(report_path) needs_attention = summary.count(*ATTENTION_STATUSES) > 0 return 1 if needs_attention or summary.interrupted else 0 if __name__ == "__main__": sys.exit(main())