A bot that joins your meetings and sends the summary and tasks to Telegram

RECALL_API_KEY=your-recall-key
RECALL_BASE=https://eu-central-1.recall.ai
ANTHROPIC_API_KEY=your-language-model-key
TELEGRAM_TOKEN=token-from-botfather
TELEGRAM_CHAT_ID=your-chat-id
GOOGLE_KEY_FILE=/home/ops/meeting-bot/google-key.json
CALENDAR_ID=you@example.com
MY_NAME=Your name as it appears in meetings
BOT_NAME=Notetaker
BOT_LANG=en

chmod 600 ~/meeting-bot/.env
pip install requests anthropic google-api-python-client google-auth

# meeting_bot.py, part 1 of 4: settings, texts and meeting links
import json
import os
import re
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path

RECALL_BASE = os.environ.get("RECALL_BASE", "https://eu-central-1.recall.ai")
RECALL_KEY = os.environ.get("RECALL_API_KEY", "")
BOT_NAME = os.environ.get("BOT_NAME", "Notetaker")
LANG = os.environ.get("BOT_LANG", "he")
STATE_FILE = Path(os.environ.get("STATE_FILE", "state.json"))
TRANSCRIPTS = Path(os.environ.get("TRANSCRIPTS_DIR", "transcripts"))
SKIP_TAGS = ("#ללא-בוט", "#no-bot")

TEXT = {
    "he": {
        "today": "הפגישות של היום",
        "tagged": "בלי בוט (מסומן)",
        "cancel_hint": "לביטול: בלי בוט <מספר>",
        "none": "אין היום פגישות עם קישור.",
        "cancelled": "בוטל הבוט לפגישה {n}.",
        "failed": "הבוט לא הצליח להקליט את {title}.",
        "summary": "סיכום",
        "decisions": "החלטות",
        "mine": "משימות שלי",
        "theirs": "משימות של אחרים",
        "follow_ups": "הצעות להמשך (שום דבר לא יוצא בלי אישור)",
        "questions": "שאלות פתוחות",
        "no_due": "בלי יעד",
        "approve_hint": "לאישור: אשר <מספר> · לביטול: בטל <מספר>",
        "approved": "אושר {n}. הנוסח הסופי:",
        "dropped": "בוטל {n}.",
    },
    "en": {
        "today": "Today's meetings",
        "tagged": "no bot (tagged)",
        "cancel_hint": "To cancel: no bot <number>",
        "none": "No meetings with a link today.",
        "cancelled": "Bot for meeting {n} cancelled.",
        "failed": "The bot could not record {title}.",
        "summary": "Summary",
        "decisions": "Decisions",
        "mine": "My tasks",
        "theirs": "Other people's tasks",
        "follow_ups": "Proposed follow-ups (nothing goes out without approval)",
        "questions": "Open questions",
        "no_due": "no due date",
        "approve_hint": "To approve: approve <number> · to drop: drop <number>",
        "approved": "Approved {n}. The final text:",
        "dropped": "Dropped {n}.",
    },
}
T = TEXT["en" if LANG == "en" else "he"]

# Zoom, Google Meet and Teams links, wherever the invite put them
MEETING_LINK = re.compile(
    r"https://(?:"
    r"(?:[\w-]+\.)?zoom\.us/(?:j|my|w)/[\w.?=&%/-]+"
    r"|meet\.google\.com/[a-z]{3}-[a-z]{4}-[a-z]{3}"
    r"|teams\.microsoft\.com/l/meetup-join/[^\s\"'<>]+"
    r"|teams\.live\.com/meet/[^\s\"'<>]+"
    r")",
    re.IGNORECASE,
)


def extract_meeting_url(event):
    """The video link of a Google Calendar event, or None."""
    if event.get("hangoutLink"):
        return event["hangoutLink"]
    for ep in (event.get("conferenceData") or {}).get("entryPoints", []):
        if ep.get("entryPointType") == "video" and ep.get("uri"):
            return ep["uri"]
    for field in ("location", "description"):
        m = MEETING_LINK.search(event.get(field) or "")
        if m:
            return m.group(0).rstrip(").,;")
    return None


def should_skip(event):
    text = " ".join(event.get(k) or "" for k in ("summary", "description"))
    return any(tag in text for tag in SKIP_TAGS)


def load_state():
    if STATE_FILE.exists():
        return json.loads(STATE_FILE.read_text(encoding="utf-8"))
    return {"events": {}, "bots": {}, "items": {}, "next_item": 1, "tg_offset": 0}


def save_state(state):
    STATE_FILE.write_text(json.dumps(state, ensure_ascii=False, indent=1), encoding="utf-8")

# meeting_bot.py, part 2 of 4: the calendar and one bot per meeting
import requests


def recall(method, path, **kwargs):
    r = requests.request(method, RECALL_BASE + path, timeout=60,
                         headers={"Authorization": "Token " + RECALL_KEY}, **kwargs)
    r.raise_for_status()
    return r.json() if r.content else {}


def todays_events():
    """Timed events from now until the end of the day (needs a service account key)."""
    from google.oauth2 import service_account
    from googleapiclient.discovery import build

    creds = service_account.Credentials.from_service_account_file(
        os.environ["GOOGLE_KEY_FILE"],
        scopes=["https://www.googleapis.com/auth/calendar.readonly"])
    service = build("calendar", "v3", credentials=creds, cache_discovery=False)
    now = datetime.now(timezone.utc)
    res = service.events().list(
        calendarId=os.environ["CALENDAR_ID"], timeMin=now.isoformat(),
        timeMax=(now + timedelta(hours=18)).isoformat(),
        singleEvents=True, orderBy="startTime").execute()
    return [e for e in res.get("items", []) if "dateTime" in e.get("start", {})]


def schedule_bot(event, url):
    start = datetime.fromisoformat(event["start"]["dateTime"])
    bot = recall("POST", "/api/v1/bot/", json={
        "meeting_url": url,
        "bot_name": BOT_NAME,
        "join_at": (start - timedelta(minutes=1)).astimezone(timezone.utc).isoformat(),
        "recording_config": {"transcript": {"provider": {"recallai_streaming": {
            "mode": "prioritize_accuracy", "language_code": "he"}}}},
    })
    return bot["id"]


def plan_day(send_list):
    """morning: schedule and send the list. sync: schedule only what was added since."""
    state = load_state()
    today = datetime.now().strftime("%Y-%m-%d")
    lines = []
    for ev in todays_events():
        url = extract_meeting_url(ev)
        start = datetime.fromisoformat(ev["start"]["dateTime"])
        if not url or start <= datetime.now(timezone.utc):
            continue
        title = ev.get("summary", "")
        if should_skip(ev):
            lines.append(f"- {start:%H:%M} {title}: {T['tagged']}")
            continue
        if ev["id"] not in state["events"]:
            n = 1 + sum(1 for b in state["bots"].values() if b["day"] == today)
            bot_id = schedule_bot(ev, url)
            state["events"][ev["id"]] = bot_id
            state["bots"][bot_id] = {"n": n, "day": today, "title": title,
                                     "start": ev["start"]["dateTime"], "status": "scheduled"}
        b = state["bots"][state["events"][ev["id"]]]
        lines.append(f"{b['n']}. {start:%H:%M} {title}")
    save_state(state)
    if send_list:
        tg("\n".join([T["today"], ""] + lines + ["", T["cancel_hint"]]) if lines else T["none"])

# meeting_bot.py, part 3 of 4: after the meeting, transcript, summary, delete
def bot_status(bot):
    changes = bot.get("status_changes") or []
    return changes[-1]["code"] if changes else "unknown"


def fetch_transcript(bot):
    for rec in bot.get("recordings") or []:
        tr = (rec.get("media_shortcuts") or {}).get("transcript") or {}
        if (tr.get("status") or {}).get("code") == "done":
            return requests.get(tr["data"]["download_url"], timeout=120).json()
    return None


def transcript_to_text(transcript):
    """One line per speaker turn: [mm:ss] Name: text"""
    lines = []
    for part in transcript:
        words = part.get("words") or []
        if not words:
            continue
        name = (part.get("participant") or {}).get("name") or "?"
        sec = int((words[0].get("start_timestamp") or {}).get("relative") or 0)
        lines.append(f"[{sec // 60:02d}:{sec % 60:02d}] {name}: " + " ".join(w["text"] for w in words))
    return "\n".join(lines)


def parse_json(text):
    s = text.strip()
    a, b = s.find("{"), s.rfind("}")
    if a < 0 or b <= a:
        raise ValueError("no JSON object in the answer")
    return json.loads(s[a:b + 1])


def _date_or_empty(v):
    v = str(v or "").strip()
    try:
        datetime.strptime(v, "%Y-%m-%d")
        return v
    except ValueError:
        return ""


def _strings(v):
    return [str(x).strip() for x in v if str(x).strip()] if isinstance(v, list) else []


def validate_summary(data):
    """Keeps only what the format allows. No date that is not a real yyyy-mm-dd."""
    if not isinstance(data, dict):
        raise ValueError("the answer is not a JSON object")
    out = {"summary": _strings(data.get("summary")), "decisions": _strings(data.get("decisions")),
           "open_questions": _strings(data.get("open_questions")),
           "tasks": {"mine": [], "theirs": []}, "follow_ups": []}
    tasks = data.get("tasks") if isinstance(data.get("tasks"), dict) else {}
    for side in ("mine", "theirs"):
        for t in tasks.get(side) or []:
            if isinstance(t, dict) and str(t.get("task") or "").strip():
                out["tasks"][side].append({
                    "task": str(t["task"]).strip(), "owner": str(t.get("owner") or "").strip(),
                    "due": _date_or_empty(t.get("due")), "quote": str(t.get("quote") or "").strip()})
    for f in data.get("follow_ups") or []:
        if isinstance(f, dict) and f.get("kind") in ("email", "message", "meeting") and str(f.get("draft") or "").strip():
            out["follow_ups"].append({k: str(f.get(k) or "").strip() for k in ("kind", "to", "subject", "draft", "why")})
    if not out["summary"]:
        raise ValueError("empty summary")
    return out


def summarize(transcript_text, title, date):
    import anthropic

    prompt = Path("summary_prompt.txt").read_text(encoding="utf-8")
    for key, value in (("{{TITLE}}", title), ("{{DATE}}", date),
                       ("{{ME}}", os.environ.get("MY_NAME", "")), ("{{TRANSCRIPT}}", transcript_text)):
        prompt = prompt.replace(key, value)
    client = anthropic.Anthropic()  # reads ANTHROPIC_API_KEY
    msg = client.beta.messages.create(
        model=os.environ.get("LLM_MODEL", "claude-opus-5-5"), max_tokens=16000,
        betas=["server-side-fallback-2026-07-01"], fallbacks="default",
        messages=[{"role": "user", "content": prompt}])
    if msg.stop_reason == "refusal":
        raise RuntimeError("the model declined this transcript")
    return validate_summary(parse_json("".join(b.text for b in msg.content if b.type == "text")))


def finish(state, title, date, text):
    TRANSCRIPTS.mkdir(exist_ok=True)
    (TRANSCRIPTS / f"{date} {title[:40]}.txt".replace("/", "-")).write_text(text, encoding="utf-8")
    send_summary(state, title, summarize(text, title, date))


def collect(state):
    for bot_id, b in state["bots"].items():
        if b["status"] != "scheduled":
            continue
        bot = recall("GET", f"/api/v1/bot/{bot_id}/")
        if bot_status(bot) == "fatal":
            b["status"] = "failed"
            tg(T["failed"].format(title=b["title"]))
            continue
        if bot_status(bot) != "done":
            continue
        transcript = fetch_transcript(bot)
        if transcript is None:
            continue  # still processing, try on the next run
        finish(state, b["title"], b["start"][:10], transcript_to_text(transcript))
        recall("POST", f"/api/v1/bot/{bot_id}/delete_media/")
        b["status"] = "done"
        save_state(state)

# meeting_bot.py, part 4 of 4: Telegram, approvals and the commands
TG = "https://api.telegram.org/bot" + os.environ.get("TELEGRAM_TOKEN", "")
OWNER = str(os.environ.get("TELEGRAM_CHAT_ID", ""))
COMMAND = re.compile(r"^\s*(בלי בוט|no bot|אשר|approve|בטל|drop)\s+(\d+)\s*$", re.IGNORECASE)


def tg(text):
    for i in range(0, len(text), 4000):  # Telegram's limit is 4096 characters
        requests.post(TG + "/sendMessage", json={"chat_id": OWNER, "text": text[i:i + 4000]},
                      timeout=30).raise_for_status()


def format_summary(title, data, first_item):
    out = [title, "", T["summary"]] + [f"- {s}" for s in data["summary"]]
    if data["decisions"]:
        out += ["", T["decisions"]] + [f"- {s}" for s in data["decisions"]]
    for side in ("mine", "theirs"):
        if data["tasks"][side]:
            out += ["", T[side]]
            for t in data["tasks"][side]:
                who = f"{t['owner']}: " if side == "theirs" else ""
                out.append(f"- {who}{t['task']} ({t['due'] or T['no_due']})")
    if data["follow_ups"]:
        out += ["", T["follow_ups"]]
        for i, f in enumerate(data["follow_ups"]):
            out.append(f"{first_item + i}. {f['kind']} → {f['to']}: {f['subject']} ({f['why']})")
        out += ["", T["approve_hint"]]
    if data["open_questions"]:
        out += ["", T["questions"]] + [f"- {s}" for s in data["open_questions"]]
    return "\n".join(out)


def send_summary(state, title, data):
    first = state["next_item"]
    for f in data["follow_ups"]:
        state["items"][str(state["next_item"])] = {**f, "status": "waiting"}
        state["next_item"] += 1
    tg(format_summary(title, data, first))


def run_follow_up(item):
    """Runs only after "approve N". Connect your mail or calendar here.
    As written, it sends you the final text so you can paste it and send it yourself."""
    tg(f"{item['kind']} → {item['to']}\n{item['subject']}\n\n{item['draft']}")


def parse_command(text):
    m = COMMAND.match(text or "")
    if not m:
        return None
    word = m.group(1).lower()
    kind = {"בלי בוט": "cancel", "no bot": "cancel", "אשר": "approve", "approve": "approve"}.get(word, "drop")
    return kind, int(m.group(2))


def handle_telegram(state):
    res = requests.get(TG + "/getUpdates", params={"offset": state["tg_offset"], "timeout": 0}, timeout=30).json()
    today = datetime.now().strftime("%Y-%m-%d")
    for upd in res.get("result", []):
        state["tg_offset"] = upd["update_id"] + 1
        msg = upd.get("message") or {}
        if str((msg.get("chat") or {}).get("id")) != OWNER:
            continue  # only you can give commands
        cmd = parse_command(msg.get("text"))
        if not cmd:
            continue
        kind, n = cmd
        if kind == "cancel":
            for bot_id, b in state["bots"].items():
                if b["day"] == today and b["n"] == n and b["status"] == "scheduled":
                    recall("DELETE", f"/api/v1/bot/{bot_id}/")
                    b["status"] = "cancelled"
                    tg(T["cancelled"].format(n=n))
            continue
        item = state["items"].get(str(n))
        if not item or item["status"] != "waiting":
            continue
        item["status"] = "approved" if kind == "approve" else "dropped"
        if kind == "approve":
            tg(T["approved"].format(n=n))
            run_follow_up(item)
        else:
            tg(T["dropped"].format(n=n))
    save_state(state)


def main(cmd, *args):
    if cmd in ("morning", "sync"):
        plan_day(send_list=(cmd == "morning"))
    elif cmd == "poll":
        state = load_state()
        handle_telegram(state)
        collect(state)
    elif cmd == "file":  # a phone or in-person recording, transcribed on your computer
        path, title = Path(args[0]), args[1]
        state = load_state()
        finish(state, title, datetime.now().strftime("%Y-%m-%d"), path.read_text(encoding="utf-8"))
        save_state(state)


if __name__ == "__main__":
    main(*sys.argv[1:])

30 7 * * 0-5  cd ~/meeting-bot && set -a && . ./.env && python3 meeting_bot.py morning
5 8-20 * * 0-5  cd ~/meeting-bot && set -a && . ./.env && python3 meeting_bot.py sync
*/5 7-22 * * 0-5  cd ~/meeting-bot && set -a && . ./.env && python3 meeting_bot.py poll

python3 meeting_bot.py file "transcripts/2026-11-12 call.txt" "Call with the print supplier"