#!/usr/bin/env python """Sync billing catalog (plans + usage packages + runtime AI) from DB to Stripe. Usage: python -m scripts.sync_billing_catalog python -m scripts.sync_billing_catalog --plans-only python -m scripts.sync_billing_catalog --packages-only python -m scripts.sync_billing_catalog --runtime-ai-only python -m scripts.sync_billing_catalog --plan free python -m scripts.sync_billing_catalog --package ai-minutes-60 python -m scripts.sync_billing_catalog --dry-run Behavior: - Plans: pushes name, description, active, and features/limits metadata to Stripe Products (no new prices). When DB recurring amounts differ from the Stripe Price already linked, creates new Prices, deactivates old ones, and updates plan_price_versions. - Usage packages: same idea for one-time prices (codegen credits). - Runtime AI packages: separate wallet product (`runtime_ai_*` slugs). Never mixed with workspace_credit_ledger / usage_credit_packages. Uses config.stripe_secret_key (from .env or environment) for Stripe API calls. Exits non-zero on sync failures. """ from __future__ import annotations import asyncio import argparse import os import sys sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) async def run_sync( plans_only: bool = False, packages_only: bool = False, runtime_ai_only: bool = False, plan_slug: str | None = None, package_slug: str | None = None, dry_run: bool = False, ) -> int: from src.config import get_settings from src.core.database import async_session_factory from src.domain.admin.repository import PlanPriceVersionRepository, SubscriptionPlanRepository from src.domain.billing.repository import UsageCreditPackageRepository from src.domain.billing.stripe_sync_service import ( PlanStripeSyncService, UsagePackageStripeSyncService, ) from src.integrations.stripe_client import StripeClientImpl settings = get_settings() stripe = StripeClientImpl() if settings.stripe_secret_key else None if not stripe and not dry_run: print("WARN: stripe_secret_key not configured - sync will create DB entries only (no Stripe API calls)") if dry_run: print("DRY-RUN: no changes will be committed") created = synced = failed = 0 async with async_session_factory() as db: plan_repo = SubscriptionPlanRepository(db) package_repo = UsageCreditPackageRepository(db) price_version_repo = PlanPriceVersionRepository(db) if not packages_only and not runtime_ai_only: plan_sync = PlanStripeSyncService(plan_repo, stripe, price_version_repo=price_version_repo) if plan_slug: plan = await plan_repo.get_by_slug(plan_slug) if not plan: print(f"FAIL: Plan '{plan_slug}' not found") return 1 plans = [plan] else: plans = await plan_repo.list_all(active_only=True) for p in plans: try: had_stripe = bool(p.stripe_product_id) await plan_sync.sync_plan(p) has_stripe = bool(p.stripe_product_id) if has_stripe and not had_stripe: created += 1 print(f" [CREATED] {p.slug}: product={p.stripe_product_id}") else: synced += 1 print(f" [OK] {p.slug}: product={p.stripe_product_id or '(none)'}") except Exception as e: failed += 1 print(f" [FAIL] {p.slug}: {e}") if not plans_only and not runtime_ai_only: pkg_sync = UsagePackageStripeSyncService(package_repo, stripe) if package_slug: pkg = await package_repo.get_by_slug(package_slug) if not pkg: print(f"FAIL: Package '{package_slug}' not found") return 1 packages = [pkg] else: packages = await package_repo.list_all(active_only=True) for p in packages: try: had_stripe = bool(p.stripe_product_id) await pkg_sync.sync_package(p) has_stripe = bool(p.stripe_product_id) if has_stripe and not had_stripe: created += 1 print(f" [CREATED] {p.slug}: product={p.stripe_product_id}") else: synced += 1 print(f" [OK] {p.slug}: product={p.stripe_product_id or '(none)'}") except Exception as e: failed += 1 print(f" [FAIL] {p.slug}: {e}") if not plans_only and not packages_only: from src.domain.runtime_ai.service import RuntimeAiService from src.domain.cloud_credits.service import CloudCreditService runtime = RuntimeAiService(db=db, stripe_client=stripe) cloud = CloudCreditService(db=db, stripe_client=stripe) if not stripe: print("WARN: skipping Runtime AI / Cloud credit Stripe product/price create (no stripe_secret_key)") else: try: await runtime.sync_packages_to_stripe(stripe) for pkg in await runtime.list_packages(): synced += 1 print( f" [OK] {pkg.slug}: product={pkg.stripe_product_id or '(none)'} " f"price={pkg.stripe_price_id or '(none)'}" ) except Exception as e: failed += 1 print(f" [FAIL] runtime_ai: {e}") try: await cloud.sync_packages_to_stripe(stripe) for pkg in await cloud.list_packages(): synced += 1 print( f" [OK] {pkg.slug}: product={pkg.stripe_product_id or '(none)'} " f"price={pkg.stripe_price_id or '(none)'}" ) except Exception as e: failed += 1 print(f" [FAIL] cloud_credits: {e}") if not dry_run and failed == 0: await db.commit() print(f"\nResult: created={created} already_synced={synced} failed={failed}") return 1 if failed > 0 else 0 def main() -> int: parser = argparse.ArgumentParser(description="Sync billing catalog to Stripe") parser.add_argument("--plans-only", action="store_true") parser.add_argument("--packages-only", action="store_true") parser.add_argument("--runtime-ai-only", action="store_true") parser.add_argument("--plan", metavar="SLUG", help="Sync only this plan by slug") parser.add_argument("--package", metavar="SLUG", help="Sync only this package by slug") parser.add_argument("--dry-run", action="store_true", help="Do not commit changes") args = parser.parse_args() plan_slug = args.plan if args.plan else None package_slug = args.package if args.package else None if plan_slug: args.plans_only = True if package_slug: args.packages_only = True return asyncio.run(run_sync( plans_only=args.plans_only, packages_only=args.packages_only, runtime_ai_only=args.runtime_ai_only, plan_slug=plan_slug, package_slug=package_slug, dry_run=args.dry_run, )) if __name__ == "__main__": sys.exit(main())