Janady13/AI_Brain / workers/stripe_consumer.py

Source identity

Field	Recorded value
Source type	GitHub source
Original record	Janady13/AI_Brain / workers/stripe_consumer.py
Date	Not specified in source
Repository	Janady13/AI_Brain
Revision	cdb497ece2d1a770ca8c83eb1de8923cfaa9eceb
Branch	feature/seeder-engine-tests
Path or archive member	workers/stripe_consumer.py
Original SHA-256	64ded90c54f9593b0faf6255be18d3ba3ae5e7916a2ac9925f40ce615df5ea1c
Git blob SHA	0913bffe2b440992db4bb60da9f5548f7463f8e5
Reading-copy SHA-256	64ded90c54f9593b0faf6255be18d3ba3ae5e7916a2ac9925f40ce615df5ea1c

Original source ↗

Recovered source transcript

Download complete source text ↓90 lines · 2.4 KB

#!/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()