refactor: rework appointment booking lifecycle and portal verification (backport #57270) (#57294)

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
Co-authored-by: Diptanil Saha <diptanil@frappe.io>
This commit is contained in:
mergify[bot]
2026-07-21 03:23:32 +05:30
committed by GitHub
parent 2cd531d099
commit 319841d59c
13 changed files with 1279 additions and 249 deletions

View File

@@ -7,7 +7,11 @@
"engine": "InnoDB",
"field_order": [
"scheduled_time",
"column_break_xaox",
"status",
"created_through_portal",
"email_verified",
"verification_token",
"customer_details_section",
"customer_name",
"customer_phone_number",
@@ -54,7 +58,8 @@
"fieldtype": "Datetime",
"in_list_view": 1,
"label": "Scheduled Time",
"reqd": 1
"reqd": 1,
"search_index": 1
},
{
"fieldname": "status",
@@ -77,6 +82,7 @@
"fieldname": "customer_email",
"fieldtype": "Data",
"label": "Email",
"options": "Email",
"reqd": 1
},
{
@@ -99,14 +105,43 @@
"fieldtype": "Dynamic Link",
"label": "Party",
"options": "appointment_with"
},
{
"default": "0",
"fieldname": "created_through_portal",
"fieldtype": "Check",
"label": "Created through Portal",
"read_only": 1,
"set_only_once": 1
},
{
"fieldname": "column_break_xaox",
"fieldtype": "Column Break"
},
{
"default": "0",
"depends_on": "eval:doc.created_through_portal === 1;",
"fieldname": "email_verified",
"fieldtype": "Check",
"label": "Email Verified",
"read_only": 1
},
{
"fieldname": "verification_token",
"fieldtype": "Data",
"label": "Verification Token",
"hidden": 1,
"read_only": 1,
"no_copy": 1,
"search_index": 1
}
],
"links": [],
"modified": "2022-12-15 11:11:02.131986",
"modified": "2026-07-20 02:00:00.000000",
"modified_by": "Administrator",
"module": "CRM",
"name": "Appointment",
"name_case": "UPPER CASE",
"naming_rule": "Expression (old style)",
"owner": "Administrator",
"permissions": [
{
@@ -158,8 +193,9 @@
}
],
"quick_entry": 1,
"sort_field": "modified",
"row_format": "Dynamic",
"sort_field": "creation",
"sort_order": "DESC",
"states": [],
"track_changes": 1
}
}

View File

@@ -3,14 +3,20 @@
from collections import Counter
from datetime import timedelta
from urllib.parse import urlencode
import frappe
from frappe import _
from frappe.desk.form.assign_to import add as add_assignment
from frappe.model.document import Document
from frappe.share import add_docshare
from frappe.utils import get_url, getdate, now
from frappe.utils.verified_command import get_signed_params
from frappe.utils import add_to_date, cint, date_diff, get_datetime, get_url, getdate, now, now_datetime
from frappe.utils.data import sha256_hash
from erpnext.setup.doctype.holiday_list.holiday_list import is_holiday
WEEKDAYS = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"]
class Appointment(Document):
@@ -24,103 +30,227 @@ class Appointment(Document):
appointment_with: DF.Link | None
calendar_event: DF.Link | None
created_through_portal: DF.Check
customer_details: DF.LongText | None
customer_email: DF.Data
customer_name: DF.Data
customer_phone_number: DF.Data | None
customer_skype: DF.Data | None
email_verified: DF.Check
party: DF.DynamicLink | None
scheduled_time: DF.Datetime
status: DF.Literal["Open", "Unverified", "Closed"]
verification_token: DF.Data | None
# end: auto-generated types
def find_lead_by_email(self):
lead_list = frappe.get_list(
"Lead", filters={"email_id": self.customer_email}, ignore_permissions=True
)
if lead_list:
return lead_list[0].name
return None
def validate(self):
self.validate_status_update()
if not self.has_value_changed("scheduled_time"):
return
def find_customer_by_email(self):
customer_list = frappe.get_list(
"Customer", filters={"email_id": self.customer_email}, ignore_permissions=True
self.validate_backdated_booking()
if is_appointment_scheduling_enabled():
self.validate_advanced_booking()
self.validate_holiday()
self.validate_slot_timing()
self.validate_available_time_slot()
def validate_status_update(self):
if not self.has_value_changed("status"):
return
if not self.created_through_portal:
if self.status == "Unverified":
frappe.throw(_("Appointments created manually cannot have 'Unverified' status."))
return
if self.status == "Unverified" and self.email_verified:
frappe.throw(_("A verified appointment cannot be moved back to 'Unverified' status."))
if self.status == "Open" and not self.email_verified:
frappe.throw(
_("An appointment booked through the portal can only be opened via email verification.")
)
def validate_backdated_booking(self):
if get_datetime(self.scheduled_time) < now_datetime():
frappe.throw(_("Appointment cannot be scheduled for a past time."))
def validate_advanced_booking(self):
advance_booking_days = cint(get_booking_settings().advance_booking_days)
if advance_booking_days and date_diff(self.scheduled_time, now_datetime()) > advance_booking_days:
frappe.throw(
_("Appointment can only be scheduled up to {0} day(s) in advance.").format(
advance_booking_days
)
)
def validate_holiday(self):
holiday_list = get_booking_settings().holiday_list
if not holiday_list:
frappe.throw(_("Please add a valid Holiday List on Appointment Booking Settings."))
if is_holiday(holiday_list, getdate(self.scheduled_time)):
frappe.throw(_("Appointment cannot be scheduled on a holiday."))
def validate_slot_timing(self):
settings = get_booking_settings()
if not settings.availability_of_slots:
frappe.throw(_("No availability of slots are found. Please add on Appointment Booking Settings."))
scheduled_time = get_datetime(self.scheduled_time)
day_of_week = WEEKDAYS[scheduled_time.weekday()]
slot_start = timedelta(
hours=scheduled_time.hour, minutes=scheduled_time.minute, seconds=scheduled_time.second
)
if customer_list:
return customer_list[0].name
return None
slot_end = slot_start + timedelta(minutes=cint(settings.appointment_duration))
for slot in settings.availability_of_slots:
if slot.day_of_week == day_of_week and slot.from_time <= slot_start and slot_end <= slot.to_time:
return
frappe.throw(_("Appointment must be scheduled within the available slot timings."))
def validate_available_time_slot(self):
settings = get_booking_settings()
if not cint(settings.number_of_agents):
return
# the locking read serializes concurrent bookings for the same window,
# so two simultaneous requests cannot both pass the capacity check
booked = count_overlapping_appointments(
self.scheduled_time,
cint(settings.appointment_duration),
exclude_appointment=self.name,
for_update=True,
)
if booked >= cint(settings.number_of_agents):
frappe.throw(_("Time slot is not available"))
def before_insert(self):
number_of_appointments_in_same_slot = frappe.db.count(
"Appointment", filters={"scheduled_time": self.scheduled_time}
)
number_of_agents = frappe.db.get_single_value("Appointment Booking Settings", "number_of_agents")
if not number_of_agents == 0:
if number_of_appointments_in_same_slot >= number_of_agents:
frappe.throw(_("Time slot is not available"))
# Link lead
if not self.party:
lead = self.find_lead_by_email()
customer = self.find_customer_by_email()
if customer:
self.appointment_with = "Customer"
self.party = customer
else:
self.appointment_with = "Lead"
self.party = lead
# Set status to "Unverified" for new Appointments.
if self.created_through_portal:
self.status = "Unverified"
return
self.link_customer_lead()
def after_insert(self):
if self.party:
# Create Calendar event
if not self.created_through_portal and self.party:
self.auto_assign()
self.create_calendar_event()
else:
# Set status to unverified
self.status = "Unverified"
# Send email to confirm
self.send_confirmation_email()
return
# Send email to confirm
self.send_confirmation_email()
def on_update(self):
# capture transitions before nested saves during materialization
# refresh the before-save snapshot
status_changed = self.has_value_changed("status")
email_just_verified = bool(
self.created_through_portal and self.email_verified
) and self.has_value_changed("email_verified")
self.link_auto_assign_and_create_calendar_event()
if email_just_verified:
self.send_appointment_confirmed_email()
if status_changed:
self.update_event_and_assignments_status()
def on_trash(self):
# the Event only references the party, not the appointment,
# so it must be cleaned up explicitly
if not self.calendar_event:
return
event = self.calendar_event
self.db_set("calendar_event", None, update_modified=False)
frappe.delete_doc("Event", event, ignore_permissions=True)
def send_confirmation_email(self):
verify_url = self._get_verify_url()
template = "confirm_appointment"
args = {
"link": verify_url,
"site_url": frappe.utils.get_url(),
"full_name": self.customer_name,
}
self.send_email_to_customer(
template="confirm_appointment",
subject=_("Appointment Confirmation"),
args={"link": self._get_verify_url(), "expiry_minutes": get_verification_link_expiry()},
)
frappe.msgprint(_("Please check your email to confirm the appointment."))
def send_appointment_confirmed_email(self):
self.send_email_to_customer(
template="appointment_confirmed",
subject=_("Appointment Confirmed"),
args={"scheduled_time": frappe.utils.format_datetime(self.scheduled_time)},
reference_doctype="Appointment",
reference_name=self.name,
)
def send_email_to_customer(self, template, subject, args, **kwargs):
frappe.sendmail(
recipients=[self.customer_email],
template=template,
args=args,
subject=_("Appointment Confirmation"),
args={"full_name": self.customer_name, "site_url": frappe.utils.get_url(), **args},
subject=subject,
**kwargs,
)
if frappe.session.user == "Guest":
frappe.msgprint(_("Please check your email to confirm the appointment"))
else:
frappe.msgprint(
_("Appointment was created. But no lead was found. Please check the email to confirm")
)
def on_change(self):
# Sync Calendar
if not self.calendar_event:
def link_auto_assign_and_create_calendar_event(self):
if self.is_new() or (self.created_through_portal and not self.email_verified):
return
if not self.calendar_event:
# first materialization: link the party, assign an agent, create the event
self.link_customer_lead()
self.auto_assign()
self.create_calendar_event()
self.sync_calendar_event()
def sync_calendar_event(self):
if not self.calendar_event or not self.has_value_changed("scheduled_time"):
return
cal_event = frappe.get_doc("Event", self.calendar_event)
cal_event.starts_on = self.scheduled_time
cal_event.save(ignore_permissions=True)
def set_verified(self, email):
if not email == self.customer_email:
frappe.throw(_("Email verification failed."))
# Create new lead
def update_event_and_assignments_status(self):
"""Close or reopen the calendar event and assignments along with the appointment."""
if self.status == "Unverified":
return
is_closed = self.status == "Closed"
new_status = "Closed" if is_closed else "Open"
if self.calendar_event:
frappe.db.set_value("Event", self.calendar_event, "status", new_status)
# only move ToDos between Open and Closed - never touch Cancelled ones
todo_filters = {
"reference_type": "Appointment",
"reference_name": self.name,
"status": "Open" if is_closed else "Closed",
}
frappe.db.set_value("ToDo", todo_filters, "status", new_status)
def link_customer_lead(self):
if not self.party:
customer = self.find_party_by_email("Customer")
self.appointment_with = "Customer" if customer else "Lead"
self.party = customer or self.find_party_by_email("Lead")
self.create_lead_and_link()
# Remove unverified status
self.status = "Open"
# Create calender event
self.auto_assign()
self.create_calendar_event()
self.save(ignore_permissions=True)
frappe.db.commit()
def find_party_by_email(self, doctype):
party = frappe.get_all(doctype, filters={"email_id": self.customer_email}, limit=1, pluck="name")
return party[0] if party else None
def create_lead_and_link(self):
# Return if already linked
@@ -139,86 +269,39 @@ class Appointment(Document):
if self.customer_details:
lead.append(
"notes",
{
"note": self.customer_details,
"added_by": frappe.session.user,
"added_on": now(),
},
{"note": self.customer_details, "added_by": frappe.session.user, "added_on": now()},
)
lead.insert(ignore_permissions=True)
# Link lead
self.party = lead.name
self.party = lead.insert(ignore_permissions=True).name
def auto_assign(self):
existing_assignee = self.get_assignee_from_latest_opportunity()
if existing_assignee:
# If the latest opportunity is assigned to someone
# Assign the appointment to the same
self.assign_agent(existing_assignee)
return
if self._assign:
return
available_agents = _get_agents_sorted_by_asc_workload(getdate(self.scheduled_time))
for agent in available_agents:
if _check_agent_availability(agent, self.scheduled_time):
self.assign_agent(agent[0])
break
if existing_assignee := self.get_assignee_from_latest_opportunity():
# assign to whoever handles the party's latest opportunity
self.assign_agent(existing_assignee)
return
busy_agents = get_busy_agents(self.scheduled_time)
for agent in _get_agents_sorted_by_asc_workload(getdate(self.scheduled_time)):
if agent not in busy_agents:
self.assign_agent(agent)
break
def get_assignee_from_latest_opportunity(self):
if not self.party:
if not self.party or not frappe.db.exists("Lead", self.party):
return None
if not frappe.db.exists("Lead", self.party):
return None
opporutnities = frappe.get_list(
opportunities = frappe.get_all(
"Opportunity",
filters={
"party_name": self.party,
},
ignore_permissions=True,
filters={"party_name": self.party},
fields=["_assign"],
order_by="creation desc",
limit=1,
)
if not opporutnities:
return None
latest_opportunity = frappe.get_doc("Opportunity", opporutnities[0].name)
assignee = latest_opportunity._assign
if not assignee:
return None
assignee = frappe.parse_json(assignee)[0]
return assignee
def create_calendar_event(self):
if self.calendar_event:
return
appointment_event = frappe.get_doc(
{
"doctype": "Event",
"subject": " ".join(["Appointment with", self.customer_name]),
"starts_on": self.scheduled_time,
"status": "Open",
"type": "Public",
"send_reminder": frappe.db.get_single_value(
"Appointment Booking Settings", "email_reminders"
),
"event_participants": [
dict(reference_doctype=self.appointment_with, reference_docname=self.party)
],
}
)
employee = _get_employee_from_user(self._assign)
if employee:
appointment_event.append(
"event_participants", dict(reference_doctype="Employee", reference_docname=employee.name)
)
appointment_event.insert(ignore_permissions=True)
self.calendar_event = appointment_event.name
self.save(ignore_permissions=True)
def _get_verify_url(self):
verify_route = "/book_appointment/verify"
params = {"email": self.customer_email, "appointment": self.name}
return get_url(verify_route + "?" + get_signed_params(params))
assignees = opportunities and frappe.parse_json(opportunities[0]._assign or "[]")
return assignees[0] if assignees else None
def assign_agent(self, agent):
if not frappe.has_permission(doc=self, user=agent):
@@ -226,45 +309,157 @@ class Appointment(Document):
add_assignment({"doctype": self.doctype, "name": self.name, "assign_to": [agent]})
def create_calendar_event(self):
if self.calendar_event:
return
event = frappe.get_doc(
{
"doctype": "Event",
"subject": f"Appointment with {self.customer_name}",
"starts_on": self.scheduled_time,
"status": "Open",
"type": "Public",
"send_reminder": cint(get_booking_settings().email_reminders),
"event_participants": self.get_event_participants(),
}
).insert(ignore_permissions=True)
self.calendar_event = event.name
self.save(ignore_permissions=True)
def get_event_participants(self):
participants = [dict(reference_doctype=self.appointment_with, reference_docname=self.party)]
if employee := _get_employee_from_user(self._assign):
participants.append(dict(reference_doctype="Employee", reference_docname=employee.name))
return participants
def _get_verify_url(self):
key = self.generate_verification_key()
return get_url("/book_appointment/verify?" + urlencode({"key": key}))
def generate_verification_key(self):
# store only the hash; the raw key lives solely in the emailed link
key = frappe.generate_hash()
self.db_set("verification_token", sha256_hash(key), update_modified=False)
return key
def get_booking_settings():
return frappe.get_cached_doc("Appointment Booking Settings")
def is_appointment_scheduling_enabled():
return bool(cint(get_booking_settings().enable_scheduling))
def get_verification_link_expiry():
"""Verification link expiry window in minutes."""
return cint(get_booking_settings().verification_link_expiry_duration)
def count_overlapping_appointments(
scheduled_time, appointment_duration, exclude_appointment=None, for_update=False
):
"""Count non-Closed appointments whose duration window overlaps `scheduled_time`.
With `for_update`, the range stays locked until commit, serializing concurrent bookings."""
# select the rows (not COUNT) so `for_update` stays valid: PostgreSQL
# rejects `FOR UPDATE` combined with an aggregate function
appointment = frappe.qb.DocType("Appointment")
query = (
frappe.qb.from_(appointment)
.select(appointment.name)
.where(appointment.scheduled_time > add_to_date(scheduled_time, minutes=-appointment_duration))
.where(appointment.scheduled_time < add_to_date(scheduled_time, minutes=appointment_duration))
.where(appointment.status != "Closed")
)
if exclude_appointment:
query = query.where(appointment.name != exclude_appointment)
if for_update:
query = query.for_update()
return len(query.run())
def handle_expired_unverified_appointments():
"""Close or delete Unverified appointments whose verification link has expired."""
expiry = get_verification_link_expiry()
if not expiry:
return
cutoff = add_to_date(now_datetime(), minutes=-expiry)
filters = {"status": "Unverified", "creation": ("<", cutoff)}
action = get_booking_settings().action_for_expired_unverified_appointments or "Mark as Closed"
if action == "Mark as Closed":
frappe.db.set_value("Appointment", filters, "status", "Closed")
elif action == "Delete Permanently":
for name in frappe.get_all("Appointment", filters=filters, pluck="name"):
frappe.delete_doc("Appointment", name, ignore_permissions=True)
def _get_agents_sorted_by_asc_workload(date):
appointments = frappe.get_all("Appointment", fields="*")
agent_list = _get_agent_list_as_strings()
if not appointments:
return agent_list
appointment_counter = Counter(agent_list)
for appointment in appointments:
assign_data = appointment._assign
if isinstance(assign_data, str):
assign_data = assign_data.strip()
if not assign_data:
continue
assigned_to = frappe.parse_json(assign_data)
if assigned_to and (assigned_to[0] in agent_list) and getdate(appointment.scheduled_time) == date:
appointment_counter[assigned_to[0]] += 1
sorted_agent_list = appointment_counter.most_common()
sorted_agent_list.reverse()
return sorted_agent_list
# count only the given day's assignments; scheduled_time is indexed so the
# date range is resolved in SQL instead of scanning every appointment ever
workload = Counter(agent.user for agent in get_booking_settings().agent_list)
assigns = frappe.get_all(
"Appointment",
filters=[
["_assign", "is", "set"],
["scheduled_time", ">=", getdate(date)],
["scheduled_time", "<", add_to_date(getdate(date), days=1)],
],
pluck="_assign",
)
for assign in assigns:
assignees = frappe.parse_json((assign or "").strip() or "[]")
if assignees and assignees[0] in workload:
workload[assignees[0]] += 1
return [agent for agent, _workload in reversed(workload.most_common())]
def _get_agent_list_as_strings():
agent_list_as_strings = []
agent_list = frappe.get_doc("Appointment Booking Settings").agent_list
for agent in agent_list:
agent_list_as_strings.append(agent.user)
return agent_list_as_strings
def get_busy_agents(scheduled_time):
"""Agents already assigned to a non-Closed appointment overlapping `scheduled_time`."""
duration = _get_appointment_duration()
assigns = frappe.get_all(
"Appointment",
filters=[
["scheduled_time", ">", add_to_date(scheduled_time, minutes=-duration)],
["scheduled_time", "<", add_to_date(scheduled_time, minutes=duration)],
["status", "!=", "Closed"],
],
pluck="_assign",
)
return {assignee for assign in assigns for assignee in frappe.parse_json(assign or "[]")}
def _check_agent_availability(agent_email, scheduled_time):
appointemnts_at_scheduled_time = frappe.get_all("Appointment", filters={"scheduled_time": scheduled_time})
for appointment in appointemnts_at_scheduled_time:
if appointment._assign == agent_email:
return False
return True
return agent_email not in get_busy_agents(scheduled_time)
def get_booked_slot_times(from_time, to_time):
"""scheduled_times of non-Closed appointments within (from_time, to_time), for slot availability."""
return frappe.get_all(
"Appointment",
filters=[
["scheduled_time", ">", from_time],
["scheduled_time", "<", to_time],
["status", "!=", "Closed"],
],
pluck="scheduled_time",
)
def _get_appointment_duration():
return cint(get_booking_settings().appointment_duration)
def _get_employee_from_user(user):
employee_docname = frappe.db.get_value("Employee", {"user_id": user})
if employee_docname:
return frappe.get_doc("Employee", employee_docname)
return None
return frappe.get_doc("Employee", employee_docname) if employee_docname else None

View File

@@ -2,37 +2,175 @@
# See license.txt
import datetime
import unittest
from unittest.mock import patch
from urllib.parse import parse_qs, urlparse
import frappe
from frappe.tests.utils import FrappeTestCase
from frappe.utils import add_to_date, getdate, now_datetime, set_request
from frappe.utils.data import get_system_timezone, sha256_hash
from erpnext.crm.doctype.appointment.appointment import (
Appointment,
_check_agent_availability,
handle_expired_unverified_appointments,
)
from erpnext.setup.doctype.holiday_list.test_holiday_list import make_holiday_list
from erpnext.www.book_appointment.index import create_appointment, get_appointment_slots
from erpnext.www.book_appointment.verify import index as verify_index
LEAD_EMAIL = "test_appointment_lead@example.com"
VERIFICATION_EXPIRY_MINUTES = 30
ALL_WEEKDAYS = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"]
def create_test_appointment():
test_appointment = frappe.get_doc(
{
"doctype": "Appointment",
"status": "Open",
"customer_name": "Test Lead",
"customer_phone_number": "666",
"customer_skype": "test",
"customer_email": LEAD_EMAIL,
"scheduled_time": datetime.datetime.now(),
"customer_details": "Hello, Friend!",
}
)
def create_test_appointment(**kwargs):
args = {
"doctype": "Appointment",
"status": "Open",
"customer_name": "Test Lead",
"customer_phone_number": "666",
"customer_skype": "test",
"customer_email": LEAD_EMAIL,
"scheduled_time": add_to_date(now_datetime(), hours=2),
"customer_details": "Hello, Friend!",
}
args.update(kwargs)
test_appointment = frappe.get_doc(args)
test_appointment.insert()
return test_appointment
class TestAppointment(unittest.TestCase):
def setUpClass():
frappe.db.delete("Lead", {"email_id": LEAD_EMAIL})
def create_lead(email, name="Existing Lead"):
frappe.db.delete("Lead", {"email_id": email})
return frappe.get_doc({"doctype": "Lead", "lead_name": name, "email_id": email}).insert(
ignore_permissions=True
)
def set_booking_setting(field, value):
frappe.db.set_single_value("Appointment Booking Settings", field, value)
def slot_on(days_from_now, hour, minute=0):
day = datetime.date.today() + datetime.timedelta(days=days_from_now)
return datetime.datetime.combine(day, datetime.time(hour, minute))
def backdate_creation(appointment_name, minutes):
frappe.db.set_value(
"Appointment",
appointment_name,
"creation",
add_to_date(now_datetime(), minutes=-minutes),
update_modified=False,
)
def get_status(appointment_name):
return frappe.db.get_value("Appointment", appointment_name, "status")
def get_assignees(appointment_name):
return frappe.parse_json(frappe.db.get_value("Appointment", appointment_name, "_assign") or "[]")
def get_todo_statuses(appointment_name):
return frappe.get_all(
"ToDo",
filters={"reference_type": "Appointment", "reference_name": appointment_name},
pluck="status",
)
def parse_verify_url(verify_url):
parsed = urlparse(verify_url)
return parsed, {key: value[0] for key, value in parse_qs(parsed.query).items()}
class TestAppointment(FrappeTestCase):
def setUp(self):
# sending an email commits the transaction (EmailQueue sets its status
# with commit=True), which would break the per-test rollback below
frappe.flags.mute_emails = 1
set_booking_setting("verification_link_expiry_duration", VERIFICATION_EXPIRY_MINUTES)
frappe.db.delete("Lead", {"email_id": LEAD_EMAIL})
self.test_appointment = create_test_appointment()
self.test_appointment.set_verified(self.test_appointment.customer_email)
def tearDown(self):
frappe.db.rollback()
frappe.clear_document_cache("Appointment Booking Settings", "Appointment Booking Settings")
frappe.flags.mute_emails = 0
def _configure_booking_settings(self, holiday_dates=None, agents=None):
holiday_list = make_holiday_list(
"_Test Appointment Holiday List",
from_date=getdate(),
to_date=add_to_date(getdate(), days=60),
holiday_dates=holiday_dates or [],
)
settings = frappe.get_doc("Appointment Booking Settings")
settings.enable_scheduling = 1
settings.enable_appointment_portal = 1
settings.appointment_duration = 30
settings.advance_booking_days = 30
settings.verification_link_expiry_duration = VERIFICATION_EXPIRY_MINUTES
settings.holiday_list = holiday_list.name
settings.set("agent_list", [])
for agent in agents or ["Administrator"]:
settings.append("agent_list", {"user": agent})
settings.set("availability_of_slots", [])
for day in ALL_WEEKDAYS:
settings.append(
"availability_of_slots", {"day_of_week": day, "from_time": "09:00:00", "to_time": "17:00:00"}
)
settings.save()
def _create_portal_appointment(self, email, days_from_now=7, time="10:00:00"):
"""Book as Guest. The verification email is mocked and kept on
``self._verification_email_mock`` for assertions."""
if not getattr(self, "_booking_settings_configured", False):
self._configure_booking_settings()
self._booking_settings_configured = True
with self.set_user("Guest"), patch.object(Appointment, "send_confirmation_email") as mock_send:
appointment = create_appointment(
date=str(datetime.date.today() + datetime.timedelta(days=days_from_now)),
time=time,
tz=get_system_timezone(),
contact={"name": "Portal Visitor", "email": email, "number": "123", "skype": "", "notes": ""},
)
self._verification_email_mock = mock_send
return appointment
def _request_verification(self, appointment, verify_url=None):
"""Simulate the GET request made by clicking the emailed verification link.
The confirmation email sent on successful verification is mocked and kept
on ``self._confirmed_email_mock`` for assertions.
"""
parsed, params = parse_verify_url(verify_url or appointment._get_verify_url())
old_request = getattr(frappe.local, "request", None)
old_form_dict = frappe.local.form_dict
old_user = frappe.session.user
try:
# the real link is clicked by an anonymous visitor; set_user resets
# form_dict, so switch the user before populating the request
frappe.set_user("Guest")
set_request(method="GET", path=f"{parsed.path}?{parsed.query}")
frappe.local.form_dict = frappe._dict(params)
context = frappe._dict()
with patch.object(Appointment, "send_appointment_confirmed_email") as mock_confirmed:
verify_index.get_context(context)
self._confirmed_email_mock = mock_confirmed
return context
finally:
frappe.set_user(old_user)
frappe.local.request = old_request
frappe.local.form_dict = old_form_dict
frappe.local.flags.commit = False
def test_calendar_event_created(self):
cal_event = frappe.get_doc("Event", self.test_appointment.calendar_event)
@@ -40,3 +178,369 @@ class TestAppointment(unittest.TestCase):
def test_lead_linked(self):
self.assertTrue(self.test_appointment.party)
def test_desk_created_appointment_skips_email_verification(self):
"""Appointments created from the desk (created_through_portal unset) must be
linked and confirmed immediately - no verification email should be sent."""
with patch.object(Appointment, "send_confirmation_email") as mock_send:
appointment = create_test_appointment(customer_email="another_desk_lead@example.com")
mock_send.assert_not_called()
self.assertEqual(appointment.status, "Open")
self.assertTrue(appointment.party)
frappe.db.delete("Lead", {"email_id": "another_desk_lead@example.com"})
def test_portal_booking_stays_unverified_for_existing_lead(self):
"""A portal booking whose email matches an existing Lead/Customer must NOT
be auto-linked - it must stay Unverified until the email is confirmed."""
create_lead("existing_lead@example.com")
appointment = self._create_portal_appointment("existing_lead@example.com", days_from_now=5)
self._verification_email_mock.assert_called_once()
self.assertTrue(appointment.created_through_portal)
self.assertEqual(appointment.status, "Unverified")
self.assertFalse(appointment.email_verified)
self.assertFalse(appointment.party)
def test_verify_url_uses_opaque_token(self):
appointment = self._create_portal_appointment("portal_visitor@example.com")
parsed, params = parse_verify_url(appointment._get_verify_url())
# the link carries only an opaque key - no email, name or signed params
self.assertEqual(set(params), {"key"})
self.assertNotIn("email", parsed.query)
# only the hash of that key is stored on the appointment
stored = frappe.db.get_value("Appointment", appointment.name, "verification_token")
self.assertEqual(stored, sha256_hash(params["key"]))
def test_email_verification_within_expiry_window(self):
# Link used within the validity window - verification succeeds and the
# appointment gets linked, assigned and added to the calendar
on_time = self._create_portal_appointment("portal_visitor_on_time@example.com")
context = self._request_verification(on_time)
self.assertTrue(context.success)
self._confirmed_email_mock.assert_called_once()
on_time.reload()
self.assertEqual(on_time.status, "Open")
self.assertTrue(on_time.email_verified)
self.assertTrue(on_time.party)
self.assertTrue(on_time.calendar_event)
# Link used after the validity window - verification fails
late = self._create_portal_appointment("portal_visitor_late@example.com", days_from_now=10)
after_expiry = add_to_date(now_datetime(), minutes=VERIFICATION_EXPIRY_MINUTES + 1)
with patch.object(verify_index, "now_datetime", return_value=after_expiry):
context = self._request_verification(late)
self.assertFalse(context.success)
self._confirmed_email_mock.assert_not_called()
late.reload()
self.assertEqual(late.status, "Unverified")
self.assertFalse(late.email_verified)
self.assertFalse(late.party)
def test_verification_link_reused_after_success(self):
appointment = self._create_portal_appointment("portal_visitor_twice@example.com")
verify_url = appointment._get_verify_url()
context = self._request_verification(appointment, verify_url=verify_url)
self.assertTrue(context.success)
self._confirmed_email_mock.assert_called_once()
# re-clicking the link is idempotent and does not send another email
context = self._request_verification(appointment, verify_url=verify_url)
self.assertTrue(context.success)
self.assertIn("already verified", context.message)
self._confirmed_email_mock.assert_not_called()
def test_verification_link_for_deleted_appointment(self):
"""A verification link can outlive its appointment - clicking it must
render a friendly message, not crash."""
appointment = self._create_portal_appointment("portal_visitor_gone@example.com")
verify_url = appointment._get_verify_url()
frappe.delete_doc("Appointment", appointment.name, ignore_permissions=True)
context = self._request_verification(appointment, verify_url=verify_url)
self.assertFalse(context.success)
self.assertIn("book the appointment again", context.message)
def test_reschedule_syncs_calendar_event(self):
new_time = add_to_date(self.test_appointment.scheduled_time, hours=1)
self.test_appointment.scheduled_time = new_time
self.test_appointment.save()
starts_on = frappe.db.get_value("Event", self.test_appointment.calendar_event, "starts_on")
self.assertEqual(starts_on, new_time)
def test_portal_endpoint_disabled(self):
self._configure_booking_settings()
set_booking_setting("enable_appointment_portal", 0)
with self.set_user("Guest"), self.assertRaises(frappe.Redirect):
create_appointment(
date=str(datetime.date.today() + datetime.timedelta(days=3)),
time="10:00:00",
tz="UTC",
contact={
"name": "Blocked",
"email": "blocked@example.com",
"number": "1",
"skype": "",
"notes": "",
},
)
def test_booked_slot_unavailable_on_portal(self):
self._configure_booking_settings()
tz = get_system_timezone()
day = datetime.date.today() + datetime.timedelta(days=2)
def get_availability():
with self.set_user("Guest"):
slots = get_appointment_slots(str(day), tz)
return {slot["time"].strftime("%H:%M"): slot["availability"] for slot in slots}
booked = create_test_appointment(
customer_email="slot_taken@example.com", scheduled_time=slot_on(2, 10)
)
availability = get_availability()
self.assertFalse(availability["10:00"])
self.assertTrue(availability["13:00"])
# closing the appointment frees its slot on the portal
booked.status = "Closed"
booked.save()
self.assertTrue(get_availability()["10:00"])
# an off-grid desk appointment blocks every portal slot it overlaps
create_test_appointment(customer_email="off_grid@example.com", scheduled_time=slot_on(2, 13, 15))
availability = get_availability()
self.assertFalse(availability["13:00"])
self.assertFalse(availability["13:30"])
self.assertTrue(availability["14:00"])
def test_expired_unverified_appointments_are_closed(self):
stale = self._create_portal_appointment("portal_visitor_stale@example.com", days_from_now=8)
fresh = self._create_portal_appointment("portal_visitor_fresh@example.com", days_from_now=9)
verify_url = stale._get_verify_url()
backdate_creation(stale.name, VERIFICATION_EXPIRY_MINUTES + 15)
set_booking_setting("action_for_expired_unverified_appointments", "Mark as Closed")
handle_expired_unverified_appointments()
self.assertEqual(get_status(stale.name), "Closed")
self.assertEqual(get_status(fresh.name), "Unverified")
# Open appointments are never touched, regardless of age
self.assertEqual(get_status(self.test_appointment.name), "Open")
# clicking the link of a closed appointment renders a friendly message
context = self._request_verification(stale, verify_url=verify_url)
self.assertFalse(context.success)
self.assertIn("closed", context.message)
def test_expired_unverified_appointments_are_deleted(self):
stale = self._create_portal_appointment("portal_visitor_purged@example.com", days_from_now=8)
fresh = self._create_portal_appointment("portal_visitor_kept@example.com", days_from_now=9)
backdate_creation(stale.name, VERIFICATION_EXPIRY_MINUTES + 15)
set_booking_setting("action_for_expired_unverified_appointments", "Delete Permanently")
handle_expired_unverified_appointments()
self.assertFalse(frappe.db.exists("Appointment", stale.name))
self.assertTrue(frappe.db.exists("Appointment", fresh.name))
self.assertTrue(frappe.db.exists("Appointment", self.test_appointment.name))
def test_cleanup_skipped_when_expiry_not_configured(self):
appointment = self._create_portal_appointment("portal_visitor_no_expiry@example.com")
backdate_creation(appointment.name, 5)
set_booking_setting("verification_link_expiry_duration", 0)
handle_expired_unverified_appointments()
self.assertEqual(get_status(appointment.name), "Unverified")
def test_status_transition_rules(self):
# desk appointments can never be Unverified
with self.assertRaises(frappe.ValidationError):
create_test_appointment(customer_email="desk_unverified@example.com", status="Unverified")
# portal appointments cannot be opened manually before verification
unverified = self._create_portal_appointment("manual_open@example.com")
unverified.status = "Open"
with self.assertRaises(frappe.ValidationError):
unverified.save(ignore_permissions=True)
# verified appointments cannot be reverted to Unverified
verified = self._create_portal_appointment("revert_unverified@example.com", days_from_now=8)
self._request_verification(verified)
verified.reload()
verified.status = "Unverified"
with self.assertRaises(frappe.ValidationError):
verified.save(ignore_permissions=True)
# both desk and verified portal appointments can be closed and reopened
for appointment in (self.test_appointment, verified):
appointment.reload()
appointment.status = "Closed"
appointment.save(ignore_permissions=True)
appointment.status = "Open"
appointment.save(ignore_permissions=True)
self.assertEqual(appointment.status, "Open")
def test_agent_auto_assignment(self):
agent_email = "appointment_agent@example.com"
if not frappe.db.exists("User", agent_email):
frappe.get_doc(
{"doctype": "User", "email": agent_email, "first_name": "Appointment Agent"}
).insert(ignore_permissions=True)
self._configure_booking_settings(agents=["Administrator", agent_email])
first = create_test_appointment(
customer_email="assigned_one@example.com", scheduled_time=slot_on(2, 11)
)
second = create_test_appointment(
customer_email="assigned_two@example.com", scheduled_time=slot_on(2, 11)
)
# both appointments in the same slot get an agent, and never the same one
self.assertTrue(get_assignees(first.name))
self.assertTrue(get_assignees(second.name))
self.assertNotEqual(get_assignees(first.name), get_assignees(second.name))
# closing an assigned appointment closes its ToDo without re-assigning
first.reload()
first.status = "Closed"
first.save()
self.assertTrue(get_todo_statuses(first.name))
self.assertTrue(all(status == "Closed" for status in get_todo_statuses(first.name)))
# reopening brings the ToDos back
first.status = "Open"
first.save()
self.assertTrue(all(status == "Open" for status in get_todo_statuses(first.name)))
def test_agent_busy_for_the_whole_appointment_duration(self):
self._configure_booking_settings()
slot = slot_on(3, 11)
appointment = create_test_appointment(customer_email="busy_agent@example.com", scheduled_time=slot)
assignee = get_assignees(appointment.name)[0]
# busy anywhere inside the 30-minute appointment window, free right after it
self.assertFalse(_check_agent_availability(assignee, slot))
self.assertFalse(_check_agent_availability(assignee, slot + datetime.timedelta(minutes=15)))
self.assertTrue(_check_agent_availability(assignee, slot + datetime.timedelta(minutes=30)))
def test_closed_appointment_closes_calendar_event(self):
self.test_appointment.status = "Closed"
self.test_appointment.save()
event_status = frappe.db.get_value("Event", self.test_appointment.calendar_event, "status")
self.assertEqual(event_status, "Closed")
# reopening the appointment reopens the calendar event
self.test_appointment.status = "Open"
self.test_appointment.save()
event_status = frappe.db.get_value("Event", self.test_appointment.calendar_event, "status")
self.assertEqual(event_status, "Open")
def test_deleting_appointment_deletes_calendar_event(self):
event = self.test_appointment.calendar_event
self.assertTrue(frappe.db.exists("Event", event))
frappe.delete_doc("Appointment", self.test_appointment.name)
self.assertFalse(frappe.db.exists("Event", event))
def test_backdated_appointment_is_rejected(self):
with self.assertRaises(frappe.ValidationError):
create_test_appointment(
customer_email="backdated@example.com",
scheduled_time=add_to_date(now_datetime(), hours=-1),
)
def test_booking_beyond_advance_window_is_rejected(self):
self._configure_booking_settings()
set_booking_setting("advance_booking_days", 7)
# within the advance booking window - allowed
within = create_test_appointment(
customer_email="advance_within@example.com", scheduled_time=slot_on(5, 10)
)
self.assertTrue(frappe.db.exists("Appointment", within.name))
# beyond the advance booking window - rejected
with self.assertRaises(frappe.ValidationError):
create_test_appointment(
customer_email="advance_beyond@example.com", scheduled_time=slot_on(8, 10)
)
def test_appointment_on_holiday_is_rejected(self):
holiday = add_to_date(getdate(), days=3)
self._configure_booking_settings(
holiday_dates=[{"holiday_date": holiday, "description": "Test Holiday"}]
)
with self.assertRaises(frappe.ValidationError):
create_test_appointment(customer_email="on_holiday@example.com", scheduled_time=slot_on(3, 10))
# the day after the holiday is bookable
after_holiday = create_test_appointment(
customer_email="after_holiday@example.com", scheduled_time=slot_on(4, 10)
)
self.assertTrue(frappe.db.exists("Appointment", after_holiday.name))
def test_appointment_outside_slot_timing_is_rejected(self):
self._configure_booking_settings()
# before the slot opens
with self.assertRaises(frappe.ValidationError):
create_test_appointment(customer_email="before_opening@example.com", scheduled_time=slot_on(2, 8))
# starts within the slot but would end after it closes
with self.assertRaises(frappe.ValidationError):
create_test_appointment(
customer_email="past_closing@example.com", scheduled_time=slot_on(2, 16, 45)
)
# within the slot timings
within = create_test_appointment(
customer_email="within_slot@example.com", scheduled_time=slot_on(2, 10)
)
self.assertTrue(frappe.db.exists("Appointment", within.name))
def test_overlapping_time_slot_capacity(self):
set_booking_setting("number_of_agents", 1)
set_booking_setting("appointment_duration", 30)
slot = slot_on(1, 10)
first = create_test_appointment(customer_email="slot_first@example.com", scheduled_time=slot)
# a booking starting inside the first appointment's duration is rejected
with self.assertRaises(frappe.ValidationError):
create_test_appointment(
customer_email="slot_overlap@example.com",
scheduled_time=slot + datetime.timedelta(minutes=15),
)
# rescheduling must not count the appointment's own booked slot
first.scheduled_time = slot + datetime.timedelta(minutes=10)
first.save()
# a booking starting exactly when the rescheduled one ends is allowed
adjacent = create_test_appointment(
customer_email="slot_adjacent@example.com",
scheduled_time=slot + datetime.timedelta(minutes=40),
)
self.assertTrue(frappe.db.exists("Appointment", adjacent.name))
# a closed (cancelled) appointment frees its slot
first.status = "Closed"
first.save()
after_cancellation = create_test_appointment(
customer_email="after_cancellation@example.com", scheduled_time=slot
)
self.assertTrue(frappe.db.exists("Appointment", after_cancellation.name))

View File

@@ -1,48 +1,56 @@
{
"actions": [],
"allow_bulk_edit": 1,
"creation": "2019-08-27 10:56:48.309824",
"doctype": "DocType",
"editable_grid": 1,
"engine": "InnoDB",
"field_order": [
"enable_scheduling",
"agent_detail_section",
"availability_of_slots",
"number_of_agents",
"agent_list",
"holiday_list",
"appointment_details_section",
"appointment_duration",
"email_reminders",
"column_break_ehiq",
"agent_list",
"number_of_agents",
"agent_detail_section",
"enable_scheduling",
"availability_of_slots",
"section_break_bkln",
"column_break_alwa",
"advance_booking_days",
"column_break_bspp",
"holiday_list",
"success_details",
"success_redirect_url"
"enable_appointment_portal",
"verification_link_expiry_duration",
"column_break_fovk",
"success_redirect_url",
"action_for_expired_unverified_appointments"
],
"fields": [
{
"depends_on": "eval:doc.enable_scheduling === 1;",
"fieldname": "availability_of_slots",
"fieldtype": "Table",
"label": "Availability Of Slots",
"options": "Appointment Booking Slots",
"reqd": 1
"mandatory_depends_on": "eval:doc.enable_scheduling === 1;",
"options": "Appointment Booking Slots"
},
{
"default": "1",
"fieldname": "number_of_agents",
"fieldtype": "Int",
"hidden": 1,
"in_list_view": 1,
"label": "Number of Concurrent Appointments",
"read_only": 1,
"reqd": 1
"read_only": 1
},
{
"depends_on": "eval:doc.enable_scheduling === 1;",
"fieldname": "holiday_list",
"fieldtype": "Link",
"in_list_view": 1,
"label": "Holiday List",
"options": "Holiday List",
"reqd": 1
"mandatory_depends_on": "eval:doc.enable_scheduling === 1;",
"options": "Holiday List"
},
{
"default": "60",
@@ -60,29 +68,31 @@
},
{
"default": "7",
"depends_on": "eval:doc.enable_scheduling === 1;",
"fieldname": "advance_booking_days",
"fieldtype": "Int",
"label": "Number of days appointments can be booked in advance",
"reqd": 1
"mandatory_depends_on": "eval:doc.enable_scheduling === 1;"
},
{
"fieldname": "agent_list",
"fieldtype": "Table MultiSelect",
"label": "Agents",
"options": "Assignment Rule User",
"reqd": 1
"mandatory_depends_on": "eval:doc.enable_scheduling === 1;",
"options": "Assignment Rule User"
},
{
"default": "0",
"fieldname": "enable_scheduling",
"fieldtype": "Check",
"label": "Enable Appointment Scheduling",
"reqd": 1
"mandatory_depends_on": "eval:doc.enable_appointment_portal === 1;"
},
{
"fieldname": "agent_detail_section",
"fieldtype": "Section Break",
"label": "Agent Details"
"hide_border": 1,
"label": "Appointment Scheduling"
},
{
"fieldname": "appointment_details_section",
@@ -92,18 +102,68 @@
{
"fieldname": "success_details",
"fieldtype": "Section Break",
"label": "Success Settings"
"label": "Appointment Booking Portal Settings"
},
{
"description": "Leave blank for home.\nThis is relative to site URL, for example \"about\" will redirect to \"https://yoursitename.com/about\"",
"fieldname": "success_redirect_url",
"fieldtype": "Data",
"label": "Success Redirect URL"
"label": "Success Redirect URL",
"permlevel": 1
},
{
"default": "30",
"depends_on": "eval: doc.enable_scheduling === 1;",
"description": "In Minutes (min: 15 mins, max: 60 mins)",
"fieldname": "verification_link_expiry_duration",
"fieldtype": "Int",
"label": "Verification Link Expiry Duration",
"mandatory_depends_on": "eval:doc.enable_appointment_portal === 1;",
"max_value": 60.0,
"min_value": 15.0,
"non_negative": 1,
"permlevel": 1
},
{
"fieldname": "column_break_ehiq",
"fieldtype": "Column Break"
},
{
"default": "0",
"fieldname": "enable_appointment_portal",
"fieldtype": "Check",
"label": "Enable Appointment Booking Through Portal",
"permlevel": 1
},
{
"fieldname": "column_break_fovk",
"fieldtype": "Column Break"
},
{
"default": "Mark as Closed",
"fieldname": "action_for_expired_unverified_appointments",
"fieldtype": "Select",
"label": "Action for Expired Unverified Appointments",
"options": "Mark as Closed\nDelete Permanently",
"permlevel": 1
},
{
"fieldname": "section_break_bkln",
"fieldtype": "Section Break"
},
{
"fieldname": "column_break_alwa",
"fieldtype": "Column Break"
},
{
"fieldname": "column_break_bspp",
"fieldtype": "Column Break"
}
],
"grid_page_length": 50,
"issingle": 1,
"links": [],
"modified": "2022-12-15 11:10:13.517742",
"modified": "2026-07-20 00:11:18.996384",
"modified_by": "Administrator",
"module": "CRM",
"name": "Appointment Booking Settings",
@@ -137,6 +197,15 @@
"role": "Sales Manager",
"share": 1,
"write": 1
},
{
"email": 1,
"permlevel": 1,
"print": 1,
"read": 1,
"role": "System Manager",
"share": 1,
"write": 1
}
],
"quick_entry": 1,
@@ -144,4 +213,4 @@
"sort_order": "DESC",
"states": [],
"track_changes": 1
}
}

View File

@@ -3,11 +3,11 @@
import datetime
import typing
import frappe
from frappe import _
from frappe.model.document import Document
from frappe.utils import getdate
class AppointmentBookingSettings(Document):
@@ -26,33 +26,43 @@ class AppointmentBookingSettings(Document):
AppointmentBookingSlots,
)
action_for_expired_unverified_appointments: DF.Literal["Mark as Closed", "Delete Permanently"]
advance_booking_days: DF.Int
agent_list: DF.TableMultiSelect[AssignmentRuleUser]
appointment_duration: DF.Int
availability_of_slots: DF.Table[AppointmentBookingSlots]
email_reminders: DF.Check
enable_appointment_portal: DF.Check
enable_scheduling: DF.Check
holiday_list: DF.Link
holiday_list: DF.Link | None
number_of_agents: DF.Int
success_redirect_url: DF.Data | None
verification_link_expiry_duration: DF.Int
# end: auto-generated types
agent_list: typing.ClassVar[list] = [] # Hack
min_date = "01/01/1970 "
format_string = "%d/%m/%Y %H:%M:%S"
def validate(self):
self.validate_availability_of_slots()
def save(self):
self.number_of_agents = len(self.agent_list)
super().save()
self.validate_appointment_scheduling()
self.validate_portal_booking()
def validate_appointment_scheduling(self):
if not self.enable_scheduling:
return
self.validate_availability_of_slots()
self.validate_holiday_list()
self.validate_advance_booking_days()
def validate_availability_of_slots(self):
if not self.availability_of_slots:
frappe.throw(
_("Please fill up the Availability of Slots table to enable Appointment Scheduling.")
)
format_string = "%Y-%m-%d %H:%M:%S"
for record in self.availability_of_slots:
from_time = datetime.datetime.strptime(self.min_date + record.from_time, self.format_string)
to_time = datetime.datetime.strptime(self.min_date + record.to_time, self.format_string)
to_time - from_time
from_time = datetime.datetime.strptime(f"1970-01-01 {record.from_time}", format_string)
to_time = datetime.datetime.strptime(f"1970-01-01 {record.to_time}", format_string)
self.validate_from_and_to_time(from_time, to_time, record)
self.duration_is_divisible(from_time, to_time)
@@ -67,3 +77,38 @@ class AppointmentBookingSettings(Document):
timedelta = to_time - from_time
if timedelta.total_seconds() % (self.appointment_duration * 60):
frappe.throw(_("The difference between from time and To Time must be a multiple of Appointment"))
def validate_holiday_list(self):
if not self.holiday_list:
frappe.throw(_("Please select a Holiday List to enable Appointment Scheduling."))
hl_from_date, hl_to_date = frappe.get_cached_value(
"Holiday List", self.holiday_list, ["from_date", "to_date"]
)
now = getdate()
if not (now >= hl_from_date and now <= hl_to_date):
frappe.throw(_("Holiday List - {0} is not valid for current date.").format(self.holiday_list))
def validate_advance_booking_days(self):
if not self.advance_booking_days:
frappe.throw(_("Advance Booking Days is mandatory for Appointment Scheduling."))
def validate_portal_booking(self):
if not self.enable_appointment_portal:
return
if not self.enable_scheduling:
frappe.throw(
_("Appointment Scheduling needs to be enabled for Appointment Booking through portal.")
)
self.validate_link_expiry_duration()
def validate_link_expiry_duration(self):
if (
not self.verification_link_expiry_duration
or self.verification_link_expiry_duration > 60
or self.verification_link_expiry_duration < 15
):
frappe.throw(_("'Verification Link Expiry Duration' must be between 15 to 60 minutes."))

View File

@@ -1,9 +1,125 @@
# Copyright (c) 2019, Frappe Technologies Pvt. Ltd. and Contributors
# See license.txt
# import frappe
import unittest
import datetime
import frappe
from frappe.tests.utils import FrappeTestCase
from frappe.utils import add_to_date, getdate
from erpnext.setup.doctype.holiday_list.test_holiday_list import make_holiday_list
class TestAppointmentBookingSettings(unittest.TestCase):
pass
class TestAppointmentBookingSettings(FrappeTestCase):
def assert_invalid(self, settings):
with self.assertRaises(frappe.ValidationError):
settings.save()
def make_settings(self, appointment_duration=30):
doc = frappe.new_doc("Appointment Booking Settings")
doc.appointment_duration = appointment_duration
return doc
def dt(self, hms):
# the controller parses times against a fixed epoch date
return datetime.datetime.strptime("1970-01-01 " + hms, "%Y-%m-%d %H:%M:%S")
def get_valid_scheduling_settings(self):
holiday_list = make_holiday_list(
"_Test Booking Settings Holiday List",
from_date=getdate(),
to_date=add_to_date(getdate(), days=30),
holiday_dates=[],
)
settings = frappe.get_doc("Appointment Booking Settings")
settings.enable_scheduling = 1
settings.appointment_duration = 30
settings.advance_booking_days = 7
settings.verification_link_expiry_duration = 30
settings.holiday_list = holiday_list.name
settings.set("agent_list", [])
settings.append("agent_list", {"user": "Administrator"})
settings.set("availability_of_slots", [])
settings.append(
"availability_of_slots",
{"day_of_week": "Monday", "from_time": "09:00:00", "to_time": "17:00:00"},
)
return settings
def test_from_time_must_precede_to_time(self):
doc = self.make_settings()
record = frappe._dict(day_of_week="Monday")
self.assertRaises(
frappe.ValidationError,
doc.validate_from_and_to_time,
self.dt("18:00:00"),
self.dt("09:00:00"),
record,
)
doc.validate_from_and_to_time(self.dt("09:00:00"), self.dt("18:00:00"), record) # valid order
def test_slot_length_must_be_a_multiple_of_the_duration(self):
doc = self.make_settings(appointment_duration=30)
# 60 minutes is two 30-minute appointments -> fine
doc.duration_is_divisible(self.dt("09:00:00"), self.dt("10:00:00"))
# 45 minutes leaves a partial appointment -> rejected
self.assertRaises(
frappe.ValidationError, doc.duration_is_divisible, self.dt("09:00:00"), self.dt("09:45:00")
)
def test_scheduling_requires_slots(self):
settings = self.get_valid_scheduling_settings()
settings.set("availability_of_slots", [])
self.assert_invalid(settings)
def test_validate_checks_every_slot(self):
settings = self.get_valid_scheduling_settings()
settings.append(
"availability_of_slots",
{"day_of_week": "Tuesday", "from_time": "09:00:00", "to_time": "09:45:00"},
)
self.assert_invalid(settings)
def test_scheduling_requires_holiday_list_covering_today(self):
settings = self.get_valid_scheduling_settings()
settings.holiday_list = None
self.assert_invalid(settings)
expired_list = make_holiday_list(
"_Test Booking Settings Expired Holiday List",
from_date=add_to_date(getdate(), days=-60),
to_date=add_to_date(getdate(), days=-30),
holiday_dates=[],
)
settings.holiday_list = expired_list.name
self.assert_invalid(settings)
def test_scheduling_requires_advance_booking_days(self):
settings = self.get_valid_scheduling_settings()
settings.advance_booking_days = 0
self.assert_invalid(settings)
def test_portal_requires_scheduling(self):
settings = frappe.get_doc("Appointment Booking Settings")
settings.enable_scheduling = 0
settings.enable_appointment_portal = 1
self.assert_invalid(settings)
def test_portal_expiry_duration_bounds(self):
settings = self.get_valid_scheduling_settings()
settings.enable_appointment_portal = 1
settings.verification_link_expiry_duration = 5
self.assert_invalid(settings)
def test_number_of_agents_derived_from_agent_list(self):
settings = self.get_valid_scheduling_settings()
settings.number_of_agents = 99
settings.save()
self.assertEqual(frappe.db.get_single_value("Appointment Booking Settings", "number_of_agents"), 1)

View File

@@ -431,6 +431,7 @@ scheduler_events = {
],
"hourly_long": [],
"hourly_maintenance": [
"erpnext.crm.doctype.appointment.appointment.handle_expired_unverified_appointments",
"erpnext.stock.doctype.repost_item_valuation.repost_item_valuation.repost_entries",
"erpnext.utilities.bulk_transaction.retry",
"erpnext.projects.doctype.project.project.collect_project_status",

View File

@@ -0,0 +1,6 @@
<p>{{_("Dear")}} {{ full_name }},</p>
<p>{{_("Your email has been verified and your appointment has been confirmed for {0}").format(scheduled_time)}}.</p>
<p>{{_("We look forward to meeting you")}}.</p>
<br>
<p style="font-size: 85%;">{{_("This email was sent from {0}").format(site_url)}}</p>

View File

@@ -1,6 +1,7 @@
<p>{{_("Dear")}} {{ full_name }}{% if last_name %} {{ last_name}}{% endif %},</p>
<p>{{_("A new appointment has been created for you with {0}").format(site_url)}}.</p>
<p>{{_("Click on the link below to verify your email and confirm the appointment")}}.</p>
<p>{{_("This link is valid for {0} minutes").format(expiry_minutes)}}.</p>
<p style="margin: 30px 0px;">
<a href="{{ link }}" rel="nofollow" style="padding: 8px 20px; background-color: #7575ff; color: #fff; border-radius: 4px; text-decoration: none; line-height: 1; border-bottom: 3px solid rgba(0, 0, 0, 0.2); font-size: 14px; font-weight: 200;">{{ _("Verify Email") }}</a>

View File

@@ -237,9 +237,9 @@ async function submit() {
frappe.show_alert(__("Appointment Created Successfully"));
}
setTimeout(() => {
let redirect_url = "/";
let redirect_url = "/book_appointment";
if (window.appointment_settings.success_redirect_url) {
redirect_url += window.appointment_settings.success_redirect_url;
redirect_url = `/${window.appointment_settings.success_redirect_url}`;
}
window.location.href = redirect_url;
}, 5000);

View File

@@ -4,6 +4,7 @@ import json
import frappe
import pytz
from frappe import _
from frappe.rate_limiter import rate_limit
from frappe.utils.data import get_system_timezone
WEEKDAYS = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"]
@@ -18,7 +19,7 @@ def get_context(context):
def handle_appointment_booking_disabled():
if not frappe.get_single_value("Appointment Booking Settings", "enable_scheduling"):
if not frappe.get_single_value("Appointment Booking Settings", "enable_appointment_portal"):
frappe.redirect_to_message(
_("Appointment Scheduling Disabled"),
_("Appointment Scheduling has been disabled for this site"),
@@ -66,6 +67,8 @@ def get_appointment_slots(date, timezone):
)
holiday_list = frappe.get_doc("Holiday List", settings.holiday_list)
timeslots = get_available_slots_between(query_start_time, query_end_time, settings)
# fetch the day's booked slots once instead of querying per timeslot
booked_times = get_booked_slot_times_for(timeslots, settings.appointment_duration)
# Filter and convert timeslots
converted_timeslots = []
@@ -76,7 +79,7 @@ def get_appointment_slots(date, timezone):
converted_timeslots.append(dict(time=converted_timeslot, availability=False))
continue
# Check availability
if check_availabilty(timeslot, settings) and converted_timeslot >= now:
if is_slot_available(timeslot, booked_times, settings) and converted_timeslot >= now:
converted_timeslots.append(dict(time=converted_timeslot, availability=True))
else:
converted_timeslots.append(dict(time=converted_timeslot, availability=False))
@@ -102,7 +105,8 @@ def get_available_slots_between(query_start_time, query_end_time, settings):
return timeslots
@frappe.whitelist(allow_guest=True)
@frappe.whitelist(allow_guest=True, methods=["POST"])
@rate_limit(limit=5, seconds=300)
def create_appointment(date, time, tz, contact):
handle_appointment_booking_disabled()
format_string = "%Y-%m-%d %H:%M:%S"
@@ -114,13 +118,13 @@ def create_appointment(date, time, tz, contact):
# Create a appointment document from form
appointment = frappe.new_doc("Appointment")
appointment.scheduled_time = scheduled_time
contact = json.loads(contact)
contact = frappe.parse_json(contact)
appointment.customer_name = contact.get("name", None)
appointment.customer_phone_number = contact.get("number", None)
appointment.customer_skype = contact.get("skype", None)
appointment.customer_details = contact.get("notes", None)
appointment.customer_email = contact.get("email", None)
appointment.status = "Open"
appointment.created_through_portal = 1
appointment.insert(ignore_permissions=True)
return appointment
@@ -150,8 +154,23 @@ def convert_to_system_timezone(guest_tz, datetimeobject):
return datetimeobject
def check_availabilty(timeslot, settings):
return frappe.db.count("Appointment", {"scheduled_time": timeslot}) < settings.number_of_agents
def get_booked_slot_times_for(timeslots, appointment_duration):
if not timeslots:
return []
from erpnext.crm.doctype.appointment.appointment import get_booked_slot_times
duration = datetime.timedelta(minutes=appointment_duration)
return get_booked_slot_times(min(timeslots) - duration, max(timeslots) + duration)
def is_slot_available(timeslot, booked_times, settings):
# mirror the server capacity check: count non-Closed appointments whose
# duration window overlaps this slot, without a per-slot query
duration = datetime.timedelta(minutes=settings.appointment_duration)
lower, upper = timeslot - duration, timeslot + duration
overlapping = sum(1 for booked in booked_times if lower < booked < upper)
return overlapping < settings.number_of_agents
def _is_holiday(date, holiday_list):

View File

@@ -12,7 +12,7 @@
</div>
{% else %}
<div class="alert alert-danger">
{{ _("Verification failed please check the link") }}
{{ message or _("Verification failed please check the link") }}
</div>
{% endif %}
{% endblock%}

View File

@@ -1,20 +1,58 @@
import frappe
from frappe.utils.verified_command import verify_request
from frappe import _
from frappe.utils import add_to_date, now_datetime
from frappe.utils.data import sha256_hash
from erpnext.crm.doctype.appointment.appointment import get_verification_link_expiry
def get_context(context):
if not verify_request():
key = frappe.form_dict.get("key")
if not key:
context.success = False
return context
email = frappe.form_dict["email"]
appointment_name = frappe.form_dict["appointment"]
appointment_name = frappe.db.get_value("Appointment", {"verification_token": sha256_hash(key)}, "name")
if not appointment_name:
context.success = False
context.message = _("This verification link is invalid. Please book the appointment again.")
return context
if email and appointment_name:
appointment = frappe.get_doc("Appointment", appointment_name)
appointment.set_verified(email)
appointment = frappe.get_doc("Appointment", appointment_name)
# report a settled status before expiry: a closed/verified appointment is
# more informative than a generic "expired" (and creation-based expiry would
# otherwise mask a sweeper-closed appointment)
if appointment.status == "Closed":
context.success = False
context.message = _("Appointment has been closed. Please book the appointment again.")
return context
if appointment.status == "Open":
context.success = True
context.message = _("Appointment is already verified.")
return context
else:
if now_datetime() > add_to_date(appointment.creation, minutes=get_verification_link_expiry()):
context.success = False
context.message = _("Verification link has expired.")
return context
verify_appointment(appointment)
# GET requests are rolled back at the end of the request unless this flag is set
frappe.local.flags.commit = True
context.success = True
return context
def verify_appointment(appointment):
# the signed link is the authorization; materializing the appointment
# (agent assignment) needs system privileges the Guest visitor lacks
visitor = frappe.session.user
try:
frappe.set_user("Administrator")
appointment.email_verified = True
appointment.status = "Open"
appointment.save(ignore_permissions=True)
finally:
frappe.set_user(visitor)