Finance Process Flows
Reimagined with AI

from azure.ai.formrecognizer import DocumentAnalysisClient
from azure.identity import DefaultAzureCredential
import os

CONFIDENCE_THRESHOLD = 0.85

def extract_invoice(blob_url: str) -> dict:
    """Extract structured fields from a PDF invoice
    using Azure AI Document Intelligence."""
    client = DocumentAnalysisClient(
        endpoint=os.environ["FORM_RECOGNIZER_ENDPOINT"],
        credential=DefaultAzureCredential()
    )
    poller = client.begin_analyze_document_from_url(
        "prebuilt-invoice", blob_url
    )
    result = poller.result()

    for invoice in result.documents:
        fields = invoice.fields
        extracted = {
            "vendor_name":  _safe(fields, "VendorName"),
            "invoice_num":  _safe(fields, "InvoiceId"),
            "invoice_date": _safe(fields, "InvoiceDate"),
            "due_date":     _safe(fields, "DueDate"),
            "total":        _safe(fields, "InvoiceTotal"),
            "line_items":   _extract_lines(fields),
            "confidence":   invoice.confidence,
        }
        if invoice.confidence < CONFIDENCE_THRESHOLD:
            extracted["needs_review"] = True
            extracted["review_reason"] = (
                f"Low confidence ({invoice.confidence:.2f})"
            )
        return extracted

    return {"error": "No invoice detected", "needs_review": True}

def _safe(fields, key):
    f = fields.get(key)
    return f.value if f else None

def _extract_lines(fields):
    items = fields.get("Items")
    if not items:
        return []
    lines = []
    for item in items.value:
        sf = item.value
        lines.append({
            "description": _safe(sf, "Description"),
            "quantity":    _safe(sf, "Quantity"),
            "unit_price":  _safe(sf, "UnitPrice"),
            "amount":      _safe(sf, "Amount"),
            "confidence":  item.confidence,
        })
    return lines

from dataclasses import dataclass
from Levenshtein import distance as lev_distance

TOLERANCE_LINE  = 0.02   # 2% per line item
TOLERANCE_TOTAL = 0.05   # 5% on invoice total

@dataclass
class MatchResult:
    status: str           # "matched" | "exception"
    score: float          # 0.0 - 1.0
    exceptions: list
    explanation: str

def three_way_match(po, invoice, grn) -> MatchResult:
    """Compare PO, Invoice, and Goods Receipt line-by-line."""
    exceptions = []

    # 1. GRN must exist before matching
    if grn is None:
        return MatchResult("exception", 0.0,
            [{"type": "MISSING_GRN",
              "detail": f"No GRN for PO {po['po_number']}"}],
            "Goods receipt not yet recorded.")

    # 2. Match line items (fuzzy on description)
    matched, total_po, total_inv = 0, 0.0, 0.0
    for po_line in po["lines"]:
        best = _fuzzy_find(po_line, invoice["lines"])
        if best is None:
            exceptions.append({"type": "MISSING_LINE",
                "detail": f"PO line '{po_line['desc']}' not on invoice"})
            continue

        # Price variance check
        diff = abs(best["amount"] - po_line["amount"])
        if diff / max(po_line["amount"], 0.01) > TOLERANCE_LINE:
            exceptions.append({"type": "PRICE_VARIANCE",
                "detail": f"{po_line['desc']}: PO ${po_line['amount']:.2f}"
                           f" vs INV ${best['amount']:.2f}"})

        # Quantity check against GRN
        grn_line = _fuzzy_find(po_line, grn["lines"])
        if grn_line and grn_line["qty"] != best.get("qty", 0):
            exceptions.append({"type": "QTY_VARIANCE",
                "detail": f"{po_line['desc']}: GRN {grn_line['qty']}"
                           f" vs INV {best.get('qty', '?')}"})

        matched += 1
        total_po  += po_line["amount"]
        total_inv += best["amount"]

    # 3. Total tolerance check
    if total_po > 0:
        var = abs(total_inv - total_po) / total_po
        if var > TOLERANCE_TOTAL:
            exceptions.append({"type": "TOTAL_VARIANCE",
                "detail": f"PO ${total_po:.2f} vs INV ${total_inv:.2f}"
                           f" ({var:.1%} variance)"})

    score = matched / max(len(po["lines"]), 1)
    status = "matched" if not exceptions else "exception"
    return MatchResult(status, score, exceptions,
        f"{matched}/{len(po['lines'])} lines, {len(exceptions)} exceptions")

def _fuzzy_find(target, candidates):
    best, best_d = None, 999
    for c in candidates:
        d = lev_distance(target["desc"].lower(), c["desc"].lower())
        if d < best_d:
            best, best_d = c, d
    return best if best_d < 3 else None

import joblib
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.linear_model import LogisticRegression
from sklearn.pipeline import Pipeline

CONFIDENCE_THRESHOLD = 0.70

# ── Training ──────────────────────────────────────────────────────
def train_gl_classifier(history_df):
    """Train on historical descriptions -> GL codes.
    Expects columns: 'description', 'gl_code'."""
    pipe = Pipeline([
        ("tfidf", TfidfVectorizer(
            max_features=5000, ngram_range=(1, 2),
            sublinear_tf=True, stop_words="english")),
        ("clf", LogisticRegression(
            max_iter=1000, C=5.0,
            class_weight="balanced", solver="lbfgs")),
    ])
    pipe.fit(history_df["description"], history_df["gl_code"])
    joblib.dump(pipe, "models/gl_classifier.pkl")
    return pipe

# ── Prediction ────────────────────────────────────────────────────
def predict_gl(description: str, pipe=None):
    """Return GL code with confidence; top-3 fallback
    when confidence < threshold."""
    if pipe is None:
        pipe = joblib.load("models/gl_classifier.pkl")

    probas  = pipe.predict_proba([description])[0]
    classes = pipe.classes_
    top_idx = probas.argsort()[::-1]
    best_code = classes[top_idx[0]]
    best_conf = probas[top_idx[0]]

    if best_conf >= CONFIDENCE_THRESHOLD:
        return {
            "gl_code":    best_code,
            "confidence": round(float(best_conf), 3),
            "method":     "auto",
        }

    # Low confidence: surface top 3 for human review
    return {
        "gl_code":    best_code,
        "confidence": round(float(best_conf), 3),
        "method":     "review",
        "suggestions": [
            {"gl_code":    classes[i],
             "confidence": round(float(probas[i]), 3)}
            for i in top_idx[:3]
        ],
    }