248 lines
10 KiB
Python
248 lines
10 KiB
Python
"""
|
|
app/tenant/appointments/routes.py
|
|
Appointment management: calendar view, create, edit, status workflow,
|
|
cancellation reason, no-show capture.
|
|
"""
|
|
import logging
|
|
from datetime import datetime, timezone, timedelta, date
|
|
from flask import Blueprint, render_template, redirect, url_for, flash, request, g, jsonify
|
|
from flask_login import login_required
|
|
from app.extensions import db
|
|
from app.models.salon import Appointment, Customer, Staff, Service, Location
|
|
from app.decorators import require_role, demo_readonly
|
|
from app.tenant.utils import log_tenant_action
|
|
|
|
logger = logging.getLogger(__name__)
|
|
appointments_bp = Blueprint("appointments", __name__, url_prefix="/appointments")
|
|
|
|
VALID_STATUSES = {"pending", "confirmed", "in_progress", "completed", "cancelled", "no_show"}
|
|
|
|
|
|
@appointments_bp.route("/")
|
|
@login_required
|
|
@require_role("tenant_admin", "tenant_manager")
|
|
def index():
|
|
# Default to today
|
|
date_str = request.args.get("date", date.today().isoformat())
|
|
try:
|
|
view_date = date.fromisoformat(date_str)
|
|
except ValueError:
|
|
view_date = date.today()
|
|
|
|
day_start = datetime.combine(view_date, datetime.min.time()).replace(tzinfo=timezone.utc)
|
|
day_end = day_start + timedelta(days=1)
|
|
|
|
appts = Appointment.query.filter_by(
|
|
tenant_id=g.tenant.id, location_id=g.location.id
|
|
).filter(
|
|
Appointment.start_time >= day_start,
|
|
Appointment.start_time < day_end,
|
|
).order_by(Appointment.start_time).all()
|
|
|
|
staff_list = Staff.query.filter_by(
|
|
tenant_id=g.tenant.id, is_active=True).filter(
|
|
Staff.deleted_at.is_(None)).order_by(Staff.name).all()
|
|
services = Service.query.filter_by(
|
|
tenant_id=g.tenant.id, is_active=True).filter(
|
|
Service.deleted_at.is_(None)).order_by(Service.name).all()
|
|
|
|
prev_date = view_date - timedelta(days=1)
|
|
next_date = view_date + timedelta(days=1)
|
|
|
|
return render_template("tenant/appointments/index.html",
|
|
appointments=appts, view_date=view_date,
|
|
prev_date=prev_date, next_date=next_date,
|
|
staff_list=staff_list, services=services)
|
|
|
|
|
|
@appointments_bp.route("/new", methods=["GET", "POST"])
|
|
@login_required
|
|
@require_role("tenant_admin", "tenant_manager")
|
|
@demo_readonly
|
|
def create():
|
|
staff_list = Staff.query.filter_by(
|
|
tenant_id=g.tenant.id, is_active=True).filter(
|
|
Staff.deleted_at.is_(None)).order_by(Staff.name).all()
|
|
services = Service.query.filter_by(
|
|
tenant_id=g.tenant.id, is_active=True).filter(
|
|
Service.deleted_at.is_(None)).order_by(Service.name).all()
|
|
customers = Customer.query.filter_by(
|
|
tenant_id=g.tenant.id, is_active=True).filter(
|
|
Customer.deleted_at.is_(None)).order_by(Customer.name).limit(500).all()
|
|
|
|
if request.method == "POST":
|
|
start_raw = request.form.get("start_time", "")
|
|
if not start_raw:
|
|
return render_template("tenant/appointments/form.html",
|
|
mode="create", staff_list=staff_list,
|
|
services=services, customers=customers,
|
|
error="Start time is required.")
|
|
try:
|
|
start_time = datetime.fromisoformat(start_raw)
|
|
if start_time.tzinfo is None:
|
|
start_time = start_time.replace(tzinfo=timezone.utc)
|
|
except ValueError:
|
|
return render_template("tenant/appointments/form.html",
|
|
mode="create", staff_list=staff_list,
|
|
services=services, customers=customers,
|
|
error="Invalid date/time format.")
|
|
|
|
service_id = request.form.get("service_id", type=int)
|
|
svc = Service.query.get(service_id) if service_id else None
|
|
duration = svc.duration_min if svc else 30
|
|
end_time = start_time + timedelta(minutes=duration)
|
|
|
|
customer_id = request.form.get("customer_id", type=int) or None
|
|
staff_id = request.form.get("staff_id", type=int) or None
|
|
is_walk_in = request.form.get("is_walk_in") == "1"
|
|
|
|
appt = Appointment(
|
|
tenant_id=g.tenant.id,
|
|
location_id=g.location.id,
|
|
customer_id=customer_id,
|
|
staff_id=staff_id,
|
|
service_id=service_id,
|
|
start_time=start_time,
|
|
end_time=end_time,
|
|
is_walk_in=is_walk_in,
|
|
status="confirmed" if not is_walk_in else "in_progress",
|
|
notes=request.form.get("notes", "").strip() or None,
|
|
rebook_source="manual",
|
|
created_by=_user_db_id(),
|
|
)
|
|
db.session.add(appt)
|
|
db.session.flush()
|
|
log_tenant_action("appointment.create", "appointment", appt.id,
|
|
{"customer": customer_id, "staff": staff_id,
|
|
"start": start_raw})
|
|
|
|
# Schedule 24h and 2h email reminders
|
|
from app.models.salon import AppointmentReminder as AR
|
|
for hours, rtype in [(24, "24h"), (2, "2h")]:
|
|
remind_at = start_time - timedelta(hours=hours)
|
|
if remind_at > datetime.now(timezone.utc):
|
|
db.session.add(AR(
|
|
tenant_id=g.tenant.id,
|
|
location_id=g.location.id,
|
|
appointment_id=appt.id,
|
|
reminder_type=rtype,
|
|
scheduled_for=remind_at,
|
|
channel="email",
|
|
status="pending",
|
|
))
|
|
|
|
db.session.commit()
|
|
flash("Appointment created.", "success")
|
|
return redirect(url_for("appointments.index",
|
|
date=start_time.date().isoformat()))
|
|
return render_template("tenant/appointments/form.html",
|
|
mode="create", staff_list=staff_list,
|
|
services=services, customers=customers)
|
|
|
|
|
|
@appointments_bp.route("/<int:appt_id>")
|
|
@login_required
|
|
@require_role("tenant_admin", "tenant_manager")
|
|
def view(appt_id):
|
|
appt = Appointment.query.filter_by(
|
|
id=appt_id, tenant_id=g.tenant.id).first_or_404()
|
|
return render_template("tenant/appointments/view.html", appointment=appt)
|
|
|
|
|
|
@appointments_bp.route("/<int:appt_id>/status", methods=["POST"])
|
|
@login_required
|
|
@require_role("tenant_admin", "tenant_manager")
|
|
@demo_readonly
|
|
def set_status(appt_id):
|
|
appt = Appointment.query.filter_by(
|
|
id=appt_id, tenant_id=g.tenant.id).first_or_404()
|
|
new_status = request.form.get("status", "").strip()
|
|
if new_status not in VALID_STATUSES:
|
|
flash("Invalid status.", "danger")
|
|
return redirect(url_for("appointments.view", appt_id=appt_id))
|
|
|
|
old_status = appt.status
|
|
appt.status = new_status
|
|
|
|
if new_status == "cancelled":
|
|
appt.cancellation_reason = request.form.get("reason", "").strip() or None
|
|
appt.cancelled_at = datetime.now(timezone.utc)
|
|
# Cancel pending reminders
|
|
from app.models.salon import AppointmentReminder
|
|
AppointmentReminder.query.filter_by(
|
|
appointment_id=appt_id, status="pending"
|
|
).update({"status": "cancelled"})
|
|
|
|
elif new_status == "no_show" and appt.customer_id:
|
|
# Increment no-show counter on customer record
|
|
Customer.query.filter_by(id=appt.customer_id).update(
|
|
{"no_show_count": Customer.no_show_count + 1}
|
|
)
|
|
|
|
log_tenant_action("appointment.set_status", "appointment", appt.id,
|
|
{"old": old_status, "new": new_status})
|
|
db.session.commit()
|
|
flash(f"Appointment status updated to '{new_status}'.", "success")
|
|
return redirect(url_for("appointments.view", appt_id=appt_id))
|
|
|
|
|
|
@appointments_bp.route("/<int:appt_id>/edit", methods=["GET", "POST"])
|
|
@login_required
|
|
@require_role("tenant_admin", "tenant_manager")
|
|
@demo_readonly
|
|
def edit(appt_id):
|
|
appt = Appointment.query.filter_by(
|
|
id=appt_id, tenant_id=g.tenant.id).first_or_404()
|
|
staff_list = Staff.query.filter_by(
|
|
tenant_id=g.tenant.id, is_active=True).filter(
|
|
Staff.deleted_at.is_(None)).order_by(Staff.name).all()
|
|
services = Service.query.filter_by(
|
|
tenant_id=g.tenant.id, is_active=True).filter(
|
|
Service.deleted_at.is_(None)).order_by(Service.name).all()
|
|
customers = Customer.query.filter_by(
|
|
tenant_id=g.tenant.id, is_active=True).filter(
|
|
Customer.deleted_at.is_(None)).order_by(Customer.name).limit(500).all()
|
|
|
|
if request.method == "POST":
|
|
start_raw = request.form.get("start_time", "")
|
|
try:
|
|
start_time = datetime.fromisoformat(start_raw)
|
|
if start_time.tzinfo is None:
|
|
start_time = start_time.replace(tzinfo=timezone.utc)
|
|
except ValueError:
|
|
return render_template("tenant/appointments/form.html",
|
|
mode="edit", appointment=appt,
|
|
staff_list=staff_list, services=services,
|
|
customers=customers,
|
|
error="Invalid date/time.")
|
|
service_id = request.form.get("service_id", type=int)
|
|
svc = Service.query.get(service_id) if service_id else None
|
|
duration = svc.duration_min if svc else 30
|
|
|
|
appt.customer_id = request.form.get("customer_id", type=int) or None
|
|
appt.staff_id = request.form.get("staff_id", type=int) or None
|
|
appt.service_id = service_id
|
|
appt.start_time = start_time
|
|
appt.end_time = start_time + timedelta(minutes=duration)
|
|
appt.notes = request.form.get("notes", "").strip() or None
|
|
|
|
log_tenant_action("appointment.edit", "appointment", appt.id,
|
|
{"start": start_raw})
|
|
db.session.commit()
|
|
flash("Appointment updated.", "success")
|
|
return redirect(url_for("appointments.view", appt_id=appt_id))
|
|
|
|
return render_template("tenant/appointments/form.html",
|
|
mode="edit", appointment=appt,
|
|
staff_list=staff_list, services=services,
|
|
customers=customers)
|
|
|
|
|
|
def _user_db_id():
|
|
from flask_login import current_user
|
|
try:
|
|
uid_str = current_user.get_id()
|
|
return int(uid_str.split(":")[1]) if uid_str and ":" in uid_str else None
|
|
except Exception:
|
|
return None
|