119 lines
No EOL
5.3 KiB
Python
119 lines
No EOL
5.3 KiB
Python
import frappe
|
|
import stripe
|
|
import json
|
|
from custom_ui.services import DbService
|
|
from frappe.utils import get_url
|
|
|
|
class StripeService:
|
|
|
|
@staticmethod
|
|
def get_stripe_settings(company: str):
|
|
"""Fetch Stripe settings for a given company."""
|
|
settings_name = frappe.get_all("Stripe Settings", pluck="name", filters={"company": company})
|
|
if not settings_name:
|
|
frappe.throw(f"Stripe Settings not found for company: {company}")
|
|
settings = frappe.get_doc("Stripe Settings", settings_name[0]) if settings_name else None
|
|
return settings
|
|
|
|
@staticmethod
|
|
def get_api_key(company: str) -> str:
|
|
"""Retrieve the Stripe API key for the specified company."""
|
|
settings = StripeService.get_stripe_settings(company)
|
|
return settings.get_password("secret_key")
|
|
|
|
@staticmethod
|
|
def get_webhook_secret(company: str) -> str:
|
|
"""Retrieve the Stripe webhook secret for the specified company."""
|
|
settings = StripeService.get_stripe_settings(company)
|
|
if not settings.webhook_secret:
|
|
frappe.throw(f"Stripe Webhook Secret not configured for company: {company}")
|
|
return settings.webhook_secret
|
|
|
|
@staticmethod
|
|
def create_checkout_session(company: str, amount: float, service: str, order_num: str, currency: str = "usd", for_advance_payment: bool = False, line_items: list | None = None) -> stripe.checkout.Session:
|
|
"""Create a Stripe Checkout Session. order_num is a Sales Order name if for_advance_payment is True, otherwise it is a Sales Invoice name."""
|
|
stripe.api_key = StripeService.get_api_key(company)
|
|
|
|
line_items = line_items if line_items and not for_advance_payment else [{
|
|
"price_data": {
|
|
"currency": currency.lower(),
|
|
"product_data": {
|
|
"name": f"Advance payment for {company}{' - ' + service if service else ''}"
|
|
},
|
|
"unit_amount": int(amount * 100),
|
|
},
|
|
"quantity": 1,
|
|
}]
|
|
|
|
session = stripe.checkout.Session.create(
|
|
mode="payment",
|
|
payment_method_types=["card"],
|
|
line_items=line_items,
|
|
metadata={
|
|
"order_num": order_num,
|
|
"company": company,
|
|
"payment_type": "advance" if for_advance_payment else "full"
|
|
},
|
|
success_url=f"{get_url()}/payment_success?session_id={{CHECKOUT_SESSION_ID}}",
|
|
cancel_url=f"{get_url()}/payment_cancelled",
|
|
)
|
|
|
|
return session
|
|
|
|
@staticmethod
|
|
def get_event(payload: bytes, sig_header: str, company: str = None) -> stripe.Event:
|
|
print("DEBUG: Stripe webhook received")
|
|
print(f"DEBUG: Signature header present: {bool(sig_header)}")
|
|
|
|
# If company not provided, try to extract from payload metadata
|
|
if not company:
|
|
try:
|
|
payload_dict = json.loads(payload)
|
|
print(f"DEBUG: Parsed payload type: {payload_dict.get('type')}")
|
|
|
|
metadata = payload_dict.get("data", {}).get("object", {}).get("metadata", {})
|
|
print(f"DEBUG: Metadata from payload: {metadata}")
|
|
|
|
company = metadata.get("company")
|
|
print(f"DEBUG: Extracted company from metadata: {company}")
|
|
except (json.JSONDecodeError, KeyError, AttributeError) as e:
|
|
print(f"DEBUG: Failed to parse payload: {str(e)}")
|
|
|
|
# If we still don't have a company, reject the webhook
|
|
if not company:
|
|
print("ERROR: Company information missing in webhook payload")
|
|
frappe.throw("Company information missing in webhook payload.")
|
|
|
|
print(f"DEBUG: Validating webhook signature for company: {company}")
|
|
|
|
# Validate webhook signature with the specified company's secret
|
|
try:
|
|
event = stripe.Webhook.construct_event(
|
|
payload=payload,
|
|
sig_header=sig_header,
|
|
secret=StripeService.get_webhook_secret(company)
|
|
)
|
|
print(f"DEBUG: Webhook signature validated successfully for company: {company}")
|
|
print(f"DEBUG: Event type: {event.type}")
|
|
print(f"DEBUG: Event ID: {event.id}")
|
|
except ValueError as e:
|
|
print(f"ERROR: Invalid payload: {str(e)}")
|
|
frappe.throw(f"Invalid payload: {str(e)}")
|
|
except stripe.error.SignatureVerificationError as e:
|
|
print(f"ERROR: Invalid signature for company {company}: {str(e)}")
|
|
frappe.throw(f"Invalid signature for company {company}: {str(e)}")
|
|
|
|
return event
|
|
|
|
@staticmethod
|
|
def get_session_and_metadata(payload: bytes, sig_header: str, company: str = None) -> tuple[stripe.checkout.Session, dict]:
|
|
"""Retrieve the Stripe Checkout Session and its metadata from a webhook payload."""
|
|
event = StripeService.get_event(payload, sig_header, company)
|
|
if event.type != "checkout.session.completed":
|
|
frappe.throw(f"Unhandled event type: {event.type}")
|
|
session = event.data.object
|
|
metadata = session["metadata"] if "metadata" in session else {}
|
|
|
|
return session, metadata
|
|
|
|
|