Janady13/AI_Brain / stripe_worker.py

Source identity

Field	Recorded value
Source type	GitHub source
Original record	Janady13/AI_Brain / stripe_worker.py
Date	Not specified in source
Repository	Janady13/AI_Brain
Revision	cdb497ece2d1a770ca8c83eb1de8923cfaa9eceb
Branch	feature/seeder-engine-tests
Path or archive member	stripe_worker.py
Original SHA-256	85c223609f6f1718bb7cb1eb0da95d774e2493f19f2826c9e96971c6738047bc
Git blob SHA	7adb3ce4de261c81c24bb720d4feac542dcb8723
Reading-copy SHA-256	85c223609f6f1718bb7cb1eb0da95d774e2493f19f2826c9e96971c6738047bc

Original source ↗

Recovered source transcript

Download complete source text ↓123 lines · 5.1 KB

import os, json
import urllib.request
from fastapi import FastAPI, Request, Header
from config.stripe_secrets import apply_env_from_stripe_db
from fastapi.responses import JSONResponse

app = FastAPI(title="Revenue Worker")

# In production, provide real keys via env; tests monkeypatch stripe module
STRIPE_SECRET_KEY = os.environ.get("STRIPE_SECRET_KEY", "")
WEBHOOK_SECRET = os.environ.get("STRIPE_WEBHOOK_SECRET", "")
AGENTS_API_BASE_URL = (os.environ.get("AGENTS_API_BASE_URL") or os.environ.get("NEXT_PUBLIC_AGENTS_API_URL") or "").strip()
REVENUE_EVENTS_PATH = os.environ.get("REVENUE_EVENTS_PATH") or os.path.expanduser("~/AI_Brain/db/revenue_events.jsonl")


@app.get("/healthz")
def healthz():
    return {"ok": True}


@app.post("/webhook/stripe")
async def webhook(req: Request, stripe_signature: str = Header(None)):
    # Ensure webhook secret available
    if not os.environ.get("STRIPE_WEBHOOK_SECRET"):
        apply_env_from_stripe_db()
    raw = await req.body()
    try:
        # Lazy import: tests may stub 'stripe' in sys.modules
        import stripe  # type: ignore
        event = stripe.Webhook.construct_event(raw, stripe_signature, WEBHOOK_SECRET)  # type: ignore
    except Exception as e:
        return JSONResponse({"error": str(e)}, status_code=400)
    # Handle key event for revenue
    etype = getattr(event, "type", None) or event.get("type")
    if etype == "checkout.session.completed":
        # Extract key fields from object
        obj = getattr(event, "data", {}).get("object") if hasattr(event, "data") else event.get("data", {}).get("object")
        session_id = (obj or {}).get("id") or "unknown"
        amount_cents = int((obj or {}).get("amount_total") or 0)
        email = (obj or {}).get("customer_email") or "unknown"
        payload = {
            "provider": "stripe",
            "type": "checkout.session.completed",
            "session_id": session_id,
            "amount_cents": amount_cents,
            "email": email,
        }
        # Try Agents API first
        if AGENTS_API_BASE_URL:
            try:
                body = json.dumps({
                    "kind": "revenue.stripe",
                    "title": f"Stripe checkout completed: {email}",
                    "dedupe_key": f"stripe:{session_id}",
                    "payload": payload,
                    "priority": 1,
                }).encode("utf-8")
                url = AGENTS_API_BASE_URL.rstrip("/") + "/tasks"
                req2 = urllib.request.Request(url, data=body, headers={"Content-Type": "application/json"}, method="POST")
                with urllib.request.urlopen(req2, timeout=5) as resp:  # nosec - internal service call
                    _ = resp.read()
            except Exception:
                pass
        # Always append to local event log as durable fallback
        try:
            events_path = os.environ.get("REVENUE_EVENTS_PATH") or REVENUE_EVENTS_PATH
            os.makedirs(os.path.dirname(events_path), exist_ok=True)
            with open(events_path, "a", encoding="utf-8") as f:
                f.write(json.dumps(payload) + "\n")
        except Exception:
            pass
    return {"ok": True, "type": etype or "unknown"}


@app.post("/stripe/checkout_session")
async def checkout_session(req: Request):
    """Create a Stripe Checkout Session.
    Expects JSON with either line_items (with price+quantity) or amount+currency+name.
    Requires STRIPE_SECRET_KEY. Returns {url: ...} on success.
    """
    data = await req.json()
    # Validate config
    if not (os.environ.get("STRIPE_SECRET_KEY") or STRIPE_SECRET_KEY):
        apply_env_from_stripe_db()
    key = os.environ.get("STRIPE_SECRET_KEY") or STRIPE_SECRET_KEY
    if not key:
        return JSONResponse({"error": "stripe_not_configured"}, status_code=400)
    try:
        import stripe  # type: ignore
        stripe.api_key = key  # type: ignore

        success_url = data.get("success_url") or os.getenv("STRIPE_SUCCESS_URL") or "https://example.com/success"
        cancel_url = data.get("cancel_url") or os.getenv("STRIPE_CANCEL_URL") or "https://example.com/cancel"
        mode = data.get("mode") or "payment"

        line_items = data.get("line_items")
        if not line_items:
            # Build from amount/currency/name
            amount = int(data.get("amount") or 0)
            currency = (data.get("currency") or "usd").lower()
            name = data.get("name") or "Payment"
            if amount <= 0:
                return JSONResponse({"error": "invalid_amount"}, status_code=400)
            line_items = [{
                "price_data": {
                    "currency": currency,
                    "product_data": {"name": name},
                    "unit_amount": amount,
                },
                "quantity": 1,
            }]

        session = stripe.checkout.Session.create(  # type: ignore
            mode=mode,
            line_items=line_items,
            success_url=success_url,
            cancel_url=cancel_url,
        )
        url = getattr(session, "url", None) or session.get("url")
        return {"url": url}
    except Exception as e:
        return JSONResponse({"error": str(e)}, status_code=400)