File size: 2,847 Bytes
e28a7b2 76af018 e28a7b2 76af018 e28a7b2 76af018 e28a7b2 76af018 e28a7b2 76af018 e28a7b2 |
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 |
"""Domain models for expenses and merge suggestions."""
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime, timezone
from decimal import Decimal
from typing import List, Mapping, Sequence
from .config import settings
def _extract_expense_time(doc: Mapping[str, object]) -> datetime:
for field in settings.time_fields:
value = doc.get(field)
if isinstance(value, datetime):
if value.tzinfo is None:
value = value.replace(tzinfo=timezone.utc)
return value
raise ValueError("Expense document missing a valid timestamp field")
def _extract_merchant(doc: Mapping[str, object]) -> str:
for field in settings.merchant_fields:
value = doc.get(field)
if value:
text = str(value).strip()
if text:
return text
return ""
@dataclass(frozen=True)
class Expense:
expense_id: str
amount: Decimal
currency: str
merchant: str
expense_time: datetime
user_id: str | None = None
source: str | None = None
metadata: Mapping[str, object] | None = None
@staticmethod
def from_document(doc: Mapping[str, object]) -> "Expense":
try:
amount_value = Decimal(str(doc["amount"]))
except KeyError as exc:
raise ValueError("Expense document missing 'amount'") from exc
expense_time = _extract_expense_time(doc)
merchant_value = _extract_merchant(doc)
return Expense(
expense_id=str(doc.get("_id")),
amount=amount_value,
currency=str(doc.get("currency", "INR")),
merchant=merchant_value,
expense_time=expense_time,
user_id=str(doc.get("user")) if doc.get("user") else None,
source=doc.get("source"),
metadata=doc.get("metadata") or {},
)
@dataclass
class MergeSuggestion:
candidate_ids: Sequence[str]
message: str
details: Mapping[str, object]
audit: Mapping[str, object]
status: str = "pending"
_id: str | None = None
def to_document(self) -> Mapping[str, object]:
return {
"candidate_ids": list(self.candidate_ids),
"message": self.message,
"details": dict(self.details),
"audit": dict(self.audit),
"status": self.status,
}
@dataclass
class DuplicateCluster:
expenses: List[Expense] = field(default_factory=list)
amount_delta_pct: float = 0.0
time_delta_minutes: float = 0.0
merchant_rule: str = "exact"
def to_details(self) -> Mapping[str, object]:
return {
"amount_delta_pct": self.amount_delta_pct,
"time_delta_minutes": self.time_delta_minutes,
"merchant_match_rule": self.merchant_rule,
}
|