#!/usr/bin/env python3 """ CE Finance Sweep — pulls new receipts from email and adds to Notion. Run from /root/.openclaw/workspace with venv active. Usage: python3 skills/ce-finance-review/scripts/finance-sweep.py """ import os, sys, re, json, base64, io, requests from datetime import datetime, date from collections import defaultdict from html.parser import HTMLParser # ── Paths ────────────────────────────────────────────────────────────── WORKSPACE = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) sys.path.insert(0, WORKSPACE) NOTION_KEY = open('/home/clawd/secrets/notion/api_key').read().strip() DB_ID = "322330c2-8646-811a-a610-e003f616ac91" NOTION_HDRS = {"Authorization": f"Bearer {NOTION_KEY}", "Notion-Version": "2022-06-28", "Content-Type": "application/json"} TOKENS = [ ('google-auth/token-assaf.json', 'assafdagan@gmail.com'), ('google-auth/token.json', 'assafdagancos@gmail.com'), ] # ── HTML stripper ─────────────────────────────────────────────────────── class MLStripper(HTMLParser): def __init__(self): super().__init__(); self.reset(); self.fed = [] def handle_data(self, d): self.fed.append(d) def get_data(self): return ' '.join(self.fed) def strip_html(html): s = MLStripper(); s.feed(html); return s.get_data() def get_text(payload): mime = payload.get('mimeType','') data = payload.get('body',{}).get('data','') text = '' if data: decoded = base64.urlsafe_b64decode(data+'==').decode('utf-8', errors='ignore') text = strip_html(decoded) if 'html' in mime else decoded for part in payload.get('parts',[]): text += get_text(part) return text # ── Notion helpers ────────────────────────────────────────────────────── def get_existing_entries(): """Return set of (vendor, month, amount) and set of invoice numbers.""" entries, invoices = set(), set() cursor = None while True: body = {"page_size": 100} if cursor: body["start_cursor"] = cursor r = requests.post(f"https://api.notion.com/v1/databases/{DB_ID}/query", headers=NOTION_HDRS, json=body) data = r.json() for page in data.get('results', []): props = page['properties'] vendor = (props.get('Vendor',{}).get('rich_text') or [{}])[0].get('plain_text','') month = (props.get('Month',{}).get('select') or {}).get('name','') amount = props.get('Amount',{}).get('number', 0) or 0 inv = (props.get('Invoice',{}).get('rich_text') or [{}])[0].get('plain_text','') entries.add((vendor, month, round(amount, 2))) if inv: invoices.add(inv) if not data.get('has_more'): break cursor = data.get('next_cursor') return entries, invoices def add_entry(name, month, category, vendor, amount, currency, date_iso, invoice='', notes='', billable=False): payload = { "parent": {"database_id": DB_ID}, "properties": { "Name": {"title": [{"text": {"content": name}}]}, "Month": {"select": {"name": month}}, "Category": {"select": {"name": category}}, "Vendor": {"rich_text": [{"text": {"content": vendor}}]}, "Amount": {"number": amount}, "Currency": {"select": {"name": currency}}, "Invoice": {"rich_text": [{"text": {"content": invoice}}]}, "Date": {"date": {"start": date_iso}}, "Billable": {"checkbox": billable}, "Notes": {"rich_text": [{"text": {"content": notes}}]}, } } r = requests.post("https://api.notion.com/v1/pages", headers=NOTION_HDRS, json=payload) return r.status_code == 200 # ── Date helpers ──────────────────────────────────────────────────────── MONTHS = {'Jan':'January','Feb':'February','Mar':'March','Apr':'April', 'May':'May','Jun':'June','Jul':'July','Aug':'August', 'Sep':'September','Oct':'October','Nov':'November','Dec':'December'} def parse_email_date(date_str): """Returns (date_iso, month_label) e.g. ('2026-03-12', 'March 2026')""" m = re.search(r'(\d{1,2})\s+(\w{3})\s+(\d{4})', date_str) if m: d, mon, y = m.group(1), m.group(2), m.group(3) try: dt = datetime.strptime(f"{d} {mon} {y}", "%d %b %Y") return dt.strftime('%Y-%m-%d'), f"{MONTHS.get(mon, mon)} {y}" except: pass return date.today().isoformat(), f"Unknown" # ── Anthropic PDF parsing ─────────────────────────────────────────────── def parse_anthropic_receipts(gmail_service, existing_invoices): """Download Anthropic PDF receipts and extract amounts.""" try: import pdfplumber except ImportError: print(" ⚠️ pdfplumber not installed — skipping PDF parsing") return [] results = [] r = gmail_service.users().messages().list( userId='me', q='from:invoice+statements anthropic receipt', maxResults=30 ).execute() for msg in r.get('messages', []): detail = gmail_service.users().messages().get(userId='me', id=msg['id'], format='full').execute() hdrs = {h['name']: h['value'] for h in detail.get('payload',{}).get('headers',[])} subject = hdrs.get('Subject','') if 'receipt' not in subject.lower(): continue inv_match = re.search(r'#([\w-]+)', subject) invoice_num = inv_match.group(1) if inv_match else '' if invoice_num in existing_invoices: continue # already logged # Download first PDF attachment for part in detail.get('payload',{}).get('parts',[]): if part.get('mimeType') == 'application/pdf': att_id = part.get('body',{}).get('attachmentId') if not att_id: continue att = gmail_service.users().messages().attachments().get( userId='me', messageId=msg['id'], id=att_id ).execute() pdf_data = base64.urlsafe_b64decode(att.get('data','') + '==') with pdfplumber.open(io.BytesIO(pdf_data)) as pdf: text = ' '.join(page.extract_text() or '' for page in pdf.pages) eur = re.search(r'Amount due\s*€([\d,]+\.\d{2})', text) usd = re.search(r'Amount due\s*\$([\d,]+\.\d{2})', text) if eur: amount, currency = float(eur.group(1).replace(',','')), 'EUR' elif usd: amount, currency = float(usd.group(1).replace(',','')), 'USD' else: continue date_iso, month = parse_email_date(hdrs.get('Date','')) results.append({ 'name': f"Anthropic — {invoice_num}", 'month': month, 'category': 'AI / APIs', 'vendor': 'Anthropic', 'amount': amount, 'currency': currency, 'date': date_iso, 'invoice': invoice_num, 'notes': 'Auto-recharge — OpenClaw/Claude API usage' }) break return results # ── Generic receipt scan ──────────────────────────────────────────────── VENDOR_MAP = { 'weavy': ('Weavy', 'Design Tools', 'USD'), # Note: Weavy receipts come from "Figma, Inc." via Stripe with weavy.ai in body # Figma Design Tool comes from support@figma.com with $35 amount # Detection: if sender is stripe AND amount ~45 → Weavy; if sender is figma.com AND amount ~35 → Figma Design Tool 'figma': ('Figma Design Tool','Design Tools', 'USD'), 'recraft': ('Recraft', 'Design Tools', 'USD'), 'hetzner': ('Hetzner', 'Dev / Infra', 'EUR'), 'x developer': ('X Developer', 'Dev / Infra', 'EUR'), 'x corp': ('X Premium', 'Comms / Marketing','EUR'), 'notion': ('Notion', 'Productivity', 'USD'), 'google*workspace': ('Google Workspace (CE)', 'Productivity', 'USD'), 'mymind': ('mymind', 'Design Tools', 'USD'), 'namecheap': ('Namecheap', 'Dev / Infra', 'USD'), 'adobe': ('Adobe', 'Design Tools', 'USD'), } SKIP_SENDERS = ['meta', 'facebook', 'newsletter', 'noreply@accounts', 'hello@mockup', 'genaipm', 'rehunt', 'every.to'] SKIP_SUBJECTS = ['marketing', 'newsletter', 'tip', 'office hours', 'level up', 'master the basics', 'before we head', 'recap:'] def scan_receipts(gmail_service, existing_entries): """Scan for non-Anthropic receipts.""" results = [] r = gmail_service.users().messages().list( userId='me', q='subject:(receipt OR invoice) -from:anthropic after:2026/01/01', maxResults=50 ).execute() seen_ids = set() for msg in r.get('messages', []): if msg['id'] in seen_ids: continue seen_ids.add(msg['id']) detail = gmail_service.users().messages().get(userId='me', id=msg['id'], format='full').execute() hdrs = {h['name']: h['value'] for h in detail.get('payload',{}).get('headers',[])} subject = hdrs.get('Subject','').lower() sender = hdrs.get('From','').lower() if any(s in sender for s in SKIP_SENDERS): continue if any(s in subject for s in SKIP_SUBJECTS): continue body = re.sub(r'\s+', ' ', get_text(detail.get('payload',{}))).strip() # Find amounts amounts_eur = re.findall(r'€\s*([\d,]+\.\d{2})', body) amounts_usd = re.findall(r'\$([\d,]+\.\d{2})', body) def clean_amounts(aa): return [float(a.replace(',','')) for a in aa if 0.5 <= float(a.replace(',','')) <= 50000] eur_vals = clean_amounts(amounts_eur) usd_vals = clean_amounts(amounts_usd) if not eur_vals and not usd_vals: continue amount = eur_vals[0] if eur_vals else usd_vals[0] currency = 'EUR' if eur_vals else 'USD' # Match vendor — special case: Weavy billed as "Figma, Inc." via Stripe vendor_key = None if 'figma' in sender and 'stripe' in sender and amount in [45.0, 15.0, 25.0]: # Weavy (billed as Figma, Inc. via Stripe account with weavy.ai) vendor, category, currency = 'Weavy', 'Design Tools', 'USD' elif 'figma' in sender and 'stripe' not in sender: # True Figma Design Tool (support@figma.com) vendor, category, currency = 'Figma Design Tool', 'Design Tools', 'USD' else: for kw in VENDOR_MAP: if kw in subject or kw in sender: vendor_key = kw; break if not vendor_key: continue vendor, category, _ = VENDOR_MAP[vendor_key] date_iso, month = parse_email_date(hdrs.get('Date','')) # Dedup check key = (vendor, month, round(amount, 2)) if key in existing_entries: continue inv_match = re.search(r'#([\w-]+)', hdrs.get('Subject','')) invoice = inv_match.group(1) if inv_match else '' results.append({ 'name': f"{vendor} — {invoice or date_iso}", 'month': month, 'category': category, 'vendor': vendor, 'amount': amount, 'currency': currency, 'date': date_iso, 'invoice': invoice, 'notes': '' }) return results # ── Main ──────────────────────────────────────────────────────────────── def main(): try: from google.oauth2.credentials import Credentials from googleapiclient.discovery import build except ImportError: print("ERROR: google-api-python-client not installed. Run from workspace venv.") sys.exit(1) print("🔍 CE Finance Sweep starting...\n") existing_entries, existing_invoices = get_existing_entries() print(f"📋 Existing Notion entries: {len(existing_entries)}") print(f"📋 Known invoice numbers: {len(existing_invoices)}\n") all_new = [] seen_msg_ids = set() for token_file, label in TOKENS: token_path = os.path.join(WORKSPACE, token_file) if not os.path.exists(token_path): print(f"⚠️ Token not found: {token_file} — skipping {label}") continue try: creds = Credentials.from_authorized_user_file(token_path) gmail = build('gmail', 'v1', credentials=creds) print(f"📧 Scanning {label}...") # Anthropic PDFs anthropic_new = parse_anthropic_receipts(gmail, existing_invoices) print(f" Anthropic: {len(anthropic_new)} new receipts") all_new.extend(anthropic_new) # Other receipts other_new = scan_receipts(gmail, existing_entries) print(f" Other: {len(other_new)} new receipts") all_new.extend(other_new) except Exception as e: print(f" ❌ Error scanning {label}: {e}") # Dedup across both accounts seen_keys = set() unique_new = [] for entry in all_new: key = (entry['vendor'], entry['month'], round(entry['amount'], 2)) inv = entry.get('invoice','') if key not in seen_keys and (not inv or inv not in existing_invoices): seen_keys.add(key) if inv: existing_invoices.add(inv) unique_new.append(entry) print(f"\n✨ {len(unique_new)} new entries to add\n") added, failed = [], [] for entry in unique_new: ok = add_entry( name=entry['name'], month=entry['month'], category=entry['category'], vendor=entry['vendor'], amount=entry['amount'], currency=entry['currency'], date_iso=entry['date'], invoice=entry.get('invoice',''), notes=entry.get('notes','') ) if ok: added.append(entry) print(f" ✅ {entry['month'][:3]} | {entry['currency']} {entry['amount']:>8.2f} | {entry['vendor']}") else: failed.append(entry) print(f" ❌ Failed: {entry['name']}") # Summary print(f"\n{'='*50}") print(f"✅ Added: {len(added)} ❌ Failed: {len(failed)}") if added: by_month = defaultdict(lambda: defaultdict(float)) for e in added: by_month[e['month']][e['currency']] += e['amount'] print("\nNew spend by month:") for month, currencies in sorted(by_month.items()): parts = [f"{cur} {amt:.2f}" for cur, amt in currencies.items()] print(f" {month}: {', '.join(parts)}") print(f"\nNotion: https://www.notion.so/{DB_ID.replace('-','')}") return added, failed if __name__ == '__main__': os.chdir(os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))) main()