""" stripeSubscription_v2.py Exporta suscripciones activas + canceladas de Stripe. Genera dos CSVs: - stripe_subscriptions.csv (active, past_due, trialing) — igual que antes - stripe_canceled.csv (canceled) — con fecha de inicio y cancelación - stripe_monthly_metrics.csv (nuevos y cancelados por mes → churn real) Uso: pip install stripe python-dotenv pytz python stripeSubscription_v2.py """ import os import csv import stripe from datetime import datetime from collections import defaultdict from dotenv import load_dotenv import pytz # ─── CONFIGURACIÓN ──────────────────────────────────────────────────────────── load_dotenv("/var/www/html/backtestingmarket/.env") stripe.api_key = os.environ.get("STRIPE_SECRET_KEY", "") NY_TZ = pytz.timezone("America/New_York") STATUSES_ACTIVE = ["active", "past_due", "trialing"] # ────────────────────────────────────────────────────────────────────────────── def fetch_subscriptions_by_status(status, extra_params=None): subs = [] params = { "status": status, "limit": 100, "expand": ["data.customer", "data.latest_invoice"], } if extra_params: params.update(extra_params) while True: page = stripe.Subscription.list(**params) subs.extend(page.data) if not page.has_more: break params["starting_after"] = page.data[-1].id return subs def ts_to_ny(ts): if ts is None: return "" dt_utc = datetime.utcfromtimestamp(ts).replace(tzinfo=pytz.utc) return dt_utc.astimezone(NY_TZ).strftime("%Y-%m-%d %H:%M:%S") def ts_to_month(ts): if ts is None: return "" dt_utc = datetime.utcfromtimestamp(ts).replace(tzinfo=pytz.utc) return dt_utc.astimezone(NY_TZ).strftime("%Y-%m") def format_amount(amount_cents, currency): symbols = {"usd": "$", "eur": "€", "gbp": "£"} symbol = symbols.get(currency.lower(), currency.upper() + " ") return f"{symbol}{amount_cents / 100:,.2f}" def get_plan_info(s): items = s["items"]["data"] item = items[0] if items else {} plan = item.get("plan") if item else None interval = "" amount_cents = 0 currency = "usd" if plan: interval = plan["interval"] if plan.get("interval_count", 1) > 1: interval = f"{plan['interval_count']} {plan['interval']}s" amount_cents = plan["amount"] * (item.get("quantity") or 1) currency = plan["currency"] return interval, amount_cents, currency, item def get_customer_name(s): customer = s["customer"] if isinstance(s["customer"], (dict, stripe.Customer)) else {} if customer: return customer.get("name") or customer.get("email") or "" return "" # ── 1. Exportar activas (igual que antes) ──────────────────────────────────── def export_active(): print("📡 Exportando suscripciones activas...") all_subs = [] for status in STATUSES_ACTIVE: subs = fetch_subscriptions_by_status(status) print(f" {len(subs):>4} [{status}]") all_subs.extend(subs) headers = ["Interval", "Amount", "Status", "Current Period End (NY)", "Customer Name", "referrer_code", "youtuber_id", "created (NY)"] rows = [] for s in all_subs: invoice = s["latest_invoice"] if isinstance(s["latest_invoice"], (dict, stripe.Invoice)) else {} metadata = s.get("metadata") or {} interval, plan_cents, plan_currency, item = get_plan_info(s) inv_due = invoice.get("amount_due", 0) if invoice else 0 inv_cur = invoice.get("currency", plan_currency) if invoice else plan_currency amount = format_amount(inv_due, inv_cur) if inv_due > 0 else format_amount(plan_cents, plan_currency) rows.append([ interval, amount, s["status"], ts_to_ny(item.get("current_period_end")), get_customer_name(s), metadata.get("referrer_code", ""), metadata.get("youtuber_id", ""), ts_to_ny(s.get("created")), ]) with open("stripe_subscriptions.csv", "w", newline="", encoding="utf-8") as f: csv.writer(f).writerow(headers) csv.writer(f).writerows(rows) print(f"✅ stripe_subscriptions.csv ({len(rows)} filas)") return all_subs # ── 2. Exportar canceladas ──────────────────────────────────────────────────── def export_canceled(): print("\n📡 Exportando suscripciones canceladas...") subs = fetch_subscriptions_by_status("canceled") print(f" {len(subs):>4} [canceled]") headers = ["Interval", "Amount", "Customer Name", "Created (NY)", "Canceled At (NY)", "referrer_code", "youtuber_id"] rows = [] for s in subs: metadata = s.get("metadata") or {} interval, plan_cents, plan_currency, _ = get_plan_info(s) amount = format_amount(plan_cents, plan_currency) rows.append([ interval, amount, get_customer_name(s), ts_to_ny(s.get("created")), ts_to_ny(s.get("canceled_at")), metadata.get("referrer_code", ""), metadata.get("youtuber_id", ""), ]) with open("stripe_canceled.csv", "w", newline="", encoding="utf-8") as f: writer = csv.writer(f) writer.writerow(headers) writer.writerows(rows) print(f"✅ stripe_canceled.csv ({len(rows)} filas)") return subs # ── 3. Métricas mensuales: nuevos vs cancelados → churn real ───────────────── def export_monthly_metrics(active_subs, canceled_subs): print("\n📊 Calculando métricas mensuales...") # Nuevos por mes (fecha created de activas + canceladas) new_by_month = defaultdict(int) for s in active_subs + canceled_subs: m = ts_to_month(s.get("created")) if m: new_by_month[m] += 1 # Cancelados por mes (canceled_at) canceled_by_month = defaultdict(int) for s in canceled_subs: m = ts_to_month(s.get("canceled_at")) if m: canceled_by_month[m] += 1 all_months = sorted(set(list(new_by_month.keys()) + list(canceled_by_month.keys()))) headers = ["Month", "New Subscriptions", "Cancellations", "Net Change", "Churn Rate (%)"] rows = [] running_total = 0 for m in all_months: new = new_by_month[m] canceled = canceled_by_month[m] net = new - canceled running_total += new # churn = cancelados / subs activos al inicio del mes (aprox) base = running_total - net churn_rate = (canceled / base * 100) if base > 0 else 0 rows.append([m, new, canceled, net, f"{churn_rate:.1f}%"]) with open("stripe_monthly_metrics.csv", "w", newline="", encoding="utf-8") as f: writer = csv.writer(f) writer.writerow(headers) writer.writerows(rows) print(f"✅ stripe_monthly_metrics.csv ({len(rows)} meses)") # Imprimir últimos 6 meses como preview print("\n Últimos 6 meses:") print(f" {'Mes':<10} {'Nuevos':>8} {'Cancelados':>12} {'Neto':>8} {'Churn':>8}") for row in rows[-6:]: print(f" {row[0]:<10} {row[1]:>8} {row[2]:>12} {row[3]:>8} {row[4]:>8}") def main(): active_subs = export_active() canceled_subs = export_canceled() export_monthly_metrics(active_subs, canceled_subs) print("\n🎯 Listo. Archivos generados:") print(" - stripe_subscriptions.csv") print(" - stripe_canceled.csv") print(" - stripe_monthly_metrics.csv") if __name__ == "__main__": main()