File size: 5,269 Bytes
7b5611f | 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 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 | from __future__ import annotations
from dataclasses import dataclass, field
from typing import Dict, List, Literal, Optional
# ── Line item ────────────────────────────────────────────────────────────────
@dataclass
class InvoiceLineItem:
product_raw: str
product_normalized: Optional[str] = None
product_id: Optional[str] = None
quantity: float = 0.0
unit_price: float = 0.0
gst_rate: float = 0.0
line_total: float = 0.0
# ── Invoice JSON ──────────────────────────────────────────────────────────────
@dataclass
class InvoiceJSON:
invoice_number: Optional[str] = None
supplier: Optional[str] = None
date: Optional[str] = None
items: List[InvoiceLineItem] = field(default_factory=list)
grand_total: float = 0.0
extraction_warnings: List[str] = field(default_factory=list)
unmatched_products: List[str] = field(default_factory=list)
NormalisedInvoice = InvoiceJSON
# ── Anomaly flags ─────────────────────────────────────────────────────────────
FlagType = Literal[
"price_anomaly",
"delivery_shortage",
"duplicate_charge",
"gst_mismatch",
"duplicate_invoice",
]
@dataclass
class AnomalyFlag:
flag_type: FlagType
product_id: Optional[str] = None
product_name: Optional[str] = None
amount_inr: float = 0.0
description: str = ""
action_item: str = ""
metadata: dict = field(default_factory=dict)
# ── Leakage report ────────────────────────────────────────────────────────────
@dataclass
class LeakageReport:
audit_run_id: str
invoice_number: Optional[str] = None
supplier: Optional[str] = None
date: Optional[str] = None
anomaly_flags: List[AnomalyFlag] = field(default_factory=list)
total_leakage_inr: float = 0.0
action_items: List[str] = field(default_factory=list)
has_delivery_data: bool = False
unexpected_deliveries: List[str] = field(default_factory=list)
low_confidence_photos: List[str] = field(default_factory=list)
skipped_price_checks: int = 0
skipped_gst_checks: int = 0
passed_all_checks: bool = False
# ── Agent trace ───────────────────────────────────────────────────────────────
@dataclass
class AgentTraceEntry:
agent_name: str
agent_version: str
audit_run_id: str
timestamp_start: str # ISO 8601 UTC
timestamp_end: str
duration_ms: int
input_summary: str
output_summary: str
CANONICAL_AGENT_ORDER: List[str] = [
"Invoice_Extractor",
"Product_Matcher",
"Pricing_Agent",
"Visual_Counter",
"Reconciliation_Agent",
"Savings_Agent",
]
# ── Catalog entry ─────────────────────────────────────────────────────────────
@dataclass
class CatalogEntry:
product_id: str
canonical_name: str
hsn_code: str
gst_rate: float
category: str
brand: str
common_aliases: List[str] = field(default_factory=list)
# ── Type aliases ──────────────────────────────────────────────────────────────
DeliveryCountMap = Dict[str, int]
# ── Progress streaming ────────────────────────────────────────────────────────
AuditStage = Literal[
"extracting",
"normalising",
"checking_prices",
"counting_products",
"reconciling",
"generating_report",
"complete",
"error",
]
@dataclass
class AuditProgressUpdate:
stage: AuditStage
message: str = ""
agent_name: str = ""
duration_ms: int = 0
# ── Pricing agent metadata ────────────────────────────────────────────────────
@dataclass
class PricingAgentMeta:
skipped_price_checks: int = 0
skipped_gst_checks: int = 0
|