validators/amount_validator.py

"""
InvoiceForge AI — validators/amount_validator.py
Invoice amount cross-validation.

Checks:
  1. Grand Total = Subtotal + GST Total  (±1% tolerance for rounding)
  2. Item subtotal sum ≈ header subtotal
  3. Individual item: Total = Qty × Rate - Discount + GST
  4. GST components: CGST+SGST+IGST ≈ declared GST Total

Returns structured ValidationResult with errors, warnings, and confidence.
"""

from __future__ import annotations

import logging
from dataclasses import dataclass, field
from typing import Any

logger = logging.getLogger(__name__)

ROUNDING_TOLERANCE: float = 1.0      # Absolute rupee tolerance for small invoices
PERCENT_TOLERANCE: float = 0.01      # 1% relative tolerance


@dataclass
class ValidationResult:
    """Result of invoice amount validation."""
    is_valid: bool
    confidence: float
    errors: list[str] = field(default_factory=list)
    warnings: list[str] = field(default_factory=list)


class AmountValidator:
    """
    Cross-validates invoice amounts for consistency.

    Usage:
        validator = AmountValidator()
        result = validator.validate(items, totals_dict, header)
    """

    def validate(
        self,
        items: list[dict],
        totals: dict,
        header: dict | None = None,
    ) -> ValidationResult:
        """
        Validate all amount fields in the extracted invoice.

        Args:
            items:  List of line item dicts with taxable_amount, gst_amount, total_amount.
            totals: Dict with subtotal, gstTotal, grandTotal, cgst, sgst, igst.
            header: Header dict (used for GSTIN validation trigger, optional).

        Returns:
            ValidationResult with is_valid, confidence, errors, warnings.
        """
        errors: list[str] = []
        warnings: list[str] = []
        confidence_penalties: list[float] = []

        subtotal = _safe(totals.get("subtotal", 0))
        gst_total = _safe(totals.get("gstTotal", 0))
        grand_total = _safe(totals.get("grandTotal", 0))
        cgst = _safe(totals.get("cgst", 0))
        sgst = _safe(totals.get("sgst", 0))
        igst = _safe(totals.get("igst", 0))

        # ── Check 1: Grand Total = Subtotal + GST ─────────────────────────
        if grand_total > 0 and subtotal > 0:
            expected_grand = round(subtotal + gst_total, 2)
            diff = abs(grand_total - expected_grand)
            tolerance = max(ROUNDING_TOLERANCE, grand_total * PERCENT_TOLERANCE)

            if diff > tolerance:
                errors.append(
                    f"Grand Total mismatch: declared {grand_total:.2f} "
                    f"≠ Subtotal + GST = {expected_grand:.2f} "
                    f"(diff {diff:.2f})."
                )
                confidence_penalties.append(0.10)

        # ── Check 2: Item subtotals sum ≈ header subtotal ─────────────────
        if items and subtotal > 0:
            computed_sub = round(
                sum(_safe(i.get("taxable_amount", i.get("taxableAmount", 0))) for i in items),
                2,
            )
            diff = abs(computed_sub - subtotal)
            tolerance = max(ROUNDING_TOLERANCE, subtotal * PERCENT_TOLERANCE)
            if diff > tolerance and computed_sub > 0:
                warnings.append(
                    f"Item subtotal sum ({computed_sub:.2f}) differs from "
                    f"header subtotal ({subtotal:.2f}) by {diff:.2f}."
                )
                confidence_penalties.append(0.05)

        # ── Check 3: GST component sum ≈ GST Total ────────────────────────
        if gst_total > 0 and (cgst + sgst + igst) > 0:
            component_sum = round(cgst + sgst + igst, 2)
            diff = abs(component_sum - gst_total)
            tolerance = max(ROUNDING_TOLERANCE, gst_total * PERCENT_TOLERANCE)
            if diff > tolerance:
                warnings.append(
                    f"GST component sum (CGST={cgst:.2f} + SGST={sgst:.2f} + "
                    f"IGST={igst:.2f} = {component_sum:.2f}) differs from "
                    f"GST Total ({gst_total:.2f})."
                )
                confidence_penalties.append(0.03)

        # ── Check 4: Individual item amounts ─────────────────────────────
        for i, item in enumerate(items):
            tax_amount = _safe(item.get("taxable_amount", item.get("taxableAmount", 0)))
            gst_amount = _safe(item.get("gst_amount", item.get("gstAmount", 0)))
            total = _safe(item.get("total_amount", item.get("totalAmount", 0)))

            if tax_amount > 0 and gst_amount >= 0 and total > 0:
                expected_total = round(tax_amount + gst_amount, 2)
                diff = abs(total - expected_total)
                tolerance = max(ROUNDING_TOLERANCE, total * PERCENT_TOLERANCE)
                if diff > tolerance:
                    desc = item.get("description", f"Item {i+1}")[:30]
                    warnings.append(
                        f"'{desc}': item total {total:.2f} ≠ "
                        f"taxable {tax_amount:.2f} + gst {gst_amount:.2f} = {expected_total:.2f}."
                    )
                    confidence_penalties.append(0.02)

        # ── Confidence calculation ────────────────────────────────────────
        total_penalty = min(sum(confidence_penalties), 0.35)
        confidence = round(1.0 - total_penalty, 4)

        # Zero amounts are a data-quality warning (not hard error)
        if grand_total == 0 and subtotal == 0:
            warnings.append("All amount fields are zero — extraction may have failed.")
            confidence = min(confidence, 0.50)

        is_valid = len(errors) == 0

        return ValidationResult(
            is_valid=is_valid,
            confidence=confidence,
            errors=errors,
            warnings=warnings,
        )

    def validate_from_result(self, result: dict) -> ValidationResult:
        """
        Convenience wrapper to validate directly from a full extraction result dict.

        Args:
            result: Full InvoiceForge result dict with 'items', 'totals', 'header'.
        """
        return self.validate(
            items=result.get("items", []),
            totals=result.get("totals", {}),
            header=result.get("header"),
        )


def _safe(value: Any) -> float:
    """Safely convert to float."""
    try:
        return float(value)
    except (TypeError, ValueError):
        return 0.0