Detecting AI-Generated Invoice Fraud with FinAudit AI API

import requests
import base64
from pathlib import Path
 
API_KEY = "YOUR_API_KEY"
BASE_URL = "https://apivult.com/finaudit/v1"
 
def analyze_invoice_for_fraud(pdf_path: str, vendor_id: str = None) -> dict:
    """
    Submit an invoice PDF to FinAudit AI for fraud detection analysis.
    Returns a fraud risk score and detailed findings.
    """
    pdf_bytes = Path(pdf_path).read_bytes()
    encoded = base64.b64encode(pdf_bytes).decode("utf-8")
 
    payload = {
        "document": encoded,
        "document_type": "invoice",
        "analysis_mode": "fraud_detection",
        "vendor_id": vendor_id,  # Optional: pass known vendor ID for comparison
        "checks": [
            "document_forensics",
            "vendor_consistency",
            "amount_anomaly",
            "network_analysis"
        ]
    }
 
    response = requests.post(
        f"{BASE_URL}/analyze",
        headers={
            "X-API-Key": API_KEY,
            "Content-Type": "application/json"
        },
        json=payload,
        timeout=60
    )
 
    if response.status_code != 200:
        raise ValueError(f"FinAudit API error {response.status_code}: {response.text}")
 
    return response.json()

def evaluate_fraud_risk(analysis_result: dict) -> dict:
    """
    Evaluate the fraud risk based on FinAudit AI's analysis output.
    Returns a structured risk assessment.
    """
    fraud_score = analysis_result.get("fraud_risk_score", 0)  # 0-100
    findings = analysis_result.get("findings", [])
 
    # Categorize risk level
    if fraud_score >= 75:
        risk_level = "CRITICAL"
        recommended_action = "REJECT - Do not process. Escalate to fraud team immediately."
    elif fraud_score >= 50:
        risk_level = "HIGH"
        recommended_action = "HOLD - Manual review required before processing."
    elif fraud_score >= 25:
        risk_level = "MEDIUM"
        recommended_action = "REVIEW - Flag for supervisor approval."
    else:
        risk_level = "LOW"
        recommended_action = "APPROVE - Standard processing."
 
    # Extract key findings
    critical_flags = [f for f in findings if f.get("severity") == "critical"]
    high_flags = [f for f in findings if f.get("severity") == "high"]
 
    return {
        "fraud_score": fraud_score,
        "risk_level": risk_level,
        "recommended_action": recommended_action,
        "critical_flags": critical_flags,
        "high_flags": high_flags,
        "all_findings": findings,
        "vendor_verified": analysis_result.get("vendor_verified", False),
        "bank_account_changed": analysis_result.get("bank_account_changed", False),
        "document_authentic": analysis_result.get("document_authentic", True)
    }
 
 
# Example usage
result = analyze_invoice_for_fraud("./invoices/vendor_invoice_20260402.pdf", vendor_id="VND-4521")
risk = evaluate_fraud_risk(result)
 
print(f"Fraud Risk: {risk['risk_level']} (Score: {risk['fraud_score']}/100)")
print(f"Action: {risk['recommended_action']}")
 
if risk["critical_flags"]:
    print("\nCritical Flags:")
    for flag in risk["critical_flags"]:
        print(f"  - {flag['type']}: {flag['description']}")
 
if risk["bank_account_changed"]:
    print("\n⚠ WARNING: Bank account details differ from vendor history!")

import os
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass
from typing import Optional
 
@dataclass
class InvoiceScanResult:
    filename: str
    vendor_id: Optional[str]
    fraud_score: int
    risk_level: str
    action: str
    flags: list
    error: Optional[str] = None
 
 
def scan_invoice_batch(invoice_dir: str, vendor_mapping: dict = None) -> list[InvoiceScanResult]:
    """
    Scan all PDF invoices in a directory for fraud signals.
    vendor_mapping: {filename_prefix: vendor_id} for known vendors
    """
    invoice_files = list(Path(invoice_dir).glob("*.pdf"))
    vendor_mapping = vendor_mapping or {}
    results = []
 
    def scan_one(pdf_path: Path) -> InvoiceScanResult:
        vendor_id = vendor_mapping.get(pdf_path.stem.split("_")[0])
        try:
            analysis = analyze_invoice_for_fraud(str(pdf_path), vendor_id)
            risk = evaluate_fraud_risk(analysis)
            return InvoiceScanResult(
                filename=pdf_path.name,
                vendor_id=vendor_id,
                fraud_score=risk["fraud_score"],
                risk_level=risk["risk_level"],
                action=risk["recommended_action"],
                flags=risk["critical_flags"] + risk["high_flags"]
            )
        except Exception as e:
            return InvoiceScanResult(
                filename=pdf_path.name,
                vendor_id=vendor_id,
                fraud_score=-1,
                risk_level="ERROR",
                action="Manual review required",
                flags=[],
                error=str(e)
            )
 
    with ThreadPoolExecutor(max_workers=4) as executor:
        futures = {executor.submit(scan_one, f): f for f in invoice_files}
        for future in as_completed(futures):
            results.append(future.result())
 
    return sorted(results, key=lambda r: r.fraud_score, reverse=True)
 
 
def route_invoices(results: list[InvoiceScanResult]):
    """Route invoices to appropriate queues based on fraud risk."""
    queues = {"CRITICAL": [], "HIGH": [], "MEDIUM": [], "LOW": [], "ERROR": []}
 
    for r in results:
        queues[r.risk_level].append(r)
 
    print("\n═══ Invoice Fraud Scan Summary ═══")
    print(f"  CRITICAL (reject):    {len(queues['CRITICAL'])} invoices")
    print(f"  HIGH (hold/review):   {len(queues['HIGH'])} invoices")
    print(f"  MEDIUM (flag):        {len(queues['MEDIUM'])} invoices")
    print(f"  LOW (approve):        {len(queues['LOW'])} invoices")
    print(f"  ERRORS:               {len(queues['ERROR'])} invoices")
 
    if queues["CRITICAL"]:
        print("\n⚠ CRITICAL FRAUD ALERTS:")
        for inv in queues["CRITICAL"]:
            print(f"  - {inv.filename} (Score: {inv.fraud_score})")
            for flag in inv.flags:
                print(f"      {flag['type']}: {flag['description']}")
 
    return queues

from fastapi import FastAPI, UploadFile, File, Form
import tempfile
 
app = FastAPI()
 
@app.post("/ap/screen-invoice")
async def screen_invoice(
    invoice: UploadFile = File(...),
    vendor_id: str = Form(None),
    invoice_amount: float = Form(None)
):
    """
    AP screening endpoint: returns fraud risk before routing to approval.
    Called by your AP system when a new invoice is received.
    """
    # Save to temp file for analysis
    with tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as tmp:
        tmp.write(await invoice.read())
        tmp_path = tmp.name
 
    try:
        analysis = analyze_invoice_for_fraud(tmp_path, vendor_id)
        risk = evaluate_fraud_risk(analysis)
 
        # Override risk level if amount is unusually high
        if invoice_amount and invoice_amount > 50000 and risk["fraud_score"] > 30:
            risk["risk_level"] = "HIGH"
            risk["recommended_action"] = "HOLD - Large amount + elevated fraud score requires CFO approval."
 
        return {
            "status": "screened",
            "fraud_score": risk["fraud_score"],
            "risk_level": risk["risk_level"],
            "recommended_action": risk["recommended_action"],
            "bank_account_changed": risk["bank_account_changed"],
            "critical_flags": risk["critical_flags"],
            "route_to_queue": risk["risk_level"].lower()
        }
    finally:
        os.unlink(tmp_path)