import os
from base64 import b32encode
import frappe
import pyotp
from frappe import _
from frappe.auth import LoginAttemptTracker
from frappe.core.doctype.sms_settings.sms_settings import send_sms
from frappe.rate_limiter import rate_limit
from frappe.translate import get_all_translations
from frappe.utils import (
days_diff,
format_date,
format_time,
get_datetime,
get_datetime_in_timezone,
get_system_timezone,
now_datetime,
today,
validate_email_address,
)
from event_manager.payments import (
get_payment_gateways_for_event,
get_payment_link_for_booking,
get_payment_link_for_sponsorship,
)
from event_manager.utils import is_app_installed
OFFLINE_PAYMENT_METHOD = "Offline"
@frappe.whitelist(allow_guest=True) # nosemgrep: frappe-semgrep-rules.rules.security.guest-whitelisted-method
@rate_limit(key="identifier", limit=5, seconds=3600)
def send_guest_booking_otp(event: int, identifier: str) -> dict:
event_doc = frappe.get_cached_doc("Pohodex Event Manager Event", event)
if not event_doc.allow_guest_booking:
frappe.throw(_("Guest booking is not enabled for this event"))
if event_doc.guest_verification_method == "None":
frappe.throw(_("OTP verification is not enabled for this event"))
channel = "phone" if event_doc.guest_verification_method == "Phone OTP" else "email"
identifier = identifier.strip()
if not identifier:
frappe.throw(_("Email or phone is required"))
if channel == "email":
identifier = identifier.lower()
validate_email_address(identifier, throw=True)
otp_secret = b32encode(os.urandom(10)).decode("utf-8")
otp_code = pyotp.HOTP(otp_secret).at(0)
cache_key = f"guest_booking_otp:{channel}:{identifier}"
if frappe.in_test:
frappe.cache.set_value(cache_key, otp_secret, expires_in_sec=600)
return {"otp": otp_code}
try:
if channel == "email":
frappe.sendmail(
recipients=[identifier],
subject=_("Your Booking Verification Code"),
message=_(
"Your verification code is: {0}
This code expires in 10 minutes."
).format(otp_code),
now=True,
)
else:
send_sms(
receiver_list=[identifier],
msg=_("Your booking verification code is: {0}. It expires in 10 minutes.").format(otp_code),
)
except Exception:
frappe.throw(_("Failed to send verification code. Please try again."))
frappe.cache.set_value(cache_key, otp_secret, expires_in_sec=600)
def verify_guest_otp(channel: str, identifier: str, otp: str):
cache_key = f"guest_booking_otp:{channel}:{identifier}"
tracker = LoginAttemptTracker(
key=f"guest_otp:{channel}:{identifier}",
max_consecutive_login_attempts=5,
lock_interval=600,
)
if not tracker.is_user_allowed():
frappe.throw(_("Too many failed attempts. Please try again later."))
otp_secret = frappe.cache.get_value(cache_key)
if not otp_secret:
frappe.throw(_("Verification code expired. Please request a new one."))
if not pyotp.HOTP(otp_secret).verify(otp.strip(), 0):
tracker.add_failure_attempt()
frappe.throw(_("Invalid verification code"))
frappe.cache.delete_value(cache_key)
tracker.add_success_attempt()
def get_or_create_guest_user(email: str, full_name: str) -> str:
email = email.lower().strip()
validate_email_address(email, throw=True)
if frappe.db.exists("User", email):
return email
name_parts = full_name.strip().split(" ", 1)
first_name = name_parts[0] if name_parts else "Guest"
last_name = name_parts[1] if len(name_parts) > 1 else ""
user = frappe.get_doc(
{
"doctype": "User",
"email": email,
"first_name": first_name,
"last_name": last_name,
"enabled": 1,
"user_type": "Website User",
"send_welcome_email": 0,
}
)
user.insert(ignore_permissions=True)
return email
@frappe.whitelist()
def get_event_payment_gateways(event: str) -> list[str]:
return get_payment_gateways_for_event(event)
def are_registrations_closed(event_doc) -> bool:
if not event_doc.registrations_close_at:
return False
event_tz = event_doc.time_zone or get_system_timezone()
now_in_event_tz = get_datetime_in_timezone(event_tz).replace(tzinfo=None)
return now_in_event_tz > get_datetime(event_doc.registrations_close_at)
def is_ticket_transfer_allowed(event_id: str | int) -> bool:
try:
event = frappe.get_cached_doc("Pohodex Event Manager Event", event_id)
settings = frappe.get_single("Pohodex Event Manager Settings")
transfer_cutoff_days = settings.get("allow_transfer_ticket_before_event_start_days", 7)
if not event.start_date:
return False
return days_diff(event.start_date, today()) >= transfer_cutoff_days
except Exception as e:
frappe.log_error(f"Error checking ticket transfer eligibility: {e!s}")
return False
def is_add_on_change_allowed(event_id: str | int) -> bool:
try:
event = frappe.get_cached_doc("Pohodex Event Manager Event", event_id)
settings = frappe.get_cached_doc("Pohodex Event Manager Settings")
add_on_change_cutoff_days = settings.get("allow_add_ons_change_before_event_start_days", 7)
if not event.start_date:
return False
return days_diff(event.start_date, today()) >= add_on_change_cutoff_days
except Exception as e:
frappe.log_error(f"Error checking add-on change eligibility: {e!s}")
return False
@frappe.whitelist()
def can_transfer_ticket(event_id: str | int) -> dict:
return {"can_transfer": is_ticket_transfer_allowed(event_id), "event_id": event_id}
@frappe.whitelist()
def can_change_add_ons(event_id: str | int) -> dict:
return {"can_change_add_ons": is_add_on_change_allowed(event_id), "event_id": event_id}
def is_cancellation_request_allowed(event_id: str | int) -> bool:
try:
event = frappe.get_cached_doc("Pohodex Event Manager Event", event_id)
settings = frappe.get_cached_doc("Pohodex Event Manager Settings")
cancellation_cutoff_days = settings.get(
"allow_ticket_cancellation_request_before_event_start_days", 7
)
if not event.start_date:
return False
return days_diff(event.start_date, today()) >= cancellation_cutoff_days
except Exception as e:
frappe.log_error(f"Error checking cancellation request eligibility: {e!s}")
return False
@frappe.whitelist()
def can_request_cancellation(event_id: str | int) -> dict:
return {"can_request_cancellation": is_cancellation_request_allowed(event_id), "event_id": event_id}
@frappe.whitelist(allow_guest=True) # nosemgrep: frappe-semgrep-rules.rules.security.guest-whitelisted-method
def get_event_booking_data(event_route: str) -> dict:
data = frappe._dict()
event_doc = frappe.get_cached_doc("Pohodex Event Manager Event", {"route": event_route})
if not event_doc.is_published:
frappe.throw(_("Event not found"), frappe.DoesNotExistError)
data.registrations_closed = are_registrations_closed(event_doc)
is_guest = frappe.session.user == "Guest"
if is_guest:
data.event_details = {
"name": event_doc.name,
"title": event_doc.title,
"route": event_doc.route,
"start_date": event_doc.start_date,
"end_date": event_doc.end_date,
"start_time": event_doc.start_time,
"end_time": event_doc.end_time,
"time_zone": event_doc.time_zone,
"venue": event_doc.venue,
"medium": event_doc.medium,
"category": event_doc.category,
"banner_image": event_doc.banner_image,
"short_description": event_doc.short_description,
"free_webinar": event_doc.free_webinar,
"send_ticket_email": event_doc.send_ticket_email,
"allow_guest_booking": event_doc.allow_guest_booking,
"guest_verification_method": event_doc.guest_verification_method,
"default_ticket_type": event_doc.default_ticket_type,
}
else:
data.event_details = event_doc
available_ticket_types = []
published_ticket_types = frappe.db.get_all(
"Event Ticket Type", filters={"is_published": True, "event": event_doc.name}, pluck="name"
)
for ticket_type in published_ticket_types:
tt = frappe.get_cached_doc("Event Ticket Type", ticket_type)
if tt.are_tickets_available(1):
available_ticket_types.append(tt)
data.available_ticket_types = available_ticket_types
add_ons = frappe.db.get_all(
"Ticket Add-on", filters={"event": event_doc.name, "enabled": 1}, fields=["*"], order_by="title"
)
for add_on in add_ons:
if add_on.user_selects_option:
add_on.options = add_on.options.split("\n")
data.available_add_ons = add_ons
data.tax_settings = {
"apply_tax": event_doc.apply_tax,
"tax_inclusive": event_doc.tax_inclusive,
"tax_label": event_doc.tax_label or "Tax",
"tax_percentage": event_doc.tax_percentage or 0,
}
custom_fields = frappe.db.get_all(
"Pohodex Event Manager Custom Field",
filters={"event": event_doc.name, "enabled": 1},
fields=["*"],
order_by="order",
)
data.custom_fields = custom_fields
payment_gateways = get_payment_gateways_for_event(event_doc.name)
offline_methods_raw = frappe.get_all(
"Offline Payment Method",
filters={"event": event_doc.name, "enabled": 1},
fields=["name", "title", "description", "collect_payment_proof"],
order_by="creation",
)
offline_methods = []
for method in offline_methods_raw:
method_custom_fields = frappe.get_all(
"Pohodex Event Manager Custom Field",
filters={
"event": event_doc.name,
"enabled": 1,
"applied_to": "Offline Payment Form",
"offline_payment_method": method.name,
},
fields=["*"],
order_by="order",
)
offline_methods.append(
{
"name": method.name,
"title": method.title,
"description": method.description,
"collect_payment_proof": method.collect_payment_proof,
"custom_fields": method_custom_fields,
}
)
payment_gateways.append(method.title)
data.payment_gateways = payment_gateways
data.offline_payment_enabled = len(offline_methods) > 0
data.offline_methods = offline_methods
return data
@frappe.whitelist(allow_guest=True) # nosemgrep: frappe-semgrep-rules.rules.security.guest-whitelisted-method
def process_booking(
attendees: list[dict],
event: str,
coupon_code: str | None = None,
booking_custom_fields: dict | None = None,
payment_gateway: str | None = None,
utm_parameters: list[dict] | None = None,
guest_email: str | None = None,
guest_full_name: str | None = None,
otp: str | None = None,
guest_phone: str | None = None,
payment_proof: str | None = None,
is_offline: bool = False,
offline_payment_method: str | None = None,
invoice_requested: bool = False,
tax_id: str | None = None,
billing_address: str | None = None,
) -> dict:
event_doc = frappe.get_cached_doc("Pohodex Event Manager Event", event)
if not event_doc.is_published:
frappe.throw(_("Event is not live"))
if are_registrations_closed(event_doc):
frappe.throw(_("Registrations for this event are closed"))
is_guest = frappe.session.user == "Guest"
if is_guest:
if not event_doc.allow_guest_booking:
frappe.throw(_("Please log in to access this feature"), frappe.AuthenticationError)
if not guest_email:
frappe.throw(_("Email is required for guest booking"))
validate_email_address(guest_email, throw=True)
email = guest_email.lower().strip()
if event_doc.guest_verification_method == "Email OTP":
if not otp:
frappe.throw(_("Verification code is required"))
verify_guest_otp("email", email, otp)
elif event_doc.guest_verification_method == "Phone OTP":
if not otp:
frappe.throw(_("Verification code is required"))
if not guest_phone:
frappe.throw(_("Phone number is required"))
verify_guest_otp("phone", guest_phone.strip(), otp)
first_name = (attendees[0].get("first_name") or "").strip()
last_name = (attendees[0].get("last_name") or "").strip()
full_name = (guest_full_name or "").strip() or f"{first_name} {last_name}".strip()
if not full_name:
frappe.throw(_("Full name is required for guest booking"))
booking_user = get_or_create_guest_user(guest_email, full_name)
else:
booking_user = frappe.session.user
booking = frappe.new_doc("Event Booking")
booking.event = event
booking.coupon_code = coupon_code
booking.user = booking_user
if event_doc.apply_tax and invoice_requested:
booking.invoice_requested = 1
booking.tax_id = tax_id
booking.billing_address = billing_address
if utm_parameters:
for utm_param in utm_parameters:
booking.append(
"utm_parameters",
{
"utm_name": utm_param.get("utm_name"),
"value": utm_param.get("value"),
},
)
if booking_custom_fields:
booking_custom_field_defs = frappe.db.get_all(
"Pohodex Event Manager Custom Field",
filters={"event": event, "enabled": 1, "applied_to": "Booking"},
fields=["fieldname", "label", "fieldtype"],
)
custom_field_map = {cf["fieldname"]: cf for cf in booking_custom_field_defs}
for field_name, field_value in booking_custom_fields.items():
if field_value and field_name in custom_field_map:
field_def = custom_field_map[field_name]
booking.append(
"additional_fields",
{
"fieldname": field_name,
"value": str(field_value),
"label": field_def["label"],
"fieldtype": field_def["fieldtype"],
},
)
if event_doc.category == "Webinars":
for attendee in attendees:
if not (attendee.get("last_name") or "").strip():
frappe.throw(_("Last name is required for all attendees in webinar events"))
for attendee in attendees:
first_name = (attendee.get("first_name") or "").strip()
last_name = (attendee.get("last_name") or "").strip()
if not first_name and attendee.get("full_name"):
name_parts = attendee["full_name"].strip().split(" ", 1)
first_name = name_parts[0]
last_name = last_name or (name_parts[1] if len(name_parts) > 1 else "")
attendee_full_name = f"{first_name} {last_name}".strip()
add_ons = attendee.get("add_ons", None)
if add_ons:
add_ons = create_add_on_doc(
attendee_name=attendee_full_name,
add_ons=add_ons,
)
custom_fields = attendee.get("custom_fields", {})
attendee_row = {
"first_name": first_name,
"last_name": last_name,
"email": attendee.get("email"),
"ticket_type": attendee.get("ticket_type"),
"add_ons": add_ons.name if add_ons else None,
"custom_fields": custom_fields if custom_fields else None,
}
booking.append("attendees", attendee_row)
booking.insert(ignore_permissions=True)
frappe.db.commit()
if booking.total_amount == 0:
booking.flags.ignore_permissions = True
booking.submit()
return {"booking_name": booking.name}
if is_offline:
method_filters = {"event": event, "enabled": 1}
if offline_payment_method:
method_filters["name"] = offline_payment_method
method_doc = frappe.db.get_value(
"Offline Payment Method", method_filters, ["name", "title"], as_dict=True
)
if not method_doc:
frappe.throw(_("Offline payment is not enabled for this event"))
booking.payment_method = OFFLINE_PAYMENT_METHOD
booking.offline_payment_method = method_doc.title
booking.status = "Approval Pending"
booking.payment_status = "Verification Pending"
booking.flags.ignore_permissions = True
booking.save()
if payment_proof:
try:
file_doc = frappe.get_doc(
{
"doctype": "File",
"file_url": payment_proof,
"attached_to_doctype": "Event Booking",
"attached_to_name": booking.name,
"is_private": 1,
}
)
file_doc.insert(ignore_permissions=True)
except Exception as e:
frappe.log_error(f"Failed to attach payment proof: {e}")
return {"booking_name": booking.name, "offline_payment": True}
return {
"payment_link": get_payment_link_for_booking(
booking.name,
redirect_to=f"/dashboard/bookings/{booking.name}?success=true",
payment_gateway=payment_gateway,
)
}
def create_add_on_doc(attendee_name: str, add_ons: list[dict]):
for add_on in add_ons:
add_on["currency"] = frappe.db.get_value("Ticket Add-on", add_on["add_on"], "currency")
return frappe.get_doc(
{"doctype": "Attendee Ticket Add-on", "add_ons": add_ons, "attendee_name": attendee_name}
).insert(ignore_permissions=True)
@frappe.whitelist()
def transfer_ticket(ticket_id: str, new_first_name: str, new_last_name: str, new_email: str):
if not frappe.db.exists("Event Ticket", ticket_id):
frappe.throw(frappe._("Ticket not found."))
ticket = frappe.get_doc("Event Ticket", ticket_id)
booking_user = frappe.db.get_value("Event Booking", ticket.booking, "user")
if (
ticket.attendee_email != frappe.session.user
and booking_user != frappe.session.user
and not frappe.has_permission("Event Ticket", "write", ticket)
):
frappe.throw(frappe._("Not permitted to transfer this ticket."))
if not is_ticket_transfer_allowed(ticket.event):
frappe.throw(frappe._("Ticket transfer is not allowed at this time. The transfer window has closed."))
old_name = ticket.attendee_name
old_email = ticket.attendee_email
new_name = f"{new_first_name} {new_last_name}".strip()
ticket.first_name = new_first_name
ticket.last_name = new_last_name
ticket.attendee_email = new_email
ticket.save(ignore_permissions=True)
send_ticket_transfer_emails(ticket.name, old_name, old_email, new_name, new_email)
def send_ticket_transfer_emails(ticket_id: str, old_name: str, old_email: str, new_name: str, new_email: str):
try:
ticket = frappe.get_doc("Event Ticket", ticket_id)
event = frappe.get_doc("Pohodex Event Manager Event", ticket.event)
booking = frappe.get_doc("Event Booking", ticket.booking)
old_attendee_subject = f"Your ticket for {event.title} has been transferred"
old_attendee_message = f"""
Dear {old_name},
This is to inform you that your ticket for {event.title} has been transferred to {new_name} ({new_email}).
Event Details:
- Event: {event.title}
- Date: {format_date(event.start_date)}
- Ticket Type: {ticket.ticket_type}
- Booking ID: {booking.name}
If you have any questions about this transfer, please contact us.
Best regards,
{event.title} Team
"""
frappe.sendmail(
recipients=[old_email], subject=old_attendee_subject, message=old_attendee_message, delayed=False
)
new_attendee_subject = f"Welcome! Your ticket for {event.title}"
new_attendee_message = f"""
Dear {new_name},
Great news! A ticket for {event.title} has been transferred to you.
Event Details:
- Event: {event.title}
- Date: {format_date(event.start_date)}
- Time: {format_time(event.start_time) if event.start_time else "TBA"}
- Venue: {event.venue or "TBA"}
- Ticket Type: {ticket.ticket_type}
- Booking ID: {booking.name}
Your Ticket Details:
- Ticket ID: {ticket.name}
- Attendee Name: {new_name}
- Email: {new_email}
Please save this email for your records. You may need to present this ticket information at the event entrance.
We look forward to seeing you at the event!
Best regards,
{event.title} Team
"""
frappe.sendmail(
recipients=[new_email], subject=new_attendee_subject, message=new_attendee_message, delayed=False
)
except Exception as e:
frappe.log_error(f"Failed to send ticket transfer emails for ticket {ticket_id}: {e!s}")
@frappe.whitelist()
def get_booking_details(booking_id: str) -> dict:
details = frappe._dict()
booking_doc = frappe.get_cached_doc("Event Booking", booking_id)
details.doc = booking_doc
tickets = frappe.db.get_all(
"Event Ticket",
filters={"booking": booking_id},
fields=[
"name",
"attendee_name",
"attendee_email",
"ticket_type.title as ticket_type",
"qr_code",
"event",
"docstatus",
],
)
add_ons = frappe.db.get_all(
"Ticket Add-on Value",
filters={"parent": ("in", [ticket.name for ticket in tickets])},
fields=[
"parent",
"name",
"add_on",
"value",
"add_on.title as add_on_title",
"add_on.user_selects_option as user_selects_option",
],
)
event_add_ons = frappe.db.get_all(
"Ticket Add-on",
filters={"event": booking_doc.event, "user_selects_option": True},
fields=["name", "title", "user_selects_option", "options"],
)
add_on_options_map = {}
for event_add_on in event_add_ons:
if event_add_on.user_selects_option:
add_on_options_map[event_add_on.name] = (
event_add_on.options.split("\n") if event_add_on.options else []
)
for ticket in tickets:
ticket.add_ons = []
for add_on in add_ons:
if add_on.parent == ticket.name:
add_on_data = {
"id": add_on.name,
"name": add_on.add_on,
"title": add_on.add_on_title,
"value": add_on.value,
"user_selects_option": add_on.user_selects_option,
"options": add_on_options_map.get(add_on.add_on, []),
}
ticket.add_ons.append(add_on_data)
ticket.add_ons = sorted(ticket.add_ons, key=lambda x: x["title"])
details.tickets = tickets
details.event = frappe.get_cached_doc("Pohodex Event Manager Event", booking_doc.event)
if details.event.venue:
details.venue = frappe.get_cached_doc("Event Venue", details.event.venue)
details.can_transfer_ticket = can_transfer_ticket(details.event.name)
details.can_change_add_ons = can_change_add_ons(details.event.name)
details.can_request_cancellation = can_request_cancellation(details.event.name)
existing_cancellation = frappe.db.get_value(
"Ticket Cancellation Request",
{"booking": booking_id},
["name", "cancel_full_booking", "creation", "status", "docstatus"],
as_dict=True,
)
details.cancellation_request = existing_cancellation
details.cancellation_requested_tickets = []
if existing_cancellation and existing_cancellation.docstatus == 0:
if existing_cancellation.cancel_full_booking:
details.cancellation_requested_tickets = [ticket.name for ticket in tickets]
else:
requested_tickets = frappe.db.get_all(
"Ticket Cancellation Item", filters={"parent": existing_cancellation.name}, fields=["ticket"]
)
details.cancellation_requested_tickets = [item.ticket for item in requested_tickets]
details.cancelled_tickets = [ticket.name for ticket in tickets if ticket.docstatus == 2]
return details
@frappe.whitelist()
def change_add_on_preference(add_on_id: str, new_value: str):
if not frappe.db.exists("Ticket Add-on Value", add_on_id):
frappe.throw(frappe._("Add-on value not found."))
add_on_value = frappe.get_cached_doc("Ticket Add-on Value", add_on_id)
ticket = frappe.get_cached_doc("Event Ticket", add_on_value.parent)
if not is_add_on_change_allowed(ticket.event):
frappe.throw(
frappe._(
"Add-on changes are not allowed at this time. The change window has closed as the event is approaching."
)
)
frappe.db.set_value(
"Ticket Add-on Value",
add_on_id,
"value",
new_value,
)
@frappe.whitelist()
def get_sponsorship_details(enquiry_id: str) -> dict:
enquiry = frappe.get_doc("Sponsorship Enquiry", enquiry_id)
if enquiry.owner != frappe.session.user and not frappe.has_permission(
"Sponsorship Enquiry", "read", enquiry
):
frappe.throw(frappe._("Not permitted to view this sponsorship enquiry"))
tier_title = ""
if enquiry.tier:
tier_title = frappe.db.get_value("Sponsorship Tier", enquiry.tier, "title") or enquiry.tier
event_details = {}
if enquiry.event:
event = frappe.get_cached_doc("Pohodex Event Manager Event", enquiry.event)
event_details = {
"title": event.title,
"short_description": getattr(event, "short_description", ""),
"about": getattr(event, "about", ""),
"start_date": event.start_date,
"end_date": getattr(event, "end_date", ""),
"venue": getattr(event, "venue", ""),
"route": getattr(event, "route", ""),
}
sponsor_details = None
sponsors = frappe.db.get_all(
"Event Sponsor",
filters={"enquiry": enquiry_id},
fields=["name", "company_name", "company_logo", "creation", "event", "tier"],
limit=1,
)
if sponsors:
sponsor_details = sponsors[0]
if sponsor_details.get("tier"):
sponsor_tier_title = frappe.db.get_value("Sponsorship Tier", sponsor_details["tier"], "title")
sponsor_details["tier_title"] = sponsor_tier_title or sponsor_details["tier"]
return {
"enquiry": {
"name": enquiry.name,
"company_name": enquiry.company_name,
"company_logo": enquiry.company_logo,
"event": enquiry.event,
"tier": enquiry.tier,
"tier_title": tier_title,
"status": enquiry.status,
"creation": enquiry.creation,
"owner": enquiry.owner,
},
"event_details": event_details,
"sponsor_details": sponsor_details,
"has_sponsor": bool(sponsor_details),
}
@frappe.whitelist()
def get_user_sponsorship_inquiries() -> list:
inquiries = frappe.db.get_all(
"Sponsorship Enquiry",
filters={"owner": frappe.session.user},
fields=["name", "company_name", "event", "tier", "status", "creation"],
order_by="creation desc",
)
for inquiry in inquiries:
if inquiry.event:
event_title = frappe.db.get_value("Pohodex Event Manager Event", inquiry.event, "title")
inquiry["event_title"] = event_title
if inquiry.tier:
tier_title = frappe.db.get_value("Sponsorship Tier", inquiry.tier, "title")
inquiry["tier_title"] = tier_title or inquiry.tier
else:
inquiry["tier_title"] = ""
inquiry_names = [inquiry.name for inquiry in inquiries]
if inquiry_names:
sponsors = frappe.db.get_all(
"Event Sponsor",
filters={"enquiry": ["in", inquiry_names]},
fields=["enquiry"],
)
sponsored_inquiries = {sponsor.enquiry for sponsor in sponsors}
for inquiry in inquiries:
inquiry["has_sponsor"] = inquiry.name in sponsored_inquiries
else:
for inquiry in inquiries:
inquiry["has_sponsor"] = False
return inquiries
@frappe.whitelist()
def create_sponsorship_payment_link(enquiry_id: str, tier_id: str, payment_gateway: str | None = None) -> str:
enquiry = frappe.get_doc("Sponsorship Enquiry", enquiry_id)
if enquiry.owner != frappe.session.user:
frappe.throw(frappe._("Not permitted to create payment for this enquiry"))
redirect_url = f"/dashboard/account/sponsorships/{enquiry_id}?success=true"
return get_payment_link_for_sponsorship(
enquiry_id, tier_id, redirect_url, payment_gateway=payment_gateway
)
@frappe.whitelist()
def withdraw_sponsorship_enquiry(enquiry_id: str):
enquiry = frappe.get_cached_doc("Sponsorship Enquiry", enquiry_id)
if enquiry.owner != frappe.session.user:
frappe.throw(frappe._("Not permitted to withdraw this enquiry"))
if enquiry.status == "Paid":
frappe.throw(frappe._("Cannot withdraw a paid sponsorship enquiry"))
if enquiry.status == "Withdrawn":
frappe.throw(frappe._("This sponsorship enquiry has already been withdrawn"))
enquiry.status = "Withdrawn"
enquiry.save(ignore_permissions=True)
@frappe.whitelist()
def get_ticket_details(ticket_id: str) -> dict:
details = frappe._dict()
ticket_doc = frappe.get_cached_doc("Event Ticket", ticket_id)
if frappe.session.user != "Administrator":
if ticket_doc.attendee_email != frappe.session.user:
frappe.throw(frappe._("Not permitted to view this ticket"))
details.doc = ticket_doc
add_ons = frappe.db.get_all(
"Ticket Add-on Value",
filters={"parent": ticket_id},
fields=[
"name",
"add_on",
"add_on.title as add_on_title",
"value",
"price",
"currency",
"add_on.user_selects_option as user_selects_option",
],
)
event_add_ons = frappe.db.get_all(
"Ticket Add-on",
filters={"event": ticket_doc.event, "user_selects_option": True},
fields=["name", "title", "user_selects_option", "options"],
)
add_on_options_map = {}
for event_add_on in event_add_ons:
if event_add_on.user_selects_option:
add_on_options_map[event_add_on.name] = (
event_add_on.options.split("\n") if event_add_on.options else []
)
enhanced_add_ons = []
for add_on in add_ons:
add_on_data = {
"id": add_on.name,
"name": add_on.add_on,
"title": add_on.add_on_title,
"value": add_on.value,
"price": add_on.price,
"currency": add_on.currency,
"user_selects_option": add_on.user_selects_option,
"options": add_on_options_map.get(add_on.add_on, []),
}
enhanced_add_ons.append(add_on_data)
details.add_ons = enhanced_add_ons
details.event = frappe.get_cached_doc("Pohodex Event Manager Event", ticket_doc.event)
booking_doc = None
if ticket_doc.booking:
booking_doc = frappe.get_cached_doc("Event Booking", ticket_doc.booking)
if booking_doc.owner == frappe.session.user:
details.booking = booking_doc
else:
details.booking = None
else:
details.booking = None
details.ticket_type = frappe.get_cached_doc("Event Ticket Type", ticket_doc.ticket_type)
details.can_transfer_ticket = (
can_transfer_ticket(details.event.name) if details.event else {"can_transfer": False}
)
details.can_change_add_ons = (
can_change_add_ons(details.event.name) if details.event else {"can_change_add_ons": False}
)
details.can_request_cancellation = (
can_request_cancellation(details.event.name) if details.event else {"can_request_cancellation": False}
)
details.zoom_join_url = None
if hasattr(ticket_doc, "zoom_webinar_registration") and ticket_doc.zoom_webinar_registration:
zoom_registration = frappe.db.get_value(
"Zoom Webinar Registration",
ticket_doc.zoom_webinar_registration,
["join_url", "webinar"],
as_dict=True,
)
if zoom_registration:
details.zoom_join_url = zoom_registration.join_url
details.zoom_webinar = zoom_registration.webinar
return details
@frappe.whitelist()
def create_cancellation_request(booking_id: str, ticket_ids: list | None = None) -> dict:
booking_doc = frappe.get_cached_doc("Event Booking", booking_id)
if booking_doc.user != frappe.session.user and not frappe.has_permission(
"Event Booking", "write", booking_doc
):
frappe.throw(frappe._("Not permitted to request cancellation for this booking."))
if not is_cancellation_request_allowed(booking_doc.event):
frappe.throw("Cancellation requests are no longer allowed for this event.")
existing_request = frappe.db.exists(
"Ticket Cancellation Request", {"booking": booking_id, "docstatus": 0}
)
if existing_request:
frappe.throw("A cancellation request already exists for this booking.")
all_tickets = frappe.db.get_all("Event Ticket", filters={"booking": booking_id}, fields=["name"])
cancel_full_booking = not ticket_ids or len(ticket_ids) == len(all_tickets)
cancellation_request = frappe.new_doc("Ticket Cancellation Request")
cancellation_request.booking = booking_id
cancellation_request.cancel_full_booking = cancel_full_booking
if not cancel_full_booking and ticket_ids:
for ticket_id in ticket_ids:
ticket_booking = frappe.db.get_value("Event Ticket", ticket_id, "booking")
if ticket_booking != booking_id:
frappe.throw(f"Ticket {ticket_id} does not belong to booking {booking_id}")
cancellation_request.append("tickets", {"ticket": ticket_id})
cancellation_request.insert(ignore_permissions=True)
@frappe.whitelist(allow_guest=True) # nosemgrep: frappe-semgrep-rules.rules.security.guest-whitelisted-method
def get_user_info() -> dict:
if frappe.session.user == "Guest":
return {"is_logged_in": False}
user = frappe.get_cached_doc("User", frappe.session.user)
return {
"name": user.name,
"is_logged_in": True,
"first_name": user.first_name,
"last_name": user.last_name,
"full_name": user.full_name,
"email": user.email,
"user_image": user.user_image,
"roles": user.roles,
"brand_image": frappe.get_single_value("Website Settings", "banner_image"),
"language": user.language,
}
@frappe.whitelist()
def validate_ticket_for_checkin(ticket_id: str) -> dict:
frappe.only_for("Frontdesk Manager", True)
if not frappe.db.exists("Event Ticket", ticket_id):
frappe.throw(_("Ticket not found"))
ticket_doc = frappe.get_cached_doc("Event Ticket", ticket_id)
if ticket_doc.docstatus == 2:
frappe.throw(_("This ticket has been cancelled and cannot be checked in"))
event_doc = frappe.get_cached_doc("Pohodex Event Manager Event", ticket_doc.event)
ticket_type_doc = (
frappe.get_cached_doc("Event Ticket Type", ticket_doc.ticket_type) if ticket_doc.ticket_type else None
)
checkin_date = frappe.utils.today()
existing_checkin = frappe.db.exists("Event Check In", {"ticket": ticket_id, "date": checkin_date})
if existing_checkin:
checkin_doc = frappe.get_doc("Event Check In", existing_checkin)
formatted_checkin_time = (
format_date(checkin_doc.creation) + " at " + format_time(checkin_doc.creation)
)
frappe.throw(_("This ticket was already checked in today ({0}).").format(formatted_checkin_time))
add_ons = frappe.db.get_all(
"Ticket Add-on Value",
filters={"parent": ticket_id},
fields=[
"add_on",
"add_on.title as add_on_title",
"add_on.user_selects_option as add_on_selects_option",
"value",
"price",
"currency",
],
)
return {
"message": _("Valid ticket ready for check-in"),
"ticket": {
"id": ticket_doc.name,
"attendee_name": ticket_doc.attendee_name,
"attendee_email": ticket_doc.attendee_email,
"event_title": event_doc.title,
"ticket_type": (ticket_type_doc.title if ticket_type_doc else ticket_doc.ticket_type),
"venue": event_doc.venue,
"start_date": event_doc.start_date,
"start_time": event_doc.start_time,
"end_date": event_doc.end_date,
"end_time": event_doc.end_time,
"is_checked_in": False,
"check_in_time": None,
"booking_id": ticket_doc.booking,
"add_ons": add_ons,
},
"payment_details": get_payment_details_for_ticket(ticket_id),
}
def get_payment_details_for_ticket(ticket_id: str) -> dict | None:
booking_id = frappe.get_cached_value("Event Ticket", ticket_id, "booking")
if not booking_id:
return None
payments = frappe.db.get_all(
"Event Payment",
filters={
"reference_doctype": "Event Booking",
"reference_docname": booking_id,
"payment_received": 1,
},
fields=["name", "amount", "currency"],
limit=1,
)
if payments:
return payments[0]
@frappe.whitelist()
def checkin_ticket(ticket_id: str) -> dict:
frappe.only_for("Frontdesk Manager", True)
checkin_date = frappe.utils.today()
validation_result = validate_ticket_for_checkin(ticket_id)
checkin_doc = frappe.new_doc("Event Check In")
checkin_doc.ticket = ticket_id
checkin_doc.date = checkin_date
checkin_doc.insert(ignore_permissions=True)
checkin_doc.submit()
return {
"message": _("Successfully checked in {attendee_name} for {checkin_date}").format(
attendee_name=validation_result["ticket"]["attendee_name"],
checkin_date=frappe.format(checkin_date, {"fieldtype": "Date"}),
),
"ticket": {
**validation_result["ticket"],
"is_checked_in": True,
"check_in_time": checkin_doc.creation,
"check_in_date": checkin_date,
},
}
@frappe.whitelist(allow_guest=True) # nosemgrep: frappe-semgrep-rules.rules.security.guest-whitelisted-method
def get_enabled_languages():
languages = frappe.get_all(
"Language",
filters={"enabled": 1},
fields=["name", "language_name", "language_code"],
order_by="language_name",
)
return languages
@frappe.whitelist()
def update_user_language(language_code: str):
if not frappe.db.exists("Language", {"language_code": language_code}):
frappe.throw(_("Invalid language"))
frappe.db.set_value("User", frappe.session.user, "language", language_code)
@frappe.whitelist(allow_guest=True) # nosemgrep: frappe-semgrep-rules.rules.security.guest-whitelisted-method
def get_translations():
if frappe.session.user != "Guest":
language = frappe.db.get_value("User", frappe.session.user, "language")
else:
language = frappe.db.get_single_value("System Settings", "language")
return get_all_translations(language)
def has_app_permission():
return True
@frappe.whitelist(allow_guest=True) # nosemgrep: frappe-semgrep-rules.rules.security.guest-whitelisted-method
def validate_coupon(coupon_code: str, event: str, user_email: str | None = None) -> dict:
event_doc = frappe.get_cached_doc("Pohodex Event Manager Event", event)
if frappe.session.user == "Guest" and not event_doc.allow_guest_booking:
frappe.throw(_("Please log in to access this feature"), frappe.AuthenticationError)
if not frappe.db.exists("Pohodex Event Manager Coupon Code", coupon_code):
return {"valid": False, "error": _("Invalid coupon code")}
coupon = frappe.get_doc("Pohodex Event Manager Coupon Code", coupon_code)
is_valid, error = coupon.is_valid_for_event(event)
if not is_valid:
return {"valid": False, "error": error}
is_available, error = coupon.is_usage_available()
if not is_available:
return {"valid": False, "error": error}
if frappe.session.user == "Guest":
check_user = user_email.lower().strip() if user_email else None
else:
check_user = frappe.session.user
is_limited, error = coupon.is_user_limit_reached(user=check_user)
if is_limited:
return {"valid": False, "error": error}
if coupon.coupon_type == "Discount":
return {
"valid": True,
"coupon_type": "Discount",
"discount_type": coupon.discount_type,
"discount_value": coupon.discount_value,
"max_discount_amount": coupon.maximum_discount_amount or 0,
"min_order_value": coupon.minimum_order_value or 0,
}
remaining = coupon.number_of_free_tickets - coupon.free_tickets_claimed
if remaining <= 0:
return {"valid": False, "error": _("All free tickets have been claimed")}
return {
"valid": True,
"coupon_type": "Free Tickets",
"ticket_type": coupon.ticket_type,
"remaining_tickets": remaining,
"free_add_ons": [a.add_on for a in coupon.free_add_ons],
}
@frappe.whitelist()
def get_campaign_details(campaign: str):
if not frappe.db.exists("Pohodex Event Manager Campaign", campaign):
frappe.throw(_("Campaign not found"), frappe.DoesNotExistError)
campaign_doc = frappe.get_cached_doc("Pohodex Event Manager Campaign", campaign)
if not campaign_doc.enabled:
frappe.throw(_("This campaign is not active"))
return {
"title": campaign_doc.title,
"description": campaign_doc.description,
"event": campaign_doc.event,
}
@frappe.whitelist()
def register_campaign_interest(campaign: str):
if frappe.session.user == "Guest":
frappe.throw(_("Please login to register your interest"))
if not is_app_installed("crm"):
frappe.throw(_("CRM integration is not available"))
if not frappe.db.exists("Pohodex Event Manager Campaign", campaign):
frappe.throw(_("Campaign not found"), frappe.DoesNotExistError)
campaign_doc = frappe.get_cached_doc("Pohodex Event Manager Campaign", campaign)
user = frappe.get_cached_doc("User", frappe.session.user)
first_name = user.first_name or user.full_name or frappe.session.user.split("@")[0]
existing_lead = frappe.db.exists(
"CRM Lead",
{"email": frappe.session.user, "buzz_campaign": campaign},
)
if existing_lead:
frappe.throw(_("You have already registered for this campaign"))
ticket = None
if campaign_doc.event:
ticket = frappe.db.get_value(
"Event Ticket",
{
"attendee_email": frappe.session.user,
"event": campaign_doc.event,
"docstatus": 1,
},
"name",
)
lead = frappe.get_doc(
{
"doctype": "CRM Lead",
"first_name": first_name,
"email": frappe.session.user,
"status": "New",
"buzz_campaign": campaign,
"event_ticket": ticket,
}
)
lead.insert(ignore_permissions=True)