Spaces:
Sleeping
Sleeping
| """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. <code>450 lunch</code>." | |
| ) | |
| 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, "" | |