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