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()