Spaces:
Running on Zero
Running on Zero
File size: 4,528 Bytes
bfd6783 | 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 147 148 149 150 151 152 153 154 155 156 157 158 159 | """Domain records. Plain dataclasses -- the storage layer maps them to SQLite."""
from __future__ import annotations
import hashlib
import re
import uuid
from dataclasses import dataclass, field
from datetime import date, datetime, timezone
from typing import Optional
from .money import format_money
#: Where a record came from. Kept on every expense so a bad import can be
#: found and reversed as a unit.
SOURCES = ("text", "photo", "pdf", "csv", "xlsx", "manual")
def _now() -> datetime:
return datetime.now(timezone.utc)
def _slug(text: str) -> str:
"""Loose normalisation for dedupe: case, punctuation and spacing-insensitive."""
return re.sub(r"[^a-z0-9]+", "", text.lower())
@dataclass
class Expense:
user_id: int
amount_minor: int
currency: str
description: str
category: str
occurred_on: date
id: str = field(default_factory=lambda: uuid.uuid4().hex)
merchant: str = ""
source: str = "text"
#: Groups rows that arrived together, so /undo can drop a whole import.
batch_id: str = ""
note: str = ""
#: 1.0 for deterministic parses, lower for model-extracted rows.
confidence: float = 1.0
created_at: datetime = field(default_factory=_now)
@property
def dedupe_key(self) -> str:
"""Stable identity for 'the same transaction'.
Deliberately excludes id, source and confidence so that the same row
re-imported from a re-downloaded statement collapses onto the original.
"""
basis = "|".join(
[
str(self.user_id),
str(self.amount_minor),
self.currency.upper(),
self.occurred_on.isoformat(),
_slug(self.merchant or self.description),
]
)
return hashlib.sha256(basis.encode("utf-8")).hexdigest()[:32]
def display(self) -> str:
amount = format_money(self.amount_minor, self.currency)
label = self.merchant or self.description or self.category
return f"{amount} 路 {label} 路 {self.category} 路 {self.occurred_on.isoformat()}"
@dataclass
class Budget:
user_id: int
category: str
amount_minor: int
currency: str
period: str = "monthly"
created_at: datetime = field(default_factory=_now)
@dataclass
class ExpenseDraft:
"""A candidate row awaiting confirmation.
Statement and receipt imports land here first. Nothing model-extracted is
written to the ledger without the user seeing the total and saying yes.
"""
amount_minor: int
currency: str
description: str
occurred_on: date
category: str = "other"
merchant: str = ""
confidence: float = 0.5
raw: str = ""
def to_expense(
self, user_id: int, source: str, batch_id: str, note: str = ""
) -> Expense:
return Expense(
user_id=user_id,
amount_minor=self.amount_minor,
currency=self.currency,
description=self.description,
category=self.category,
occurred_on=self.occurred_on,
merchant=self.merchant,
source=source,
batch_id=batch_id,
note=note,
confidence=self.confidence,
)
@dataclass
class PendingImport:
"""Drafts held between 'here is your statement' and '/confirm'."""
user_id: int
drafts: list[ExpenseDraft]
source: str
batch_id: str = field(default_factory=lambda: uuid.uuid4().hex)
created_at: datetime = field(default_factory=_now)
#: Rows dropped because they already exist in the ledger.
duplicates: int = 0
#: Rows the extractor emitted but could not be made sense of.
skipped: int = 0
label: str = ""
def total_by_currency(self) -> dict[str, int]:
totals: dict[str, int] = {}
for d in self.drafts:
totals[d.currency] = totals.get(d.currency, 0) + d.amount_minor
return totals
@dataclass
class Summary:
"""Aggregated spend for a period, ready to render."""
label: str
start: date
end: date
currency: str
total_minor: int
count: int
by_category: list[tuple[str, int]]
previous_total_minor: Optional[int] = None
@property
def delta_pct(self) -> Optional[float]:
if self.previous_total_minor is None or self.previous_total_minor == 0:
return None
return (
(self.total_minor - self.previous_total_minor)
/ self.previous_total_minor
* 100.0
)
|