| """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",
|
| },
|
| }
|
|
|
