#!/usr/bin/env python3 import json import os import re import subprocess from collections import defaultdict from datetime import datetime from pathlib import Path import requests WORKSPACE = Path('/Users/ahmad/.openclaw/workspace') KB = WORKSPACE / 'knowledge_base' OUT_DAILY = KB / 'email_triage_daily.md' OUT_HEATMAP = KB / 'email_sender_heatmap_weekly.md' SENDER_MEMORY = WORKSPACE / 'memory' / 'email_sender_intelligence.json' ACCOUNTS = os.getenv('EMAIL_OPS_ACCOUNTS', 'ahmad@bigalc.com').split(',') ACCOUNTS = [a.strip() for a in ACCOUNTS if a.strip()] CLICKUP_CRED = WORKSPACE / 'clickup_credentials.md' FOLLOWUP_LIST_ID = '901613526439' # Meetings & Follow-ups BASE = 'https://api.clickup.com/api/v2' def run(cmd): p = subprocess.run(cmd, capture_output=True, text=True) if p.returncode != 0: raise RuntimeError(p.stderr.strip() or 'command failed') return p.stdout def extract_email(from_field: str) -> str: m = re.search(r'<([^>]+)>', from_field or '') if m: return m.group(1).lower().strip() # fallback m2 = re.search(r'[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}', from_field or '') return (m2.group(0).lower().strip() if m2 else (from_field or '').lower().strip()) def load_clickup_token(): txt = CLICKUP_CRED.read_text(encoding='utf-8') if CLICKUP_CRED.exists() else '' m = re.search(r'\bpk_[A-Za-z0-9_]+\b', txt) return m.group(0) if m else None def clickup_get_tasks(token): r = requests.get(f'{BASE}/list/{FOLLOWUP_LIST_ID}/task', headers={'Authorization': token}, params={'include_closed': 'true'}, timeout=30) r.raise_for_status() return r.json().get('tasks', []) def clickup_create_task(token, name, description): payload = {'name': name, 'description': description, 'tags': ['email-ops', 'follow-up']} r = requests.post(f'{BASE}/list/{FOLLOWUP_LIST_ID}/task', headers={'Authorization': token, 'Content-Type': 'application/json'}, json=payload, timeout=30) r.raise_for_status() return r.json().get('id') def classify(subject: str, sender: str) -> str: s = (subject or '').lower() f = (sender or '').lower() urgent_kw = ['urgent', 'asap', 'invoice', 'payment', 'contract', 'proposal', 'meeting', 'follow up', 'follow-up', 'action required', 'approval'] action_kw = ['re:', 'fwd:', 'review', 'question', 'confirm', 'schedule', 'invite', 'deadline'] ignore_kw = ['newsletter', 'discount', 'sale', 'offer', 'masterclass', 'promo', 'invitation', 'deal of the', 'unsubscribe'] if any(k in s for k in urgent_kw): return 'P1 urgent' if any(k in s for k in action_kw): return 'P2 needs response' if any(k in s for k in ignore_kw) or any(x in f for x in ['noreply', 'no-reply', 'mailer-daemon', 'rewards@', 'news@', 'updates@']): return 'P4 archive/newsletter' return 'P3 FYI' def fetch_threads(): # combine unread + recent across configured accounts then dedupe threads = [] for account in ACCOUNTS: try: q1 = json.loads(run(['gog', 'gmail', 'search', '--account', account, 'is:unread newer_than:2d', '--max', '80', '--json'])) q2 = json.loads(run(['gog', 'gmail', 'search', '--account', account, 'newer_than:1d', '--max', '80', '--json'])) batch = (q1.get('threads', []) if isinstance(q1, dict) else []) + (q2.get('threads', []) if isinstance(q2, dict) else []) for t in batch: t['_account'] = account threads.extend(batch) except Exception: continue seen = set() out = [] for t in threads: tid = t.get('id') key = f"{t.get('_account','')}-{tid}" if not tid or key in seen: continue seen.add(key) out.append(t) return out def thread_summary(thread_id: str, account: str): # header-level summary to avoid heavy body parsing try: raw = json.loads(run(['gog', 'gmail', 'thread', 'get', '--account', account, thread_id, '--json'])) thread = raw.get('thread', {}) msgs = thread.get('messages', []) subjects = [] participants = set() for m in msgs: headers = {h.get('name', '').lower(): h.get('value', '') for h in (m.get('payload', {}).get('headers', []) or [])} sub = headers.get('subject', '').strip() frm = headers.get('from', '').strip() to = headers.get('to', '').strip() if sub: subjects.append(sub) if frm: participants.add(extract_email(frm)) if to: for p in re.findall(r'[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}', to): participants.add(p.lower()) latest_sub = subjects[-1] if subjects else '(no subject)' return { 'message_count': len(msgs), 'latest_subject': latest_sub, 'participants': sorted(list(participants))[:8] } except Exception: return {'message_count': 0, 'latest_subject': '(unavailable)', 'participants': []} def draft_reply(item): sender = item['from'] subj = item['subject'] if item['class'] == 'P1 urgent': return f"Subject: Re: {subj}\n\nHi,\n\nThanks for your message. I reviewed this and I’m prioritizing it now. I’ll send you a concrete update shortly today.\n\nBest,\nAhmad" return f"Subject: Re: {subj}\n\nHi,\n\nThanks for reaching out. Got it — I’ll review and get back to you with the next step soon.\n\nBest,\nAhmad" def update_sender_memory(items): SENDER_MEMORY.parent.mkdir(parents=True, exist_ok=True) mem = {} if SENDER_MEMORY.exists(): try: mem = json.loads(SENDER_MEMORY.read_text(encoding='utf-8')) except Exception: mem = {} for it in items: em = extract_email(it['from']) rec = mem.get(em, {'count': 0, 'p1': 0, 'p2': 0, 'p3': 0, 'p4': 0, 'last_subject': '', 'last_date': ''}) rec['count'] += 1 if it['class'].startswith('P1'): rec['p1'] += 1 elif it['class'].startswith('P2'): rec['p2'] += 1 elif it['class'].startswith('P3'): rec['p3'] += 1 else: rec['p4'] += 1 rec['last_subject'] = it['subject'] rec['last_date'] = it['date'] # weighted importance rec['score'] = rec['p1'] * 8 + rec['p2'] * 5 + rec['p3'] * 2 + rec['count'] mem[em] = rec SENDER_MEMORY.write_text(json.dumps(mem, indent=2), encoding='utf-8') return mem def write_heatmap(mem): ranked = sorted(mem.items(), key=lambda kv: kv[1].get('score', 0), reverse=True) lines = [ '# Email Sender Intelligence Heatmap (Weekly)', f'- Generated: {datetime.now().isoformat(timespec="seconds")}', '', '## Top Senders by Importance Score', ] if not ranked: lines.append('- No data yet.') else: for em, r in ranked[:25]: lines.append(f"- {em} | score {r.get('score',0)} | P1:{r.get('p1',0)} P2:{r.get('p2',0)} P3:{r.get('p3',0)} | last: {r.get('last_subject','')[:80]}") OUT_HEATMAP.write_text('\n'.join(lines), encoding='utf-8') def calendar_linkage(items): # lightweight v3 linkage: compare sender names/domains to upcoming calendar summaries try: cal = json.loads(run(['gog', 'calendar', 'list', '--json'])) events = cal.get('events', [])[:60] except Exception: return [] links = [] for it in items: subj = (it['subject'] or '').lower() sender_email = extract_email(it['from']) sender_domain = sender_email.split('@')[-1] if '@' in sender_email else '' for ev in events: es = (ev.get('summary') or '').lower() if sender_domain and sender_domain.split('.')[0] in es: links.append((it['subject'], ev.get('summary', ''), ev.get('start', {}).get('dateTime', ev.get('start', {}).get('date', '')))) break if subj and any(tok in es for tok in [w for w in re.findall(r'[a-z]{4,}', subj)[:3]]): links.append((it['subject'], ev.get('summary', ''), ev.get('start', {}).get('dateTime', ev.get('start', {}).get('date', '')))) break return links[:15] def main(): KB.mkdir(parents=True, exist_ok=True) threads = fetch_threads() items = [] for t in threads[:120]: item = { 'id': t.get('id', ''), 'account': t.get('_account', ''), 'date': t.get('date', ''), 'from': t.get('from', ''), 'subject': t.get('subject', ''), 'class': classify(t.get('subject', ''), t.get('from', '')), 'messageCount': t.get('messageCount', 0), } item['thread_summary'] = thread_summary(item['id'], item['account'] or ACCOUNTS[0]) items.append(item) p1 = [x for x in items if x['class'] == 'P1 urgent'] p2 = [x for x in items if x['class'] == 'P2 needs response'] p3 = [x for x in items if x['class'] == 'P3 FYI'] p4 = [x for x in items if x['class'] == 'P4 archive/newsletter'] # v2 reply drafts for P1/P2 drafts = [{'id': x['id'], 'subject': x['subject'], 'draft': draft_reply(x)} for x in (p1 + p2)[:12]] # follow-up extraction + ClickUp create token = load_clickup_token() created = [] if token: existing = {t.get('name', '').strip().lower() for t in clickup_get_tasks(token)} for x in (p1 + p2)[:25]: name = f"Email Follow-up: {x['subject'][:120]}" if name.lower() in existing: continue who = x['from'] when = 'Today' what = x['subject'] desc = f"Who: {who}\nWhat: {what}\nWhen: {when}\nThread ID: {x['id']}\nPriority: {x['class']}\n\nThread summary:\n- messages: {x['thread_summary']['message_count']}\n- latest: {x['thread_summary']['latest_subject']}\n- participants: {', '.join(x['thread_summary']['participants'])}\n\nAction: review + respond + close loop." try: tid = clickup_create_task(token, name, desc) created.append((name, tid)) except Exception: pass mem = update_sender_memory(items) write_heatmap(mem) links = calendar_linkage(items) lines = [] lines.append('# Email Ops v1/v2/v3 Daily Digest') lines.append(f'- Generated: {datetime.now().isoformat(timespec="seconds")}') lines.append('') lines.append('## Summary') lines.append(f'- Total threads scanned: {len(items)}') lines.append(f'- P1 urgent: {len(p1)}') lines.append(f'- P2 needs response: {len(p2)}') lines.append(f'- P3 FYI: {len(p3)}') lines.append(f'- P4 archive/newsletter: {len(p4)}') lines.append(f'- Reply drafts generated (P1/P2): {len(drafts)}') lines.append(f'- ClickUp follow-up tasks created: {len(created)}') lines.append('') lines.append('## Key Findings') for title, arr in [('P1 urgent', p1), ('P2 needs response', p2), ('P3 FYI', p3)]: lines.append(f'### {title}') if not arr: lines.append('- None') else: for x in arr[:8]: ts = x['thread_summary'] lines.append(f"- [{x['date']}] {x['subject']} — {x['from']} (thread {x['id']})") lines.append(f" - Thread summary: {ts['message_count']} msgs | latest: {ts['latest_subject']}") lines.append('') lines.append('## Actions') if created: for n, tid in created[:20]: lines.append(f'- Created ClickUp follow-up: {n} -> https://app.clickup.com/t/{tid}') else: lines.append('- No new follow-up tasks created (none needed or already existed).') lines.append('') lines.append('### Reply Drafts (P1/P2)') if drafts: for d in drafts[:10]: lines.append(f"- Thread {d['id']} | {d['subject']}") lines.append('```') lines.append(d['draft']) lines.append('```') else: lines.append('- No P1/P2 reply drafts needed this run.') lines.append('') lines.append('### V3 linkage (Email ↔ Calendar/Fathom context proxy)') if links: for a, b, c in links[:10]: lines.append(f'- Email: {a} -> Event: {b} @ {c}') else: lines.append('- No strong linkage found this run.') lines.append('') lines.append('## Risks / Gaps') lines.append('- Classification is heuristic and should be tuned with explicit allow/deny sender lists.') lines.append('- Drafts are suggested only; no auto-send is performed.') lines.append('- Fathom linkage currently uses lightweight proxy matching (full API linkage can be added).') lines.append('') lines.append('## Next Run Improvements') lines.append('- Add sender-specific rules and domain trust scores.') lines.append('- Add promised-followup detection from thread text snippets when available.') lines.append('- Add full CRM timeline rollup per contact in dedicated daily/weekly files.') OUT_DAILY.write_text('\n'.join(lines), encoding='utf-8') print(str(OUT_DAILY)) if __name__ == '__main__': main()