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