Back up to GitHub with Telegram alerts

import base64, json
from clickruntime import on_trigger, credentials, http

def gh(method, path, **kw):
    resp = http.request(method, f"https://api.github.com{path}", headers={
        "Authorization": f"Bearer {credentials.connection('github', 'work')}", "Accept": "application/vnd.github+json"}, **kw)
    return resp.raise_for_status() or (resp.json() if resp.content else None)
CHAT_ID = "123456789"

def telegram(text):
    http.post(f"https://api.telegram.org/bot{credentials.secret('TELEGRAM_BOT_TOKEN')}/sendMessage",
              json={"chat_id": CHAT_ID, "text": text, "disable_web_page_preview": True}).raise_for_status()

REPO, EXPORT_URL = "acme/backups", "https://tools.acme.example/api/export"

@on_trigger
def run(event):
    items = http.get(EXPORT_URL, headers={"Authorization": f"Bearer {credentials.secret('EXPORT_TOKEN')}"}).json()
    changed = []
    for item in items:
        path, content = f"exports/{item['id']}.json", json.dumps(item, indent=2, sort_keys=True) + "\n"
        try:
            current = gh("GET", f"/repos/{REPO}/contents/{path}")
        except Exception:
            current = None
        if current and base64.b64decode(current["content"]).decode() == content:
            continue
        gh("PUT", f"/repos/{REPO}/contents/{path}", json={"message": f"Backup {path}", "content": base64.b64encode(content.encode()).decode(),
           **({"sha": current["sha"]} if current else {})})
        changed.append(item.get("name", path))
    telegram(f"Backup done: {len(changed)} changed\n" + "\n".join(f"• {c}" for c in changed) if changed else "Backup done: no changes")
    return {"changed": len(changed)}