"""One-shot ERPNext payments setup CLI. Configures ERPNext for the three live payment gateways: **Stripe**, **PayPal**, **SHKeeper crypto**. Adyen is intentionally deferred (future work — merchant account application still pending). What this script does --------------------- 1. Creates / updates the ``Payment Gateway Account`` records: - ``Stripe-Card``, ``Stripe-ACH``, ``Stripe-Klarna``, ``Stripe-PayPal`` - ``Crypto-Crypto`` (SHKeeper via frappe_crypto) 2. Configures the SHKeeper ``Crypto Payment Settings`` DocType from env (``SHKEEPER_URL`` + ``SHKEEPER_API_KEY``). 3. Verifies that the Subscription Plan fixtures imported by ``bench migrate`` are present and at the expected pricing. 4. Prints a final readiness checklist for anything that still requires human action (e.g. registering the Stripe webhook endpoint). Usage ----- bench --site crm.performancewest.net execute \\ scripts.setup_erpnext_payments.run Or, if running outside bench, it falls back to the ``frappe-client`` API via ERPNEXT_URL / ERPNEXT_API_KEY / ERPNEXT_API_SECRET. Required env vars ----------------- SHKEEPER_URL — default https://pay.performancewest.net SHKEEPER_API_KEY — required for crypto go-live STRIPE_SECRET_KEY — not written to ERPNext; only used for readiness check """ from __future__ import annotations import logging import os import sys from dataclasses import dataclass logger = logging.getLogger(__name__) @dataclass class GatewayAccount: name: str # Payment Gateway Account.name (primary key) gateway: str # Payment Gateway DocType name to link currency: str = "USD" payment_account: str = "" # Chart of Accounts node — operator maps this payment_channel: str = "Email" PAYMENT_GATEWAYS: list[GatewayAccount] = [ # Stripe — live GatewayAccount("Stripe-Card", "Stripe-Card"), GatewayAccount("Stripe-ACH", "Stripe-ACH"), GatewayAccount("Stripe-Klarna", "Stripe-Klarna"), GatewayAccount("Stripe-PayPal", "Stripe-PayPal"), # SHKeeper crypto — live GatewayAccount("Crypto-Crypto", "Crypto"), ] def _bench_available() -> bool: try: import frappe # noqa: F401 return True except ImportError: return False def _in_bench_and_initialized() -> bool: if not _bench_available(): return False import frappe return bool(getattr(frappe, "local", None) and getattr(frappe.local, "site", None)) def run() -> None: """Main entry point. Delegates to bench or REST client based on context.""" if _in_bench_and_initialized(): _run_in_bench() else: _run_via_rest() _print_readiness_checklist() # ─── bench-native path (preferred) ──────────────────────────────────────── def _run_in_bench() -> None: import frappe logger.info("setup_erpnext_payments: running in-bench") for acct in PAYMENT_GATEWAYS: if frappe.db.exists("Payment Gateway Account", acct.name): logger.info(" • Payment Gateway Account exists: %s", acct.name) continue doc = frappe.get_doc({ "doctype": "Payment Gateway Account", "payment_gateway": acct.gateway, "currency": acct.currency, "payment_channel": acct.payment_channel, "is_default": 0, }) try: doc.insert(ignore_permissions=True) logger.info(" + Created Payment Gateway Account: %s", doc.name) except Exception as exc: logger.error( " ! Could not create %s — %s. The underlying Payment Gateway " "record (%s) must be installed by its Frappe app first.", acct.name, exc, acct.gateway, ) # Crypto Payment Settings (SHKeeper endpoint + API key). shk_url = os.environ.get("SHKEEPER_URL", "https://pay.performancewest.net") shk_key = os.environ.get("SHKEEPER_API_KEY", "") if shk_key and frappe.db.exists("DocType", "Crypto Payment Settings"): try: settings = frappe.get_single("Crypto Payment Settings") settings.shkeeper_url = shk_url settings.shkeeper_api_key = shk_key settings.save(ignore_permissions=True) logger.info(" + Configured Crypto Payment Settings") except Exception as exc: logger.error(" ! Could not save Crypto Payment Settings: %s", exc) elif not shk_key: logger.warning(" · SHKEEPER_API_KEY not set — skipping Crypto Payment Settings") # Sanity check on subscription plans expected_plans = [ ("RA Renewal (Annual — $99)", 99.0), ("RA Renewal Wyoming (Annual — $49)", 49.0), ("Annual Report Filing (Annual — $99)", 99.0), ("CRTC Annual Maintenance ($349)", 349.0), ("US Formation Maintenance Bundle ($179)", 179.0), ("CA Formation Maintenance Bundle ($179)", 179.0), ] for plan_name, expected_cost in expected_plans: found = frappe.db.get_value("Subscription Plan", plan_name, "cost") if found is None: logger.warning(" · Subscription Plan missing: %s", plan_name) elif float(found) != expected_cost: logger.warning( " · Subscription Plan price mismatch: %s is %s (expected %s)", plan_name, found, expected_cost, ) else: logger.info(" • Subscription Plan present: %s ($%s)", plan_name, expected_cost) # ─── REST fallback path ─────────────────────────────────────────────────── def _run_via_rest() -> None: logger.info("setup_erpnext_payments: running via REST (frappe-client)") base = os.environ.get("ERPNEXT_URL") key = os.environ.get("ERPNEXT_API_KEY") secret = os.environ.get("ERPNEXT_API_SECRET") if not base or not key or not secret: logger.error( "Not running inside a bench and ERPNEXT_URL / ERPNEXT_API_KEY / " "ERPNEXT_API_SECRET are not all set. Set them and re-run, or run " "via `bench --site execute scripts.setup_erpnext_payments.run`." ) sys.exit(1) import urllib.request import urllib.error import json headers = { "Authorization": f"token {key}:{secret}", "Content-Type": "application/json", } for acct in PAYMENT_GATEWAYS: # Existence check url = f"{base.rstrip('/')}/api/resource/Payment Gateway Account/{acct.name}" req = urllib.request.Request(url, headers=headers) try: urllib.request.urlopen(req, timeout=10).read() logger.info(" • Payment Gateway Account exists: %s", acct.name) continue except urllib.error.HTTPError as err: if err.code != 404: logger.error(" ! Unexpected error looking up %s: %s", acct.name, err) continue # Create create_url = f"{base.rstrip('/')}/api/resource/Payment Gateway Account" body = json.dumps({ "payment_gateway": acct.gateway, "currency": acct.currency, "payment_channel": acct.payment_channel, }).encode() try: urllib.request.urlopen( urllib.request.Request(create_url, data=body, headers=headers, method="POST"), timeout=15, ) logger.info(" + Created Payment Gateway Account: %s", acct.name) except Exception as exc: logger.error(" ! Could not create %s: %s", acct.name, exc) # ─── Readiness checklist ────────────────────────────────────────────────── def _print_readiness_checklist() -> None: checklist = [ ("STRIPE_SECRET_KEY", os.environ.get("STRIPE_SECRET_KEY")), ("STRIPE_WEBHOOK_SECRET", os.environ.get("STRIPE_WEBHOOK_SECRET")), ("STRIPE_IDENTITY_WEBHOOK_SECRET", os.environ.get("STRIPE_IDENTITY_WEBHOOK_SECRET")), ("PAYPAL_CLIENT_ID", os.environ.get("PAYPAL_CLIENT_ID")), ("PAYPAL_CLIENT_SECRET", os.environ.get("PAYPAL_CLIENT_SECRET")), ("SHKEEPER_API_KEY", os.environ.get("SHKEEPER_API_KEY")), ("CUSTOMER_JWT_SECRET", os.environ.get("CUSTOMER_JWT_SECRET")), ] print("\n=== Payment readiness checklist ===") for name, value in checklist: mark = "✓" if value else "✗" print(f" [{mark}] {name}") print( "\nRemaining human steps:" "\n • Register Stripe webhook at " "https://api.performancewest.net/api/v1/webhooks/stripe covering " "checkout.session.completed, payment_intent.succeeded, balance.available, " "identity.verification_session.verified" "\n • Verify Subscription Plan items (RA-RENEWAL, ANNUAL-REPORT, " "CRTC-MAINT-ANNUAL, …) exist and are priced correctly." ) if __name__ == "__main__": logging.basicConfig(level=logging.INFO, format="%(message)s") run()