""" CultivaFlow — Integración de Pagos con Stripe

Stack: FastAPI + Stripe SDK (Python) Dominio: https://app.cultivaflow.io

Planes:

• Starter 29 €/mes (price_starter_monthly)
• Growth 79 €/mes (price_growth_monthly)
• Agency 249 €/mes (price_agency_monthly)

Cubre: Checkout Sessions, Suscripciones, Webhooks, Portal Cliente, Reembolsos """

import hashlib import hmac import os from typing import Optional

import stripe from fastapi import FastAPI, HTTPException, Request, status from fastapi.responses import JSONResponse, RedirectResponse

──────────────────────────────────────────────────────────

Configuración

──────────────────────────────────────────────────────────

stripe.api_key = os.environ.get("STRIPE_SECRET_KEY", "sk_test_...") STRIPE_WEBHOOK_SECRET = os.environ.get("STRIPE_WEBHOOK_SECRET", "whsec_...") DOMAIN = "https://app.cultivaflow.io"

PLANS = { "starter": { "price_id": "price_starter_monthly", "name": "Starter", "amount_eur": 2900, # céntimos }, "growth": { "price_id": "price_growth_monthly", "name": "Growth", "amount_eur": 7900, }, "agency": { "price_id": "price_agency_monthly", "name": "Agency", "amount_eur": 24900, }, }

app = FastAPI(title="CultivaFlow Payments API")

──────────────────────────────────────────────────────────

1. Crear Checkout Session (Stripe Hosted Page)

──────────────────────────────────────────────────────────

@app.post("/api/stripe/checkout") async def create_checkout_session(plan: str, user_id: str, email: str): """ Redirige al usuario a la página de pago de Stripe. Soporta SCA / 3D Secure automáticamente. """ if plan not in PLANS: raise HTTPException(status_code=400, detail=f"Plan desconocido: {plan}")

plan_data = PLANS[plan]

# Crear o recuperar customer de Stripe
customers = stripe.Customer.list(email=email, limit=1).data
if customers:
    customer = customers[0]
else:
    customer = stripe.Customer.create(
        email=email,
        metadata={"user_id": user_id, "platform": "cultivaflow"},
    )

session = stripe.checkout.Session.create(
    customer=customer.id,
    line_items=[
        {
            "price": plan_data["price_id"],
            "quantity": 1,
        }
    ],
    mode="subscription",
    # SCA / 3D Secure se maneja automáticamente con Checkout Sessions
    payment_method_types=["card"],
    allow_promotion_codes=True,
    billing_address_collection="auto",
    success_url=f"{DOMAIN}/dashboard?session_id={{CHECKOUT_SESSION_ID}}&plan={plan}",
    cancel_url=f"{DOMAIN}/pricing?cancelled=true",
    metadata={
        "user_id": user_id,
        "plan": plan,
    },
    subscription_data={
        "metadata": {"user_id": user_id, "plan": plan},
    },
    # Guardamos método de pago por defecto para renovaciones
    payment_method_options={
        "card": {"request_three_d_secure": "automatic"}
    },
)

return {"checkout_url": session.url, "session_id": session.id}

──────────────────────────────────────────────────────────

2. Portal del Cliente (gestión de suscripción)

──────────────────────────────────────────────────────────

@app.post("/api/stripe/portal") async def create_portal_session(customer_id: str): """ Genera URL del Billing Portal de Stripe donde el cliente puede: - Cambiar de plan (upgrade / downgrade) - Actualizar tarjeta - Cancelar suscripción - Descargar facturas """ portal_session = stripe.billing_portal.Session.create( customer=customer_id, return_url=f"{DOMAIN}/account", ) return {"portal_url": portal_session.url}

──────────────────────────────────────────────────────────

3. Webhook (eventos críticos de Stripe)

──────────────────────────────────────────────────────────

@app.post("/api/stripe/webhook") async def stripe_webhook(request: Request): """ Endpoint seguro para eventos de Stripe. Verifica firma HMAC antes de procesar. Implementa idempotencia: cada event_id se procesa UNA sola vez. """ payload = await request.body() sig_header = request.headers.get("stripe-signature", "")

try:
    event = stripe.Webhook.construct_event(
        payload, sig_header, STRIPE_WEBHOOK_SECRET
    )
except ValueError:
    raise HTTPException(status_code=400, detail="Payload inválido")
except stripe.error.SignatureVerificationError:
    raise HTTPException(status_code=400, detail="Firma inválida — posible replay attack")

# Idempotencia: comprobar si ya procesamos este evento
event_id = event["id"]
if _is_event_already_processed(event_id):
    return JSONResponse({"status": "duplicado, ignorado"})

# Despachar según tipo de evento
event_type = event["type"]
data_object = event["data"]["object"]

if event_type == "invoice.payment_succeeded":
    _handle_invoice_paid(data_object)

elif event_type == "invoice.payment_failed":
    _handle_invoice_failed(data_object)

elif event_type == "customer.subscription.updated":
    _handle_subscription_updated(data_object)

elif event_type == "customer.subscription.deleted":
    _handle_subscription_canceled(data_object)

elif event_type == "payment_intent.succeeded":
    _handle_payment_intent_succeeded(data_object)

elif event_type == "charge.refunded":
    _handle_charge_refunded(data_object)

_mark_event_processed(event_id)
return JSONResponse({"status": "ok"})

──────────────────────────────────────────────────────────

Handlers de eventos

──────────────────────────────────────────────────────────

def _handle_invoice_paid(invoice: dict): """Pago de suscripción exitoso → activar / renovar plan en BD.""" customer_id = invoice.get("customer") subscription_id = invoice.get("subscription") metadata = invoice.get("subscription_details", {}).get("metadata", {}) user_id = metadata.get("user_id") plan = metadata.get("plan")

print(f"[INVOICE PAID] user={user_id} plan={plan} sub={subscription_id}")
# TODO: db.users.update(user_id, plan=plan, stripe_customer_id=customer_id, active=True)
# TODO: send_email_confirmacion(user_id, plan)

def _handle_invoice_failed(invoice: dict): """Pago fallido → notificar al usuario, no revocar acceso de inmediato.""" customer_id = invoice.get("customer") attempt_count = invoice.get("attempt_count", 1) next_payment_attempt = invoice.get("next_payment_attempt")

print(f"[INVOICE FAILED] customer={customer_id} intento={attempt_count}")
if attempt_count >= 3:
    # Stripe habrá cancelado la suscripción — esperar el evento deleted
    print(f"  → Stripe cancelará la suscripción automáticamente")
else:
    # Notificar para que actualice tarjeta
    print(f"  → Próximo intento: {next_payment_attempt}")
# TODO: send_email_pago_fallido(customer_id, attempt_count)

def _handle_subscription_updated(subscription: dict): """Cambio de plan → actualizar en BD.""" customer_id = subscription.get("customer") status = subscription.get("status") metadata = subscription.get("metadata", {}) user_id = metadata.get("user_id") new_plan = metadata.get("plan")

print(f"[SUB UPDATED] user={user_id} nuevo_plan={new_plan} status={status}")
# TODO: db.users.update(user_id, plan=new_plan, sub_status=status)

def _handle_subscription_canceled(subscription: dict): """Suscripción cancelada → degradar a free tier.""" customer_id = subscription.get("customer") metadata = subscription.get("metadata", {}) user_id = metadata.get("user_id")

print(f"[SUB CANCELED] user={user_id} customer={customer_id}")
# TODO: db.users.update(user_id, plan='free', active=False)
# TODO: send_email_baja(user_id)

def _handle_payment_intent_succeeded(payment_intent: dict): """Pago único exitoso (si se usa PaymentIntents en lugar de Checkout).""" amount = payment_intent.get("amount") currency = payment_intent.get("currency") print(f"[PAYMENT INTENT OK] {amount} {currency.upper()}")

def _handle_charge_refunded(charge: dict): """Reembolso procesado → actualizar historial en BD.""" amount_refunded = charge.get("amount_refunded") customer_id = charge.get("customer") print(f"[REFUND] customer={customer_id} importe={amount_refunded/100:.2f} €") # TODO: db.refunds.create(customer_id=customer_id, amount=amount_refunded)

──────────────────────────────────────────────────────────

4. Reembolsos (desde panel admin de CultivaFlow)

──────────────────────────────────────────────────────────

@app.post("/api/stripe/refund") async def create_refund( payment_intent_id: str, amount_cents: Optional[int] = None, reason: str = "requested_by_customer", ): """ Emite un reembolso total o parcial. reason: 'requested_by_customer' | 'duplicate' | 'fraudulent' """ params = { "payment_intent": payment_intent_id, "reason": reason, } if amount_cents: params["amount"] = amount_cents # Reembolso parcial

try:
    refund = stripe.Refund.create(**params)
    return {
        "refund_id": refund.id,
        "status": refund.status,
        "amount_refunded": refund.amount / 100,
        "currency": refund.currency.upper(),
    }
except stripe.error.InvalidRequestError as e:
    raise HTTPException(status_code=400, detail=str(e.user_message))

──────────────────────────────────────────────────────────

5. Gestión de clientes

──────────────────────────────────────────────────────────

@app.get("/api/stripe/customer/{customer_id}/payment-methods") async def list_payment_methods(customer_id: str): """Lista las tarjetas guardadas de un cliente.""" methods = stripe.PaymentMethod.list(customer=customer_id, type="card").data return [ { "id": pm.id, "brand": pm.card.brand, "last4": pm.card.last4, "exp_month": pm.card.exp_month, "exp_year": pm.card.exp_year, } for pm in methods ]

──────────────────────────────────────────────────────────

Helpers de idempotencia (sustituir por Redis/BD en prod)

──────────────────────────────────────────────────────────

_processed_events: set = set()

def _is_event_already_processed(event_id: str) -> bool: return event_id in _processed_events

def _mark_event_processed(event_id: str) -> None: _processed_events.add(event_id)

──────────────────────────────────────────────────────────

Tarjetas de test (modo test Stripe)

──────────────────────────────────────────────────────────

TEST_CARDS = { "pago_ok": "4242 4242 4242 4242", "declinada": "4000 0000 0000 0002", "3d_secure": "4000 0025 0000 3155", # SCA / PSD2 "fondos_insuf": "4000 0000 0000 9995", "cargo_dispute": "4000 0000 0000 0259", }