"""Telegram message rendering. Uses Telegram's **HTML** parse mode rather than MarkdownV2. MarkdownV2 requires escaping ``. - ( ) ! +`` and more, which appear in virtually every line of financial output (``₹1,234.56``, ``-12%``, ``(30%)``); one missed escape and Telegram rejects the whole message. HTML needs only ``< > &`` escaped, so the failure surface is far smaller. """ from __future__ import annotations import html from datetime import date from typing import Iterable from .analytics import BudgetStatus, Recurring from .categories import label as category_label from .models import Expense, PendingImport, Summary from .money import format_money BAR_WIDTH = 10 def escape(text: str) -> str: return html.escape(str(text or ""), quote=False) def bar(fraction: float, width: int = BAR_WIDTH) -> str: filled = max(0, min(width, round(fraction * width))) return "█" * filled + "░" * (width - filled) def money(minor: int, currency: str) -> str: return escape(format_money(minor, currency)) def expense_line(expense: Expense) -> str: label = expense.merchant or expense.description or "unlabelled" return ( f"{money(expense.amount_minor, expense.currency)} · {escape(label)} " f"{escape(category_label(expense.category))} · " f"{expense.occurred_on.isoformat()}" ) def logged(expense: Expense, note: str = "") -> str: parts = [ f"✅ Logged {money(expense.amount_minor, expense.currency)} — " f"{escape(expense.description or expense.merchant or 'unlabelled')}", f"{escape(category_label(expense.category))} · {expense.occurred_on.isoformat()}", ] if note: parts.append(escape(note)) parts.append("/undo to remove") return "\n".join(parts) def summary(summ: Summary) -> str: if summ.count == 0: return ( f"{escape(summ.label)}\nNothing logged " f"({summ.start.isoformat()} → {summ.end.isoformat()})." ) lines = [ f"{escape(summ.label)} · {money(summ.total_minor, summ.currency)} " f"({summ.count} txns)" ] if (delta := summ.delta_pct) is not None: arrow = "▲" if delta > 0 else "▼" lines.append( f"{arrow} {abs(delta):.0f}% vs previous " f"({money(summ.previous_total_minor or 0, summ.currency)})" ) lines.append("") top = summ.by_category[:8] for name, amount in top: fraction = amount / summ.total_minor if summ.total_minor else 0 lines.append( f"{bar(fraction)} {escape(category_label(name))} " f"{money(amount, summ.currency)} ({fraction * 100:.0f}%)" ) if len(summ.by_category) > len(top): rest = sum(amount for _, amount in summ.by_category[len(top):]) lines.append(f"+{len(summ.by_category) - len(top)} more · {money(rest, summ.currency)}") return "\n".join(lines) def expense_list(expenses: Iterable[Expense], title: str, limit: int = 20) -> str: items = list(expenses)[:limit] if not items: return f"{escape(title)}\nNothing to show." lines = [f"{escape(title)}"] lines.extend(f"• {expense_line(e)}" for e in items) return "\n".join(lines) def pending_import(pending: PendingImport) -> str: totals = pending.total_by_currency() total_str = " + ".join(money(v, k) for k, v in sorted(totals.items())) lines = [ f"Found {len(pending.drafts)} transactions in that " f"{escape(pending.source.upper())} · {total_str}", ] detail = [] if pending.duplicates: detail.append(f"{pending.duplicates} already logged (skipped)") if pending.skipped: detail.append(f"{pending.skipped} unreadable") if pending.label: detail.append(pending.label) if detail: lines.append(f"{escape(' · '.join(detail))}") lines.append("") preview = pending.drafts[:8] for draft in preview: lines.append( f"• {money(draft.amount_minor, draft.currency)} · " f"{escape(draft.description[:40])} · " f"{escape(category_label(draft.category))} · " f"{draft.occurred_on.isoformat()}" ) if len(pending.drafts) > len(preview): lines.append(f"…and {len(pending.drafts) - len(preview)} more") lines.append("") lines.append("These were read by a model and not saved yet.") lines.append("/confirm to save · /cancel to discard") return "\n".join(lines) def recurring_list(items: list[Recurring], currency: str) -> str: if not items: return ( "Recurring charges\nNone detected yet — this needs a few " "months of history to spot a pattern." ) lines = ["Recurring charges"] for item in items[:15]: lines.append( f"• {money(item.typical_minor, item.currency)}/mo · {escape(item.label)} " f"({item.months} months, last {item.last_seen.isoformat()})" ) annual = sum(r.annualised_minor for r in items if r.currency == currency) if annual: lines.append("") lines.append(f"≈ {money(annual, currency)} a year if all continue.") return "\n".join(lines) def budget_list(statuses: list[BudgetStatus]) -> str: if not statuses: return ( "Budgets\nNone set.\n" "/budget food_dining 8000 to set one." ) lines = ["Budgets this month"] for status in statuses: mark = "⚠️" if status.over else ("🟡" if status.pct >= 80 else "🟢") lines.append( f"{mark} {bar(status.pct / 100)} " f"{escape(category_label(status.category))} " f"{money(status.spent_minor, status.currency)}/" f"{money(status.limit_minor, status.currency)} " f"({status.pct:.0f}%)" ) return "\n".join(lines) def csv_export(expenses: list[Expense]) -> str: import csv import io buffer = io.StringIO() writer = csv.writer(buffer) writer.writerow( ["date", "amount", "currency", "category", "merchant", "description", "source"] ) for expense in expenses: from .money import from_minor writer.writerow( [ expense.occurred_on.isoformat(), f"{from_minor(expense.amount_minor, expense.currency):.2f}", expense.currency, expense.category, expense.merchant, expense.description, expense.source, ] ) return buffer.getvalue() def period_footer(start: date, end: date) -> str: return f"{start.isoformat()} → {end.isoformat()}"