# AUTOGENERATED! DO NOT EDIT! File to edit: Stripe.ipynb.

# %% auto 0
__all__ = ['DOMAIN_URL', 'app_nm', 'price_list', 'price', 'bware', 'app', 'rt', 'WEBHOOK_SECRET', 'db', 'payments',
           'create_price', 'archive_price', 'before', 'home', 'create_checkout_session', 'Payment', 'post', 'success',
           'cancel', 'refund', 'account_management']

# %% Stripe.ipynb
from fasthtml.common import *
import os

# %% Stripe.ipynb
import stripe

# %% Stripe.ipynb
stripe.api_key = os.environ.get("STRIPE_SECRET_KEY")
DOMAIN_URL = os.environ.get("DOMAIN_URL", "http://localhost:5001")

# %% Stripe.ipynb
def _search_app(app_nm:str, limit=1): 
    "Checks for product based on app_nm and returns the product if it exists"
    return stripe.Product.search(query=f"name:'{app_nm}' AND active:'True'", limit=limit).data

def create_price(app_nm:str, amt:int, currency="usd") -> list[stripe.Price]:
    "Create a product and bind it to a price object. If product already exist just return the price list."
    existing_product = _search_app(app_nm)
    if existing_product: 
        return stripe.Price.list(product=existing_product[0].id).data
    else:
        product = stripe.Product.create(name=f"{app_nm}")
        return [stripe.Price.create(product=product.id, unit_amount=amt, currency=currency)]

def archive_price(app_nm:str):
    "Archive a price - useful for cleanup if testing."
    existing_products = _search_app(app_nm, limit=50)
    for product in existing_products:
        for price in stripe.Price.list(product=product.id).data: 
            stripe.Price.modify(price.id, active=False)
        stripe.Product.modify(product.id, active=False)

# %% Stripe.ipynb
app_nm = "[FastHTML Docs] Demo Product"
price_list = create_price(app_nm, amt=1999)
assert len(price_list) == 1, 'For this tutorial, we only have one price bound to our product.'
price = price_list[0]

# %% Stripe.ipynb
def before(sess): sess['auth'] = 'hamel@hamel.com'
bware = Beforeware(before, skip=['/webhook'])
app, rt = fast_app(before=bware)

# %% Stripe.ipynb
WEBHOOK_SECRET = os.getenv("STRIPE_LOCAL_TEST_WEBHOOK_SECRET")

# %% Stripe.ipynb
@rt("/")
def home(sess):
    auth = sess['auth']
    return Titled(
        "Buy Now", 
        Div(H2("Demo Product - $19.99"),
            P(f"Welcome, {auth}"),
            Button("Buy Now", hx_post="/create-checkout-session", hx_swap="none"),
            A("View Account", href="/account")))

# %% Stripe.ipynb
@rt("/create-checkout-session", methods=["POST"])
async def create_checkout_session(sess):
    checkout_session = stripe.checkout.Session.create(
        line_items=[{'price': price.id, 'quantity': 1}],
        mode='payment',
        payment_method_types=['card'],
        customer_email=sess['auth'],
        metadata={'app_name': app_nm, 
                  'AnyOther': 'Metadata',},
        # CHECKOUT_SESSION_ID is a special variable Stripe fills in for you
        success_url=DOMAIN_URL + '/success?checkout_sid={CHECKOUT_SESSION_ID}',
        cancel_url=DOMAIN_URL + '/cancel')
    return Redirect(checkout_session.url)

# %% Stripe.ipynb
# Database Table
class Payment:
    checkout_session_id: str  # Stripe checkout session ID (primary key)
    email: str
    amount: int  # Amount paid in cents
    payment_status: str  # paid, pending, failed
    created_at: int # Unix timestamp
    metadata: str  # Additional payment metadata as JSON

# %% Stripe.ipynb
db = Database("stripe_payments.db")
payments = db.create(Payment, pk='checkout_session_id', transform=True)

# %% Stripe.ipynb
@rt("/webhook")
async def post(req):
    payload = await req.body()
    # Verify the event came from Stripe
    try:
        event = stripe.Webhook.construct_event(
            payload, req.headers.get("stripe-signature"), WEBHOOK_SECRET)
    except Exception as e:
        print(f"Webhook error: {e}")
        return
    if event and event.type == "checkout.session.completed":
        event_data = event.data.object
        if event_data.metadata.get('app_name') == app_nm:
            payment = Payment(
                checkout_session_id=event_data.id,
                email=event_data.customer_email,
                amount=event_data.amount_total,
                payment_status=event_data.payment_status,
                created_at=event_data.created,
                metadata=str(event_data.metadata))
            payments.insert(payment)
            print(f"Payment recorded for user: {event_data.customer_email}")
            
    # Do not worry about refunds yet, we will cover how to do this later in the tutorial
    elif event and event.type == "charge.refunded":
        event_data = event.data.object
        payment_intent_id = event_data.payment_intent
        sessions = stripe.checkout.Session.list(payment_intent=payment_intent_id)
        if sessions and sessions.data:
            checkout_sid = sessions.data[0].id
            payments.update(Payment(checkout_session_id= checkout_sid, payment_status="refunded"))
            print(f"Refund recorded for payment: {checkout_sid}")

# %% Stripe.ipynb
@rt("/success")
def success(sess, checkout_sid:str):    
    # Get payment record from database (saved in the webhook)
    payment = payments[checkout_sid]

    if not payment or payment.payment_status != 'paid': 
        return Titled("Error", P("Payment not found"))

    return Titled(
        "Success",
        Div(H2("Payment Successful!"),
            P(f"Thank you for your purchase, {sess['auth']}"),
            P(f"Amount Paid: ${payment.amount / 100:.2f}"),
            P(f"Status: {payment.payment_status}"),
            P(f"Transaction ID: {payment.checkout_session_id}"),
            A("Back to Home", href="/")))

# %% Stripe.ipynb
@rt("/cancel")
def cancel():
    return Titled(
        "Cancelled",
        Div(H2("Payment Cancelled"),
            P("Your payment was cancelled."),
            A("Back to Home", href="/")))

# %% Stripe.ipynb
@rt("/refund", methods=["POST"])
async def refund(sess, checkout_sid:str):
    # Get payment record from database
    payment = payments[checkout_sid]
    
    if not payment or payment.payment_status != 'paid':
        return P("Error: Payment not found or not eligible for refund")
    
    try:
        # Get the payment intent ID from the checkout session
        checkout_session = stripe.checkout.Session.retrieve(checkout_sid)
        
        # Process the refund
        refund = stripe.Refund.create(payment_intent=checkout_session.payment_intent, reason="requested_by_customer")
        
        # Update payment status in database
        payments.update(Payment(checkout_session_id= checkout_sid, payment_status="refunded"))
        
        return Div(
            P("Refund processed successfully!"),
            P(f"Refund ID: {refund.id}"))
    
    except Exception as e: return P(f"Refund failed: {str(e)}")

# %% Stripe.ipynb
@rt("/account")
def account_management(sess):
    user_email = sess['auth']
    user_payments = payments("email=?", (user_email,))
    # Create table rows for each payment
    payment_rows = []
    for payment in user_payments:
        action_button = ""
        if payment.payment_status == 'paid':
            action_button = Button("Request Refund", hx_post=f"/refund?checkout_sid={payment.checkout_session_id}",hx_target="#refund-status")
        elif payment.payment_status == 'refunded': action_button = "Refunded"
        
        # Add row to table
        payment_rows.append(
            Tr(*map(Td, (payment.created_at, payment.amount, payment.payment_status, action_button))))
    
    # Create payment history table
    payment_history = Table(
        Thead(Tr(*map(Th, ("Date", "Amount", "Status", "Action")))),
        Tbody(*payment_rows))
    
    return Titled(
        "Account Management",
        Div(H2(f"Account: {user_email}"),
            H3("Payment History"),
            payment_history,
            Div(id="refund-status"),  # Target for refund status messages
            A("Back to Home", href="/")))

# %% Stripe.ipynb
serve()
