GenAI Foundations: NON-AI vs AI (Docling + Groq)

This comprehensive technical blueprint implements a Config-Driven, Enterprise-Style Dual-Engine Hybrid Architecture to parse Distribution Company (DISCOM) utility invoices at scale.

When managing massive multi-regional infrastructures, processing operations face two contrasting data engineering hurdles:

1. Deterministic Pipelines (NON-AI): Brittle regex models break during multi-line structural collapses or line-wrap variations.
2. Pure GenAI Pipelines (AI): Blind layout parsing structures introduce substantial LLM token computing fees and hallucination liabilities.

To bridge this gap, this system relies on a unified framework. Rather than hardcoding parsing constraints for individual layout formats, all external supplier characteristics (such as regular expressions, positional anchor text lines, and language mapping aliases) live in isolated JSON files. Adding support for a new energy provider requires adding a configuration file without altering any core Python engine scripts.

Technical Flow & Execution Matrix

Every processed document flows sequentially through our dual engines to populate a unified comparison sheet:

[Incoming Utility Bill PDF]
                              │
                              ▼
                ┌───────────────────────────┐
                │   Stage 1: PDF Extraction │ ───► Streams text layers using pdfplumber
                └───────────┬───────────────┘
                              │
                              ▼
                ┌───────────────────────────┐
                │ Stage 2: Vendor Routing   │ ───► Resolves rules via all_of / none_of
                └───────────┬───────────────┘
                              │
                              ▼
                ┌───────────────────────────┐
                │ Stage 3: Field Extraction │
                └───────────┬───────────────┘
                              │
            ┌─────────────────┴─────────────────┐
            ▼                                   ▼
[Engine 1: Rule-Based Determinism]  [Engine 2: Layout-Aware AI Fallback]
 • Runs multi-pattern regex and      • Converts files to layout-preserved
   exact relative line-offsets         Markdown via Docling
 • Execution time < 1 second         • Structural extraction via Groq (Llama 3.3)
            │                        • Maps fields through semantic aliases
            │                                   │
            └─────────────────┬─────────────────┘
                              │
                              ▼
                ┌───────────────────────────┐
                │ Stage 4: Cross-Engine App │ ───► Enforces types via dynamic Pydantic models
                └───────────┬───────────────┘
                              │
                              ▼
                ┌───────────────────────────┐
                │ Stage 5: Target Writeout  │ ───► Maps to tracking workbook via TemplateMapper
                └───────────────────────────┘      (Rows 1–3 protected; Row-1 checked dynamically)

The Output Row Contract

To facilitate immediate differential analysis, the TemplateMapper module dynamically targets rows within a single spreadsheet according to this contract:

Row Index Row Designation Functional Characteristics Row 1 Column Headers Canonical token definitions — Untouched / Protected Row 2 LT Template Definitions Reference definitions for low-tension billing — Protected Row 3 HT Template Definitions Reference definitions for high-tension billing — Protected Row 4 LT NON-AI Extraction Populated by pdfplumber + regular expression line mappings Row 5 HT NON-AI Extraction Populated by pdfplumber + regular expression line mappings Row 6 LT AI Extraction Populated by Docling + Groq Llama-3.3-70b-versatile Row 7 HT AI Extraction Populated by Docling + Groq Llama-3.3-70b-versatile

Non-AI Pipeline(Deterministic)

Cell 1: Environment Bootstrapping & Dependency Assembly

This cell downloads core platform tools: pdfplumber for stream processing, openpyxl for low-level spreadsheet operations, docling for visual markdown rendering, and pydantic for schema validation.

Python

# Install required libraries for both extraction systems
!pip install pdfplumber pandas openpyxl pydantic docling rapidfuzz groq anthropic

Cell 2: Workspace Synchronization & Directory Mount

This cell provisions isolated directories in Google Drive for assets, processing layers, templates, and final outputs.

Python

import os
from google.colab import drive
# Establish persistent storage connections
drive.mount('/content/drive')

# Configure production workspace paths
BASE_PATH = "/content/drive/MyDrive/GenAI/DISCOM_PARSER"
INPUT_PATH = f"{BASE_PATH}/input_pdfs"
OUTPUT_PATH = f"{BASE_PATH}/output"
TEMPLATE_PATH = f"{BASE_PATH}/templates"
CONFIG_PATH = f"{BASE_PATH}/configs"

# Generate physical runtime environment structures
for folder in [INPUT_PATH, OUTPUT_PATH, TEMPLATE_PATH, CONFIG_PATH]:
    os.makedirs(folder, exist_ok=True)

print("Environment synchronized. Root storage context targeted at:")
print(f" -> Config Workspace Path: {CONFIG_PATH}")

Cell 3: Initializing the Global Canonical Schema Layer

This cell creates a stable dictionary that decouples variable supplier terminology from downstream systems, mapping fixed tokens directly to target workbook columns.

Python

import json

canonical_fields = {
  "_comment": "Canonical field dictionary. Internal names are stable and DISCOM-agnostic. 'template_column' = Row-1 header token in the Excel template (null = extracted for audit but no column, left blank). DISCOM configs reference these names and may override template_column. New field = add here; new DISCOM = add a config JSON only.",
  "version": 1,
  "fields": {
    "consumer_number":        {"template_column": "CanSerNo",          "value_type": "string", "description": "Consumer / service connection number"},
    "consumer_name":          {"template_column": "ConsumerName",      "value_type": "string", "description": "Registered consumer name"},
    "address":                {"template_column": "Address",           "value_type": "string", "description": "Supply address"},
    "mobile_email":           {"template_column": None,                "value_type": "string", "description": "Contact mobile / email (no template column)"},
    "bill_number":            {"template_column": "BillNo",            "value_type": "string", "description": "Bill number"},
    "bill_month":             {"template_column": "BillMonth",         "value_type": "string", "description": "Billing month / period (MON-YYYY)"},
    "bill_period":            {"template_column": None,                "value_type": "string", "description": "Bill period text (no template column)"},
    "bill_date":              {"template_column": "BillDate",          "value_type": "date",   "description": "Bill issue date"},
    "due_date":               {"template_column": "BillDueDate",       "value_type": "date",   "description": "Payment due date"},
    "supply_date":            {"template_column": None,                "value_type": "date",   "description": "Date of supply (no template column)"},
    "bill_amount":            {"template_column": None,                "value_type": "number", "description": "Headline bill amount (no template column)"},
    "tariff_category":        {"template_column": "ConCat",            "value_type": "string", "description": "Tariff / category"},
    "tariff_code":            {"template_column": None,                "value_type": "string", "description": "Tariff code (no template column)"},
    "sanctioned_load":        {"template_column": "ConnLd",            "value_type": "string", "description": "Sanctioned load"},
    "connected_load":         {"template_column": None,                "value_type": "string", "description": "Connected load (no distinct column)"},
    "contract_demand":        {"template_column": "ContDmd",           "value_type": "number", "description": "Contract demand (KVA)"},
    "billing_demand":         {"template_column": "BillDemd",          "value_type": "number", "description": "Billed demand (KVA)"},
    "max_demand_recorded":    {"template_column": "MaxDemReco",        "value_type": "number", "description": "Recorded maximum demand"},
    "feeder_voltage":         {"template_column": None,                "value_type": "string", "description": "Feeder voltage (no template column)"},
    "feeder_name":            {"template_column": None,                "value_type": "string", "description": "Feeder name (no template column)"},
    "substation_name":        {"template_column": None,                "value_type": "string", "description": "Substation name (no template column)"},
    "billing_unit":           {"template_column": None,                "value_type": "string", "description": "Billing unit code (no template column)"},
    "meter_number":           {"template_column": "MetrNo",            "value_type": "string", "description": "Meter number"},
    "meter_status":           {"template_column": "MeterStatus",       "value_type": "string", "description": "Meter status"},
    "multiplying_factor":     {"template_column": "MulFac",            "value_type": "number", "description": "Meter multiplying factor"},
    "previous_reading_date":  {"template_column": "OmrDt",             "value_type": "date",   "description": "Previous meter reading date"},
    "current_reading_date":   {"template_column": "CmrDt",             "value_type": "date",   "description": "Current meter reading date"},
    "previous_reading_kwh":   {"template_column": "OmrKwh",            "value_type": "number", "description": "Previous reading KWH"},
    "current_reading_kwh":    {"template_column": "CmrKwh",            "value_type": "number", "description": "Current reading KWH"},
    "previous_reading_kvah":  {"template_column": "OmrKvah",           "value_type": "number", "description": "Previous reading KVAH"},
    "current_reading_kvah":   {"template_column": "CmrKvah",           "value_type": "number", "description": "Current reading KVAH"},
    "consumed_units":         {"template_column": "ConsUnits",         "value_type": "number", "description": "Consumed units (KWH)"},
    "consumed_units_kvah":    {"template_column": "ConsUnitsKvah",     "value_type": "number", "description": "Consumed units (KVAH)"},
    "billed_units":           {"template_column": "BilledUnit",        "value_type": "number", "description": "Billed units"},
    "final_consumed_units":   {"template_column": "FinalConsUnits",    "value_type": "number", "description": "Final consumed units"},
    "fixed_charges":          {"template_column": "FxdChg",            "value_type": "number", "description": "Fixed / demand charges"},
    "energy_charges":         {"template_column": "EngyChg",           "value_type": "number", "description": "Energy charges"},
    "fac":                    {"template_column": "FulCstAdj",         "value_type": "number", "description": "Fuel adjustment charge (F.A.C.)"},
    "electricity_duty":       {"template_column": "EleDuty",           "value_type": "number", "description": "Electricity duty amount"},
    "tax_on_sale":            {"template_column": "P11",               "value_type": "number", "description": "Tax on sale amount"},
    "wheeling_charges":       {"template_column": "P10",               "value_type": "number", "description": "Wheeling charges amount"},
    "tod_charges":            {"template_column": "P06",               "value_type": "number", "description": "Time-of-day tariff energy charges"},
    "net_bill_amount":        {"template_column": "CurAmtPay",         "value_type": "number", "description": "Total current bill payable"},
    "rounded_bill":           {"template_column": "GrosAmt",           "value_type": "number", "description": "Rounded / gross bill amount"},
    "dpc":                    {"template_column": "CurrentLPS",        "value_type": "number", "description": "Delayed payment charge / current LPS"},
    "arrears":                {"template_column": "Arrears",           "value_type": "number", "description": "Principal / total arrears"},
    "average_pf":             {"template_column": "AvgPf",             "value_type": "number", "description": "Average / billed power factor"},
    "load_factor":            {"template_column": None,                "value_type": "number", "description": "Load factor (no template column)"},
    "security_deposit":       {"template_column": "SecDep",            "value_type": "number", "description": "Security deposit held"},
    "additional_security_deposit": {"template_column": "AddSecDep",    "value_type": "number", "description": "Additional security deposit demanded"},
    "last_receipt_amount":    {"template_column": "LastAmountPaid",    "value_type": "number", "description": "Last receipt / payment amount"},
    "last_receipt_date":      {"template_column": "LastAmountPaidDate","value_type": "date",   "description": "Last receipt / payment date"},
    "prompt_payment_discount":{"template_column": "P40",               "value_type": "number", "description": "Prompt payment discount amount"},
    "prompt_gross_amount":    {"template_column": "PromptGrossAmt",    "value_type": "number", "description": "Gross amount if paid by prompt date"},
    "prompt_pay_by_date":     {"template_column": "PromptPayByDate",   "value_type": "date",   "description": "Prompt payment by date"},
    "payment_considered_upto":{"template_column": None,                "value_type": "date",   "description": "Payment considered upto (no template column)"},
    "current_rkvah_lag":      {"template_column": "P18",               "value_type": "number", "description": "Current RKVAH (LAG)"},
    "current_rkvah_lead":     {"template_column": "P21",               "value_type": "number", "description": "Current RKVAH (LEAD)"},
    "gstin":                  {"template_column": None,                "value_type": "string", "description": "GSTIN (no template column)"},
    "pan":                    {"template_column": None,                "value_type": "string", "description": "PAN (no template column)"}
  }
}

path = f"{CONFIG_PATH}/canonical_fields.json"
with open(path, "w", encoding="utf-8") as f:
    json.dump(canonical_fields, f, indent=2)

print(f"Wrote canonical layer specifications: {path}")
print(f"Total metrics monitored: {len(canonical_fields['fields'])}")

Cell 4: Compiling Low-Tension (LT) Target Rules

This cell isolates low-tension characteristics into an individual config file, applying exact relative layout anchors to capture positional fields like name and address blocks.

Python

msedcl_lt = {
  "discom": "MSEDCL",
  "tariff": "LT",
  "template_name": "MSEDCL_LT",
  "target_row": 4,
  "description": "MSEDCL Low-Tension. All DISCOM-specific logic lives here; the engine stays generic.",
  "detect": {
    "all_of": ["Bill of Supply For:", "MSEDCL"],
    "none_of": ["BILL OF SUPPLY FOR THE MONTH OF"]
  },
  "fields": {
    "consumer_number":   {"template_column": "CanSerNo",     "extraction_type": "regex",       "aliases": ["Consumer No", "Service No"],          "regex_patterns": [r"Consumer No:\s*(\d+)"]},
    "consumer_name":     {"template_column": "ConsumerName",  "extraction_type": "line_offset", "aliases": ["Consumer Name"],                       "anchor_text": "Consumer No:", "line_offset": 1},
    "address":           {"template_column": "Address",       "extraction_type": "line_offset", "aliases": ["Address"],                             "anchor_text": "Consumer No:", "line_offset": 2},
    "bill_number":       {"template_column": "BillNo",        "extraction_type": "regex",       "aliases": ["BILL NO.(GGN)"],                       "regex_patterns": [r"BILL NO\.\(GGN\):\s*(\d+)"]},
    "bill_month":        {"template_column": "BillMonth",     "extraction_type": "regex",       "aliases": ["Bill of Supply For"],                  "regex_patterns": [r"Bill of Supply For:\s*([A-Z]{3}-\d{4})"]},
    "bill_date":         {"template_column": "BillDate",      "extraction_type": "regex",       "aliases": ["Bill Date"],                           "regex_patterns": [r"Bill Date:\s*(\d{1,2}-[A-Z]{3}-\d{2,4})"], "value_type": "date"},
    "due_date":          {"template_column": "BillDueDate",   "extraction_type": "regex",       "aliases": ["Due Date"],                            "regex_patterns": [r"Due Date:\s*(\d{1,2}-[A-Z]{3}-\d{2,4})"], "value_type": "date"},
    "bill_amount":       {"template_column": None,            "extraction_type": "regex",       "aliases": ["Bill Amount Rs"],                      "regex_patterns": [r"Bill Amount Rs:\s*([\d,]+\.\d{2})"], "value_type": "number"},
    "tariff_category":   {"template_column": "ConCat",        "extraction_type": "regex",       "aliases": ["Tariff/Category"],                     "regex_patterns": [r"Tariff/Category:\s*([^\n]+?)\s+Sanct"]},
    "sanctioned_load":   {"template_column": "ConnLd",        "extraction_type": "regex",       "aliases": ["Sanct. Load"],                         "regex_patterns": [r"Sanct\. Load:\s*([\w\s\.]+?)(?:\s+Pole|\s+Security|\n)"]},
    "security_deposit":  {"template_column": "SecDep",        "extraction_type": "regex",       "aliases": ["Security Deposit(Rs)"],                "regex_patterns": [r"Security Deposit\(Rs\):\s*([\d,]+\.\d{2})"], "value_type": "number"},
    "meter_number":      {"template_column": "MetrNo",        "extraction_type": "regex",       "aliases": ["Meter No"],                            "regex_patterns": [r"Meter No:\s*(\S+)"]},
    "meter_status":      {"template_column": "MeterStatus",   "extraction_type": "regex",       "aliases": ["Meter status"],                        "regex_patterns": [r"Meter status:\s*(\w+)"]},
    "current_reading_date":  {"template_column": "CmrDt",     "extraction_type": "regex",       "aliases": ["Current Reading Date"],                "regex_patterns": [r"Current Reading Date:\s*(\d{1,2}-[A-Z]{3}-\d{2,4})"], "value_type": "date"},
    "previous_reading_date": {"template_column": "OmrDt",     "extraction_type": "regex",       "aliases": ["Previous Reading Date"],               "regex_patterns": [r"Previous Reading Date:\s*(\d{1,2}-[A-Z]{3}-\d{2,4})"], "value_type": "date"},
    "consumed_units":        {"template_column": "ConsUnits", "extraction_type": "regex",       "aliases": ["Units", "Consumption"],                "regex_patterns": [r"\d+(?:\.\d+)?\s+\d+(?:\.\d+)?\s+\d+\s+(\d+)\s+\d+\s+\d+\s*\n"], "value_type": "number"},
    "billed_units":          {"template_column": "BilledUnit","extraction_type": "regex",       "aliases": ["Billed Units"],                        "regex_patterns": [r"\d+(?:\.\d+)?\s+\d+(?:\.\d+)?\s+\d+\s+\d+\s+\d+\s+(\d+)\s*\n"], "value_type": "number"},
    "final_consumed_units":  {"template_column": "FinalConsUnits","extraction_type": "regex",   "aliases": ["Total"],                               "regex_patterns": [r"\d+(?:\.\d+)?\s+\d+(?:\.\d+)?\s+\d+\s+\d+\s+\d+\s+(\d+)\s*\n"], "value_type": "number"},
    "fixed_charges":     {"template_column": "FxdChg",        "extraction_type": "regex",       "aliases": ["Fixed Charges"],                       "regex_patterns": [r"Fixed Charges\s+([\d,]+\.\d{2})"], "value_type": "number"},
    "energy_charges":    {"template_column": "EngyChg",       "extraction_type": "regex",       "aliases": ["Energy Charges"],                      "regex_patterns": [r"Energy Charges\s+([\d,]+\.\d{2})"], "value_type": "number"},
    "fac":               {"template_column": "FulCstAdj",     "extraction_type": "regex",       "aliases": ["F.A.C.", "Fuel Adjustment Charge"],    "regex_patterns": [r"F\.A\.C\.\s+([\d,]+\.\d{2})"], "value_type": "number"},
    "electricity_duty":  {"template_column": "EleDuty",       "extraction_type": "regex",       "aliases": ["Electricity Duty"],                    "regex_patterns": [r"Electricity Duty\s*\([^)]*\)\s+([\d,]+\.\d{2})"], "value_type": "number"},
    "tax_on_sale":       {"template_column": "P11",           "extraction_type": "regex",       "aliases": ["Tax on Sale"],                         "regex_patterns": [r"Tax on Sale @\s*[\d.]+\s*Paise/Unit\s+([\d,]+\.\d{2})"], "value_type": "number"},
    "wheeling_charges":  {"template_column": "P10",           "extraction_type": "regex",       "aliases": ["Wheeling Charges"],                    "regex_patterns": [r"Wheeling Charges\s+[\d.]+/\s*Unit\s+([\d,]+\.\d{2})"], "value_type": "number"},
    "net_bill_amount":   {"template_column": "CurAmtPay",     "extraction_type": "regex",       "aliases": ["Total Current Bill(Rs)"],              "regex_patterns": [r"Total Current Bill\(Rs\)\s+([\d,]+\.\d{2})"], "value_type": "number"},
    "rounded_bill":      {"template_column": "GrosAmt",       "extraction_type": "regex",       "aliases": ["Rounded Bill(Rs)"],                    "regex_patterns": [r"Rounded Bill\(Rs\)\s+([\d,]+\.\d{2})"], "value_type": "number"},
    "dpc":               {"template_column": "CurrentLPS",    "extraction_type": "regex",       "aliases": ["DPC"],                                 "regex_patterns": [r"DPC:\s*([\d,]+\.\d{2})"], "value_type": "number"},
    "arrears":           {"template_column": "Arrears",       "extraction_type": "regex",       "aliases": ["Total Arrears"],                       "regex_patterns": [r"Total Arrears\s+(-?[\d,]+\.\d{2})"], "value_type": "number"},
    "last_receipt_amount": {"template_column": "LastAmountPaid",    "extraction_type": "regex", "aliases": ["Last Receipt Amount"],                 "regex_patterns": [r"Last Receipt Amount\s+([\d,]+\.\d{2})"], "value_type": "number"},
    "last_receipt_date":   {"template_column": "LastAmountPaidDate","extraction_type": "regex", "aliases": ["Last Receipt Date"],                   "regex_patterns": [r"Last Receipt Date\s+(\d{1,2}-[A-Z]{3}-\d{2,4})"], "value_type": "date"},
    "prompt_payment_discount": {"template_column": "P40",     "extraction_type": "regex",       "aliases": ["Prompt Payment Discount"],             "regex_patterns": [r"Prompt Payment Discount:\s*Rs\.\s*([\d,]+\.\d{2})"], "value_type": "number"},
    "prompt_pay_by_date":      {"template_column": "PromptPayByDate","extraction_type": "regex","aliases": ["if bill is paid on or before"],         "regex_patterns": [r"if bill is paid on or before\s+(\d{1,2}-[A-Z]{3}-\d{2,4})"], "value_type": "date"},
    "billing_unit":      {"template_column": None,            "extraction_type": "regex",       "aliases": ["Billing Unit"],                        "regex_patterns": [r"Billing Unit:\s*(\d+)"]}
  }
}

path = f"{CONFIG_PATH}/msedcl_lt.json"
with open(path, "w", encoding="utf-8") as f:
    json.dump(msedcl_lt, f, indent=2)

print(f"LT rules structured and written out to: {path}")

Cell 5: Compiling High-Tension (HT) Target Rules

This cell establishes structural rule mappings tailored to dense, high-tension industrial layout connections.

Python

msedcl_ht = {
  "discom": "MSEDCL",
  "tariff": "HT",
  "template_name": "MSEDCL_HT",
  "target_row": 5,
  "description": "MSEDCL High-Tension. All DISCOM-specific logic lives here; the engine stays generic.",
  "detect": {
    "all_of": ["BILL OF SUPPLY FOR THE MONTH OF"],
    "any_of": ["HT", "High Tension", "HIGH TENSION"]
  },
  "fields": {
    "consumer_number":   {"template_column": "CanSerNo",     "extraction_type": "regex",       "aliases": ["Consumer No."],                        "regex_patterns": [r"Consumer No\.\s*:\s*(\d+)"]},
    "consumer_name":     {"template_column": "ConsumerName",  "extraction_type": "regex",       "aliases": ["Consumer Name"],                       "regex_patterns": [r"Consumer Name\s*:\s*([^\n]+)"]},
    "address":           {"template_column": "Address",       "extraction_type": "line_offset", "aliases": ["Address"],                             "anchor_text": "Consumer Name", "line_offset": 1},
    "bill_month":        {"template_column": "BillMonth",     "extraction_type": "regex",       "aliases": ["BILL OF SUPPLY FOR THE MONTH OF"],     "regex_patterns": [r"BILL OF SUPPLY FOR THE MONTH OF\s+([A-Z]{3}-\d{4})"]},
    "bill_number":       {"template_column": "BillNo",        "extraction_type": "regex",       "aliases": ["Bill No"],                             "regex_patterns": [r"BILL OF SUPPLY FOR THE MONTH OF\s+[A-Z]{3}-\d{4}\s+(\d+)"]},
    "bill_date":         {"template_column": "BillDate",      "extraction_type": "regex",       "aliases": ["BILL DATE"],                           "regex_patterns": [r"BILL DATE\s+(\d{2}/\d{2}/\d{4})"], "value_type": "date"},
    "due_date":          {"template_column": "BillDueDate",   "extraction_type": "regex",       "aliases": ["DUE DATE"],                            "regex_patterns": [r"DUE DATE\s+(\d{2}/\d{2}/\d{4})"], "value_type": "date"},
    "tariff_category":   {"template_column": "ConCat",        "extraction_type": "regex",       "aliases": ["Tariff"],                              "regex_patterns": [r"Tariff\s*:\s*([^\n]+?)(?:\s+Meter|\s+Old)"]},
    "sanctioned_load":   {"template_column": "ConnLd",        "extraction_type": "regex",       "aliases": ["Sanctioned Load (KW)"],                "regex_patterns": [r"Sanctioned Load \(KW\)\s*:\s*([\d,]+(?:\.\d+)?)"]},
    "contract_demand":   {"template_column": "ContDmd",       "extraction_type": "regex",       "aliases": ["Contract Demand (KVA)"],               "regex_patterns": [r"Contract Demand \(KVA\)\s*:\s*([\d,]+\.\d{2})"], "value_type": "number"},
    "connected_load":    {"template_column": None,            "extraction_type": "regex",       "aliases": ["Connected Load (KW)"],                 "regex_patterns": [r"Connected Load \(KW\)\s*:\s*([\d,]+\.\d{2})"]},
    "max_demand_recorded": {"template_column": "MaxDemReco",  "extraction_type": "regex",       "aliases": ["Recorded MD"],                         "regex_patterns": [r"Recorded MD \(except A\s*\n?Zone\)\s*\n?\s*(\d+(?:\.\d+)?)"], "value_type": "number"},
    "billing_demand":    {"template_column": "BillDemd",      "extraction_type": "regex",       "aliases": ["Billed Demand (KVA)"],                 "regex_patterns": [r"Billed Demand \(KVA\)\s*\n?\s*(\d+(?:\.\d+)?)"], "value_type": "number"},
    "multiplying_factor":{"template_column": "MulFac",        "extraction_type": "regex",       "aliases": ["Multiplying Factor"],                  "regex_patterns": [r"Multiplying Factor\s+([\d.]+)"], "value_type": "number"},
    "meter_number":      {"template_column": "MetrNo",        "extraction_type": "regex",       "aliases": ["Meter No"],                            "regex_patterns": [r"Meter No\s*:\s*(\S+)"]},
    "security_deposit":  {"template_column": "SecDep",        "extraction_type": "regex",       "aliases": ["Security Deposit Held Rs."],           "regex_patterns": [r"Security Deposit Held Rs\.\s*:\s*([\d,]+\.\d{2})"], "value_type": "number"},
    "additional_security_deposit": {"template_column": "AddSecDep", "extraction_type": "regex", "aliases": ["Addl. S.D. Demanded Rs."],             "regex_patterns": [r"Addl\. S\.D\. Demanded Rs\.\s*:\s*([\d,]+\.\d{2})"], "value_type": "number"},
    "consumed_units":      {"template_column": "ConsUnits",     "extraction_type": "regex",     "aliases": ["Total Consumption KWH", "KWH"],        "regex_patterns": [r"Total Consumption\s+([\d,]+(?:\.\d+)?)"], "value_type": "number"},
    "consumed_units_kvah": {"template_column": "ConsUnitsKvah", "extraction_type": "regex",     "aliases": ["Total Consumption KVAH", "KVAH"],      "regex_patterns": [r"Total Consumption\s+[\d,]+(?:\.\d+)?\s+([\d,]+(?:\.\d+)?)"], "value_type": "number"},
    "current_rkvah_lag":   {"template_column": "P18",          "extraction_type": "regex",     "aliases": ["Current RKVAH (LAG)"],                 "regex_patterns": [r"Total Consumption\s+[\d,]+(?:\.\d+)?\s+[\d,]+(?:\.\d+)?\s+([\d,]+(?:\.\d+)?)"], "value_type": "number"},
    "fixed_charges":     {"template_column": "FxdChg",        "extraction_type": "regex",       "aliases": ["Demand Charges"],                      "regex_patterns": [r"Demand Charges @ Rs\.\d+\s+([\d,]+\.\d{2})"], "value_type": "number"},
    "energy_charges":    {"template_column": "EngyChg",       "extraction_type": "regex",       "aliases": ["Energy Charges"],                      "regex_patterns": [r"Energy Charges\s+([\d,]+\.\d{2})"], "value_type": "number"},
    "tod_charges":       {"template_column": "P06",           "extraction_type": "regex",       "aliases": ["TOD Tariff EC"],                       "regex_patterns": [r"TOD Tariff EC\s+([\d,]+\.\d{2})"], "value_type": "number"},
    "wheeling_charges":  {"template_column": "P10",           "extraction_type": "regex",       "aliases": ["Wheeling Charge"],                     "regex_patterns": [r"Wheeling Charge @\s*[\d.]+\s*Rs/U\s+([\d,]+\.\d{2})"], "value_type": "number"},
    "fac":               {"template_column": "FulCstAdj",     "extraction_type": "regex",       "aliases": ["FAC", "F.A.C."],                       "regex_patterns": [r"FAC @\s*[\d.]+\s*Ps\./U\s+([\d,]+\.\d{2})"], "value_type": "number"},
    "electricity_duty":  {"template_column": "EleDuty",       "extraction_type": "regex",       "aliases": ["Electricity Duty"],                    "regex_patterns": [r"Electricity Duty\s+([\d,]+\.\d{2})"], "value_type": "number"},
    "tax_on_sale":       {"template_column": "P11",           "extraction_type": "regex",       "aliases": ["Tax on Sale"],                         "regex_patterns": [r"Tax on Sale @\s*[\d.]+\s*Ps\./U\s+([\d,]+\.\d{2})"], "value_type": "number"},
    "net_bill_amount":   {"template_column": "CurAmtPay",     "extraction_type": "regex",       "aliases": ["TOTAL CURRENT BILL AS PER TARIFF"],    "regex_patterns": [r"TOTAL CURRENT BILL AS PER TARIFF\s+([\d,]+\.\d{2})"], "value_type": "number"},
    "dpc":               {"template_column": "CurrentLPS",    "extraction_type": "regex",       "aliases": ["Delay Payment Charges Rs."],           "regex_patterns": [r"Delay Payment Charges Rs\.\s+([\d,]+\.\d{2})"], "value_type": "number"},
    "arrears":           {"template_column": "Arrears",       "extraction_type": "regex",       "aliases": ["Principal Arrears"],                   "regex_patterns": [r"Principal Arrears\s+(-?\s*[\d,]+\.\d{2})"], "value_type": "number"},
    "rounded_bill":      {"template_column": "GrosAmt",       "extraction_type": "regex",       "aliases": ["After PPD upto Due Date"],              "regex_patterns": [r"Total Bill Amount Payable[^\n]*\n[^\n]*?[\d,]+\s+([\d,]+)\s+[\d,]+"], "value_type": "number"},
    "average_pf":        {"template_column": "AvgPf",         "extraction_type": "regex",       "aliases": ["Billed PF", "Power Factor"],           "regex_patterns": [r"Billed PF\s*:\s*([\d.]+)"], "value_type": "number"},
    "last_receipt_amount": {"template_column": "LastAmountPaid",    "extraction_type": "regex", "aliases": ["Last Month Payment"],                  "regex_patterns": [r"Last Month Payment\s*:\s*([\d,]+\.\d{2})"], "value_type": "number"},
    "last_receipt_date":   {"template_column": "LastAmountPaidDate","extraction_type": "regex", "aliases": ["Last Receipt No./Date"],               "regex_patterns": [r"Last Receipt No\./Date\s*:\s*\d+\s*/\s*(\d{2}-\d{2}-\d{4})"], "value_type": "date"},
    "prompt_payment_discount": {"template_column": "P40",     "extraction_type": "regex",       "aliases": ["Prompt Payment Discount"],             "regex_patterns": [r"Prompt Payment Discount\s+-\s*([\d,]+\.\d{2})"], "value_type": "number"},
    "prompt_pay_by_date":      {"template_column": "PromptPayByDate","extraction_type": "regex","aliases": ["IF PAID ON OR BEFORE"],                 "regex_patterns": [r"IF PAID ON OR BEFORE\s+(\d{1,2}-[A-Z]{3}-\d{2,4})"], "value_type": "date"}
  }
}

path = f"{CONFIG_PATH}/msedcl_ht.json"
with open(path, "w", encoding="utf-8") as f:
    json.dump(msedcl_ht, f, indent=2)

print(f"HT rules compiled and written out to: {path}")

Cell 6: Defining Core Engine Modules & Template Mappers

This cell contains the complete engine code for our deterministic pipeline. It features PDF processing, configuration discovery, data scrubbers, and the cell writer module.

Python

import json, re, shutil, glob, os
from pathlib import Path

# ---------- 1. Generic PDF Reader ----------
def read_pdf_text(pdf_path):
    import pdfplumber
    parts = []
    with pdfplumber.open(str(pdf_path)) as pdf:
        for page in pdf.pages:
            parts.append(page.extract_text() or "")
    return "\n".join(parts)

# ---------- 2. Config Loading & DISCOM Routing ----------
def load_canonical(configs_dir):
    with open(Path(configs_dir) / "canonical_fields.json", encoding="utf-8") as f:
        return json.load(f)["fields"]
def load_discom_configs(configs_dir):
    cfgs = []
    for path in sorted(Path(configs_dir).glob("*.json")):
        if path.name == "canonical_fields.json":
            continue
        with open(path, encoding="utf-8") as f:
            c = json.load(f)
        c["_source"] = path.name
        cfgs.append(c)
    return cfgs

def detect_config(text, configs):
    for cfg in configs:
        r = cfg.get("detect", {})
        if any(t not in text for t in r.get("all_of", [])):           continue
        if r.get("any_of") and not any(t in text for t in r["any_of"]): continue
        if any(t in text for t in r.get("none_of", [])):              continue
        return cfg
    return None

# ---------- Generic Data Scrubbing Helpers ----------
def _vtype(fc, name, canon):
    return fc.get("value_type") or canon.get(name, {}).get("value_type", "string")

def _tcol(fc, name, canon):
    return fc["template_column"] if "template_column" in fc else canon.get(name, {}).get("template_column")

def _clean(raw, vtype):
    if raw is None: return None
    s = str(raw).strip()
    if not s: return None
    if vtype == "number": s = s.replace(",", "").replace(" ", "")
    return s

# ---------- 3. NON-AI Engine Mappings ----------
def _extract_regex(text, fc):
    g = fc.get("group", 1)
    for pat in fc.get("regex_patterns", []):
        m = re.search(pat, text)
        if m:
            try: return m.group(g)
            except IndexError: return m.group(0)
    return None

def _extract_line_offset(text, fc):
    anchor = fc.get("anchor_text")
    off = int(fc.get("line_offset", 1))
    if not anchor: return None
    lines = text.splitlines()
    for i, line in enumerate(lines):
        if anchor in line:
            j = i + off
            if 0 <= j < len(lines):
                val = lines[j].strip()
                for pat in fc.get("regex_patterns", []):
                    m = re.search(pat, val)
                    if m: return m.group(fc.get("group", 1))
                return val or None
            return None
    return None

def extract_non_ai(text, cfg, canon):
    out = {}
    for name, fc in cfg["fields"].items():
        et = fc.get("extraction_type", "regex")
        raw = _extract_regex(text, fc) if et == "regex" else \
              _extract_line_offset(text, fc) if et == "line_offset" else None
        out[name] = _clean(raw, _vtype(fc, name, canon))
    return out

# ---------- 4. Dynamic Template Mapper Engine ----------
class TemplateMapper:
    def __init__(self, template_path, output_path):
        self.output_path = Path(output_path)
        if not self.output_path.exists():
            shutil.copy(str(template_path), str(self.output_path))
        from openpyxl import load_workbook
        wb = load_workbook(self.output_path); ws = wb.active
        self.h2c = {str(c.value).strip(): c.column for c in ws[1] if c.value not in (None, "")}
        self.max_col = ws.max_column
        wb.close()

    def write_row(self, row, values, cfg, canon, method_label):
        from openpyxl import load_workbook
        if row <= 3:
            raise ValueError("Rows 1-3 are protected (headers + definitions).")
        wb = load_workbook(self.output_path); ws = wb.active
        for c in range(1, self.max_col + 1):
            ws.cell(row=row, column=c, value=None)
        written = 0
        for name, value in values.items():
            tcol = _tcol(cfg["fields"][name], name, canon)
            if tcol is None: continue
            ci = self.h2c.get(tcol)
            if ci is None or value in (None, ""): continue
            ws.cell(row=row, column=ci, value=value); written += 1
        if "TemplateName" in self.h2c: ws.cell(row=row, column=self.h2c["TemplateName"], value=cfg["template_name"])
        if "Remarks"      in self.h2c: ws.cell(row=row, column=self.h2c["Remarks"], value=method_label)
        if "rid"          in self.h2c: ws.cell(row=row, column=self.h2c["rid"], value=row)
        wb.save(self.output_path)
        return written

print("Core utility libraries compiled successfully.")

Cell 7: Running Engine 1 (The Deterministic Pipeline)

This cell scans input folders for source documents, automatically applies structural configuration assignments, and writes extracted values directly to the target workbook rows.

Python

# Read config templates
canon   = load_canonical(CONFIG_PATH)
configs = load_discom_configs(CONFIG_PATH)

pdfs   = sorted(glob.glob(f"{INPUT_PATH}/*.pdf"))
tmpls  = sorted(glob.glob(f"{TEMPLATE_PATH}/*.xlsx"))

if not tmpls:
    raise FileNotFoundError(f"Missing master configuration spreadsheet inside {TEMPLATE_PATH}")

TEMPLATE = tmpls[0]
OUTPUT   = f"{OUTPUT_PATH}/FINAL_COMPARISON.xlsx"

# Always clear staging copies to ensure a fresh processing pass
if os.path.exists(OUTPUT):
    os.remove(OUTPUT)

mapper  = TemplateMapper(TEMPLATE, OUTPUT)
results = []

print(f"Beginning processing pass over {len(pdfs)} source invoice targets...")
for pdf in pdfs:
    text = read_pdf_text(pdf)
    cfg  = detect_config(text, configs)
    if cfg is None:
        print(f" -> UNRECOGNISED METADATA FOR FILE: {os.path.basename(pdf)}"); continue

    values  = extract_non_ai(text, cfg, canon)
    row     = cfg["target_row"] # LT=4, HT=5
    written = mapper.write_row(
        row, values, cfg, canon,
        f"{cfg['template_name']} | NON-AI (pdfplumber+config)")

    filled = sum(1 for v in values.values() if v not in (None, ""))
    results.append((os.path.basename(pdf), cfg["template_name"], row, filled, written, values))
    print(f"Processed {os.path.basename(pdf)} -> Target Row {row} | Extracted fields: {filled}")

Cell 8: Executing Pydantic Type & Format Validation

This cell handles run-time validation using a dynamically compiled Pydantic model built straight from our global schema definitions.

Python

from datetime import datetime
from typing import Optional
from pydantic import create_model, ValidationError

_DATE_FORMATS = ("%d-%b-%y", "%d-%b-%Y", "%d/%m/%Y", "%d-%m-%Y", "%Y-%m-%d")

def _build_model(canon):
    type_map = {"number": float, "string": str, "date": str}
    fields = {
        name: (Optional[type_map.get(meta.get("value_type", "string"), str)], None)
        for name, meta in canon.items()
    }
    return create_model("CanonicalBill", **fields)

CanonicalBill = _build_model(canon)

def _date_ok(s):
    for fmt in _DATE_FORMATS:
        try:
            datetime.strptime(str(s).strip(), fmt); return True
        except ValueError: continue
    return False

def validate_record(values, canon):
    problems = []
    # 1. Evaluate string-to-numeric coercion safety via Pydantic model structure
    try:
        CanonicalBill.model_validate(values)
    except ValidationError as e:
        for err in e.errors():
            fld = err["loc"][0]
            problems.append((fld, values.get(fld), err["msg"]))
    # 2. Reconcile date string formats
    for name, meta in canon.items():
        if meta.get("value_type") == "date":
            v = values.get(name)
            if v not in (None, "") and not _date_ok(v):
                problems.append((name, v, "unrecognised date format"))
    return problems

# Evaluate current extraction rows against types
for fname, tmpl, row, filled, written, values in results:
    probs = validate_record(values, canon)
    print(f"\nAudit Log for {fname} ({tmpl}, Row {row})")
    print(f"Extracted properties: {filled} | Format issues found: {len(probs)}")
    for fld, val, msg in probs:
        print(f"  [WARN] Field key: {fld} = {val!r} -> Failure type: {msg}")

Cell 9: Reading Row 1–3 Mappings

This verification cell extracts the generated rows and validates cell structures to confirm the protection of our metadata definitions.

Python

import openpyxl
import pandas as pd

wb = openpyxl.load_workbook(OUTPUT)
ws = wb.active

headers = {c.column: str(c.value).strip() for c in ws[1] if c.value not in (None, "")}
col_of  = {h: ci for ci, h in headers.items()}

ROW_LABELS = {2: "LT def (template)", 3: "HT def (template)",
              4: "LT NON-AI",         5: "HT NON-AI",
              6: "LT AI",             7: "HT AI"}
SHOW = ["CanSerNo","ConsumerName","BillDate","BillDueDate","ConCat",
        "ConsUnits","FxdChg","EngyChg","FulCstAdj","EleDuty",
        "CurAmtPay","GrosAmt","Arrears","AvgPf"]

def cell(r, name):
    ci = col_of.get(name)
    v = ws.cell(row=r, column=ci).value if ci else None
    return "" if v in (None, "") else str(v)

# Verify structural boundaries are untouched
print(f"Row 2 LT verification matches: {ws['B2'].value == 'MSEDCL_LT'}")
print(f"Row 3 HT verification matches: {ws['B3'].value == 'MSEDCL_HT'}\n")

df = pd.DataFrame({ROW_LABELS[r]: {n: cell(r, n) for n in SHOW} for r in (2, 4, 3, 5)})
display(df)

AI Pipeline

Cell 10: Authenticating API Workspace Contexts (Groq Client Setup)

This cell connects to the Groq processing engine, managing missing secrets by providing an explicit prompt fallback to capture keys securely.

Python

from groq import Groq
groq_key = None
try:
    from google.colab import userdata
    groq_key = userdata.get('GROQ_API_KEY')
except Exception: pass

if not groq_key:
    import getpass
    groq_key = getpass.getpass('Enter valid GROQ_API_KEY value: ').strip()

groq_client = Groq(api_key=groq_key)
GROQ_MODEL  = "llama-3.3-70b-versatile"

# Verify connection health
_ping = groq_client.chat.completions.create(
    model=GROQ_MODEL,
    messages=[{"role": "user", "content": "reply with the single word: ok"}],
    temperature=0, max_tokens=5,
)
print(f"System connection established. Status token: {_ping.choices[0].message.content.strip()}")

Cell 11: Blueprinting Engine 2 (The Structural GenAI Pipeline)

This cell structures Engine 2, using layout-aware Markdown generation alongside structural schemas to manage format variations across documents.

Python

_DOCLING = None

def pdf_to_structured_markdown(pdf_path):
    """Docling Engine: Serializes composite structures into layout-aware Markdown."""
    global _DOCLING
    if _DOCLING is None:
        from docling.document_converter import DocumentConverter
        _DOCLING = DocumentConverter()
    return _DOCLING.convert(str(pdf_path)).document.export_to_markdown()

def _field_catalog(cfg, canon):
    """Generates structural context hints to ensure semantic mapping alignment."""
    lines = []
    for name, fc in cfg["fields"].items():
        vt = _vtype(fc, name, canon)
        al = ", ".join(fc.get("aliases", [])) or name
        lines.append(f'- "{name}" ({vt}) — bill label(s): {al}')
    return "\n".join(lines)

def extract_ai(pdf_path, cfg, canon, client, model):
    markdown = pdf_to_structured_markdown(pdf_path)
    field_names = list(cfg["fields"].keys())
    catalog = _field_catalog(cfg, canon)
    system = (
        "You extract structured fields from Indian electricity (DISCOM) bills. "
        "You are given a LAYOUT-AWARE structured document (Docling: headings, "
        "tables, key/value blocks preserved). Match each field by its aliases and "
        "surrounding context (section, table header, adjacent label).\n"
        "STRICT RULES:\n"
        "1. If a field is not present on the bill, its value MUST be null. "
        "NEVER output 0, 0.0, '', 'N/A', or a guessed value for a missing field. "
        "0 is allowed ONLY if the bill literally prints 0 for that field.\n"
        "2. Do not invent, infer, or carry values across fields.\n"
        "3. Numbers: copy the bill's digits, strip only commas/currency/units "
        "(keep the real decimals).\n"
        "4. Dates: copy exactly as printed.\n"
        "Respond with ONE JSON object containing EXACTLY the requested keys."
    )
    user = (
        f"Bill type: {cfg['template_name']}.\n"
        f"Return a JSON object with exactly these keys:\n{field_names}\n\n"
        f"Field meanings (canonical name -> bill aliases):\n{catalog}\n\n"
        f"=== STRUCTURED DOCUMENT (Docling) ===\n{markdown}\n=== END DOCUMENT ==="
    )
    resp = client.chat.completions.create(
        model=model,
        messages=[{"role": "system", "content": system},
                  {"role": "user", "content": user}],
        response_format={"type": "json_object"},
        temperature=0, max_tokens=4000,
    )
    raw = json.loads(resp.choices[0].message.content)
    out = {}
    for name, fc in cfg["fields"].items():
        out[name] = _clean(raw.get(name), _vtype(fc, name, canon))
    return out

print("Engine 2 structural routing initialization complete.")

Cell 12: Running Engine 2 AI Extractions & Processing Mappings

This cell initiates our semantic parsing layer, rendering incoming documents to Markdown before executing structured schema lookups over our target variables.

Python

import time
# Sync mapper instances to output paths
mapper = TemplateMapper(TEMPLATE, OUTPUT)
ai_results = []

print("Running Engine 2 structural processing pipeline...")
for pdf in pdfs:
    fname = os.path.basename(pdf)
    text  = read_pdf_text(pdf)
    cfg   = detect_config(text, configs)
    if cfg is None: continue

    # Map target rows according to safety offsets (LT 4->6, HT 5->7)
    ai_row = cfg["target_row"] + 2
    t0 = time.time()
    try:
        values  = extract_ai(pdf, cfg, canon, groq_client, GROQ_MODEL)
        written = mapper.write_row(
            ai_row, values, cfg, canon,
            f"{cfg['template_name']} | AI (Docling+Groq:{GROQ_MODEL})")
        filled = sum(1 for v in values.values() if v not in (None, ""))
        ai_results.append((fname, cfg["template_name"], ai_row, filled, written, values))
        print(f"Processed AI Target {fname} -> Row {ai_row} | Fields: {filled} ({time.time()-t0:.1f}s)")
    except Exception as e:
        print(f"Critical execution error for target {fname}: {type(e).__name__} - {e}")

Cell 13: Executing Cross-Engine Comparison Mappings

This validation module parses cell variables side-by-side, normalizing numbers and date blocks to surface variance alerts across extraction passes.

Python

from datetime import datetime
_DF = ("%d-%b-%y","%d-%b-%Y","%d/%m/%Y","%d-%m-%Y","%Y-%m-%d")

def _norm(val, vtype):
    if val in (None, ""): return None
    s = str(val).strip()
    if vtype == "number":
        try: return round(float(s.replace(",", "")), 2)
        except ValueError: return s.lower()
    if vtype == "date":
        for f in _DF:
            try: return datetime.strptime(s, f).date()
            except ValueError: pass
        return s.lower()
    return " ".join(s.lower().split())

def compare2(tariff, na_row, ai_row):
    agree = differ = only_na = only_ai = blank = 0
    print(f"\n===== Reconciling {tariff} Connections: Row {na_row} (NON-AI) vs Row {ai_row} (AI) =====")
    for cname, meta in canon.items():
        tcol = meta.get("template_column")
        if tcol not in col_of: continue
        vt = meta.get("value_type","string")
        na = _norm(cell(na_row, tcol), vt)
        ai = _norm(cell(ai_row, tcol), vt)

        if na is None and ai is None: 
            blank += 1; continue
        if na is not None and ai is None: 
            only_na += 1
            print(f"  [Variance] NON-AI exclusive property -> {tcol:<18} Value: {na!r}")
            continue
        if ai is not None and na is None: 
            only_ai += 1
            print(f"  [Variance] AI exclusive property -> {tcol:<18} Value: {ai!r}")
            continue
        if na == ai: 
            agree += 1
        else: 
            differ += 1
            print(f"  [ALERT] Structural cell mismatch -> {tcol:<18} NON-AI: {na!r} | AI: {ai!r}")

    total = agree + differ + only_na + only_ai
    print(f"  --> Processing Summary: Matches={agree}/{total} | Discrepancies={differ} | Both Blank={blank}")

# Execute validation passes
compare2("LT", 4, 6)
compare2("HT", 5, 7)

Workspace Layout Topology

/content/drive/MyDrive/GenAI/DISCOM_PARSER/
├── input_pdfs/                 # Location for incoming billing source PDFs
│   ├── LT_Commercial_Bill.pdf
│   └── HT_Industrial_Bill.pdf
├── templates/                  # Base tracking workbook directory
│   └── MASTER_TEMPLATE.xlsx
├── configs/                    # Auto-provisioned runtime layout parameters
│   ├── canonical_fields.json   # Base dictionary specifications
│   ├── msedcl_lt.json          # LT rules configurations
│   └── msedcl_ht.json          # HT rules configurations
└── output/
    └── FINAL_COMPARISON.xlsx   # Real-time generated analytics output file

Architectural & Strategic Benefits

1. Optimized Resource Consumption: Over 90% of layout formats are instantly processed by Engine 1 on standard CPU resources, running in less than 1 second per document without incurring LLM call fees.
2. Resilience against Layout Shifts: If a provider updates invoice designs or wraps a table mid-page, Engine 2 correctly flags and extracts semantic entities based on layout context.
3. Streamlined Supplier Onboarding: Scaling tracking operations across dozens of new energy providers becomes a data entry exercise rather than a software redesign task. To onboard a new layout type, you only need to create a single JSON file outlining its rule structures. No platform engine updates are ever required.