Spaces:
Sleeping
Sleeping
| """ | |
| PNA Assistant — Stripe Webhook Handler | |
| POST /webhook → Stripe sends checkout.session.completed events here | |
| → We generate a PNA-PRO-XXXXX code and email it to the customer | |
| Run standalone: uvicorn webhook:app --host 0.0.0.0 --port 8080 | |
| In Docker: separate service in docker-compose.yml | |
| """ | |
| import hashlib | |
| import hmac | |
| import json | |
| import logging | |
| import os | |
| import random | |
| import string | |
| from datetime import datetime, timezone | |
| from pathlib import Path | |
| import boto3 | |
| from botocore.exceptions import ClientError | |
| from fastapi import FastAPI, HTTPException, Request | |
| from fastapi.responses import JSONResponse | |
| # ─── Config ────────────────────────────────────────────────────────────────── | |
| STRIPE_WEBHOOK_SECRET = os.getenv("STRIPE_WEBHOOK_SECRET", "") | |
| AWS_REGION = os.getenv("AWS_REGION", "eu-west-2") | |
| SES_FROM_EMAIL = os.getenv("SES_FROM_EMAIL", "lincoln@clinyqai.com") | |
| # Where we persist issued codes (append-only JSONL) | |
| CODES_LOG = Path("/app/codes.jsonl") | |
| logging.basicConfig(level=logging.INFO) | |
| log = logging.getLogger("pna-webhook") | |
| app = FastAPI(title="PNA Assistant Webhook", docs_url=None, redoc_url=None) | |
| # ─── Helpers ───────────────────────────────────────────────────────────────── | |
| def _generate_code(tier: str = "PRO") -> str: | |
| """Generate a unique activation code e.g. PNA-PRO-A7K2M9""" | |
| suffix = "".join(random.choices(string.ascii_uppercase + string.digits, k=6)) | |
| return f"PNA-{tier}-{suffix}" | |
| def _verify_stripe_signature(payload: bytes, sig_header: str, secret: str) -> bool: | |
| """ | |
| Validate Stripe-Signature header. | |
| Stripe uses HMAC-SHA256 with the raw request body. | |
| """ | |
| if not secret: | |
| log.warning("STRIPE_WEBHOOK_SECRET not set — skipping signature verification") | |
| return True # dev mode only; set the secret in production | |
| try: | |
| parts = {k: v for part in sig_header.split(",") for k, v in [part.split("=", 1)]} | |
| timestamp = parts.get("t", "") | |
| v1_sig = parts.get("v1", "") | |
| signed_payload = f"{timestamp}.".encode() + payload | |
| expected = hmac.new( | |
| secret.encode(), | |
| signed_payload, | |
| hashlib.sha256, | |
| ).hexdigest() | |
| return hmac.compare_digest(expected, v1_sig) | |
| except Exception as exc: | |
| log.error("Signature verification error: %s", exc) | |
| return False | |
| def _log_code(email: str, code: str, plan: str, session_id: str) -> None: | |
| """Append issued code to local JSONL log for admin audit trail.""" | |
| CODES_LOG.parent.mkdir(parents=True, exist_ok=True) | |
| record = { | |
| "ts": datetime.now(timezone.utc).isoformat(), | |
| "email": email, | |
| "code": code, | |
| "plan": plan, | |
| "session_id": session_id, | |
| } | |
| with CODES_LOG.open("a") as f: | |
| f.write(json.dumps(record) + "\n") | |
| log.info("Code issued: %s → %s (%s)", email, code, plan) | |
| def _send_activation_email(email: str, code: str, plan: str) -> bool: | |
| """ | |
| Send activation code to customer via Amazon SES. | |
| Returns True if sent, False on error. | |
| """ | |
| label = "Pro" if plan == "PRO" else "Institution" | |
| price = "£9.99/month" if plan == "PRO" else "£99/month" | |
| body_html = f""" | |
| <!DOCTYPE html> | |
| <html> | |
| <body style="font-family:Inter,sans-serif;max-width:600px;margin:0 auto;padding:2rem;color:#1e293b"> | |
| <div style="background:linear-gradient(135deg,#1a2460,#0d9488);padding:2rem;border-radius:12px;margin-bottom:2rem"> | |
| <h1 style="color:white;font-size:1.5rem;margin:0">👨🏾⚕️ PNA Assistant</h1> | |
| <p style="color:rgba(255,255,255,0.85);margin:0.5rem 0 0"> | |
| Professional Nurse Advocate AI — {label} Plan | |
| </p> | |
| </div> | |
| <p>Thank you for subscribing to PNA Assistant <strong>{label}</strong> ({price}).</p> | |
| <p>Your activation code is:</p> | |
| <div style="background:#f0fdf4;border:2px solid #0d9488;border-radius:12px;padding:1.5rem;text-align:center;margin:1.5rem 0"> | |
| <span style="font-family:monospace;font-size:1.75rem;font-weight:700;color:#0d9488;letter-spacing:0.1em"> | |
| {code} | |
| </span> | |
| </div> | |
| <p><strong>How to activate:</strong></p> | |
| <ol> | |
| <li>Go to <a href="https://pna.nursingcitizendevelopment.com">pna.nursingcitizendevelopment.com</a></li> | |
| <li>Open the sidebar and find <em>Subscription</em></li> | |
| <li>Paste your code in the <em>Activation code</em> box</li> | |
| <li>Click Enter — your {label} features will unlock immediately</li> | |
| </ol> | |
| <p>Your {label} plan includes:</p> | |
| <ul> | |
| <li>📄 Download supervision notes (.docx)</li> | |
| <li>📋 NMC revalidation CPD log export (.docx)</li> | |
| <li>💾 Save and resume sessions</li> | |
| <li>📧 Monthly PNA newsletter</li> | |
| {"<li>🏫 Multi-user access</li><li>🎨 Custom branding</li>" if plan == "INST" else ""} | |
| </ul> | |
| <p style="font-size:0.875rem;color:#64748b"> | |
| If you have any questions, reply to this email or contact | |
| <a href="mailto:lincoln@clinyqai.com">lincoln@clinyqai.com</a>.<br><br> | |
| This tool is for educational purposes only and does not replace clinical supervision | |
| or your employer's formal governance processes.<br><br> | |
| Contains public sector information licensed under the | |
| <a href="https://www.nationalarchives.gov.uk/doc/open-government-licence/version/3/"> | |
| Open Government Licence v3.0 | |
| </a> — NHS England. | |
| </p> | |
| </body> | |
| </html> | |
| """ | |
| body_text = ( | |
| f"Thank you for subscribing to PNA Assistant {label} ({price}).\n\n" | |
| f"Your activation code is: {code}\n\n" | |
| "To activate:\n" | |
| "1. Go to pna.nursingcitizendevelopment.com\n" | |
| "2. Open the sidebar → Subscription\n" | |
| "3. Paste your code in the Activation code box\n\n" | |
| "Questions? Email lincoln@clinyqai.com" | |
| ) | |
| try: | |
| ses = boto3.client("ses", region_name=AWS_REGION) | |
| ses.send_email( | |
| Source = f"PNA Assistant <{SES_FROM_EMAIL}>", | |
| Destination = {"ToAddresses": [email]}, | |
| Message = { | |
| "Subject": {"Data": f"Your PNA Assistant {label} activation code 🎉"}, | |
| "Body": { | |
| "Text": {"Data": body_text}, | |
| "Html": {"Data": body_html}, | |
| }, | |
| }, | |
| ) | |
| log.info("Activation email sent to %s", email) | |
| return True | |
| except ClientError as exc: | |
| log.error("SES send failed: %s", exc.response["Error"]["Message"]) | |
| return False | |
| except Exception as exc: | |
| log.error("Email error: %s", exc) | |
| return False | |
| # ─── Routes ────────────────────────────────────────────────────────────────── | |
| async def health(): | |
| return {"status": "ok"} | |
| async def stripe_webhook(request: Request): | |
| payload = await request.body() | |
| sig_header = request.headers.get("Stripe-Signature", "") | |
| # 1. Verify Stripe signature | |
| if not _verify_stripe_signature(payload, sig_header, STRIPE_WEBHOOK_SECRET): | |
| log.warning("Invalid Stripe signature — rejected") | |
| raise HTTPException(status_code=400, detail="Invalid signature") | |
| # 2. Parse event | |
| try: | |
| event = json.loads(payload) | |
| except json.JSONDecodeError: | |
| raise HTTPException(status_code=400, detail="Invalid JSON") | |
| event_type = event.get("type", "") | |
| log.info("Received Stripe event: %s", event_type) | |
| # 3. Handle checkout completion | |
| if event_type == "checkout.session.completed": | |
| session = event["data"]["object"] | |
| email = session.get("customer_details", {}).get("email") or session.get("customer_email") | |
| amount = session.get("amount_total", 0) # pence/cents | |
| session_id = session.get("id", "unknown") | |
| if not email: | |
| log.warning("No email in session %s — cannot send code", session_id) | |
| return JSONResponse({"received": True, "warning": "no_email"}) | |
| # Determine tier from amount (999 = £9.99, 9900 = £99.00) | |
| tier = "INST" if amount >= 9900 else "PRO" | |
| code = _generate_code(tier) | |
| # 4. Log the code (always) | |
| _log_code(email, code, tier, session_id) | |
| # 5. Email the customer | |
| sent = _send_activation_email(email, code, tier) | |
| if not sent: | |
| log.warning( | |
| "SES send failed for %s — code %s logged to %s", | |
| email, code, CODES_LOG, | |
| ) | |
| return JSONResponse({"received": True}) | |