Ty
15f6a6c713
Add operator health checks (make doctor) wrapping platform CLIs, and the fixtures-only daily board skill library with unit tests (make verify).
280 lines
9.6 KiB
Python
280 lines
9.6 KiB
Python
"""Deterministic board builder.
|
||
|
||
Takes a list of Appointment objects and produces a DayBoard with:
|
||
- Sorted appointments
|
||
- Computed gaps between consecutive appointments per staff member
|
||
- Confirmation flags
|
||
- Offline/fixture labeling
|
||
|
||
All logic is deterministic — no model inference.
|
||
See design/det-vs-inf.md.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
import warnings
|
||
from datetime import datetime, time
|
||
from typing import Any
|
||
|
||
from lumina_skills.domain import Appointment, AppointmentStatus, DayBoard, Gap
|
||
|
||
|
||
def build_board(
|
||
appointments: list[Appointment],
|
||
date: str,
|
||
salon_name: str,
|
||
source: str = "fixtures",
|
||
business_hours: dict[str, str] | None = None,
|
||
) -> DayBoard:
|
||
"""Build a complete DayBoard from a list of appointments.
|
||
|
||
Args:
|
||
appointments: Raw appointment list (from fixtures or live source).
|
||
date: The board date in YYYY-MM-DD format.
|
||
salon_name: Display name of the salon.
|
||
source: Data source label — "fixtures", "offline", "vagaro", "square".
|
||
business_hours: Optional {"open": "HH:MM", "close": "HH:MM"} to
|
||
compute gaps at day boundaries.
|
||
|
||
Returns:
|
||
A fully populated DayBoard.
|
||
"""
|
||
is_offline = source in ("fixtures", "offline")
|
||
|
||
# Filter out cancelled appointments for the board view.
|
||
active = [a for a in appointments if a.status != AppointmentStatus.CANCELLED]
|
||
|
||
# Sort by start time.
|
||
active.sort(key=lambda a: a.start_time)
|
||
|
||
# Compute gaps per staff member.
|
||
gaps = _compute_gaps(active, date, business_hours)
|
||
|
||
# Confirmation flags: pending appointments that need confirmation.
|
||
needs_confirmation = [a for a in active if a.needs_confirmation]
|
||
|
||
# Totals.
|
||
total_booked = sum(a.duration_minutes() for a in active)
|
||
total_gap = sum(g.duration_minutes for g in gaps)
|
||
|
||
return DayBoard(
|
||
date=date,
|
||
salon_name=salon_name,
|
||
source=source,
|
||
is_offline=is_offline,
|
||
appointments=active,
|
||
gaps=gaps,
|
||
needs_confirmation=needs_confirmation,
|
||
total_booked_minutes=total_booked,
|
||
total_gap_minutes=total_gap,
|
||
)
|
||
|
||
|
||
def _compute_gaps(
|
||
appointments: list[Appointment],
|
||
date: str,
|
||
business_hours: dict[str, str] | None = None,
|
||
) -> list[Gap]:
|
||
"""Compute unbooked gaps between consecutive appointments per staff.
|
||
|
||
Gaps are computed per staff member. If business_hours is provided,
|
||
gaps from open→first appointment and last appointment→close are
|
||
included (only if >= 30 minutes).
|
||
|
||
Args:
|
||
appointments: Sorted list of active appointments.
|
||
date: Board date string (YYYY-MM-DD).
|
||
business_hours: Optional {"open": "HH:MM", "close": "HH:MM"}.
|
||
|
||
Returns:
|
||
List of Gap objects.
|
||
"""
|
||
gaps: list[Gap] = []
|
||
|
||
# Group appointments by staff.
|
||
staff_apts: dict[str, list[Appointment]] = {}
|
||
for apt in appointments:
|
||
staff_apts.setdefault(apt.staff_name, []).append(apt)
|
||
|
||
for staff_name, apts in staff_apts.items():
|
||
# apts is already sorted by start_time from the caller.
|
||
open_time = None
|
||
close_time = None
|
||
if business_hours:
|
||
open_str = business_hours.get("open", "")
|
||
close_str = business_hours.get("close", "")
|
||
if open_str:
|
||
open_time = _parse_time_str(open_str)
|
||
if open_time is None:
|
||
warnings.warn(
|
||
f"Invalid business_hours.open format: {open_str!r} "
|
||
f"(expected HH:MM). Skipping open boundary gap.",
|
||
UserWarning,
|
||
stacklevel=2,
|
||
)
|
||
if close_str:
|
||
close_time = _parse_time_str(close_str)
|
||
if close_time is None:
|
||
warnings.warn(
|
||
f"Invalid business_hours.close format: {close_str!r} "
|
||
f"(expected HH:MM). Skipping close boundary gap.",
|
||
UserWarning,
|
||
stacklevel=2,
|
||
)
|
||
|
||
# Gap from open to first appointment.
|
||
if open_time and apts:
|
||
first_start = apts[0].start_time_only()
|
||
gap_mins = _time_diff_minutes(open_time, first_start)
|
||
if gap_mins >= 30:
|
||
gaps.append(Gap(
|
||
start_time=open_time,
|
||
end_time=first_start,
|
||
duration_minutes=gap_mins,
|
||
staff_name=staff_name,
|
||
following_appointment_id=apts[0].appointment_id,
|
||
))
|
||
|
||
# Gaps between consecutive appointments.
|
||
for i in range(len(apts) - 1):
|
||
current_end = apts[i].end_time_only()
|
||
next_start = apts[i + 1].start_time_only()
|
||
gap_mins = _time_diff_minutes(current_end, next_start)
|
||
if gap_mins < 0:
|
||
warnings.warn(
|
||
f"Overlapping appointments for {staff_name}: "
|
||
f"{apts[i].appointment_id} ends at {current_end} but "
|
||
f"{apts[i + 1].appointment_id} starts at {next_start} "
|
||
f"({abs(gap_mins)} min overlap). Gap skipped.",
|
||
UserWarning,
|
||
stacklevel=2,
|
||
)
|
||
continue
|
||
if gap_mins >= 30:
|
||
gaps.append(Gap(
|
||
start_time=current_end,
|
||
end_time=next_start,
|
||
duration_minutes=gap_mins,
|
||
staff_name=staff_name,
|
||
preceding_appointment_id=apts[i].appointment_id,
|
||
following_appointment_id=apts[i + 1].appointment_id,
|
||
))
|
||
|
||
# Gap from last appointment to close.
|
||
if close_time and apts:
|
||
last_end = apts[-1].end_time_only()
|
||
gap_mins = _time_diff_minutes(last_end, close_time)
|
||
if gap_mins >= 30:
|
||
gaps.append(Gap(
|
||
start_time=last_end,
|
||
end_time=close_time,
|
||
duration_minutes=gap_mins,
|
||
staff_name=staff_name,
|
||
preceding_appointment_id=apts[-1].appointment_id,
|
||
))
|
||
|
||
return gaps
|
||
|
||
|
||
def _parse_time_str(raw: str) -> time | None:
|
||
"""Parse an HH:MM string into a time object.
|
||
|
||
Returns None if the format is invalid (not HH:MM with valid ranges).
|
||
"""
|
||
m = re.fullmatch(r"(\d{2}):(\d{2})", raw)
|
||
if m is None:
|
||
return None
|
||
h, mi = int(m.group(1)), int(m.group(2))
|
||
if h > 23 or mi > 59:
|
||
return None
|
||
return time(h, mi)
|
||
|
||
|
||
def _time_diff_minutes(start: time, end: time) -> int:
|
||
"""Minutes between two time objects (same day assumed).
|
||
|
||
Returns a negative value when *end* is before *start* (overlap).
|
||
Callers should check for negative results and warn.
|
||
"""
|
||
diff = datetime.combine(datetime.today(), end) - datetime.combine(datetime.today(), start)
|
||
return int(diff.total_seconds() // 60)
|
||
|
||
|
||
def format_board_text(board: DayBoard) -> str:
|
||
"""Format a DayBoard as structured text for chat display.
|
||
|
||
This is deterministic formatting — no model inference.
|
||
The model may rephrase when presenting to the owner, but the
|
||
facts come from this function.
|
||
"""
|
||
lines: list[str] = []
|
||
|
||
# Header with offline label.
|
||
source_label = "📋 FIXTURE DATA" if board.is_offline else "📅 LIVE DATA"
|
||
lines.append(f"═══ {board.salon_name} — {board.date} ═══")
|
||
lines.append(f"[{source_label}]")
|
||
lines.append("")
|
||
|
||
# Appointments.
|
||
lines.append("── Appointments ──")
|
||
if not board.appointments:
|
||
lines.append(" No appointments.")
|
||
else:
|
||
for apt in board.appointments:
|
||
start_str = apt.start_time.strftime("%H:%M")
|
||
end_str = apt.end_time.strftime("%H:%M")
|
||
status_icon = _status_icon(apt.status)
|
||
confirm_flag = " ⚠️ CONFIRM" if apt.needs_confirmation else ""
|
||
lines.append(
|
||
f" {start_str}–{end_str} {status_icon} {apt.client_name}"
|
||
f" — {apt.service_name} ({apt.staff_name}){confirm_flag}"
|
||
)
|
||
if apt.notes:
|
||
lines.append(f" 📝 {apt.notes}")
|
||
lines.append("")
|
||
|
||
# Gaps.
|
||
lines.append("── Gaps (≥30 min) ──")
|
||
if not board.gaps:
|
||
lines.append(" No significant gaps.")
|
||
else:
|
||
for gap in board.gaps:
|
||
start_str = gap.start_time.strftime("%H:%M")
|
||
end_str = gap.end_time.strftime("%H:%M")
|
||
lines.append(
|
||
f" {start_str}–{end_str} ({gap.duration_minutes} min) "
|
||
f"— {gap.staff_name}"
|
||
)
|
||
lines.append("")
|
||
|
||
# Confirmation needed.
|
||
if board.needs_confirmation:
|
||
lines.append("── Needs Confirmation ──")
|
||
for apt in board.needs_confirmation:
|
||
start_str = apt.start_time.strftime("%H:%M")
|
||
lines.append(
|
||
f" ⚠️ {apt.client_name} — {apt.service_name} at {start_str}"
|
||
)
|
||
lines.append("")
|
||
|
||
# Summary.
|
||
lines.append("── Summary ──")
|
||
lines.append(f" Booked: {board.total_booked_minutes} min | Gaps: {board.total_gap_minutes} min")
|
||
lines.append(f" Appointments: {len(board.appointments)} | "
|
||
f"Need confirmation: {len(board.needs_confirmation)}")
|
||
|
||
return "\n".join(lines)
|
||
|
||
|
||
def _status_icon(status: AppointmentStatus) -> str:
|
||
"""Emoji icon for appointment status."""
|
||
icons = {
|
||
AppointmentStatus.CONFIRMED: "✅",
|
||
AppointmentStatus.PENDING: "⏳",
|
||
AppointmentStatus.COMPLETED: "✔️",
|
||
AppointmentStatus.NO_SHOW: "❌",
|
||
AppointmentStatus.CANCELLED: "🚫",
|
||
}
|
||
return icons.get(status, "❓")
|