#!/usr/bin/env python3 """ Abigael Meeting Prep — No More Chores ====================================== Scans recent Slack activity and Google Calendar to generate a structured agenda for Mike's weekly meetings with Abigael (scheduler/VA). USAGE: python3 scripts/meeting-prep.py # Generate prep for next meeting python3 scripts/meeting-prep.py --date 2026-03-27 # Override meeting date python3 scripts/meeting-prep.py --dry-run # Print to stdout only (no Telegram, no save) python3 scripts/meeting-prep.py --weeks 6 # Look back 6 weeks instead of 4 SECRETS: ~/.openclaw/secrets/slack-token.txt ~/.openclaw/secrets/google-calendar-sa.json ~/.openclaw/secrets/telegram-bot-token.txt CHANNELS SCANNED: emilys-team (C02NFPKNZ1U) customer-feedback (C0AJQTVST1B) calls (CDP3Y6YB0) """ import base64 import json import os import re import subprocess import sys import tempfile import time import urllib.error import urllib.parse import urllib.request from datetime import datetime, timedelta, timezone from typing import Optional # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) WORKSPACE_DIR = os.path.dirname(SCRIPT_DIR) MEMORY_DIR = os.path.join(WORKSPACE_DIR, "memory") SECRETS_DIR = os.path.expanduser("~/.openclaw/secrets") SLACK_TOKEN_FILE = os.path.join(SECRETS_DIR, "slack-token.txt") GCAL_SA_FILE = os.path.join(SECRETS_DIR, "google-calendar-sa.json") TELEGRAM_TOKEN_FILE = os.path.join(SECRETS_DIR, "telegram-bot-token.txt") TELEGRAM_CHAT_ID = 8792051045 PREVIEW_BASE_URL = "https://macbookair.taila88920.ts.net/memory" CALENDAR_ID = "mike@nomorechores.com" GCAL_SCOPE = "https://www.googleapis.com/auth/calendar.readonly" CHANNELS = { "emilys-team": "C02NFPKNZ1U", "customer-feedback": "C0AJQTVST1B", "calls": "CDP3Y6YB0", } DEFAULT_LOOKBACK_WEEKS = 4 COMPLAINT_KEYWORDS = [ "complaint", "mistake", "wrong", "missed", "late", "cancel", "error", "issue", "unhappy", "re-clean", "reclean", "rebook", "escalat", "refund", "upset", "angry", "frustrated", "problem", "no-show", "noshow", ] SCHEDULING_KEYWORDS = [ "wrong time", "missed arrival", "contractor late", "contractor missing", "didn't show", "did not show", "no show", "no-show", "late arrival", "arrival time", "scheduling", "reschedule", "time change", "wrong date", "wrong address", "booking error", "double book", "overlap", ] ABIGAEL_KEYWORDS = [ "abigael", "abby", "scheduler", "va ", "virtual assistant", ] SNIPPET_LEN = 120 # --------------------------------------------------------------------------- # Logging # --------------------------------------------------------------------------- def log(msg: str) -> None: ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S") print(f"[{ts}] {msg}", file=sys.stderr) # --------------------------------------------------------------------------- # Secrets # --------------------------------------------------------------------------- def read_secret(path: str) -> str: try: with open(path) as f: return f.read().strip() except FileNotFoundError: log(f"ERROR: Secret file not found: {path}") sys.exit(1) # --------------------------------------------------------------------------- # HTTP helpers # --------------------------------------------------------------------------- _USER_AGENT = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) HarveyBot/1.0" def _do_request(req: urllib.request.Request, retry_on_5xx: bool = True) -> Optional[dict]: if "User-Agent" not in req.headers: req.add_header("User-Agent", _USER_AGENT) try: with urllib.request.urlopen(req, timeout=15) as resp: return json.loads(resp.read().decode()) except urllib.error.HTTPError as e: body = e.read().decode(errors="replace") if retry_on_5xx and e.code >= 500: log(f"WARN: HTTP {e.code} — retrying in 2s…") time.sleep(2) return _do_request(req, retry_on_5xx=False) log(f"ERROR: HTTP {e.code} for {req.full_url}: {body[:200]}") return None except urllib.error.URLError as e: log(f"ERROR: URL error for {req.full_url}: {e.reason}") return None except Exception as e: log(f"ERROR: Unexpected error: {e}") return None def slack_get(token: str, endpoint: str, params: dict) -> Optional[dict]: qs = urllib.parse.urlencode(params) url = f"https://slack.com/api/{endpoint}?{qs}" req = urllib.request.Request(url, headers={"Authorization": f"Bearer {token}"}) return _do_request(req) # --------------------------------------------------------------------------- # Google Calendar (SA JWT — same pattern as check-calendar.py) # --------------------------------------------------------------------------- def get_gcal_token() -> Optional[str]: try: with open(GCAL_SA_FILE) as f: sa = json.load(f) except FileNotFoundError: log(f"WARN: Google Calendar SA file not found: {GCAL_SA_FILE}") return None except json.JSONDecodeError as e: log(f"WARN: Could not parse SA file: {e}") return None header = base64.urlsafe_b64encode( json.dumps({"alg": "RS256", "typ": "JWT"}).encode() ).rstrip(b"=").decode() now = int(time.time()) claims = { "iss": sa["client_email"], "sub": CALENDAR_ID, "scope": GCAL_SCOPE, "aud": "https://oauth2.googleapis.com/token", "iat": now, "exp": now + 3600, } payload = base64.urlsafe_b64encode( json.dumps(claims).encode() ).rstrip(b"=").decode() msg = f"{header}.{payload}".encode() try: with tempfile.NamedTemporaryFile(mode="w", suffix=".pem", delete=False) as kf: kf.write(sa["private_key"]) kf_path = kf.name result = subprocess.run( ["openssl", "dgst", "-sha256", "-sign", kf_path], input=msg, capture_output=True, ) sig = base64.urlsafe_b64encode(result.stdout).rstrip(b"=").decode() except Exception as e: log(f"WARN: JWT signing failed: {e}") return None finally: try: os.unlink(kf_path) except Exception: pass data = urllib.parse.urlencode({ "grant_type": "urn:ietf:params:oauth:grant-type:jwt-bearer", "assertion": f"{header}.{payload}.{sig}", }).encode() try: resp = json.loads( urllib.request.urlopen( urllib.request.Request("https://oauth2.googleapis.com/token", data=data), timeout=15, ).read() ) return resp.get("access_token") except Exception as e: log(f"WARN: Could not get GCal access token: {e}") return None def get_abigael_events(days_ahead: int = 14) -> list[dict]: """Return upcoming calendar events containing 'abigael' in summary/description.""" token = get_gcal_token() if not token: log("WARN: Skipping calendar check (no token)") return [] now = datetime.now(timezone.utc) time_min = now.strftime("%Y-%m-%dT%H:%M:%SZ") time_max = (now + timedelta(days=days_ahead)).strftime("%Y-%m-%dT%H:%M:%SZ") params = urllib.parse.urlencode({ "timeMin": time_min, "timeMax": time_max, "singleEvents": "true", "orderBy": "startTime", "maxResults": 50, "q": "abigael", }) url = ( f"https://www.googleapis.com/calendar/v3/calendars/" f"{urllib.parse.quote(CALENDAR_ID)}/events?{params}" ) req = urllib.request.Request(url, headers={"Authorization": f"Bearer {token}"}) try: data = json.loads(urllib.request.urlopen(req, timeout=15).read()) except Exception as e: log(f"WARN: Calendar API error: {e}") return [] events = [] for e in data.get("items", []): summary = e.get("summary", "") description = e.get("description", "") or "" combined = (summary + " " + description).lower() if "abigael" not in combined: continue start = e.get("start", {}).get("dateTime") or e.get("start", {}).get("date", "") events.append({ "summary": summary, "start": start, "description": description[:200], "location": e.get("location", ""), }) log(f"Calendar: found {len(events)} Abigael event(s) in next {days_ahead} days") return events # --------------------------------------------------------------------------- # Slack scanning # --------------------------------------------------------------------------- def _make_link(channel_id: str, ts: str) -> str: ts_nodot = ts.replace(".", "") return f"https://slack.com/archives/{channel_id}/p{ts_nodot}" def _matches_abigael(text: str) -> bool: lower = text.lower() return any(kw in lower for kw in ABIGAEL_KEYWORDS) def _classify_message(text: str) -> Optional[str]: """Return 'complaint', 'scheduling', or None.""" lower = text.lower() if any(kw in lower for kw in SCHEDULING_KEYWORDS): return "scheduling" if any(kw in lower for kw in COMPLAINT_KEYWORDS): return "complaint" return None def scan_slack(token: str, lookback_weeks: int) -> dict: """ Returns: { "complaints": [{"channel", "channel_id", "date", "snippet", "link", "text"}], "scheduling": [...], "other": [...], } """ oldest = str((datetime.now(timezone.utc) - timedelta(weeks=lookback_weeks)).timestamp()) results = {"complaints": [], "scheduling": [], "other": []} seen = set() for channel_name, channel_id in CHANNELS.items(): log(f"Scanning #{channel_name} ({channel_id}) back {lookback_weeks} weeks…") cursor = None while True: params: dict = { "channel": channel_id, "oldest": oldest, "limit": 200, } if cursor: params["cursor"] = cursor resp = slack_get(token, "conversations.history", params) if not resp or not resp.get("ok"): log(f" WARN: Could not fetch #{channel_name}: {resp}") break messages = resp.get("messages", []) log(f" {len(messages)} messages fetched") for msg in messages: text = msg.get("text", "") ts = msg.get("ts", "") # Skip bot messages if msg.get("bot_id") or msg.get("subtype") == "bot_message": continue dedup_key = f"{channel_id}:{ts}" if dedup_key in seen: continue seen.add(dedup_key) lower = text.lower() mentions_abigael = _matches_abigael(text) category = _classify_message(text) if not mentions_abigael and not category: continue dt = datetime.fromtimestamp(float(ts), tz=timezone.utc) date_str = dt.strftime("%Y-%m-%d") snippet = text[:SNIPPET_LEN].replace("\n", " ") link = _make_link(channel_id, ts) item = { "channel": channel_name, "channel_id": channel_id, "date": date_str, "ts": float(ts), "snippet": snippet, "link": link, "text": text, } if category == "scheduling": results["scheduling"].append(item) elif category == "complaint": results["complaints"].append(item) else: # Mentions Abigael but no specific category results["other"].append(item) # Pagination next_cursor = resp.get("response_metadata", {}).get("next_cursor", "") if not next_cursor: break cursor = next_cursor # Sort each list newest-first for key in results: results[key].sort(key=lambda x: x["ts"], reverse=True) log( f"Slack scan complete: {len(results['complaints'])} complaints, " f"{len(results['scheduling'])} scheduling issues, " f"{len(results['other'])} other items" ) return results # --------------------------------------------------------------------------- # Previous meeting open items # --------------------------------------------------------------------------- def get_open_items_from_previous_meetings() -> list[str]: """Scan memory/meeting-*.md files for ## headings to surface as open items.""" if not os.path.isdir(MEMORY_DIR): return [] meeting_files = sorted( [f for f in os.listdir(MEMORY_DIR) if f.startswith("meeting-") and f.endswith(".md")], reverse=True, ) items = [] for fname in meeting_files[:3]: # Last 3 meeting files fpath = os.path.join(MEMORY_DIR, fname) try: with open(fpath) as f: for line in f: line = line.rstrip() # Grab ## headings (but skip the standard structural ones) if line.startswith("## ") and not any( skip in line for skip in [ "Meeting:", "Open Items", "Customer Issues", "Scheduling Issues", "Other Notable", "Discussion Points", ] ): items.append(f"- {line[3:]} _(from {fname})_") except OSError: pass return items # --------------------------------------------------------------------------- # Agenda generation # --------------------------------------------------------------------------- def _format_item(item: dict) -> str: return ( f"- **{item['date']}** | #{item['channel']} | " f"[view]({item['link']})\n" f" > {item['snippet']}" ) def generate_agenda( meeting_date: str, slack_results: dict, calendar_events: list[dict], open_items: list[str], ) -> str: dt = datetime.strptime(meeting_date, "%Y-%m-%d") weekday = dt.strftime("%A") lines = [f"## Meeting: {weekday}, {meeting_date}", ""] # --- Open Items --- lines.append("## Open Items from Last Meeting") lines.append("") if open_items: lines.extend(open_items) else: lines.append("Nothing flagged this week.") lines.append("") # --- Upcoming Calendar Events --- lines.append("## Upcoming Abigael Calendar Events") lines.append("") if calendar_events: for e in calendar_events: start = e["start"] summary = e["summary"] loc = f" @ {e['location']}" if e["location"] else "" lines.append(f"- **{start}** — {summary}{loc}") else: lines.append("No upcoming Abigael events found in the next 14 days.") lines.append("") # --- Customer Issues & Complaints --- lines.append("## Customer Issues & Complaints") lines.append("") if slack_results["complaints"]: for item in slack_results["complaints"]: lines.append(_format_item(item)) lines.append("") else: lines.append("Nothing flagged this week.") lines.append("") # --- Scheduling Issues --- lines.append("## Scheduling Issues") lines.append("") if slack_results["scheduling"]: for item in slack_results["scheduling"]: lines.append(_format_item(item)) lines.append("") else: lines.append("Nothing flagged this week.") lines.append("") # --- Other Notable Activity --- lines.append("## Other Notable Activity") lines.append("") if slack_results["other"]: for item in slack_results["other"]: lines.append(_format_item(item)) lines.append("") else: lines.append("Nothing flagged this week.") lines.append("") # --- Discussion Points --- lines.append("## Discussion Points") lines.append("") lines.append("_(Mike fills in)_") lines.append("") return "\n".join(lines) # --------------------------------------------------------------------------- # Telegram # --------------------------------------------------------------------------- def send_telegram(token: str, chat_id: int, text: str) -> bool: data = json.dumps({ "chat_id": chat_id, "text": text, "parse_mode": "Markdown", }).encode() req = urllib.request.Request( f"https://api.telegram.org/bot{token}/sendMessage", data=data, headers={"Content-Type": "application/json"}, ) try: with urllib.request.urlopen(req, timeout=15) as resp: result = json.loads(resp.read().decode()) return result.get("ok", False) except Exception as e: log(f"WARN: Telegram send failed: {e}") return False def build_telegram_message( meeting_date: str, slack_results: dict, calendar_events: list[dict], agenda_filename: str, ) -> str: dt = datetime.strptime(meeting_date, "%Y-%m-%d") weekday = dt.strftime("%A") n_complaints = len(slack_results["complaints"]) n_scheduling = len(slack_results["scheduling"]) n_other = len(slack_results["other"]) n_cal = len(calendar_events) preview_url = f"{PREVIEW_BASE_URL}/{agenda_filename}" lines = [ f"*Meeting prep for Abigael ({weekday} {meeting_date})*", "", f"• {n_complaints} complaint(s) flagged", f"• {n_scheduling} scheduling issue(s) flagged", f"• {n_other} other Abigael mention(s)", f"• {n_cal} upcoming calendar event(s)", "", f"Preview: {preview_url}", f"Full agenda saved to `memory/{agenda_filename}`", ] return "\n".join(lines) # --------------------------------------------------------------------------- # Next meeting date detection # --------------------------------------------------------------------------- def next_thursday() -> str: """Return the date string for the next upcoming Thursday.""" today = datetime.now(timezone.utc).date() days_ahead = (3 - today.weekday()) % 7 # Thursday = 3 if days_ahead == 0: days_ahead = 7 return (today + timedelta(days=days_ahead)).strftime("%Y-%m-%d") # --------------------------------------------------------------------------- # Entrypoint # --------------------------------------------------------------------------- def main() -> None: args = sys.argv[1:] dry_run = "--dry-run" in args lookback_weeks = DEFAULT_LOOKBACK_WEEKS # --weeks N meeting_date = None i = 0 while i < len(args): if args[i] == "--date" and i + 1 < len(args): meeting_date = args[i + 1] i += 2 elif args[i] == "--weeks" and i + 1 < len(args): try: lookback_weeks = int(args[i + 1]) except ValueError: log(f"WARN: Invalid --weeks value '{args[i+1]}', using default {DEFAULT_LOOKBACK_WEEKS}") i += 2 else: i += 1 if not meeting_date: meeting_date = next_thursday() log(f"No --date provided, defaulting to next Thursday: {meeting_date}") log(f"=== Meeting Prep: {meeting_date} (lookback={lookback_weeks}w, dry_run={dry_run}) ===") # Load secrets slack_token = read_secret(SLACK_TOKEN_FILE) telegram_token = None if dry_run else read_secret(TELEGRAM_TOKEN_FILE) # Scan Slack slack_results = scan_slack(slack_token, lookback_weeks) # Check calendar calendar_events = get_abigael_events(days_ahead=14) # Previous meeting open items open_items = get_open_items_from_previous_meetings() log(f"Found {len(open_items)} open item(s) from previous meeting files") # Generate agenda agenda = generate_agenda(meeting_date, slack_results, calendar_events, open_items) agenda_filename = f"meeting-{meeting_date}.md" agenda_path = os.path.join(MEMORY_DIR, agenda_filename) # Print to stdout print(agenda) if dry_run: log("Dry run — skipping file save and Telegram.") return # Save to memory/ os.makedirs(MEMORY_DIR, exist_ok=True) try: with open(agenda_path, "w") as f: f.write(agenda) log(f"Agenda saved to {agenda_path}") except OSError as e: log(f"ERROR: Could not save agenda: {e}") # Send Telegram tg_msg = build_telegram_message( meeting_date, slack_results, calendar_events, agenda_filename ) ok = send_telegram(telegram_token, TELEGRAM_CHAT_ID, tg_msg) if ok: log("Telegram message sent successfully") else: log("WARN: Telegram message may not have been sent") if __name__ == "__main__": main()