Intelligent Document
Processing

import cv2
import numpy as np
from azure.ai.formrecognizer import DocumentAnalysisClient
from azure.identity import DefaultAzureCredential
import pytesseract
from PIL import Image

class DualEngineOCR:
    """Multi-engine OCR: Azure primary, Tesseract fallback for low-confidence pages."""

    def __init__(self, endpoint: str, confidence_threshold: float = 0.85):
        credential = DefaultAzureCredential()
        self.client = DocumentAnalysisClient(endpoint, credential)
        self.threshold = confidence_threshold

    def preprocess(self, image_path: str) -> np.ndarray:
        """Deskew and binarize for Tesseract fallback."""
        img = cv2.imread(image_path, cv2.IMREAD_GRAYSCALE)
        coords = np.column_stack(np.where(img < 128))
        angle = cv2.minAreaRect(coords)[-1]
        angle = -(90 + angle) if angle < -45 else -angle
        h, w = img.shape
        M = cv2.getRotationMatrix2D((w // 2, h // 2), angle, 1.0)
        img = cv2.warpAffine(img, M, (w, h), flags=cv2.INTER_CUBIC,
                             borderMode=cv2.BORDER_REPLICATE)
        img = cv2.adaptiveThreshold(img, 255, cv2.ADAPTIVE_THRESH_GAUSSIAN_C,
                                    cv2.THRESH_BINARY, 31, 11)
        return img

    async def extract(self, doc_path: str) -> dict:
        """Run Azure first; fall back to Tesseract on low-confidence pages."""
        with open(doc_path, "rb") as f:
            poller = await self.client.begin_analyze_document(
                "prebuilt-invoice", document=f
            )
        result = await poller.result()

        pages, fallback_pages = [], []
        for page in result.pages:
            avg_conf = np.mean([w.confidence for w in page.words]) if page.words else 0
            if avg_conf >= self.threshold:
                pages.append({"engine": "azure", "confidence": avg_conf,
                              "words": [w.content for w in page.words]})
            else:
                fallback_pages.append(page.page_number)

        for page_num in fallback_pages:
            preprocessed = self.preprocess(f"{doc_path}_page{page_num}.png")
            text = pytesseract.image_to_data(
                Image.fromarray(preprocessed), output_type=pytesseract.Output.DICT
            )
            conf_vals = [int(c) for c in text["conf"] if int(c) > 0]
            pages.append({"engine": "tesseract", "page": page_num,
                          "confidence": np.mean(conf_vals) / 100 if conf_vals else 0,
                          "words": [w for w, c in zip(text["text"], text["conf"])
                                    if int(c) > 40 and w.strip()]})

        return {"pages": pages, "fallback_count": len(fallback_pages),
                "avg_confidence": np.mean([p["confidence"] for p in pages])}

import torch
from torch.utils.data import DataLoader, WeightedRandomSampler
from transformers import LayoutLMv3ForTokenClassification, LayoutLMv3Processor
from datetime import datetime, timedelta

class ActiveLearningPipeline:
    """Ingest HITL corrections and fine-tune extraction model weekly."""

    def __init__(self, model_name: str, corrections_db):
        self.processor = LayoutLMv3Processor.from_pretrained(model_name)
        self.model = LayoutLMv3ForTokenClassification.from_pretrained(model_name)
        self.corrections_db = corrections_db

    def ingest_corrections(self, since: timedelta = timedelta(days=7)) -> list:
        """Extract labeled correction pairs from HITL review queue."""
        cutoff = datetime.utcnow() - since
        corrections = self.corrections_db.find({
            "reviewed_at": {"$gte": cutoff},
            "status": "corrected"
        })
        return [
            {
                "document_id": c["document_id"],
                "original_extraction": c["model_output"],
                "corrected_extraction": c["human_correction"],
                "field": c["field_name"],
                "confidence_delta": c["original_confidence"]
            }
            for c in corrections
        ]

    def confidence_weighted_loss(self, logits, labels, confidences):
        """Weight loss inversely by original confidence.
        Low-confidence errors that humans corrected contribute more to learning."""
        base_loss = torch.nn.functional.cross_entropy(
            logits.view(-1, logits.size(-1)), labels.view(-1), reduction="none"
        )
        # High-confidence wrong predictions penalized 3x more
        weights = 1.0 - confidences.view(-1).clamp(0.5, 0.99)
        weights = weights / weights.sum() * len(weights)
        return (base_loss * weights).mean()

    def retrain(self, training_pairs: list, epochs: int = 3, lr: float = 2e-5) -> dict:
        """Fine-tune on HITL corrections with confidence-weighted loss."""
        dataset = self._build_dataset(training_pairs)
        # Oversample rare correction types proportional to their surprise
        weights = [1.0 / max(p["confidence_delta"], 0.01) for p in training_pairs]
        sampler = WeightedRandomSampler(weights, len(weights))
        loader = DataLoader(dataset, batch_size=8, sampler=sampler)
        optimizer = torch.optim.AdamW(self.model.parameters(), lr=lr)
        self.model.train()
        epoch_losses = []
        for epoch in range(epochs):
            total = 0
            for batch in loader:
                logits = self.model(**batch["inputs"]).logits
                loss = self.confidence_weighted_loss(
                    logits, batch["labels"], batch["confidences"]
                )
                loss.backward()
                optimizer.step()
                optimizer.zero_grad()
                total += loss.item()
            epoch_losses.append(total / len(loader))
        return {"corrections_used": len(training_pairs),
                "final_loss": epoch_losses[-1],
                "epoch_losses": epoch_losses}

import imagehash
from dataclasses import dataclass
from typing import Optional
from PIL import Image

@dataclass
class MatchResult:
    status: str      # "matched" | "variance" | "no_po" | "no_receipt" | "duplicate"
    variance_pct: float
    details: dict

class ThreeWayMatcher:
    """Cross-reference PO + Goods Receipt + Invoice; catch visual duplicates via pHash."""

    AMOUNT_TOLERANCE = 0.02    # 2% for currency rounding differences
    PHASH_THRESHOLD  = 6       # Hamming distance threshold for visual duplicate

    def __init__(self, po_store, receipt_store, invoice_store):
        self.po_store      = po_store
        self.receipt_store = receipt_store
        self.invoice_store = invoice_store

    def detect_visual_duplicate(self, invoice_path: str) -> Optional[str]:
        """Perceptual hash to catch re-submitted PDFs with modified filenames/numbers."""
        current_hash = imagehash.phash(Image.open(invoice_path))
        for inv in self.invoice_store.find({"status": "processed"}):
            if current_hash - imagehash.hex_to_hash(inv["phash"]) <= self.PHASH_THRESHOLD:
                return inv["invoice_number"]
        return None

    def match(self, invoice: dict) -> MatchResult:
        # 1. Visual dedup via pHash (catches re-submissions with altered metadata)
        if invoice.get("file_path"):
            dup = self.detect_visual_duplicate(invoice["file_path"])
            if dup:
                return MatchResult("duplicate", 0,
                                   {"duplicate_of": dup, "method": "perceptual_hash"})

        # 2. Exact-match dedup (invoice number + vendor)
        existing = self.invoice_store.find_one({
            "invoice_number": invoice["invoice_number"],
            "vendor_id":      invoice["vendor_id"],
            "status":         {"$ne": "rejected"}
        })
        if existing:
            return MatchResult("duplicate", 0,
                               {"duplicate_of": existing["invoice_number"],
                                "method": "invoice_number+vendor"})

        # 3. PO existence check
        po = self.po_store.find_one({"po_number": invoice["po_number"]})
        if not po:
            return MatchResult("no_po", 0, {"po_number": invoice["po_number"]})

        # 4. Goods receipt confirmation
        receipt = self.receipt_store.find_one({
            "po_number": invoice["po_number"], "status": "received"
        })
        if not receipt:
            return MatchResult("no_receipt", 0, {"po_number": invoice["po_number"],
                                                  "reason": "goods not yet received"})

        # 5. Amount tolerance check (2%)
        variance = abs(invoice["total_amount"] - po["total_amount"]) / po["total_amount"]
        if variance > self.AMOUNT_TOLERANCE:
            return MatchResult("variance", round(variance * 100, 2),
                               {"po_amount": po["total_amount"],
                                "invoice_amount": invoice["total_amount"],
                                "threshold": "2%"})

        return MatchResult("matched", round(variance * 100, 2),
                           {"po":      po["po_number"],
                            "receipt": receipt["receipt_id"],
                            "invoice": invoice["invoice_number"],
                            "amount":  invoice["total_amount"]})