File size: 3,298 Bytes
af24ae8 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 | """Pydantic data models for invoice extraction and validation."""
from __future__ import annotations
from datetime import datetime
from enum import Enum
from typing import Any
from pydantic import BaseModel, Field, field_validator
class ValidationSeverity(str, Enum):
ERROR = "error"
WARNING = "warning"
INFO = "info"
class LineItem(BaseModel):
description: str = ""
quantity: float | None = None
unit_price: float | None = None
total: float | None = None
confidence: float = Field(default=0.0, ge=0.0, le=1.0)
class ValidationIssue(BaseModel):
field: str
severity: ValidationSeverity
message: str
rule_id: str
class ValidationReport(BaseModel):
is_valid: bool
score: float = Field(ge=0.0, le=1.0)
issues: list[ValidationIssue] = Field(default_factory=list)
def to_markdown(self) -> str:
if not self.issues:
return "✅ All business rules passed."
lines = ["### Validation Report", ""]
for issue in self.issues:
icon = {"error": "❌", "warning": "⚠️", "info": "ℹ️"}[issue.severity.value]
lines.append(f"- {icon} **{issue.field}** (`{issue.rule_id}`): {issue.message}")
lines.append("")
lines.append(f"**Overall score:** {self.score:.0%} | **Status:** {'PASS' if self.is_valid else 'REVIEW REQUIRED'}")
return "\n".join(lines)
class ExtractedInvoice(BaseModel):
vendor_name: str | None = None
vendor_tax_id: str | None = None
invoice_number: str | None = None
invoice_date: str | None = None
invoice_date_jalali: str | None = None
buyer_name: str | None = None
subtotal: float | None = None
tax_amount: float | None = None
total_amount: float | None = None
currency: str = "IRR"
line_items: list[LineItem] = Field(default_factory=list)
raw_text: str = ""
confidence: float = Field(default=0.0, ge=0.0, le=1.0)
extraction_method: str = "hybrid_ocr_ner"
processed_at: datetime = Field(default_factory=datetime.utcnow)
@field_validator("currency", mode="before")
@classmethod
def normalize_currency(cls, value: Any) -> str:
if not value:
return "IRR"
return str(value).upper()
def to_export_dict(self) -> dict[str, Any]:
return {
"vendor": {
"name": self.vendor_name,
"tax_id": self.vendor_tax_id,
},
"invoice": {
"number": self.invoice_number,
"date_gregorian": self.invoice_date,
"date_jalali": self.invoice_date_jalali,
"buyer": self.buyer_name,
},
"amounts": {
"subtotal": self.subtotal,
"tax": self.tax_amount,
"total": self.total_amount,
"currency": self.currency,
},
"line_items": [item.model_dump() for item in self.line_items],
"metadata": {
"confidence": round(self.confidence, 3),
"extraction_method": self.extraction_method,
"processed_at": self.processed_at.isoformat() + "Z",
},
}
|