"""Turn user input into expense drafts. Order of preference throughout: deterministic parsing first, model second. The regex parser handles "coffee 250" instantly, for free, and identically every time. The model is spent only on what actually needs judgement -- reading a receipt photo, or working out which column of an unfamiliar bank export is the amount. Statement imports never write straight to the ledger. They produce a :class:`~finbot.models.PendingImport` that the user confirms, because a model that misreads a column would otherwise quietly poison months of history. """ from __future__ import annotations import logging from datetime import date, datetime from typing import Any, Optional from .categories import CATEGORIES, categorise, coerce from .extract import ExtractedDocument, rows_to_text from .llm import LLMClient, LLMUnavailable from .models import ExpenseDraft, PendingImport from .money import to_minor from .parser import parse_expense log = logging.getLogger(__name__) _CATEGORY_LIST = ", ".join(CATEGORIES) _STATEMENT_SYSTEM = f"""You extract transactions from bank and card statements. The statement may come from any bank in any country, in any currency, with any column order or language. Work out the structure yourself from the content. Return a JSON object with one key, "transactions", whose value is an array. Each element must have exactly these fields: "date": ISO 8601 date, YYYY-MM-DD "description": the merchant or narration, cleaned of reference numbers "amount": a positive number in major units (e.g. 1234.56), never a string "currency": ISO 4217 code, e.g. INR, USD, EUR "direction": "debit" if money left the account, "credit" if it came in "category": one of: {_CATEGORY_LIST} Rules: - Infer the currency from symbols or codes in the document. If genuinely absent, use "{{default_currency}}". - Amounts are always positive; use "direction" to express the sign. - Ambiguous dates are day-first unless the document proves otherwise. - Skip opening/closing balances, subtotals, carried-forward lines and headers. - If you cannot read a row confidently, omit it rather than guessing. - Return only the JSON object.""" _RECEIPT_SYSTEM = f"""You read receipts and bills from photographs. Return a JSON object with these fields: "merchant": the shop or business name, "" if unreadable "date": ISO 8601 date YYYY-MM-DD, or "" if not visible "amount": the FINAL TOTAL paid, positive, in major units, as a number "currency": ISO 4217 code inferred from the receipt "category": one of: {_CATEGORY_LIST} "confidence": 0.0-1.0, how sure you are of the total Take the grand total actually paid -- not a subtotal, not a single line item, not the tax. If the image is not a receipt, set amount to 0.""" _FREETEXT_SYSTEM = f"""You turn a person's shorthand note into one expense record. Return a JSON object with: "amount": positive number in major units "currency": ISO 4217 code "description": short, human-readable "date": ISO 8601 YYYY-MM-DD "category": one of: {_CATEGORY_LIST} If the note contains no spend at all, return {{"amount": 0}}.""" def _parse_any_date(value: Any, fallback: date) -> date: if isinstance(value, date) and not isinstance(value, datetime): return value if isinstance(value, datetime): return value.date() text = str(value or "").strip() if not text: return fallback try: return date.fromisoformat(text[:10]) except ValueError: pass try: from dateutil import parser as dateparser return dateparser.parse(text, dayfirst=True, fuzzy=True).date() except Exception: return fallback def _coerce_amount(value: Any) -> Optional[float]: if isinstance(value, (int, float)): return abs(float(value)) text = str(value or "").strip() if not text: return None # Strip everything that isn't numeric, then reuse the shared separator logic. from .money import _normalise_number # local import: private helper cleaned = "".join(ch for ch in text if ch.isdigit() or ch in ".,-") cleaned = cleaned.lstrip("-") normalised = _normalise_number(cleaned) if normalised is None: return None try: return abs(float(normalised)) except ValueError: return None def _draft_from_obj( obj: dict, default_currency: str, today: date ) -> Optional[ExpenseDraft]: """Validate one model-emitted record into a draft, or drop it.""" if not isinstance(obj, dict): return None amount = _coerce_amount(obj.get("amount")) if not amount or amount <= 0: return None currency = str(obj.get("currency") or default_currency).upper()[:3] if not currency.isalpha(): currency = default_currency.upper() description = str(obj.get("description") or obj.get("merchant") or "").strip() description = " ".join(description.split())[:200] or "unlabelled" category = coerce(str(obj.get("category") or "")) if category == "other": # The model's label was unusable; keyword rules often still know this one. guessed, hit = categorise(description) if hit: category = guessed return ExpenseDraft( amount_minor=to_minor(amount, currency), currency=currency, description=description, occurred_on=_parse_any_date(obj.get("date"), today), category=category, merchant=str(obj.get("merchant") or "").strip()[:120], confidence=float(obj.get("confidence") or 0.7), ) class Ingestor: def __init__(self, llm: LLMClient, default_currency: str = "INR"): self.llm = llm self.default_currency = default_currency # ------------------------------------------------------------ free text def from_text(self, text: str, today: Optional[date] = None) -> Optional[ExpenseDraft]: """Deterministic parse, with a model fallback only if that fails.""" today = today or date.today() draft = parse_expense(text, self.default_currency, today) if draft is not None: return draft if not self.llm.enabled: return None try: obj = self.llm.chat_json( _FREETEXT_SYSTEM, f"Today is {today.isoformat()}.\n\nNote: {text}", max_tokens=400, ) except (LLMUnavailable, ValueError) as exc: log.info("Free-text model fallback unavailable: %s", exc) return None return _draft_from_obj(obj, self.default_currency, today) # -------------------------------------------------------------- receipts def from_receipt( self, image_bytes: bytes, mime_type: str, today: Optional[date] = None ) -> tuple[Optional[ExpenseDraft], str]: """Read a receipt photo. Returns (draft, error_message).""" today = today or date.today() if not self.llm.enabled: return None, ( "Reading receipt photos needs an inference token. " "Set HF_TOKEN in the Space secrets, or type the amount instead." ) try: obj = self.llm.vision_json( _RECEIPT_SYSTEM, f"Today is {today.isoformat()}. Read this receipt.", image_bytes, mime_type, ) except (LLMUnavailable, ValueError) as exc: # An infrastructure failure is not the user's fault, so don't tell # them to retake the photo, and don't surface raw exception text. log.warning("Receipt read failed: %s", exc) return None, ( "Something went wrong on my side reading that image — it's been " "logged. You can type the amount instead, e.g. 450 lunch." ) draft = _draft_from_obj(obj, self.default_currency, today) if draft is None: return None, ( "I couldn't find a total on that image. If it is a receipt, " "try a straighter, better-lit photo, or just type the amount." ) return draft, "" # ------------------------------------------------------------ statements def from_document( self, doc: ExtractedDocument, user_id: int, today: Optional[date] = None, label: str = "", ) -> tuple[Optional[PendingImport], str]: """Map an extracted statement into drafts. Returns (pending, error).""" today = today or date.today() if doc.kind == "unsupported" or doc.is_empty: return None, doc.note or "There was nothing readable in that file." if not self.llm.enabled: return None, ( "Reading statements needs an inference token. Set HF_TOKEN in " "the Space secrets and send the file again." ) payload = rows_to_text(doc.rows) if doc.rows else doc.text if not payload.strip(): return None, "There was nothing readable in that file." system = _STATEMENT_SYSTEM.replace("{default_currency}", self.default_currency) try: result = self.llm.chat_json( system, f"Today is {today.isoformat()}. Extract every transaction.\n\n{payload}", max_tokens=8000, ) except (LLMUnavailable, ValueError) as exc: log.warning("Statement extraction failed: %s", exc) return None, ( "Something went wrong on my side reading that statement — it's " "been logged. Try again in a moment, or send a CSV export." ) records = result.get("transactions") if isinstance(result, dict) else result if not isinstance(records, list): return None, "The extractor returned an unexpected shape. Try again?" drafts: list[ExpenseDraft] = [] skipped = 0 credits = 0 for record in records: if not isinstance(record, dict): skipped += 1 continue # Credits are money coming in. Storing them alongside spend would # corrupt every "total spent" figure, so they are counted and # reported rather than written. if str(record.get("direction", "debit")).lower().startswith("cred"): credits += 1 continue draft = _draft_from_obj(record, self.default_currency, today) if draft is None: skipped += 1 continue drafts.append(draft) if not drafts: detail = f" ({credits} credits, {skipped} unreadable rows)" if (credits or skipped) else "" return None, f"I found no spending transactions in that file{detail}." pending = PendingImport( user_id=user_id, drafts=drafts, source=doc.kind, skipped=skipped, label=label, ) if credits: pending.label = (pending.label + f" · {credits} credits skipped").strip(" ·") if doc.truncated: pending.label = (pending.label + " · file truncated").strip(" ·") return pending, ""