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