"""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, "❓")