File size: 3,786 Bytes
2edb151 | 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 | from __future__ import annotations
import json
import re
from datetime import date as Date
from datetime import datetime as DateTime
from decimal import Decimal, ROUND_HALF_UP
from enum import StrEnum
from typing import Any
from pydantic import BaseModel, ConfigDict, Field, field_validator
from app.config import CATEGORIES, DOC_KINDS
_FENCE_RE = re.compile(r"```(?:json)?\s*(.*?)\s*```", re.DOTALL | re.IGNORECASE)
class ReceiptStatus(StrEnum):
queued = "queued"
processing = "processing"
needs_ocr = "needs_ocr"
needs_extract = "needs_extract"
needs_review = "needs_review"
confirmed = "confirmed"
failed = "failed"
duplicate = "duplicate"
class MatchBand(StrEnum):
exact = "exact"
auto = "auto"
review = "review"
unmatched = "unmatched"
confirmed = "confirmed"
class LineItem(BaseModel):
model_config = ConfigDict(extra="forbid")
description: str
qty: float | None = None
unit_price: Decimal | None = None
amount: Decimal | None = None
sku: str | None = None
class ReceiptExtract(BaseModel):
model_config = ConfigDict(extra="forbid")
doc_kind: str = "receipt"
category: str = "other"
vendor: str | None = None
date: Date | None = None
tax: Decimal | None = None
total: Decimal | None = None
currency: str | None = None
line_items: list[LineItem] = Field(default_factory=list)
@field_validator("doc_kind")
@classmethod
def _doc_kind(cls, value: str) -> str:
kind = (value or "receipt").strip().lower()
return kind if kind in DOC_KINDS else "document"
@field_validator("category")
@classmethod
def _category(cls, value: str) -> str:
cat = (value or "other").strip().lower()
return cat if cat in CATEGORIES else "other"
@field_validator("date", mode="before")
@classmethod
def _date(cls, value: object) -> object:
if value in (None, "", "null"):
return None
if isinstance(value, Date):
return value
text = str(value).strip()[:10]
return DateTime.strptime(text, "%Y-%m-%d").date()
class MatchHit(BaseModel):
catalog_id: int | None = None
sku: str | None = None
vendor: str | None = None
description: str | None = None
similarity: float
band: MatchBand
reason: str
class ProcessResult(BaseModel):
receipt_id: int | None = None
status: ReceiptStatus
source_path: str
extract: ReceiptExtract | None = None
matches: list[MatchHit] = Field(default_factory=list)
error: str | None = None
def to_cents(value: Decimal | float | int | None) -> int | None:
if value is None:
return None
quantized = (Decimal(str(value)) * 100).quantize(Decimal("1"), rounding=ROUND_HALF_UP)
return int(quantized)
def cents_to_decimal(cents: int | None) -> Decimal | None:
if cents is None:
return None
return (Decimal(cents) / Decimal("100")).quantize(Decimal("0.01"))
def extract_json_object(text: str) -> str:
stripped = text.strip()
fenced = _FENCE_RE.search(stripped)
if fenced:
stripped = fenced.group(1).strip()
start = stripped.find("{")
end = stripped.rfind("}")
if start < 0 or end <= start:
raise ValueError("no JSON object in model output")
return stripped[start : end + 1]
def parse_extract_json(text: str) -> ReceiptExtract:
payload = json.loads(extract_json_object(text))
if not isinstance(payload, dict):
raise ValueError("extract JSON must be an object")
if "line_items" not in payload or payload["line_items"] is None:
payload["line_items"] = []
return ReceiptExtract.model_validate(payload)
EXTRACT_JSON_SCHEMA: dict[str, Any] = ReceiptExtract.model_json_schema()
|