import json import pprint import time import uuid from datetime import datetime from typing import Any from xmlrpc.client import DateTime import stripe from sqlalchemy import func from sqlalchemy.orm import joinedload from core.config import get_db, settings from core.email.config import SendEnrollmentEmail from core.security.exceptions import GenericError from core.utility import send_email,send_order_summary_email from models import Cart, Course, BursaryDiscountApplication, Participant, BursaryCourseAllocation, Payment, \ Parent, BursaryDiscountInfo, WalletHistory, PaymentDetails, Attendance from models.discount_coupon import DiscountCoupon from schema.payment import PaymentDetailsBase, CartBaseSchema class PaymentService: currency = "gbp" def add_to_cart(self, request, user): cart_items = [] discount_amount = 0 with get_db() as db: db.begin() try: for row in request.participant_id: course_instance = db.query(Course).get(request.course_id) discount_query = (db.query(BursaryCourseAllocation) .join(BursaryDiscountApplication, BursaryCourseAllocation.bursary_discount_application_id == BursaryDiscountApplication.id).filter( BursaryDiscountApplication.is_approved == True, BursaryDiscountApplication.participant_id == row, BursaryDiscountApplication.academic_year_id == course_instance.academic_year_id, BursaryCourseAllocation.is_used == False, BursaryCourseAllocation.course_id == request.course_id)) if discount_query.count() > 0: discount_instance = discount_query.first() if db.query(BursaryDiscountInfo).filter(BursaryDiscountInfo.end_date >= datetime.today().date(), BursaryDiscountInfo.start_date <= datetime.today().date(), BursaryDiscountInfo.academic_year_id == discount_instance.application.academic_year_id).count() > 0: discount_amount = discount_instance.discount_price course_amount=self.get_course_amount(db,course_instance) final_amount = (float(course_amount) - float(discount_amount)) cart_items.append({ "cart_line_item_id": str(uuid.uuid4()), "cart_owner": user['id'], "user_type": user['user_type'], "parent_id": user['id'] if user['user_type'] == 'parent' else None, "participant_id": row, "course_id": request.course_id, "course_original_amount": course_instance.amount, "discount_amount": discount_amount, "final_amount": final_amount, "is_discount_applied": True if discount_amount > 0 else False, }) db.bulk_insert_mappings(Cart, cart_items) db.commit() except Exception as e: db.rollback() print(e) raise GenericError(status_code=422, exc="Could not add to cart") def get_course_amount(self,db,course): course_amount = course.amount if course.start_date < datetime.now().date(): no_of_attendance = db.query(Attendance.attendance_date).filter(Attendance.course_id == course.id).distinct().count() print(no_of_attendance) if all([ no_of_attendance != 0, course.no_of_classes is not None, course.rate_per_class is not None ]): remaining_class = course.no_of_classes - no_of_attendance course_amount = max(remaining_class, 0) * course.rate_per_class return course_amount def list_cart_items(self, user): with get_db() as db: cart_data = [] participant_list_query = db.query(Cart.participant_id).distinct().filter(Cart.cart_owner == user['id'], Cart.is_cart_processed == False).all() participant_list = [row[0] for row in participant_list_query] for participant_id in participant_list: participant = db.query(Participant).get(participant_id) cart_item_query = db.query(Cart).filter(Cart.cart_owner == user['id'], Cart.participant_id == participant_id, Cart.is_cart_processed == False).all() course_details = [CartBaseSchema.model_validate(item).model_dump() for item in cart_item_query] cart_data.append({ "participant_id": participant_id, "participant_first_name": participant.first_name, "participant_last_name": participant.last_name, "total_amount": sum(cart_row.get("final_amount", 0) or 0 for cart_row in course_details), "course_details": course_details, }) return cart_data def remove_cart_item(self, cart_line_item_id, user): with get_db() as db: db.query(Cart).filter(Cart.cart_line_item_id == cart_line_item_id, Cart.cart_owner == user['id']).delete() db.commit() def generate_transaction_id(self, user_id): # Get current time in milliseconds, take last 8 digits timestamp = int(time.time() * 1000) # Milliseconds base_id = str(timestamp)[-8:] # Last 8 digits return f"{base_id}-{user_id}-lwxbu" def fetch_wallet_amount(self, request, user): wallet_amount = 0 if request.use_wallet: with get_db() as db: if user['user_type'] == 'parent': wallet_amount = db.query(Parent).get(user['id']).wallet_amount elif user['user_type'] == 'participant': wallet_amount = db.query(Participant).filter(Participant.id == user['id'], Participant.parent_id.is_(None)).first().wallet_amount return wallet_amount def apply_discount(self, cart_total_amount, discount_coupon_instance, user_id): if discount_coupon_instance: # First check if customer has exceeded usage limit with get_db() as db: usage_count = db.query(Payment).filter( Payment.discount_coupon_id == discount_coupon_instance.id, Payment.created_by == user_id, Payment.is_payment_completed==True ).count() if usage_count >= discount_coupon_instance.uses_per_customer: return float(cart_total_amount), 0 calculated_discount = 0 if discount_coupon_instance.rule_type: if float(cart_total_amount) >= discount_coupon_instance.min_cart_value: if discount_coupon_instance.discount_type == 'percentage': calculated_discount = float(cart_total_amount) * float(discount_coupon_instance.discount_value) / 100 else: calculated_discount = float(discount_coupon_instance.discount_value) else: if discount_coupon_instance.discount_type == 'percentage': calculated_discount = float(cart_total_amount) * float(discount_coupon_instance.discount_value) / 100 else: calculated_discount = float(discount_coupon_instance.discount_value) if hasattr(discount_coupon_instance, 'max_discount') and discount_coupon_instance.max_discount: calculated_discount = min(calculated_discount, float(discount_coupon_instance.max_discount)) # Cap the discount at the cart total amount to prevent negative values calculated_discount = min(calculated_discount, float(cart_total_amount)) cart_total_amount = max(0, float(cart_total_amount) - calculated_discount) return float(cart_total_amount), calculated_discount def create_checkout_session(self, request, user): session_url = None session_id = None payment_status = "pending" wallet_amount=0 used_wallet_amount=0 paid_on=None is_payment_completed = False discount_coupon_instance=None calculated_coupon_discount=0 print("session request data",request) with get_db() as db: try: # Fetch cart items cart_items = db.query(Cart).filter(Cart.cart_owner == user['id'], Cart.is_cart_processed == False) if cart_items.count() == 0: raise GenericError(status_code=404, exc="Cart is empty") # validating the discount coupon code if request.discount_coupon_code and not any(item.discount_amount > 0 for item in cart_items): discount_coupon_instance = db.query(DiscountCoupon).filter( DiscountCoupon.code == request.discount_coupon_code, DiscountCoupon.status==True, DiscountCoupon.start_date<=datetime.now().date(), DiscountCoupon.end_date>=datetime.now().date(), DiscountCoupon.is_applicable_for_course==True, ).first() if not discount_coupon_instance: raise GenericError(status_code=404, exc="Discount coupon is not valid or not applicable to this cart") # Creating batch id & calculate sum of total from cart transaction_batch_id = self.generate_transaction_id(user_id=user['id']) cart_items = cart_items.all() cart_total_amount = sum([item.final_amount for item in cart_items]) print("cart_total_amount",cart_total_amount) if discount_coupon_instance and not any(item.discount_amount > 0 for item in cart_items): cart_total_amount,calculated_coupon_discount = self.apply_discount(cart_total_amount, discount_coupon_instance,user['id']) if request.charity: # add 5% to cart_total_amount cart_total_amount = cart_total_amount + (cart_total_amount * 0.05) print("cart_total_amount",cart_total_amount) if request.use_wallet and cart_total_amount > 0: wallet_amount = max(0.0, float(self.fetch_wallet_amount(request, user))) # Ensure non-negative if wallet_amount < cart_total_amount: used_wallet_amount = wallet_amount cart_total_amount = cart_total_amount - wallet_amount wallet_amount = 0 else: used_wallet_amount = cart_total_amount wallet_amount = wallet_amount - cart_total_amount cart_total_amount = 0 cart_total_amount = round(cart_total_amount, 2) print("cart_total_amount",cart_total_amount) if cart_total_amount > 0: session = self.generate_link(cart_items, cart_total_amount, transaction_batch_id) session_url = session.url else: payment_status='success' paid_on = datetime.now() is_payment_completed=True payment_data={ "request":request, "user":user, "cart_items":cart_items, "cart_total_amount":cart_total_amount, "wallet_amount":wallet_amount, "used_wallet_amount":used_wallet_amount, "session_url":session_url, "payment_status":payment_status, "paid_on":paid_on, "is_payment_completed":is_payment_completed, "transaction_batch_id":transaction_batch_id, "calculated_coupon_discount":calculated_coupon_discount, "discount_coupon_id":discount_coupon_instance.id if discount_coupon_instance else None, "charity":request.charity, } self.save_payment_data(db,request, payment_data) return { "payment_url": session_url, "payment_status": payment_status, "is_payment_completed": False, "is_wallet_used": request.use_wallet, "transaction_batch_id": transaction_batch_id, "success_url": f"{settings.stripe_success_url}/{transaction_batch_id}", "cancel_url": f"{settings.stripe_cancel_url}/{transaction_batch_id}", } except Exception as e: db.rollback() raise GenericError(status_code=422, exc=f"Could not create checkout session: {e}") def create_webhook(self, stripe, webhook_url): print("creating webhook") webhook = stripe.WebhookEndpoint.create( url=webhook_url, enabled_events=["payment_intent.succeeded", "checkout.session.completed", "payment_intent.payment_failed", ], ) return webhook def generate_link(self, cart_items, total_amount, transaction_batch_id): description = ",".join([item.course.name for item in cart_items]) line_items = [ {"quantity": 1, "price_data": {"currency": self.currency, "unit_amount": int(total_amount * 100), "product_data": {"name": "Wac Course Purchase", "description": f"Course(s):{description}"}, }} ] stripe.api_key = settings.stripe_secret_key session = stripe.checkout.Session.create(payment_method_types=["card"], line_items=line_items, mode="payment", metadata={"transaction_batch_id": transaction_batch_id}, success_url=f"{settings.stripe_success_url}/{transaction_batch_id}", cancel_url=f"{settings.stripe_cancel_url}/{transaction_batch_id}", ) print("sucess url", f"{settings.stripe_success_url}/{transaction_batch_id}") print("cancel url", f"{settings.stripe_cancel_url}/{transaction_batch_id}") if session.url: webhook_url = f"{settings.stripe_webhook_url}/payment/stripe/webhook" existing_webhooks = stripe.WebhookEndpoint.list() is_webhook_exists = False for webhook in existing_webhooks: if webhook.url == webhook_url: is_webhook_exists = True break print(is_webhook_exists) if not is_webhook_exists: self.create_webhook(stripe, webhook_url) return session def save_payment_data(self, db, request, payment_data): payment_details = [] processed_cart_item = [] for item in payment_data['cart_items']: processed_cart_item.append(item.id) pd = PaymentDetails( purchased_for_id=item.participant_id, course_id=item.course_id, course_original_amount=item.course_original_amount, final_amount=item.final_amount, ) payment_details.append(pd) # Update cart items with transaction batch ID db.query(Cart).filter( Cart.id.in_(processed_cart_item) ).update({ 'transaction_batch_id': payment_data['transaction_batch_id'], 'is_cart_processed': True }, synchronize_session=False) payment = Payment( created_by=payment_data['user']['id'], user_type=payment_data['user']['user_type'], stripe_url=payment_data['session_url'], total_paid_amount=payment_data['cart_total_amount'], is_wallet_used=request.use_wallet, used_wallet_amount=payment_data["used_wallet_amount"], payment_status=payment_data["payment_status"], payment_method='CARD', intent_created_on=datetime.now(), paid_on=payment_data["paid_on"], payment_id=None, payment_gateway_session_id=None, is_payment_completed=payment_data["is_payment_completed"], transaction_batch_id=payment_data["transaction_batch_id"], charity=payment_data["charity"], coupon_discounted_amount=payment_data["calculated_coupon_discount"], discount_coupon_id=payment_data["discount_coupon_id"], payment_details=payment_details ) db.add(payment) db.commit() if payment_data["is_payment_completed"] == True: send_order_summary_email(payment.id) # Deduct wallet amount immediately if payment was fully covered by wallet if payment_data["used_wallet_amount"] > 0: if payment_data['user']['user_type'] == "parent": paid_user = db.query(Parent).get(payment_data['user']['id']) paid_user.wallet_amount -= payment_data["used_wallet_amount"] else: paid_user = db.query(Participant).get(payment_data['user']['id']) paid_user.wallet_amount -= payment_data["used_wallet_amount"] db.add(WalletHistory( payment_id=payment.id, user_type=payment_data['user']['user_type'], refund_for_id=payment_data['user']['id'], amount=payment_data["used_wallet_amount"], refund_type="purchase", transaction_id=payment.transaction_batch_id, note="N/A", is_credit=False, message=f"Purchased course(s)", )) db.commit() def update_payment(self, payload: str, sig_header: str, background_tasks: Any): with get_db() as db: event = json.loads(payload.decode("utf-8")) if event["type"] == "checkout.session.completed": transaction_batch_id = event["data"]["object"]["metadata"]["transaction_batch_id"] if event["data"]["object"]["payment_status"] == "paid": payment_instance = db.query(Payment).filter( Payment.transaction_batch_id == transaction_batch_id).first() if not payment_instance: return if payment_instance.user_type == "parent": paid_user = db.query(Parent).get(payment_instance.created_by) paid_user.wallet_amount -= payment_instance.used_wallet_amount else: paid_user = db.query(Participant).get(payment_instance.created_by) paid_user.wallet_amount -= payment_instance.used_wallet_amount db.query(Payment).filter(Payment.transaction_batch_id == transaction_batch_id).update( { "payment_status": "success", "payment_id": event["data"]["object"]["payment_intent"], "is_payment_completed": True, "paid_on": datetime.now() }, synchronize_session="fetch" ) db.query(Cart).filter(Cart.transaction_batch_id == transaction_batch_id).update( {"is_cart_processed": True} ) else: db.query(Payment).filter(Payment.transaction_batch_id == transaction_batch_id).update( { "payment_status": "failed", "is_payment_completed": False }, synchronize_session="fetch" ) if payment_instance.is_wallet_used: db.add(WalletHistory( payment_id=payment_instance.id, user_type=payment_instance.user_type, refund_for_id=payment_instance.created_by, amount=payment_instance.used_wallet_amount, refund_type="purchase", transaction_id=payment_instance.transaction_batch_id, note="N/A", is_credit=False, message=f"Purchased course(s)", )) db.commit() if payment_instance.is_payment_completed: send_order_summary_email(payment_instance.id) def verify_payment(self, transaction_batch_id: str): with get_db() as db: query = db.query(Payment).filter(Payment.transaction_batch_id == transaction_batch_id, Payment.is_payment_completed == True) return dict(is_payment_completed=query.count() > 0) def payment_history(self, user, request): with get_db() as db: data_without_duplicate = [] added_transactions = [] query = db.query(Payment).options( joinedload(Payment.payment_details), joinedload(Payment.discount_coupon) ).filter(Payment.is_payment_completed == True) print(user["user_type"]) if user['user_type'] == 'parent': query = query.filter(Payment.created_by == user['id']) elif user['user_type'] == 'participant': query = query.filter(Payment.user_type == 'participant', Payment.created_by == user['id']) if request.email: parent_ids = db.query(Parent.id).filter(Parent.email.icontains(request.email)).all() parent_ids = [row[0] for row in parent_ids] # print(parent_ids) participant_ids = db.query(Participant.id).filter(Participant.email.icontains(request.email)).all() participant_ids = [row[0] for row in participant_ids] parent_ids.extend(participant_ids) query = query.filter(Payment.created_by.in_(parent_ids)) if request.status: query=query.filter(Payment.payment_status.ilike(request.status)) if request.mobile_number: parent_ids = db.query(Parent.id).filter(Parent.mobile_number.icontains(request.mobile_number)).all() parent_ids = [row[0] for row in parent_ids] # print(parent_ids) participant_ids = db.query(Participant.id).filter( Participant.mobile_number.icontains(request.mobile_number)).all() participant_ids = [row[0] for row in participant_ids] parent_ids.extend(participant_ids) query = query.filter(Payment.created_by.in_(parent_ids)) for row in query.all(): if row.transaction_batch_id not in added_transactions: if row.user_type == 'parent': guardian = db.query(Parent).get(row.created_by) else: guardian = db.query(Participant).get(row.created_by) discount_details = None if row.discount_coupon_id: discount_details = { "coupon_code": row.discount_coupon.code, "discount_amount": row.discount_coupon.discount_value, "discount_type": row.discount_coupon.discount_type, "discount_value": row.discount_coupon.discount_value } data_without_duplicate.append({ "payment_id": row.id, "transaction_batch_id": row.transaction_batch_id, "parent_or_participant_full_name": f"{guardian.first_name} {guardian.last_name}", "parent_or_participant_email": guardian.email, "parent_or_participant_mobile_number": guardian.mobile_number, "total_amount": row.total_paid_amount, "paid_on": row.paid_on, "payment_status": row.payment_status, "is_payment_completed": row.is_payment_completed, "is_wallet_user": row.is_wallet_used, "used_wallet_amount": row.used_wallet_amount, "discount_details": discount_details }) added_transactions.append(row.transaction_batch_id) return data_without_duplicate def payment_history_details(self, transaction_id): with (get_db() as db): history_list = [] payment=db.query(Payment).filter(Payment.transaction_batch_id==transaction_id).first() query = db.query(PaymentDetails).options(joinedload(PaymentDetails.course)).filter( PaymentDetails.payment_id == payment.id).all() for row in query: data = PaymentDetailsBase.model_validate(row).model_dump() data["participant"] = {"first_name": row.purchased_participant.first_name, "last_name": row.purchased_participant.last_name} data["course"]={ "id": row.course.id, "name": row.course.name, "age_group": row.course.age_group.label, "term": row.course.term.name, "academic_year": row.course.academic_year.title, } history_list.append(data) payments_details_all={ "payment_id": payment.id, "transaction_batch_id": payment.transaction_batch_id, "total_amount": payment.total_paid_amount, "paid_on": payment.paid_on, "payment_status": payment.payment_status, "is_payment_completed": payment.is_payment_completed, "payment_details":history_list, } return payments_details_all def get_purchased_courses(self, user, request): """ List all purchased courses for the current user with optional filters """ with get_db() as db: # Start with base query joining necessary tables query = (db.query(PaymentDetails) .join(Payment, PaymentDetails.payment_id == Payment.id) .join(Course, PaymentDetails.course_id == Course.id) .join(Participant, PaymentDetails.purchased_for_id == Participant.id) .filter(Payment.is_payment_completed == True)) # Filter by user if user['user_type'] == 'parent': query = query.filter(Payment.created_by == user['id']) elif user['user_type'] == 'participant': query = query.filter( Payment.user_type == 'participant', Payment.created_by == user['id'] ) # Apply filters if provided if request.participant_name: query = query.filter( (func.concat(Participant.first_name, ' ', Participant.last_name) .ilike(f'%{request.participant_name}%')) ) if request.academic_year_id: query = query.filter(Course.academic_year_id == request.academic_year_id) if request.term_id: query = query.filter(Course.term_id == request.term_id) # Execute query and format results results = [] for item in query.all(): results.append({ "payment_id": item.payment_id, "course_id": item.course_id, "course_name": item.course.name, "course_start_date":item.course.start_date, "course_end_date":item.course.end_date, "participant_name": f"{item.purchased_participant.first_name} {item.purchased_participant.last_name}", "purchase_date": item.payment.paid_on, "amount_paid": item.final_amount, "academic_year": item.course.academic_year.title if item.course.academic_year else None, "term": item.course.term.name if item.course.term else None, "payment_status": item.payment.payment_status }) return results def get_payment_by_id(self, payment_id: str): """ Get payment and its details by payment ID """ with get_db() as db: payment = db.query(Payment).options( joinedload(Payment.payment_details) ).filter(Payment.id == payment_id).first() if not payment: raise HTTPException(status_code=404, detail="Payment not found") # Get payment details payment_details = [] for detail in payment.payment_details: payment_details.append({ "id": detail.id, "purchased_for_id": detail.purchased_for_id, "course_id": detail.course_id, "course_original_amount": detail.course_original_amount, "final_amount": detail.final_amount, "participant_name": detail.purchased_participant.full_name if detail.purchased_participant else None, "course_name": detail.course.name if detail.course else None, "academic_year": detail.course.academic_year.title if detail.course and detail.course.academic_year else None, "term": detail.course.term.name if detail.course and detail.course.term else None }) return { "id": payment.id, "transaction_batch_id": payment.transaction_batch_id, "total_paid_amount": payment.total_paid_amount, "paid_on": payment.paid_on, "payment_status": payment.payment_status, "is_payment_completed": payment.is_payment_completed, "payment_details": payment_details } def get_payment_details(self,payment_details_id: str): with get_db() as db: payment_details = db.query(PaymentDetails).filter( PaymentDetails.id == payment_details_id ).first() if not payment_details: raise HTTPException(status_code=404, detail="Payment details not found") return { "id": payment_details.id, "purchased_for_id": payment_details.purchased_for_id, "course_id": payment_details.course_id, "course_original_amount": payment_details.course_original_amount, "final_amount": payment_details.final_amount, "participant_name": payment_details.purchased_participant.full_name if payment_details.purchased_participant else None, "course_name": payment_details.course.name if payment_details.course else None, "academic_year": payment_details.course.academic_year.title if payment_details.course and payment_details.course.academic_year else None, "term": payment_details.course.term.name if payment_details.course and payment_details.course.term else None }