Log receipts sent to Telegram in a Notion expenses database

import base64
from datetime import date, timedelta
from clickruntime import on_event, on_trigger, credentials, http, ai

DB_ID, CHAT_ID = "your-notion-expenses-db-id", "your-telegram-chat-id"
SCHEMA = {"type": "object", "properties": {"vendor": {"type": "string"}, "date": {"type": "string"}, "amount": {"type": "number"}, "category": {"type": "string"}}}

def tg(method, **kw):
    return http.post(f"https://api.telegram.org/bot{credentials.secret('TELEGRAM_BOT_TOKEN')}/{method}", json=kw).json()

def notion(path, body):
    return http.post(f"https://api.notion.com/v1/{path}", json=body, headers={"Authorization": f"Bearer {credentials.connection('notion', 'work')}", "Notion-Version": "2022-06-28"}).json()

@on_event
def receipt(event):
    photo = ((event.json or {}).get("message") or {}).get("photo")
    if not photo:
        return {"skipped": True}
    path = tg("getFile", file_id=photo[-1]["file_id"])["result"]["file_path"]
    img = http.get(f"https://api.telegram.org/file/bot{credentials.secret('TELEGRAM_BOT_TOKEN')}/{path}").content
    ocr = http.post("https://vision.googleapis.com/v1/images:annotate", headers={"Authorization": f"Bearer {credentials.connection('google', 'work')}"},
                    json={"requests": [{"image": {"content": base64.b64encode(img).decode()}, "features": [{"type": "TEXT_DETECTION"}]}]}).json()
    text = ocr["responses"][0].get("fullTextAnnotation", {}).get("text", "")[:3000]
    try:
        r = ai.generate(f"Receipt text:\n{text}\nExtract vendor, date (YYYY-MM-DD), total and category (Travel, Equipment, Meals, Office, Other).", json_schema=SCHEMA, max_output_tokens=100).json
    except ai.AIError:
        tg("sendMessage", chat_id=CHAT_ID, text="Could not read that receipt right now; please log it by hand.")
        return {"logged": False}
    notion("pages", {"parent": {"database_id": DB_ID}, "properties": {"Vendor": {"title": [{"text": {"content": r["vendor"]}}]}, "Date": {"date": {"start": r["date"]}},
                     "Amount": {"number": r["amount"]}, "Category": {"select": {"name": r["category"]}}}})
    tg("sendMessage", chat_id=CHAT_ID, text=f"Logged {r['vendor']}, {r['amount']:.2f} ({r['category']}, {r['date']})")
    return {"logged": True}

@on_trigger
def weekly(event):
    since = str(date.today() - timedelta(days=7))
    pages = notion(f"databases/{DB_ID}/query", {"filter": {"property": "Date", "date": {"on_or_after": since}}})["results"]
    totals = {}
    for p in pages:
        cat = (p["properties"]["Category"]["select"] or {}).get("name", "Other")
        totals[cat] = totals.get(cat, 0) + (p["properties"]["Amount"]["number"] or 0)
    tg("sendMessage", chat_id=CHAT_ID, text="This week: " + ", ".join(f"{k} {v:.2f}" for k, v in totals.items()) if totals else "No expenses this week.")
    return {"expenses": len(pages)}