f5b627256f
Full backend implementation with custom Odoo modules: - encoach_api: Core API, user management, JWT auth - encoach_exam: Exam generation (reading, writing, listening, speaking) - encoach_evaluate: AI-powered evaluation (writing, speaking) - encoach_training: Training tips and walkthrough - encoach_storage: File storage management - encoach_payment: Stripe, PayPal, Paymob integration - encoach_mail: Email notifications Made-with: Cursor
134 lines
4.7 KiB
Python
134 lines
4.7 KiB
Python
import hashlib
|
|
import hmac
|
|
import json
|
|
import logging
|
|
import time
|
|
|
|
import httpx
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
STRIPE_API_BASE = "https://api.stripe.com/v1"
|
|
|
|
|
|
class EncoachStripeService:
|
|
|
|
def __init__(self, env):
|
|
self.env = env
|
|
params = env["ir.config_parameter"].sudo()
|
|
self.api_key = params.get_param("encoach.stripe_secret_key", "")
|
|
self.webhook_secret = params.get_param("encoach.stripe_webhook_secret", "")
|
|
|
|
def _headers(self):
|
|
return {
|
|
"Authorization": f"Bearer {self.api_key}",
|
|
"Content-Type": "application/x-www-form-urlencoded",
|
|
}
|
|
|
|
def create_checkout_session(self, user, package, success_url, cancel_url, discount_code=None):
|
|
if not self.api_key:
|
|
return {"error": "Stripe secret key not configured"}
|
|
|
|
amount = package.price
|
|
if discount_code:
|
|
discount = self.env["encoach.discount"].sudo().search([
|
|
("code", "=", discount_code),
|
|
], limit=1)
|
|
if discount and discount.is_valid():
|
|
amount = amount * (1 - discount.percentage / 100.0)
|
|
|
|
amount_cents = int(round(amount * 100))
|
|
|
|
data = {
|
|
"mode": "payment",
|
|
"line_items[0][price_data][currency]": (package.currency or "usd").lower(),
|
|
"line_items[0][price_data][unit_amount]": str(amount_cents),
|
|
"line_items[0][price_data][product_data][name]": package.name,
|
|
"line_items[0][quantity]": "1",
|
|
"success_url": success_url,
|
|
"cancel_url": cancel_url,
|
|
"client_reference_id": str(user.id),
|
|
"customer_email": user.login,
|
|
"metadata[package_id]": str(package.id),
|
|
"metadata[user_id]": str(user.id),
|
|
}
|
|
|
|
try:
|
|
resp = httpx.post(
|
|
f"{STRIPE_API_BASE}/checkout/sessions",
|
|
headers=self._headers(),
|
|
data=data,
|
|
timeout=30,
|
|
)
|
|
resp.raise_for_status()
|
|
session = resp.json()
|
|
return {"sessionId": session["id"], "url": session["url"]}
|
|
except httpx.HTTPStatusError as e:
|
|
_logger.error("Stripe API error: %s", e.response.text)
|
|
return {"error": f"Stripe error: {e.response.status_code}"}
|
|
except Exception as e:
|
|
_logger.exception("Stripe checkout error")
|
|
return {"error": str(e)}
|
|
|
|
def verify_webhook_signature(self, payload, sig_header):
|
|
if not self.webhook_secret:
|
|
_logger.warning("No Stripe webhook secret configured")
|
|
return False
|
|
|
|
try:
|
|
elements = {k: v for k, v in (
|
|
item.split("=", 1) for item in sig_header.split(",")
|
|
)}
|
|
timestamp = elements.get("t", "")
|
|
signature = elements.get("v1", "")
|
|
|
|
signed_payload = f"{timestamp}.{payload}"
|
|
expected = hmac.new(
|
|
self.webhook_secret.encode("utf-8"),
|
|
signed_payload.encode("utf-8"),
|
|
hashlib.sha256,
|
|
).hexdigest()
|
|
|
|
if not hmac.compare_digest(expected, signature):
|
|
return False
|
|
|
|
if abs(time.time() - int(timestamp)) > 300:
|
|
_logger.warning("Stripe webhook timestamp too old")
|
|
return False
|
|
|
|
return True
|
|
except Exception:
|
|
_logger.exception("Stripe webhook signature verification failed")
|
|
return False
|
|
|
|
def handle_webhook(self, payload, sig_header):
|
|
if not self.verify_webhook_signature(payload, sig_header):
|
|
return {"error": "Invalid signature"}
|
|
|
|
event = json.loads(payload)
|
|
event_type = event.get("type")
|
|
|
|
if event_type == "checkout.session.completed":
|
|
session = event["data"]["object"]
|
|
user_id = int(session.get("client_reference_id", 0))
|
|
package_id = int(session.get("metadata", {}).get("package_id", 0))
|
|
transaction_id = session.get("payment_intent") or session.get("id")
|
|
|
|
if user_id and package_id:
|
|
self._complete_payment(
|
|
user_id=user_id,
|
|
package_id=package_id,
|
|
provider="stripe",
|
|
transaction_id=transaction_id,
|
|
amount=session.get("amount_total", 0) / 100.0,
|
|
currency=(session.get("currency") or "usd").upper(),
|
|
)
|
|
|
|
return {"ok": True}
|
|
|
|
def _complete_payment(self, user_id, package_id, provider, transaction_id, amount, currency):
|
|
from .subscription_helper import extend_subscription
|
|
extend_subscription(
|
|
self.env, user_id, package_id, provider, transaction_id, amount, currency,
|
|
)
|