core/storage.py

"""Persistent per-user transaction storage on the HF Dataset.

Phase 4: each user's transactions live in their own append-only file
(data/<user>.jsonl), so one person's save never clobbers another's. A
"transaction" is the unified record for a parsed receipt, a payment screenshot,
or a manual entry — see the schema in _build_entry(). Model-independent, so it
stays unit-testable in isolation. Low-level dataset IO is in core.hubio.
"""

from __future__ import annotations

import uuid
from datetime import datetime, timezone
from typing import Any

from core import hubio

USER_DATA_DIR = "data"
BUDGET_DIR = "budgets"


def _num(value: Any) -> float:
    try:
        return float(value)
    except (TypeError, ValueError):
        return 0.0


def _norm_user(user: Any) -> str:
    return str(user or "").strip().lower()


def _user_path(user: str) -> str:
    return f"{USER_DATA_DIR}/{_norm_user(user)}.jsonl"


# --------------------------------------------------------------------------- #
# Build + save
# --------------------------------------------------------------------------- #
def _build_entry(record: dict[str, Any], user: str) -> dict[str, Any]:
    """Project a UI record into the persisted transaction schema (v2)."""
    category = str(record.get("category") or record.get("receipt_category") or "Other")
    entry: dict[str, Any] = {
        "id": uuid.uuid4().hex,
        "saved_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
        "user": _norm_user(user),
        "source": str(record.get("source", "receipt") or "receipt"),
        "vendor": str(record.get("vendor", "") or ""),
        "date": str(record.get("date", "") or ""),
        "currency": str(record.get("currency", "") or ""),
        "note": str(record.get("note", "") or ""),
        "total": _num(record.get("total", 0)),
        "category": category,
        "receipt_category": category,  # back-compat
        "understanding": str(record.get("understanding", "") or ""),
        "line_items": [
            {
                "name": str(it.get("name", "") or ""),
                "qty": _num(it.get("qty", 1)) or 1,
                "amount": _num(it.get("amount", 0)),
                "category": str(it.get("category") or category),
            }
            for it in (record.get("line_items") or [])
        ],
        "charges": [
            {"label": str(c.get("label", "") or "Charge"), "amount": _num(c.get("amount", 0))}
            for c in (record.get("charges") or [])
        ],
    }
    return entry


# --------------------------------------------------------------------------- #
# Per-user budget
# --------------------------------------------------------------------------- #
def _budget_path(user: str) -> str:
    return f"{BUDGET_DIR}/{_norm_user(user)}.jsonl"


def get_budget(user: str, force: bool = False) -> float:
    if not _norm_user(user):
        return 0.0
    rows = hubio.read_jsonl(_budget_path(user), force=force)
    return _num(rows[-1].get("amount")) if rows else 0.0


def set_budget(user: str, amount: float) -> float:
    if not _norm_user(user):
        raise ValueError("No user.")
    hubio.write_jsonl(_budget_path(user), [{"amount": _num(amount)}], "set budget")
    return _num(amount)


def save(user: str, record: dict[str, Any]) -> dict[str, Any]:
    """Append a transaction to the user's file. Returns the persisted entry.

    Raises on a hard failure so the UI can surface it.
    """
    if not _norm_user(user):
        raise ValueError("No user — sign in before saving.")
    entry = _build_entry(record, user)
    hubio.append_jsonl(
        _user_path(user), entry, message=f"Add {entry['source']} {entry['id'][:8]}"
    )
    return entry


# --------------------------------------------------------------------------- #
# Load + view
# --------------------------------------------------------------------------- #
def load(user: str, force: bool = False) -> list[dict[str, Any]]:
    """All of this user's transactions, oldest first ([] if none)."""
    if not _norm_user(user):
        return []
    return hubio.read_jsonl(_user_path(user), force=force)


def to_table(records: list[dict[str, Any]]) -> list[list[Any]]:
    """Rows for a transactions table: [date, vendor, total, category]. Recent first."""
    rows: list[list[Any]] = []
    for r in reversed(records):
        rows.append(
            [
                r.get("date", ""),
                r.get("vendor", ""),
                _num(r.get("total", 0)),
                r.get("receipt_category", "Other"),
            ]
        )
    return rows


def load_table(user: str, force: bool = False) -> list[list[Any]]:
    """Convenience: load this user's transactions, formatted for the table."""
    return to_table(load(user, force=force))