#!/usr/bin/env python3 import os import json import time import requests AGENTS_API = os.getenv("AGENTS_API_BASE_URL", "http://127.0.0.1:3001") REVENUE_LOG = os.getenv( "REVENUE_EVENTS_PATH", os.path.expanduser("~/AI_Brain/db/revenue_events.jsonl"), ) def read_events(path): if not os.path.exists(path): return [] with open(path, "r", encoding="utf-8") as f: for line in f: line = (line or "").strip() if not line: continue try: yield json.loads(line) except Exception: continue def _event_type(ev: dict) -> str: # Accept flat or nested event structures t = ev.get("type") if t: return str(t) data = ev.get("data") or {} if isinstance(data, dict): obj = data.get("object") or {} return str(obj.get("type") or "") return "" def _event_id(ev: dict) -> str: # Prefer explicit session_id, else nested id, else id/created for key in ("session_id", "id", "created"): v = ev.get(key) if v: return str(v) data = ev.get("data") or {} if isinstance(data, dict): obj = data.get("object") or {} v = obj.get("id") or obj.get("session_id") if v: return str(v) return "" def process_event(ev: dict) -> None: kind = _event_type(ev) if kind.startswith("checkout.session.completed") or kind == "checkout.session.completed": sid = _event_id(ev) print(f"[stripe_consumer] revenue event → {sid}") try: requests.post( f"{AGENTS_API}/tasks", json={ "kind": "revenue.stripe.analyze", "title": f"stripe:{sid}" if sid else "stripe:completed", "payload": ev, "priority": 1, }, timeout=5, ) except Exception as e: print(f"[stripe_consumer] post failed: {e}") def main(interval: int = 10) -> None: seen = set() print(f"[stripe_consumer] watching {REVENUE_LOG}") while True: for ev in read_events(REVENUE_LOG): key = _event_id(ev) or json.dumps(ev, sort_keys=True)[:128] if key not in seen: seen.add(key) process_event(ev) time.sleep(interval) if __name__ == "__main__": main()