|
#!/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 = "<PDF_FOLDER>/_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<len(rows):
|
|
rr=rows[i+1]
|
|
vals=[self._clean(x) for x in rr if self._clean(x)]
|
|
nums=[x for x in vals if self._float(x) is not None]
|
|
if len(nums)>=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):
|
|
rr=rows[i+1]
|
|
for j,x in enumerate(rr):
|
|
x=self._clean(x)
|
|
if not x: continue
|
|
if self._compact(r[j] if j<len(r) else "")=="pkg": result["pkg"] = self._integer(x)
|
|
if self._compact(r[j] if j<len(r) else "")=="gwt": result["gwt"] = self._float(x) if self._float(x) is not None else result.get("gwt",0)
|
|
if self._compact(r[j] if j<len(r) else "") in {"kgs","kg"}: result["gwt_unit"]="KGS"
|
|
text=self._row_text(rr).upper()
|
|
if "KGS" in text: result["gwt_unit"]="KGS"
|
|
if result.get("pkg",0)==0:
|
|
m2=re.search(r"PKG\s+(\d+(?:\.\d+)?)", self._row_text(rr), re.I)
|
|
if m2: result["pkg"]=self._integer(m2.group(1))
|
|
if result.get("gwt",0)==0:
|
|
m2=re.search(r"G\.?WT\s+(?:KGS\s+)?(\d+(?:\.\d+)?)", self._row_text(rr), re.I)
|
|
if m2: result["gwt"]=self._float(m2.group(1))
|
|
# Final safe fallback for the top header only.
|
|
if not result.get("gwt"):
|
|
self._recover_gwt_from_text(result)
|
|
if not result.get("gwt_unit"): result["gwt_unit"]="KGS" if re.search(r"G\.?WT\s+KGS",head,re.I) else ""
|
|
def _extract_statuses(self, result):
|
|
for t in self.tables:
|
|
rows=t["rows"]
|
|
for i,row in enumerate(rows):
|
|
compact=self._compact(self._row_text(row))
|
|
if "1mode" not in compact or "2assess" not in compact or "3exmn" not in compact: continue
|
|
if 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<len(values):
|
|
v=self._clean_status(values[j])
|
|
if token=="p_1a_mode":
|
|
if v and re.fullmatch(r"[A-Z]{2,12}",v): mapping[token]=v
|
|
elif v in {"Y","N"}: mapping[token]=v
|
|
for f in self.STATUS_FIELDS:
|
|
if self._is_missing(f,result.get(f)): result[f]="N"
|
|
for f,v in mapping.items(): result[f]=v; self._trace(f,v,t["page"])
|
|
return
|
|
# Coordinate fallback: find labels then the nearest single status token.
|
|
for f in self.STATUS_FIELDS + ["p_1a_mode"]:
|
|
if not self._is_missing(f,result.get(f)): continue
|
|
aliases=self.PARENT_FIELD_ALIASES.get(f, self.LABELS.get(f,[]))
|
|
val,_=self._semantic_value(aliases, lambda x: self._clean_status(x) in {"Y","N"} or (f=="p_1a_mode" and bool(re.fullmatch(r"[A-Za-z]{2,12}",x))))
|
|
if val: result[f]=self._clean_status(val) if f!="p_1a_mode" else val.upper()
|
|
|
|
def _extract_locations(self,result):
|
|
fields=["port_of_loading","state_of_origin","port_of_finaldstn","cntry_of_finaldstn","port_of_discharge","cntry_of_discharge"]
|
|
for f in fields:
|
|
aliases=self.PARENT_FIELD_ALIASES.get(f,self.LABELS.get(f,[]))
|
|
val=self._table_field_value(aliases, field=f)
|
|
if val is None: val,_=self._semantic_value(aliases,lambda x:self._valid_location_candidate(f,x),max_dx=700,max_dy=100)
|
|
if val is not None: result[f]=self._clean_location(val)
|
|
|
|
def _extract_parties(self,result):
|
|
for t in self.tables:
|
|
rows=t["rows"]
|
|
for i,r in enumerate(rows):
|
|
exp_idx=next((j for j,c in enumerate(r) if "exportersnameaddress" in self._compact(c)),None)
|
|
con_idx=next((j for j,c in enumerate(r) if "consigneenameaddress" in self._compact(c)),None)
|
|
if exp_idx is None or con_idx is None: continue
|
|
# Only consume the three party-address rows immediately below
|
|
# the labels. This prevents AD/CB/bank fields from leaking in.
|
|
left=[]; right=[]
|
|
for rr in rows[i+1:i+5]:
|
|
cells=[self._clean(x) for x in rr]
|
|
left += [x for x in cells[exp_idx:con_idx] if x]
|
|
right += [x for x in cells[con_idx:] if x]
|
|
right=[x for x in right if x not in {"P","O","SA","C","Y"}]
|
|
if left:
|
|
result["p_1b_exporter_name"]=self._clean_party_value("p_1b_exporter_name",left[0])
|
|
result["p_1b_exporter_address"]=self._clean_address("\n".join(left[1:]))
|
|
if right:
|
|
result["p_1b_consignee_name"]=self._clean_party_value("p_1b_consignee_name",right[0])
|
|
result["p_1b_consignee_address"]=self._clean_address("\n".join(right[1:]))
|
|
def next_nonempty(rr,j):
|
|
for k in range(j+1,len(rr)):
|
|
v=self._clean(rr[k])
|
|
if not v: continue
|
|
if self._looks_like_any_known_label(v): return ""
|
|
if re.match(r"^[0-9]+\s*\.",v): return ""
|
|
return v
|
|
return ""
|
|
# Scalar party/bank fields must be paired on the same visual row.
|
|
for rr in rows[i+1:i+10]:
|
|
for j,c in enumerate(rr):
|
|
cc=self._compact(c); v=next_nonempty(rr,j)
|
|
if cc in {"2type","type"} and v.lower() in {"private","government","public"}: result["p_1b_type"]=v
|
|
elif "3adcode" in cc and self._valid_code(v): result["p_1b_ad_code"]=v
|
|
elif "5cbname" in cc and v: result["p_1b_cb_name"]=self._clean_party_value("p_1b_cb_name",v.replace("C ","",1) if v.startswith("C ") else v)
|
|
elif "9forexbankacno" in cc and v: result["p_1b_forex_ac_no"]=v
|
|
elif "10dbkbankacno" in cc and v: result["p_1b_dbk_ac_no"]=v
|
|
elif "11ifscno" in cc and v and re.fullmatch(r"[A-Z0-9]{4,20}",v,re.I): result["p_1b_ifsc_no"]=v
|
|
elif "6aeo" in cc and v and self._valid_clean_aeo(v): result["p_1b_aeo"]=v
|
|
elif "4rbiwaiverno" in cc and v: result["p_1b_rbi_wr_no"]=v
|
|
break
|
|
if result.get("p_1b_consignee_name"): result["buyer_name"]=result["p_1b_consignee_name"]
|
|
def _party_block(self,aliases,field):
|
|
label=self._best_label(aliases,page=1)
|
|
if not label: return None,None
|
|
row=label["row"]
|
|
# Table extraction is authoritative for the standard two-column block.
|
|
for t in self.tables:
|
|
if t["page"]!=label["page"]: continue
|
|
rows=t["rows"]
|
|
for i,r in enumerate(rows):
|
|
txt=self._row_text(r)
|
|
if self._compact(aliases[0]) in self._compact(txt) and i+1<len(rows):
|
|
# Exporter occupies left half; consignee occupies right half.
|
|
left=[]; right=[]
|
|
for rr in rows[i+1:i+5]:
|
|
cells=[self._clean(c) for c in rr]
|
|
mid=max(1,len(cells)//2)
|
|
left += [c for c in cells[:mid] if c]
|
|
right += [c for c in cells[mid:] if c]
|
|
raw=(left if field=="p_1b_exporter_name" else right)
|
|
raw=[x for x in raw if not self._looks_like_stop_text(x)]
|
|
if raw:
|
|
name=raw[0]
|
|
addr="\n".join(raw[1:])
|
|
return self._clean_party_value(field,name),self._clean_address(addr)
|
|
# Word-based fallback: collect lines below the label until a known stop.
|
|
lines=[]
|
|
for r in self.page_rows.get(label["_page"],[]):
|
|
if r["_y"]<=label["_y"]: continue
|
|
text=self._clean(r["text"])
|
|
if not text or self._looks_like_stop_text(text):
|
|
if lines: break
|
|
continue
|
|
lines.append(text)
|
|
if len(lines)>=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<len(rows) else []
|
|
self._map_financial_row(result,row,vals)
|
|
if "6deductions" in compact and "7pc" in compact and "8duty" in compact:
|
|
vals=rows[i+1] if i+1<len(rows) else []
|
|
self._map_financial_row(result,row,vals)
|
|
return
|
|
|
|
def _map_financial_row(self,result,headers,values):
|
|
for j,hraw in enumerate(headers):
|
|
h=self._compact(hraw)
|
|
if not h: continue
|
|
f=None
|
|
if "fobvalue" in h: f="p_1c_fob_val"
|
|
elif "freight" in h: f="p_1c_freight"
|
|
elif "insuranc" in h: f="p_1c_insurance"
|
|
elif "discou" in h: f="p_1c_discount"
|
|
elif h.endswith("com") or "commission" in h: f="p_1c_com"
|
|
elif "deductions" in h or h.endswith("deduct"): f="p_1c_deductions"
|
|
elif "pc" in h and ("7" in h or h in {"pc","p c"}): f="p_1c_p_c"
|
|
elif "8duty" in h or h.endswith("duty"): f="p_1c_duty"
|
|
elif "9cess" in h or h.endswith("cess"): f="p_1c_cess"
|
|
elif "dbkclaim" in h: f="p_1d_dbk"
|
|
elif "igstamt" in h: f="p_1d_igst_amt"
|
|
elif "cessamt" in h: f="p_1d_cess_amt"
|
|
elif "igstvalue" in h: f="p_1d_igst_val"
|
|
elif "rodtepamt" in h: f="p_1d_rodtep_amt"
|
|
elif "rosctlamt" in h or "roslamt" in h: f="p_1d_rosctle_amt"
|
|
if f and j<len(values):
|
|
v=self._float(values[j])
|
|
if v is not None: result[f]=v
|
|
|
|
def _extract_dates_times(self,result):
|
|
for t in self.tables:
|
|
rows=t["rows"]
|
|
for i,row in enumerate(rows):
|
|
c=self._compact(self._row_text(row))
|
|
if "1event" not in c or "2date" not in c or "3time" not in c: continue
|
|
# The LEO number is printed in the header/value row beside
|
|
# the '4.LEO NO.' heading.
|
|
for j,v in enumerate(row):
|
|
if "4leono" in self._compact(v):
|
|
leo_value = next((self._clean(x) for x in row[j+1:]
|
|
if re.fullmatch(r"\d+\s*/\s*\d+", self._clean(x))), "")
|
|
# Prefer the actual word-level value when the table extractor
|
|
# clips a value at a cell boundary (for example 2/838 -> 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(rows):
|
|
vals=rows[header_i+1]
|
|
hm=self._header_map(rows[header_i],"invoice_details")
|
|
self._map_row_to_record(inv,"invoice_details",hm,vals)
|
|
# The combined invoice number/date cell needs splitting.
|
|
raw=self._cell_for_header(rows[header_i],vals,["invoice no","invoice no dt","invoice number"])
|
|
if raw:
|
|
m=re.search(r"(\S+)\s+(\d{1,2}/\d{1,2}/\d{2,4})",raw)
|
|
if m: inv["p_2a_inv_no"]=m.group(1); inv["p_2a_inv_dt"]=self._format_date(m.group(2))
|
|
inv["inv_sn"]=self._integer(inv.get("inv_sn") or self._cell_for_header(rows[header_i],vals,["s no","sno"]) or 1)
|
|
# party/value blocks on page 2
|
|
for i,r in enumerate(rows):
|
|
c=self._compact(self._row_text(r))
|
|
if "exportersnameaddress" in c and i+1<len(rows):
|
|
exp_idx=next((j for j,x in enumerate(r) if "exportersnameaddress" in self._compact(x)),1)
|
|
buyer_idx=next((j for j,x in enumerate(r) if "buyersnameaddress" in self._compact(x)),9)
|
|
left=[]; right=[]
|
|
for rr in rows[i+1:i+5]:
|
|
cells=[self._clean(x) for x in rr]
|
|
left += [x for x in cells[exp_idx:buyer_idx] if x]
|
|
right += [x for x in cells[buyer_idx:] if x]
|
|
right=[x for x in right if x not in {"Y","P","O","C","SA"}]
|
|
if left:
|
|
inv["p_2b_exporter_name"]=self._clean_party_value("p_2b_exporter_name",left[0])
|
|
inv["p_2b_exporter_addr"]=self._clean(" ".join(left[1:]))
|
|
if right:
|
|
inv["p_2b_buyer_name"]=self._clean_party_value("p_2b_buyer_name",right[0])
|
|
inv["p_2b_buyer_addr"]=self._clean_address("\n".join(right[1:]))
|
|
if "invoicevalue" in c and "exchangerate" in c and i+1<len(rows):
|
|
self._map_invoice_value_grid(inv,r,rows[i+1])
|
|
if i+2 < len(rows): self._map_invoice_currency_row(inv, r, rows[i+2])
|
|
if self._invoice_has_identity(inv): result["invoice_details"]=[inv]
|
|
|
|
def _map_invoice_currency_row(self,inv,headers,row):
|
|
field_by_header={
|
|
"invoicevalue":"p_2c_invoice_curr","fobvalue":"p_2c_fob_curr","freight":"p_2c_frieght_curr",
|
|
"insurance":"p_2c_insurance_curr","discount":"p_2c_discount_curr","commison":"p_2c_commison_curr",
|
|
"commission":"p_2c_commison_curr","deduct":"p_2c_deduct_curr","pc":"p_2c_p_c_curr",
|
|
}
|
|
for j,h in enumerate(headers):
|
|
if j>=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(data):
|
|
vals=data[data.index(r)+1]
|
|
hm2=self._header_map(r,"item_details"); self._map_row_to_record(item,"item_details",hm2,vals)
|
|
if "19scheme" in c and data.index(r)+1<len(data):
|
|
self._map_row_to_record(item,"item_details",self._header_map(r,"item_details"),data[data.index(r)+1])
|
|
if "24ptabroad" in c and data.index(r)+1<len(data):
|
|
self._map_row_to_record(item,"item_details",self._header_map(r,"item_details"),data[data.index(r)+1])
|
|
if self._item_has_identity(item): result["item_details"]=[item]
|
|
|
|
# ================================================================
|
|
# known sections + generic table mapper
|
|
# ================================================================
|
|
|
|
def _extract_known_sections(self,result):
|
|
self._extract_manifest(result); self._extract_annex(result); self._extract_single_window(result)
|
|
# Empty optional sections remain empty; populated sections are mapped generically.
|
|
for table_name in self.CHILD_TABLES:
|
|
if table_name in {"manifest_details","annex_details","invoice_details","item_details","single_window_declaration"}: continue
|
|
self._extract_child_section(result,table_name)
|
|
|
|
def _extract_manifest(self,result):
|
|
for t in self.tables:
|
|
rows=t["rows"]
|
|
for i,r in enumerate(rows):
|
|
c=self._compact(self._row_text(r))
|
|
if not ("mawbno" in c and "mawbdt" in c and "hawbno" in c): continue
|
|
rec={"doctype":"Manifest Detail"}
|
|
if i+1<len(rows):
|
|
self._map_row_to_record(rec,"manifest_details",self._header_map(r,"manifest_details"),rows[i+1])
|
|
# HAWB/NOC values are often OCR noise; retain only values
|
|
# matching the expected airway/reference shapes.
|
|
if rec.get("p_1e_hawb_no") in {"R","S","Y","P"}: rec["p_1e_hawb_no"]=""
|
|
if i+3<len(rows):
|
|
r2=rows[i+2]; r3=rows[i+3]
|
|
self._map_row_to_record(rec,"manifest_details",self._header_map(r2,"manifest_details"),r3)
|
|
if self._manifest_has_identity(rec):
|
|
result["manifest_details"]=[rec]
|
|
return
|
|
def _extract_annex(self,result):
|
|
for t in self.tables:
|
|
rows=t["rows"]
|
|
for i,r in enumerate(rows):
|
|
c=self._compact(self._row_text(r))
|
|
if not ("sealtype" in c and "natureofcargo" in c and "loosepackets" in c): continue
|
|
rec={"doctype":"Annex Detail"}
|
|
if i+1<len(rows): self._map_row_to_record(rec,"annex_details",self._header_map(r,"annex_details"),rows[i+1])
|
|
# Marks & Numbers is a separate labelled row; its value is
|
|
# normally in the following cell on that same row.
|
|
for rr in rows[i+2:i+4]:
|
|
for j,x in enumerate(rr):
|
|
if "marks" in self._compact(x):
|
|
for y in rr[j+1:]:
|
|
y=self._clean(y)
|
|
if y and "marks" not in self._compact(y) and y not in {".","-"}: rec["p_1i_marks_numbers"]=y; break
|
|
if self._child_identity("annex_details",rec): result["annex_details"]=[rec]
|
|
return
|
|
def _extract_single_window(self,result):
|
|
for t in self.tables:
|
|
rows=t["rows"]
|
|
for i,r in enumerate(rows):
|
|
c=self._compact(self._row_text(r))
|
|
if "invsn" in c and "itmsn" in c and "qualifier" in c and "infomsr" in c and "uqc" in c:
|
|
records=[]
|
|
for rr in rows[i+1:]:
|
|
if self._is_major_section_row(rr): break
|
|
if not any(self._clean(x) for x in rr): continue
|
|
rec={"doctype":"Single Window Declaration"}; self._map_row_to_record(rec,"single_window_declaration",self._header_map(r,"single_window_declaration"),rr)
|
|
if rec.get("p_4d_qualifier"): rec["p_4d_qualifier"]=re.sub(r"^[A-Z]\s+(SQC|GCESS|RDT|DOO|EPT|STO)$", r"\1", str(rec["p_4d_qualifier"]).strip(), flags=re.I)
|
|
if self._single_window_identity(rec): records.append(rec)
|
|
if records: result["single_window_declaration"]=records
|
|
return
|
|
|
|
def _extract_child_section(self,result,table_name):
|
|
headings=self.SECTION_HEADINGS.get(table_name,[])
|
|
if not headings: return
|
|
for t in self.tables:
|
|
rows=t["rows"]
|
|
for i,r in enumerate(rows):
|
|
c=self._compact(self._row_text(r))
|
|
if not any(self._compact(h) in c for h in headings): continue
|
|
# Header is normally the next row. If the heading row also
|
|
# contains labels, use it directly.
|
|
header_idx=i+1
|
|
if header_idx>=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+1<len(rows):
|
|
vals=[self._clean(x) for x in rows[i+1]]
|
|
codes=[v for v in vals if self._valid_port(v)]
|
|
if codes: result["port_code"]=codes[0]
|
|
for v in vals:
|
|
if self._valid_sb_number(v): result["shipping_bill_no"]=v
|
|
if self._valid_date(v): result["shipping_bill_date"]=self._format_date(v)
|
|
return
|
|
|
|
def _reconcile_status_authority(self,result):
|
|
self._extract_statuses(result)
|
|
|
|
def _reconcile_locations_authority(self,result):
|
|
self._extract_locations(result)
|
|
|
|
def _reconcile_party_authority(self,result):
|
|
self._extract_parties(result)
|
|
# Clean known column contamination.
|
|
for f in ("p_1b_exporter_address","p_1b_consignee_address"):
|
|
result[f]=self._clean_address(result.get(f,""))
|
|
if result.get("p_1b_consignee_name"): result["buyer_name"]=result["p_1b_consignee_name"]
|
|
|
|
def _reconcile_financial_authority(self,result):
|
|
self._extract_financials(result)
|
|
for f in self.NUMBER_FIELDS:
|
|
if result.get(f) is None: result[f]=0
|
|
|
|
def _reconcile_event_authority(self,result):
|
|
self._extract_dates_times(result)
|
|
|
|
def _reconcile_invoice_authority(self,result):
|
|
self._extract_invoices(result)
|
|
invs=result.get("invoice_details") or []
|
|
# Invoice count is the number of real invoice rows, not boilerplate prose.
|
|
invs=[r for r in invs if self._invoice_has_identity(r)]
|
|
for i,r in enumerate(invs,1): r["inv_sn"]=i
|
|
result["invoice_details"]=invs; result["inv"]=len(invs) or result.get("inv") or 0
|
|
if invs:
|
|
r=invs[0]
|
|
if self._is_missing("p_2a_ad_code",r.get("p_2a_ad_code")) and result.get("p_1b_ad_code"): r["p_2a_ad_code"]=result["p_1b_ad_code"]
|
|
self._inherit_invoice_currencies(r)
|
|
r["p_2b_buyer_addr"] = self._clean_address(r.get("p_2b_buyer_addr", ""))
|
|
r["p_2b_exporter_addr"] = self._clean_address(r.get("p_2b_exporter_addr", ""))
|
|
r["p_2b_buyer_addr"] = self._clean_address(r.get("p_2b_buyer_addr", ""))
|
|
r["p_2b_exporter_addr"] = self._clean_address(r.get("p_2b_exporter_addr", ""))
|
|
|
|
def _reconcile_item_authority(self,result):
|
|
self._extract_items(result)
|
|
items=[r for r in result.get("item_details") or [] if self._item_has_identity(r)]
|
|
for i,r in enumerate(items,1):
|
|
r["p_3a_invsno"]=self._integer(r.get("p_3a_invsno") or 1)
|
|
r["p_3a_itemsn"]=self._integer(r.get("p_3a_itemsn") or i)
|
|
result["item_details"]=items; result["item"]=len(items) or result.get("item") or 0
|
|
|
|
def _reconcile_annex_authority(self,result): self._extract_annex(result)
|
|
def _reconcile_manifest_authority(self,result): self._extract_manifest(result)
|
|
def _reconcile_single_window_authority(self,result): self._extract_single_window(result)
|
|
|
|
def _identity_fields(self,table):
|
|
if table=="invoice_details": return ["p_2a_inv_no","inv_sn"]
|
|
if table=="item_details": return ["p_3a_invsno","p_3a_itemsn","p_3a_cth"]
|
|
if table=="manifest_details": return ["p_1e_mawb_no","p_1e_cin_no"]
|
|
if table=="single_window_declaration": return ["p_4d_invsn","p_4d_itmsn","p_4d_info","p_4d_qualifier"]
|
|
return self.COMPLETE_CHILD_FIELDS.get(table,[])[:1]
|
|
|
|
def _child_identity(self,table,rec):
|
|
if table=="invoice_details": return bool(rec.get("p_2a_inv_no"))
|
|
if table=="item_details": return bool(rec.get("p_3a_cth") or rec.get("p_3a_item_desc"))
|
|
if table=="manifest_details": return bool(rec.get("p_1e_mawb_no") or rec.get("p_1e_cin_no"))
|
|
if table=="annex_details": return bool(rec.get("p_1i_seal_typ") or rec.get("p_1i_nature_cargo") or rec.get("p_1i_no_of_pkgs"))
|
|
if table=="single_window_declaration": return bool(rec.get("p_4d_info") or rec.get("p_4d_qualifier"))
|
|
return False
|
|
|
|
def _invoice_has_identity(self,r): return bool(self._clean(r.get("p_2a_inv_no")))
|
|
def _item_has_identity(self,r): return bool(self._clean(r.get("p_3a_cth")) or self._clean(r.get("p_3a_item_desc")))
|
|
def _manifest_has_identity(self,r): return bool(self._clean(r.get("p_1e_mawb_no")) or self._clean(r.get("p_1e_cin_no")))
|
|
def _single_window_identity(self,r): return bool(self._clean(r.get("p_4d_info")) or self._clean(r.get("p_4d_qualifier")))
|
|
def _stronger_child_value(self,field,new,old): return self._is_missing(field,old) or (isinstance(new,str) and len(new)>len(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<len(data) else ""
|
|
return ""
|
|
def _table_field_value(self,aliases,field=None):
|
|
aa=[self._compact(a) for a in aliases]
|
|
for t in self.tables:
|
|
rows=t["rows"]
|
|
for i,r in enumerate(rows):
|
|
for j,c in enumerate(r):
|
|
cc=self._compact(c)
|
|
if not cc or not any(a==cc or a in cc for a in aa): continue
|
|
# Shipping Summary fields are laid out horizontally:
|
|
# label cell followed by its value cell on the same row.
|
|
if field in {"port_of_loading","state_of_origin","port_of_finaldstn","cntry_of_finaldstn","port_of_discharge","cntry_of_discharge"}:
|
|
for k in range(j+1,len(r)):
|
|
v=self._clean(r[k])
|
|
if not v: continue
|
|
if self._looks_like_any_known_label(v): continue
|
|
if self._valid_location_candidate(field,v): return v
|
|
# Generic labelled table fallback: same column on next row.
|
|
if i+1<len(rows) and j<len(rows[i+1]):
|
|
v=self._clean(rows[i+1][j])
|
|
if v and not self._looks_like_any_known_label(v): return v
|
|
for k in range(j+1,len(r)):
|
|
v=self._clean(r[k])
|
|
if v and not self._looks_like_any_known_label(v): return v
|
|
return None
|
|
def _valid_port(self,v): return bool(re.fullmatch(r"[A-Z]{3,8}\d{1,3}",str(v or "").strip().upper()))
|
|
|
|
|
|
# =============================================================================
|
|
# SETTINGS
|
|
# =============================================================================
|
|
|
|
class ConfigError(Exception):
|
|
"""Raised when the configuration is incomplete or invalid."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Settings:
|
|
frappe_url: str
|
|
api_key: str
|
|
api_secret: str
|
|
pdf_folder: Path
|
|
include_subfolders: bool
|
|
doctype: str
|
|
naming_series: str
|
|
duplicate_action: str
|
|
duplicate_match_fields: tuple[str, ...]
|
|
attach_pdf: bool
|
|
attach_private: bool
|
|
validate_against_doctype: bool
|
|
required_fields: dict[str, str]
|
|
report_folder: Path
|
|
open_report_when_done: bool
|
|
save_json_files: bool
|
|
json_output_folder: Path
|
|
log_folder: Path
|
|
request_timeout: int
|
|
max_retries: int
|
|
verify_ssl: bool
|
|
max_consecutive_connection_failures: int
|
|
dry_run: bool
|
|
|
|
@classmethod
|
|
def load(cls, config: dict[str, Any], args: argparse.Namespace) -> "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())
|