A R R R Associates
458 lines
19 KiB
Python
458 lines
19 KiB
Python
from __future__ import annotations
|
|
|
|
import csv
|
|
import hashlib
|
|
import io
|
|
import json
|
|
import re
|
|
from datetime import date, datetime
|
|
from pathlib import Path
|
|
|
|
from openpyxl import load_workbook
|
|
|
|
|
|
MAX_UPLOAD_BYTES = 25 * 1024 * 1024
|
|
|
|
|
|
def _s(value) -> str:
|
|
if value is None:
|
|
return ""
|
|
if isinstance(value, float) and value.is_integer():
|
|
return str(int(value))
|
|
return str(value).strip()
|
|
|
|
|
|
def _norm(value) -> str:
|
|
text = _s(value).lower().replace("\n", " ").replace("\r", " ")
|
|
text = re.sub(r"[_\-]+", " ", text)
|
|
text = re.sub(r"[^\w/(). ]+", " ", text)
|
|
return re.sub(r"\s+", " ", text).strip()
|
|
|
|
|
|
def _gstin(value) -> str:
|
|
return re.sub(r"[^A-Z0-9]", "", _s(value).upper())[:15]
|
|
|
|
|
|
def _hsn(value) -> str:
|
|
return re.sub(r"\D", "", _s(value))[:8]
|
|
|
|
|
|
def _money(value) -> float:
|
|
if value in (None, ""):
|
|
return 0.0
|
|
if isinstance(value, (int, float)):
|
|
return float(value)
|
|
text = _s(value).replace(",", "").replace("₹", "")
|
|
text = re.sub(r"^\((.*)\)$", r"-\1", text)
|
|
try:
|
|
return float(text)
|
|
except Exception:
|
|
return 0.0
|
|
|
|
|
|
def _date(value) -> str:
|
|
if value in (None, ""):
|
|
return ""
|
|
if isinstance(value, datetime):
|
|
return value.date().isoformat()
|
|
if isinstance(value, date):
|
|
return value.isoformat()
|
|
text = _s(value)
|
|
for fmt in (
|
|
"%d/%m/%Y", "%d-%m-%Y", "%Y-%m-%d", "%d-%b-%Y", "%d %b %Y",
|
|
"%d.%m.%Y", "%d/%m/%y", "%d-%m-%y",
|
|
):
|
|
try:
|
|
return datetime.strptime(text, fmt).date().isoformat()
|
|
except Exception:
|
|
pass
|
|
return text[:20]
|
|
|
|
|
|
def _doc_type(value: str, number: str = "") -> str:
|
|
text = f"{value} {number}".upper()
|
|
if "CREDIT" in text or "CR NOTE" in text:
|
|
return "credit_note"
|
|
if "DEBIT" in text or "DR NOTE" in text:
|
|
return "debit_note"
|
|
return "invoice"
|
|
|
|
|
|
def _sha(value: str) -> str:
|
|
return hashlib.sha256(value.encode("utf-8", "ignore")).hexdigest()
|
|
|
|
|
|
def file_sha256(content: bytes) -> str:
|
|
return hashlib.sha256(content).hexdigest()
|
|
|
|
|
|
def _record_key(source_type: str, row: dict) -> str:
|
|
external = row.get("external_reference", "")
|
|
if external:
|
|
return f"{source_type}|REF|{external}"
|
|
return "|".join([
|
|
source_type,
|
|
row.get("supplier_gstin", ""),
|
|
row.get("document_number", ""),
|
|
row.get("document_date", ""),
|
|
row.get("document_type", ""),
|
|
])
|
|
|
|
|
|
def _finalize(source_type: str, row: dict, items: list[dict]):
|
|
row["source_type"] = source_type
|
|
row["supplier_gstin"] = _gstin(row.get("supplier_gstin"))
|
|
row["recipient_gstin"] = _gstin(row.get("recipient_gstin"))
|
|
row["document_number"] = _s(row.get("document_number"))[:160]
|
|
row["document_date"] = _date(row.get("document_date"))
|
|
row["document_type"] = _doc_type(_s(row.get("document_type")), row["document_number"])
|
|
row["supplier_name"] = _s(row.get("supplier_name"))[:260]
|
|
row["external_reference"] = _s(row.get("external_reference"))[:180]
|
|
for key in ("taxable_value", "igst", "cgst", "sgst", "cess", "invoice_value"):
|
|
row[key] = _money(row.get(key))
|
|
row["place_of_supply"] = _s(row.get("place_of_supply"))[:120]
|
|
row["transport_mode"] = _s(row.get("transport_mode"))[:80]
|
|
row["vehicle_number"] = _s(row.get("vehicle_number"))[:40]
|
|
row["transporter_id"] = _s(row.get("transporter_id"))[:40]
|
|
|
|
normalized_items = []
|
|
for index, item in enumerate(items or [], start=1):
|
|
product_name = _s(item.get("product_name"))[:500]
|
|
description = _s(item.get("description_text"))[:4000]
|
|
hsn = _hsn(item.get("hsn_code"))
|
|
normalized_items.append({
|
|
"line_number": int(item.get("line_number") or index),
|
|
"product_name": product_name,
|
|
"description_text": description,
|
|
"hsn_code": hsn,
|
|
"quantity": _money(item.get("quantity")),
|
|
"unit": _s(item.get("unit"))[:30],
|
|
"unit_price": _money(item.get("unit_price")),
|
|
"taxable_value": _money(item.get("taxable_value")),
|
|
"gst_rate": _money(item.get("gst_rate")),
|
|
"igst": _money(item.get("igst")),
|
|
"cgst": _money(item.get("cgst")),
|
|
"sgst": _money(item.get("sgst")),
|
|
"cess": _money(item.get("cess")),
|
|
})
|
|
|
|
if not row["taxable_value"]:
|
|
row["taxable_value"] = sum(x["taxable_value"] for x in normalized_items)
|
|
if not row["igst"]:
|
|
row["igst"] = sum(x["igst"] for x in normalized_items)
|
|
if not row["cgst"]:
|
|
row["cgst"] = sum(x["cgst"] for x in normalized_items)
|
|
if not row["sgst"]:
|
|
row["sgst"] = sum(x["sgst"] for x in normalized_items)
|
|
if not row["cess"]:
|
|
row["cess"] = sum(x["cess"] for x in normalized_items)
|
|
if not row["invoice_value"]:
|
|
row["invoice_value"] = row["taxable_value"] + row["igst"] + row["cgst"] + row["sgst"] + row["cess"]
|
|
|
|
if not row["document_number"] or not row["document_date"]:
|
|
return None
|
|
if not row["supplier_gstin"] and not row["supplier_name"]:
|
|
return None
|
|
|
|
row["source_document_key"] = _record_key(source_type, row)
|
|
compact = json.dumps({"row": row, "items": normalized_items}, sort_keys=True, default=str, ensure_ascii=False)
|
|
row["source_row_hash"] = _sha(compact)
|
|
return row, normalized_items
|
|
|
|
|
|
def _path(data, *keys, default=None):
|
|
cur = data
|
|
for key in keys:
|
|
if not isinstance(cur, dict):
|
|
return default
|
|
candidates = (key, key.lower(), key.upper(), key.capitalize())
|
|
found = None
|
|
for cand in candidates:
|
|
if cand in cur:
|
|
found = cur[cand]
|
|
break
|
|
if found is None:
|
|
# case-insensitive fallback
|
|
kmap = {str(k).lower(): k for k in cur.keys()}
|
|
real = kmap.get(str(key).lower())
|
|
if real is None:
|
|
return default
|
|
found = cur[real]
|
|
cur = found
|
|
return cur
|
|
|
|
|
|
def _parse_einvoice_object(obj: dict):
|
|
doc = _path(obj, "DocDtls", default={}) or {}
|
|
seller = _path(obj, "SellerDtls", default={}) or {}
|
|
buyer = _path(obj, "BuyerDtls", default={}) or {}
|
|
val = _path(obj, "ValDtls", default={}) or {}
|
|
tran = _path(obj, "TranDtls", default={}) or {}
|
|
irn = _path(obj, "Irn") or _path(obj, "IRN") or _path(obj, "AckNo") or ""
|
|
|
|
row = {
|
|
"external_reference": irn,
|
|
"supplier_gstin": _path(seller, "Gstin") or _path(obj, "SellerGstin") or _path(obj, "supplier_gstin"),
|
|
"supplier_name": _path(seller, "TrdNm") or _path(seller, "LglNm") or _path(obj, "SellerName") or _path(obj, "supplier_name"),
|
|
"recipient_gstin": _path(buyer, "Gstin") or _path(obj, "BuyerGstin") or _path(obj, "recipient_gstin"),
|
|
"document_number": _path(doc, "No") or _path(obj, "DocNo") or _path(obj, "invoice_number"),
|
|
"document_date": _path(doc, "Dt") or _path(obj, "DocDt") or _path(obj, "invoice_date"),
|
|
"document_type": _path(doc, "Typ") or _path(obj, "DocTyp") or "invoice",
|
|
"taxable_value": _path(val, "AssVal") or _path(obj, "TaxableValue"),
|
|
"igst": _path(val, "IgstVal") or _path(obj, "IgstVal"),
|
|
"cgst": _path(val, "CgstVal") or _path(obj, "CgstVal"),
|
|
"sgst": _path(val, "SgstVal") or _path(obj, "SgstVal"),
|
|
"cess": _path(val, "CesVal") or _path(obj, "CessVal"),
|
|
"invoice_value": _path(val, "TotInvVal") or _path(obj, "TotInvVal") or _path(obj, "invoice_value"),
|
|
"place_of_supply": _path(buyer, "Pos") or _path(obj, "Pos"),
|
|
"transport_mode": _path(tran, "TransMode") or _path(obj, "TransMode"),
|
|
"vehicle_number": _path(obj, "VehNo") or "",
|
|
"transporter_id": _path(tran, "TransId") or _path(obj, "TransId"),
|
|
}
|
|
raw_items = _path(obj, "ItemList", default=[]) or _path(obj, "items", default=[]) or []
|
|
items = []
|
|
for i, item in enumerate(raw_items if isinstance(raw_items, list) else [], start=1):
|
|
ass = _path(item, "AssAmt") or _path(item, "TaxableValue") or 0
|
|
rate = _path(item, "GstRt") or _path(item, "GSTRate") or 0
|
|
items.append({
|
|
"line_number": _path(item, "SlNo") or i,
|
|
"product_name": _path(item, "PrdDesc") or _path(item, "ProductName") or _path(item, "Nm"),
|
|
"description_text": _path(item, "PrdDesc") or _path(item, "Desc"),
|
|
"hsn_code": _path(item, "HsnCd") or _path(item, "HSN"),
|
|
"quantity": _path(item, "Qty"),
|
|
"unit": _path(item, "Unit"),
|
|
"unit_price": _path(item, "UnitPrice"),
|
|
"taxable_value": ass,
|
|
"gst_rate": rate,
|
|
"igst": _path(item, "IgstAmt"),
|
|
"cgst": _path(item, "CgstAmt"),
|
|
"sgst": _path(item, "SgstAmt"),
|
|
"cess": _path(item, "CesAmt"),
|
|
})
|
|
return _finalize("e_invoice", row, items)
|
|
|
|
|
|
def _parse_ewaybill_object(obj: dict):
|
|
row = {
|
|
"external_reference": _path(obj, "ewbNo") or _path(obj, "EwbNo") or _path(obj, "ewayBillNo"),
|
|
"supplier_gstin": _path(obj, "fromGstin") or _path(obj, "supplierGstin") or _path(obj, "supplier_gstin"),
|
|
"supplier_name": _path(obj, "fromTrdName") or _path(obj, "fromPlace") or _path(obj, "supplierName"),
|
|
"recipient_gstin": _path(obj, "toGstin") or _path(obj, "recipientGstin") or _path(obj, "recipient_gstin"),
|
|
"document_number": _path(obj, "docNo") or _path(obj, "documentNumber") or _path(obj, "invoice_number"),
|
|
"document_date": _path(obj, "docDate") or _path(obj, "documentDate") or _path(obj, "invoice_date"),
|
|
"document_type": _path(obj, "docType") or "invoice",
|
|
"taxable_value": _path(obj, "totalValue") or _path(obj, "taxableValue"),
|
|
"igst": _path(obj, "igstValue"),
|
|
"cgst": _path(obj, "cgstValue"),
|
|
"sgst": _path(obj, "sgstValue"),
|
|
"cess": _path(obj, "cessValue"),
|
|
"invoice_value": _path(obj, "totInvValue") or _path(obj, "invoiceValue"),
|
|
"place_of_supply": _path(obj, "toStateCode") or _path(obj, "placeOfSupply"),
|
|
"transport_mode": _path(obj, "transMode") or _path(obj, "transportMode"),
|
|
"vehicle_number": _path(obj, "vehicleNo") or _path(obj, "vehNo"),
|
|
"transporter_id": _path(obj, "transporterId") or _path(obj, "transporterGstin"),
|
|
}
|
|
raw_items = _path(obj, "itemList", default=[]) or _path(obj, "items", default=[]) or []
|
|
items = []
|
|
for i, item in enumerate(raw_items if isinstance(raw_items, list) else [], start=1):
|
|
items.append({
|
|
"line_number": i,
|
|
"product_name": _path(item, "productName") or _path(item, "productDesc"),
|
|
"description_text": _path(item, "productDesc") or _path(item, "description"),
|
|
"hsn_code": _path(item, "hsnCode") or _path(item, "hsn"),
|
|
"quantity": _path(item, "quantity") or _path(item, "qty"),
|
|
"unit": _path(item, "qtyUnit") or _path(item, "unit"),
|
|
"unit_price": _path(item, "unitPrice"),
|
|
"taxable_value": _path(item, "taxableAmount") or _path(item, "taxableValue"),
|
|
"gst_rate": (
|
|
_money(_path(item, "igstRate"))
|
|
or _money(_path(item, "cgstRate")) + _money(_path(item, "sgstRate"))
|
|
),
|
|
"igst": _path(item, "igstValue") or _path(item, "igstAmount"),
|
|
"cgst": _path(item, "cgstValue") or _path(item, "cgstAmount"),
|
|
"sgst": _path(item, "sgstValue") or _path(item, "sgstAmount"),
|
|
"cess": _path(item, "cessValue") or _path(item, "cessAmount"),
|
|
})
|
|
return _finalize("e_way_bill", row, items)
|
|
|
|
|
|
HEADER_ALIASES = {
|
|
"external_reference": {"irn", "ack no", "ackno", "e way bill no", "eway bill no", "ewb no", "ewbno"},
|
|
"supplier_gstin": {"supplier gstin", "seller gstin", "from gstin", "fromgstin", "gstin of supplier"},
|
|
"supplier_name": {"supplier name", "seller name", "from trade name", "fromtrdname", "trade/legal name"},
|
|
"recipient_gstin": {"recipient gstin", "buyer gstin", "to gstin", "togstin"},
|
|
"document_number": {"document number", "doc no", "docno", "invoice number", "invoice no"},
|
|
"document_date": {"document date", "doc date", "docdate", "invoice date"},
|
|
"document_type": {"document type", "doc type", "doctype", "invoice type"},
|
|
"taxable_value": {"taxable value", "taxable amount", "total value", "totalvalue"},
|
|
"igst": {"igst", "igst value", "igst amount"},
|
|
"cgst": {"cgst", "cgst value", "cgst amount"},
|
|
"sgst": {"sgst", "sgst value", "sgst amount"},
|
|
"cess": {"cess", "cess value", "cess amount"},
|
|
"invoice_value": {"invoice value", "total invoice value", "tot inv value", "totinvvalue"},
|
|
"place_of_supply": {"place of supply", "pos", "to state code"},
|
|
"transport_mode": {"transport mode", "trans mode", "transmode"},
|
|
"vehicle_number": {"vehicle number", "vehicle no", "veh no"},
|
|
"transporter_id": {"transporter id", "trans id", "transporter gstin"},
|
|
"hsn_code": {"hsn", "hsn code", "hsn/sac"},
|
|
"product_name": {"product name", "item name", "product"},
|
|
"description_text": {"description", "product description", "item description"},
|
|
"quantity": {"quantity", "qty"},
|
|
"unit": {"unit", "qty unit", "uom"},
|
|
"unit_price": {"unit price", "rate"},
|
|
}
|
|
ALIAS = {a: k for k, vals in HEADER_ALIASES.items() for a in vals}
|
|
|
|
|
|
def _canon_header(value):
|
|
n = _norm(value)
|
|
if n in ALIAS:
|
|
return ALIAS[n]
|
|
for alias, key in ALIAS.items():
|
|
if len(alias) >= 8 and alias in n:
|
|
return key
|
|
return None
|
|
|
|
|
|
def _sheet_records(content: bytes, source_type: str):
|
|
wb = load_workbook(io.BytesIO(content), read_only=True, data_only=True)
|
|
out = []
|
|
try:
|
|
for ws in wb.worksheets:
|
|
rows = [list(r) for r in ws.iter_rows(values_only=True)]
|
|
best = None
|
|
for idx, vals in enumerate(rows[:40]):
|
|
mapping = {}
|
|
for c, v in enumerate(vals):
|
|
key = _canon_header(v)
|
|
if key and key not in mapping:
|
|
mapping[key] = c
|
|
score = sum(k in mapping for k in ("document_number", "document_date", "supplier_gstin"))
|
|
if score >= 2 and (best is None or score > best[0]):
|
|
best = (score, idx, mapping)
|
|
if not best:
|
|
continue
|
|
_, header_idx, mapping = best
|
|
for row_no, vals in enumerate(rows[header_idx + 1:], start=header_idx + 2):
|
|
def get(k):
|
|
c = mapping.get(k)
|
|
return vals[c] if c is not None and c < len(vals) else None
|
|
row = {k: get(k) for k in (
|
|
"external_reference", "supplier_gstin", "supplier_name", "recipient_gstin",
|
|
"document_number", "document_date", "document_type", "taxable_value", "igst",
|
|
"cgst", "sgst", "cess", "invoice_value", "place_of_supply", "transport_mode",
|
|
"vehicle_number", "transporter_id",
|
|
)}
|
|
item = {
|
|
"line_number": row_no,
|
|
"product_name": get("product_name"),
|
|
"description_text": get("description_text"),
|
|
"hsn_code": get("hsn_code"),
|
|
"quantity": get("quantity"),
|
|
"unit": get("unit"),
|
|
"unit_price": get("unit_price"),
|
|
"taxable_value": get("taxable_value"),
|
|
}
|
|
finalized = _finalize(source_type, row, [item] if any(_s(v) for v in item.values()) else [])
|
|
if finalized:
|
|
out.append(finalized)
|
|
finally:
|
|
wb.close()
|
|
return out
|
|
|
|
|
|
def _csv_records(content: bytes, source_type: str):
|
|
text = None
|
|
for enc in ("utf-8-sig", "utf-8", "cp1252", "latin-1"):
|
|
try:
|
|
text = content.decode(enc)
|
|
break
|
|
except UnicodeDecodeError:
|
|
pass
|
|
text = text or content.decode("utf-8", "replace")
|
|
try:
|
|
dialect = csv.Sniffer().sniff(text[:8192], delimiters=",;\t|")
|
|
except Exception:
|
|
dialect = csv.excel
|
|
rows = list(csv.reader(io.StringIO(text), dialect))
|
|
best = None
|
|
for idx, vals in enumerate(rows[:40]):
|
|
mapping = {}
|
|
for c, v in enumerate(vals):
|
|
key = _canon_header(v)
|
|
if key and key not in mapping:
|
|
mapping[key] = c
|
|
score = sum(k in mapping for k in ("document_number", "document_date", "supplier_gstin"))
|
|
if score >= 2 and (best is None or score > best[0]):
|
|
best = (score, idx, mapping)
|
|
if not best:
|
|
return []
|
|
_, header_idx, mapping = best
|
|
result = []
|
|
for row_no, vals in enumerate(rows[header_idx + 1:], start=header_idx + 2):
|
|
def get(k):
|
|
c = mapping.get(k)
|
|
return vals[c] if c is not None and c < len(vals) else None
|
|
row = {k: get(k) for k in (
|
|
"external_reference", "supplier_gstin", "supplier_name", "recipient_gstin",
|
|
"document_number", "document_date", "document_type", "taxable_value", "igst",
|
|
"cgst", "sgst", "cess", "invoice_value", "place_of_supply", "transport_mode",
|
|
"vehicle_number", "transporter_id",
|
|
)}
|
|
item = {
|
|
"line_number": row_no,
|
|
"product_name": get("product_name"),
|
|
"description_text": get("description_text"),
|
|
"hsn_code": get("hsn_code"),
|
|
"quantity": get("quantity"),
|
|
"unit": get("unit"),
|
|
"unit_price": get("unit_price"),
|
|
"taxable_value": get("taxable_value"),
|
|
}
|
|
finalized = _finalize(source_type, row, [item] if any(_s(v) for v in item.values()) else [])
|
|
if finalized:
|
|
result.append(finalized)
|
|
return result
|
|
|
|
|
|
def _json_objects(payload):
|
|
if isinstance(payload, list):
|
|
return payload
|
|
if not isinstance(payload, dict):
|
|
return []
|
|
for key in ("data", "result", "records", "invoices", "ewayBills", "ewaybills", "items"):
|
|
val = payload.get(key)
|
|
if isinstance(val, list) and val and isinstance(val[0], dict):
|
|
return val
|
|
return [payload]
|
|
|
|
|
|
def parse_enrichment(content: bytes, filename: str, source_type: str):
|
|
if source_type not in {"e_invoice", "e_way_bill"}:
|
|
raise ValueError("Source type must be E-Invoice or E-Way Bill.")
|
|
if not content:
|
|
raise ValueError("Uploaded enrichment file is empty.")
|
|
if len(content) > MAX_UPLOAD_BYTES:
|
|
raise ValueError("Enrichment upload exceeds the 25 MB limit.")
|
|
|
|
suffix = Path(filename or "").suffix.lower()
|
|
records = []
|
|
if suffix == ".json":
|
|
payload = json.loads(content.decode("utf-8-sig"))
|
|
for obj in _json_objects(payload):
|
|
if not isinstance(obj, dict):
|
|
continue
|
|
item = _parse_einvoice_object(obj) if source_type == "e_invoice" else _parse_ewaybill_object(obj)
|
|
if item:
|
|
records.append(item)
|
|
elif suffix in {".xlsx", ".xlsm"}:
|
|
records = _sheet_records(content, source_type)
|
|
elif suffix in {".csv", ".txt"}:
|
|
records = _csv_records(content, source_type)
|
|
else:
|
|
raise ValueError("Upload a .json, .xlsx, .xlsm or .csv file.")
|
|
|
|
if not records:
|
|
raise ValueError("No usable E-Invoice/E-Way Bill document records were detected.")
|
|
return records
|