407 lines
16 KiB
Python
407 lines
16 KiB
Python
"""
|
|
app/tenant/pos/routes.py
|
|
POS / Checkout: new transaction, line item entry, promotion auto-apply,
|
|
tip, gift card redemption, void, receipt.
|
|
Rebook-at-checkout creates a new pending appointment.
|
|
"""
|
|
import logging
|
|
from datetime import datetime, timezone, timedelta
|
|
from decimal import Decimal, ROUND_HALF_UP
|
|
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 (
|
|
Transaction, TransactionItem, Appointment, Customer, Staff,
|
|
Service, Product, GiftCard, AppointmentReminder,
|
|
)
|
|
from app.decorators import require_role, demo_readonly
|
|
from app.tenant.utils import log_tenant_action, get_active_promotion, apply_promotion_to_price
|
|
|
|
logger = logging.getLogger(__name__)
|
|
pos_bp = Blueprint("pos", __name__, url_prefix="/pos")
|
|
|
|
PAYMENT_METHODS = ["cash", "zelle", "venmo", "cashapp", "gift_card", "other"]
|
|
|
|
|
|
@pos_bp.route("/")
|
|
@login_required
|
|
@require_role("tenant_admin", "tenant_manager")
|
|
def checkout():
|
|
"""New POS transaction entry form."""
|
|
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()
|
|
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.category, Service.name).all()
|
|
products = Product.query.filter_by(
|
|
tenant_id=g.tenant.id, is_active=True).filter(
|
|
Product.deleted_at.is_(None)).order_by(Product.name).all()
|
|
# Pre-fill from appointment if provided
|
|
appt_id = request.args.get("appointment_id", type=int)
|
|
appointment = None
|
|
if appt_id:
|
|
appointment = Appointment.query.filter_by(
|
|
id=appt_id, tenant_id=g.tenant.id).first()
|
|
return render_template("tenant/pos/checkout.html",
|
|
customers=customers, staff_list=staff_list,
|
|
services=services, products=products,
|
|
appointment=appointment,
|
|
payment_methods=PAYMENT_METHODS)
|
|
|
|
|
|
@pos_bp.route("/submit", methods=["POST"])
|
|
@login_required
|
|
@require_role("tenant_admin", "tenant_manager")
|
|
@demo_readonly
|
|
def submit():
|
|
"""Process a completed checkout."""
|
|
customer_id = request.form.get("customer_id", type=int) or None
|
|
staff_id = request.form.get("staff_id", type=int) or None
|
|
appt_id = request.form.get("appointment_id", type=int) or None
|
|
payment_method = request.form.get("payment_method", "cash")
|
|
payment_reference = request.form.get("payment_reference", "").strip() or None
|
|
tip_raw = request.form.get("tip_amount", "0")
|
|
gc_code = request.form.get("gift_card_code", "").strip().upper() or None
|
|
|
|
if payment_method not in PAYMENT_METHODS:
|
|
flash("Invalid payment method.", "danger")
|
|
return redirect(url_for("pos.checkout"))
|
|
|
|
try:
|
|
tip_amount = Decimal(tip_raw or "0").quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
|
|
except Exception:
|
|
tip_amount = Decimal("0")
|
|
|
|
# Parse line items from form: service_ids[] and product_ids[]
|
|
service_ids = request.form.getlist("service_ids")
|
|
product_ids = request.form.getlist("product_ids")
|
|
|
|
if not service_ids and not product_ids:
|
|
flash("Please add at least one service or product.", "danger")
|
|
return redirect(url_for("pos.checkout"))
|
|
|
|
# Resolve gift card
|
|
gift_card = None
|
|
gc_applied = Decimal("0")
|
|
if gc_code:
|
|
gift_card = GiftCard.query.filter_by(
|
|
tenant_id=g.tenant.id, code=gc_code, is_active=True).first()
|
|
if not gift_card or gift_card.remaining_balance <= 0:
|
|
flash(f"Gift card {gc_code} is invalid or has no balance.", "danger")
|
|
return redirect(url_for("pos.checkout"))
|
|
|
|
# Build transaction
|
|
txn = Transaction(
|
|
tenant_id=g.tenant.id,
|
|
location_id=g.location.id,
|
|
appointment_id=appt_id,
|
|
customer_id=customer_id,
|
|
staff_id=staff_id,
|
|
payment_method=payment_method,
|
|
payment_reference=payment_reference,
|
|
tip_amount=float(tip_amount),
|
|
gift_card_id=gift_card.id if gift_card else None,
|
|
subtotal=0, discount=0, gift_card_amount=0, total=0,
|
|
)
|
|
db.session.add(txn)
|
|
db.session.flush()
|
|
|
|
subtotal = Decimal("0")
|
|
total_discount = Decimal("0")
|
|
|
|
# Add service line items
|
|
for sid_str in service_ids:
|
|
try:
|
|
sid = int(sid_str)
|
|
except ValueError:
|
|
continue
|
|
svc = Service.query.filter_by(
|
|
id=sid, tenant_id=g.tenant.id).first()
|
|
if not svc:
|
|
continue
|
|
original_price = Decimal(str(svc.price))
|
|
promo = get_active_promotion(g.tenant.id, sid, "service")
|
|
unit_price_f, disc_pct = apply_promotion_to_price(float(original_price), promo)
|
|
unit_price = Decimal(str(unit_price_f))
|
|
discount_amt = original_price - unit_price
|
|
item = TransactionItem(
|
|
transaction_id=txn.id,
|
|
service_id=sid, product_id=None,
|
|
qty=1, unit_price=float(unit_price),
|
|
original_price=float(original_price),
|
|
discount_percent=disc_pct,
|
|
promotion_id=promo.id if promo else None,
|
|
)
|
|
db.session.add(item)
|
|
subtotal += unit_price
|
|
total_discount += discount_amt
|
|
|
|
# Add product line items
|
|
for pid_str in product_ids:
|
|
try:
|
|
pid = int(pid_str)
|
|
except ValueError:
|
|
continue
|
|
prod = Product.query.filter_by(
|
|
id=pid, tenant_id=g.tenant.id).first()
|
|
if not prod:
|
|
continue
|
|
original_price = Decimal(str(prod.sale_price))
|
|
promo = get_active_promotion(g.tenant.id, pid, "product")
|
|
unit_price_f, disc_pct = apply_promotion_to_price(float(original_price), promo)
|
|
unit_price = Decimal(str(unit_price_f))
|
|
discount_amt = original_price - unit_price
|
|
item = TransactionItem(
|
|
transaction_id=txn.id,
|
|
service_id=None, product_id=pid,
|
|
qty=1, unit_price=float(unit_price),
|
|
original_price=float(original_price),
|
|
discount_percent=disc_pct,
|
|
promotion_id=promo.id if promo else None,
|
|
)
|
|
db.session.add(item)
|
|
subtotal += unit_price
|
|
total_discount += discount_amt
|
|
|
|
# Apply gift card
|
|
if gift_card:
|
|
gc_available = Decimal(str(gift_card.remaining_balance))
|
|
gc_applied = min(gc_available, subtotal + tip_amount)
|
|
gift_card.remaining_balance = float(gc_available - gc_applied)
|
|
if gift_card.remaining_balance <= 0:
|
|
gift_card.is_active = False
|
|
|
|
total = subtotal + tip_amount - gc_applied
|
|
if total < 0:
|
|
total = Decimal("0")
|
|
|
|
txn.subtotal = float(subtotal)
|
|
txn.discount = float(total_discount)
|
|
txn.tip_amount = float(tip_amount)
|
|
txn.gift_card_amount = float(gc_applied)
|
|
txn.total = float(total)
|
|
|
|
# Mark appointment completed if linked
|
|
if appt_id:
|
|
Appointment.query.filter_by(
|
|
id=appt_id, tenant_id=g.tenant.id
|
|
).update({"status": "completed"})
|
|
|
|
# Commission log if staff has commission enabled
|
|
if staff_id:
|
|
staff = Staff.query.get(staff_id)
|
|
if staff and staff.commission_enabled and staff.commission_rate:
|
|
from app.models.salon import CommissionLog
|
|
commission_amount = float(
|
|
(Decimal(str(subtotal)) * Decimal(str(staff.commission_rate)) /
|
|
Decimal("100")).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
|
|
)
|
|
from datetime import date
|
|
period = date.today().strftime("%Y-W%U")
|
|
comm_log = CommissionLog(
|
|
tenant_id=g.tenant.id,
|
|
location_id=g.location.id,
|
|
staff_id=staff_id,
|
|
transaction_id=txn.id,
|
|
amount=commission_amount,
|
|
period=period,
|
|
)
|
|
db.session.add(comm_log)
|
|
|
|
# ── Inventory auto-deduction for product line items ─────
|
|
for item in db.session.query(TransactionItem).filter_by(
|
|
transaction_id=txn.id
|
|
).filter(TransactionItem.product_id.isnot(None)).all():
|
|
from app.models.salon import Inventory, InventoryLog, Product as _Prod
|
|
prod = _Prod.query.get(item.product_id)
|
|
if prod and prod.sku:
|
|
inv = Inventory.query.filter_by(
|
|
tenant_id=g.tenant.id,
|
|
location_id=g.location.id,
|
|
sku=prod.sku,
|
|
).first()
|
|
elif prod:
|
|
inv = Inventory.query.filter_by(
|
|
tenant_id=g.tenant.id,
|
|
location_id=g.location.id,
|
|
name=prod.name,
|
|
).first()
|
|
else:
|
|
inv = None
|
|
|
|
if inv and inv.qty_on_hand > 0:
|
|
inv.qty_on_hand -= item.qty
|
|
db.session.add(InventoryLog(
|
|
tenant_id=g.tenant.id,
|
|
location_id=g.location.id,
|
|
inventory_id=inv.id,
|
|
delta=-item.qty,
|
|
reason=f"POS sale txn#{txn.id}",
|
|
))
|
|
|
|
log_tenant_action("transaction.create", "transaction", txn.id,
|
|
{"total": float(total), "payment": payment_method})
|
|
db.session.commit()
|
|
|
|
flash(f"Checkout complete — Total: ${float(total):.2f}", "success")
|
|
|
|
# Rebook at checkout
|
|
rebook = request.form.get("rebook") == "1"
|
|
if rebook:
|
|
return redirect(url_for("pos.rebook", transaction_id=txn.id))
|
|
|
|
return redirect(url_for("pos.receipt", transaction_id=txn.id))
|
|
|
|
|
|
@pos_bp.route("/receipt/<int:transaction_id>")
|
|
@login_required
|
|
@require_role("tenant_admin", "tenant_manager")
|
|
def receipt(transaction_id):
|
|
txn = Transaction.query.filter_by(
|
|
id=transaction_id, tenant_id=g.tenant.id).first_or_404()
|
|
items = list(txn.items)
|
|
return render_template("tenant/pos/receipt.html",
|
|
transaction=txn, items=items, tenant=g.tenant,
|
|
location=g.location)
|
|
|
|
|
|
@pos_bp.route("/rebook/<int:transaction_id>", methods=["GET", "POST"])
|
|
@login_required
|
|
@require_role("tenant_admin", "tenant_manager")
|
|
@demo_readonly
|
|
def rebook(transaction_id):
|
|
"""Next-visit scheduling at checkout."""
|
|
txn = Transaction.query.filter_by(
|
|
id=transaction_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()
|
|
|
|
if request.method == "POST":
|
|
start_raw = request.form.get("start_time", "")
|
|
service_id = request.form.get("service_id", type=int)
|
|
staff_id = request.form.get("staff_id", type=int) or None
|
|
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/pos/rebook.html",
|
|
transaction=txn, staff_list=staff_list,
|
|
services=services,
|
|
error="Invalid date/time.")
|
|
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)
|
|
|
|
appt = Appointment(
|
|
tenant_id=g.tenant.id, location_id=g.location.id,
|
|
customer_id=txn.customer_id,
|
|
staff_id=staff_id, service_id=service_id,
|
|
start_time=start_time, end_time=end_time,
|
|
is_walk_in=False, status="pending",
|
|
rebook_source="checkout",
|
|
rebooked_from_transaction_id=txn.id,
|
|
created_by=_user_db_id(),
|
|
)
|
|
db.session.add(appt)
|
|
db.session.flush()
|
|
|
|
# Schedule 24h reminder
|
|
reminder_time = start_time - timedelta(hours=24)
|
|
if reminder_time > datetime.now(timezone.utc):
|
|
reminder = AppointmentReminder(
|
|
tenant_id=g.tenant.id, location_id=g.location.id,
|
|
appointment_id=appt.id, reminder_type="24h",
|
|
scheduled_for=reminder_time, channel="email", status="pending",
|
|
)
|
|
db.session.add(reminder)
|
|
|
|
log_tenant_action("appointment.rebook", "appointment", appt.id,
|
|
{"from_transaction": txn.id, "start": start_raw})
|
|
db.session.commit()
|
|
flash("Next visit scheduled.", "success")
|
|
return redirect(url_for("pos.receipt", transaction_id=txn.id))
|
|
|
|
return render_template("tenant/pos/rebook.html",
|
|
transaction=txn, staff_list=staff_list,
|
|
services=services)
|
|
|
|
|
|
@pos_bp.route("/void/<int:transaction_id>", methods=["GET", "POST"])
|
|
@login_required
|
|
@require_role("tenant_admin")
|
|
@demo_readonly
|
|
def void(transaction_id):
|
|
txn = Transaction.query.filter_by(
|
|
id=transaction_id, tenant_id=g.tenant.id).first_or_404()
|
|
if txn.voided_at:
|
|
flash("This transaction is already voided.", "warning")
|
|
return redirect(url_for("pos.receipt", transaction_id=transaction_id))
|
|
|
|
if request.method == "POST":
|
|
reason = request.form.get("reason", "").strip()
|
|
if not reason:
|
|
return render_template("tenant/pos/void.html",
|
|
transaction=txn, error="Reason is required.")
|
|
txn.voided_at = datetime.now(timezone.utc)
|
|
txn.voided_by = _user_db_id()
|
|
txn.void_reason = reason
|
|
|
|
# Reverse gift card balance if applicable
|
|
if txn.gift_card_id and txn.gift_card_amount > 0:
|
|
gc = GiftCard.query.get(txn.gift_card_id)
|
|
if gc:
|
|
gc.remaining_balance = float(
|
|
Decimal(str(gc.remaining_balance)) +
|
|
Decimal(str(txn.gift_card_amount))
|
|
)
|
|
gc.is_active = True
|
|
|
|
log_tenant_action("transaction.void", "transaction", txn.id,
|
|
{"reason": reason, "total": txn.total})
|
|
db.session.commit()
|
|
flash("Transaction voided.", "success")
|
|
return redirect(url_for("pos.receipt", transaction_id=transaction_id))
|
|
|
|
return render_template("tenant/pos/void.html", transaction=txn)
|
|
|
|
|
|
@pos_bp.route("/transactions")
|
|
@login_required
|
|
@require_role("tenant_admin", "tenant_manager")
|
|
def transactions():
|
|
from datetime import date
|
|
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)
|
|
txns = Transaction.query.filter_by(
|
|
tenant_id=g.tenant.id, location_id=g.location.id
|
|
).filter(
|
|
Transaction.created_at >= day_start,
|
|
Transaction.created_at < day_end,
|
|
Transaction.voided_at.is_(None),
|
|
).order_by(Transaction.created_at.desc()).all()
|
|
return render_template("tenant/pos/transactions.html",
|
|
transactions=txns, view_date=view_date)
|
|
|
|
|
|
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
|