Spaces:
Runtime error
Runtime error
| """Calendar route — produces FullCalendar-compatible events from appointments + provider blocks.""" | |
| from datetime import datetime, timedelta | |
| from fastapi import APIRouter, Depends | |
| from core import data_manager | |
| from core import settings_manager as settings | |
| from core.scheduling_engine import add_minutes | |
| from api.deps import get_current_session | |
| router = APIRouter(prefix="/api/calendar", tags=["calendar"]) | |
| # Deterministic color palette assigned to providers in load order. | |
| _PROVIDER_COLORS = [ | |
| "#2563eb", "#16a34a", "#dc2626", "#9333ea", "#ea580c", | |
| "#0891b2", "#ca8a04", "#db2777", "#4f46e5", "#65a30d", | |
| ] | |
| def _provider_color_map(providers: list[dict]) -> dict[str, str]: | |
| return {p["id"]: _PROVIDER_COLORS[i % len(_PROVIDER_COLORS)] for i, p in enumerate(providers)} | |
| def get_events( | |
| start: str | None = None, | |
| end: str | None = None, | |
| provider_ids: str | None = None, | |
| show_unavailable: bool = True, | |
| show_travel: bool = True, | |
| show_oncall: bool = True, | |
| session: dict = Depends(get_current_session), | |
| ): | |
| providers = data_manager.get_all_providers(active_only=False) | |
| color_map = _provider_color_map(providers) | |
| selected_ids: set[str] | None = None | |
| if session.get("user_role") == "provider": | |
| selected_ids = {session.get("provider_id")} | |
| elif provider_ids: | |
| selected_ids = set(provider_ids.split(",")) if provider_ids else None | |
| visible_providers = [p for p in providers if (selected_ids is None or p["id"] in selected_ids)] | |
| events = [] | |
| is_provider_view = session.get("user_role") == "provider" | |
| # Appointments — include both scheduled and completed; render completed in grey. | |
| appointments = [a for a in data_manager.get_all_appointments() if a.get("status") in ("scheduled", "completed")] | |
| for a in appointments: | |
| pid = a.get("provider_id") | |
| if selected_ids is not None and pid not in selected_ids: | |
| continue | |
| status = a.get("status", "scheduled") | |
| is_completed = status == "completed" | |
| color = "#94a3b8" if is_completed else color_map.get(pid, "#2563eb") | |
| if is_provider_view: | |
| title = f"Visit — {a.get('visit_level', '')}" | |
| else: | |
| title = f"{a.get('patient_name', '')} — {a.get('provider_name', '')}" | |
| if is_completed: | |
| title = f"✓ {title}" | |
| events.append({ | |
| "id": f"appt_{a['id']}", | |
| "title": title, | |
| "start": f"{a['date']}T{a['start_time']}:00", | |
| "end": f"{a['date']}T{a['end_time']}:00", | |
| "backgroundColor": color, | |
| "borderColor": color, | |
| "classNames": ["event-completed"] if is_completed else [], | |
| "extendedProps": { | |
| "type": "appointment", | |
| "status": status, | |
| "appointment_id": a["id"], | |
| "patient_name": a.get("patient_name", ""), | |
| "provider_name": a.get("provider_name", ""), | |
| "provider_id": pid, | |
| "visit_level": str(a.get("visit_level", "")), | |
| "visit_address": a.get("visit_address", ""), | |
| "recommended_departure_time": a.get("recommended_departure_time", ""), | |
| "start_time": a.get("start_time", ""), | |
| "end_time": a.get("end_time", ""), | |
| "date": a.get("date", ""), | |
| "completed_at": a.get("completed_at", ""), | |
| }, | |
| }) | |
| # Travel block (30 min before, if departure time exists) — skip for completed visits. | |
| if show_travel and not is_completed: | |
| departure = a.get("recommended_departure_time") | |
| if departure: | |
| events.append({ | |
| "id": f"travel_{a['id']}", | |
| "title": f"Travel to {a.get('patient_name', 'visit')}", | |
| "start": f"{a['date']}T{departure}:00", | |
| "end": f"{a['date']}T{a['start_time']}:00", | |
| "backgroundColor": color, | |
| "borderColor": color, | |
| "classNames": ["event-travel"], | |
| "extendedProps": { | |
| "type": "travel", | |
| "provider_id": pid, | |
| "provider_name": a.get("provider_name", ""), | |
| }, | |
| }) | |
| # Days-off blocks | |
| if show_unavailable: | |
| for p in visible_providers: | |
| for off in p.get("days_off", []) or []: | |
| if isinstance(off, str): | |
| off = {"date": off, "all_day": True, "title": "Day Off"} | |
| d = off.get("date") | |
| if not d: | |
| continue | |
| title = off.get("title", "Unavailable") | |
| if off.get("all_day", True): | |
| start_iso = f"{d}T00:00:00" | |
| end_iso = f"{d}T23:59:59" | |
| else: | |
| start_iso = f"{d}T{off.get('start_time','00:00')}:00" | |
| end_iso = f"{d}T{off.get('end_time','23:59')}:00" | |
| events.append({ | |
| "id": f"off_{p['id']}_{d}_{title}", | |
| "title": f"{p['name']} — {title}", | |
| "start": start_iso, | |
| "end": end_iso, | |
| "backgroundColor": "#fecaca", | |
| "borderColor": "#dc2626", | |
| "textColor": "#7f1d1d", | |
| "extendedProps": { | |
| "type": "days_off", | |
| "provider_id": p["id"], | |
| "provider_name": p["name"], | |
| "title": title, | |
| }, | |
| }) | |
| # Recurring weekdays — render as background events for whole date range | |
| recurring = p.get("recurring_days_off", []) or [] | |
| if recurring and start and end: | |
| weekday_to_num = {"Monday":0,"Tuesday":1,"Wednesday":2,"Thursday":3,"Friday":4,"Saturday":5,"Sunday":6} | |
| try: | |
| s_dt = datetime.strptime(start[:10], "%Y-%m-%d") | |
| e_dt = datetime.strptime(end[:10], "%Y-%m-%d") | |
| except ValueError: | |
| s_dt = e_dt = None | |
| if s_dt and e_dt: | |
| cur = s_dt | |
| while cur < e_dt: | |
| if cur.strftime("%A") in recurring: | |
| d_str = cur.strftime("%Y-%m-%d") | |
| events.append({ | |
| "id": f"rec_{p['id']}_{d_str}", | |
| "title": f"{p['name']} — Standing Day Off", | |
| "start": f"{d_str}T00:00:00", | |
| "end": f"{d_str}T23:59:59", | |
| "display": "background", | |
| "backgroundColor": "#e5e7eb", | |
| "extendedProps": {"type": "recurring_off", "provider_id": p["id"]}, | |
| }) | |
| cur += timedelta(days=1) | |
| # On-call dates highlight — 24h window from on_call_start_hour on date D | |
| # to the same hour on D+1 (matches scheduling_engine.is_oncall_for_slot). | |
| start_hour = settings.get_on_call_start_hour() # "HH:MM" | |
| for d in (p.get("on_call", []) or []) if show_oncall else []: | |
| try: | |
| window_start = datetime.strptime(f"{d} {start_hour}", "%Y-%m-%d %H:%M") | |
| except ValueError: | |
| continue | |
| window_end = window_start + timedelta(hours=24) | |
| events.append({ | |
| "id": f"oncall_{p['id']}_{d}", | |
| "title": f"{p['name']} — ON CALL", | |
| "start": window_start.strftime("%Y-%m-%dT%H:%M:%S"), | |
| "end": window_end.strftime("%Y-%m-%dT%H:%M:%S"), | |
| "backgroundColor": "#10b981", | |
| "borderColor": "#047857", | |
| "textColor": "#ffffff", | |
| "extendedProps": {"type": "on_call", "provider_id": p["id"]}, | |
| }) | |
| return { | |
| "events": events, | |
| "providers": [ | |
| {"id": p["id"], "name": p["name"], "color": color_map[p["id"]]} | |
| for p in providers | |
| ], | |
| } | |