finbot/models.py

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