"""Calendar overlap detection service. BE-CAL-006: Validates that a new or edited TimeSlot does not overlap with existing slots on the same day for the same user. Overlap is defined as two time ranges ``[start, start + duration)`` having a non-empty intersection. Cancelled/aborted slots are excluded from conflict checks (they no longer occupy calendar time). For the **create** scenario, all existing non-cancelled slots on the target date are checked. For the **edit** scenario, the slot being edited is excluded from the candidate set so it doesn't conflict with its own previous position. """ from __future__ import annotations from datetime import date, time, timedelta, datetime from typing import Optional from sqlalchemy.orm import Session from app.models.calendar import SlotStatus, TimeSlot from app.services.plan_slot import get_virtual_slots_for_date # Statuses that no longer occupy calendar time — excluded from overlap checks. _INACTIVE_STATUSES = {SlotStatus.SKIPPED, SlotStatus.ABORTED} # --------------------------------------------------------------------------- # Internal helpers # --------------------------------------------------------------------------- def _time_to_minutes(t: time) -> int: """Convert a ``time`` to minutes since midnight.""" return t.hour * 60 + t.minute def _ranges_overlap( start_a: int, end_a: int, start_b: int, end_b: int, ) -> bool: """Return *True* if two half-open intervals ``[a, a+dur)`` overlap.""" return start_a < end_b and start_b < end_a # --------------------------------------------------------------------------- # Conflict data class # --------------------------------------------------------------------------- class SlotConflict: """Describes a single overlap conflict.""" __slots__ = ("conflicting_slot_id", "conflicting_virtual_id", "scheduled_at", "estimated_duration", "slot_type", "message") def __init__( self, *, conflicting_slot_id: Optional[int] = None, conflicting_virtual_id: Optional[str] = None, scheduled_at: time, estimated_duration: int, slot_type: str, message: str, ): self.conflicting_slot_id = conflicting_slot_id self.conflicting_virtual_id = conflicting_virtual_id self.scheduled_at = scheduled_at self.estimated_duration = estimated_duration self.slot_type = slot_type self.message = message def to_dict(self) -> dict: d: dict = { "scheduled_at": self.scheduled_at.isoformat(), "estimated_duration": self.estimated_duration, "slot_type": self.slot_type, "message": self.message, } if self.conflicting_slot_id is not None: d["conflicting_slot_id"] = self.conflicting_slot_id if self.conflicting_virtual_id is not None: d["conflicting_virtual_id"] = self.conflicting_virtual_id return d # --------------------------------------------------------------------------- # Core overlap detection # --------------------------------------------------------------------------- def _format_time_range(start: time, duration: int) -> str: """Format a slot time range for human-readable messages.""" start_min = _time_to_minutes(start) end_min = start_min + duration end_h, end_m = divmod(end_min, 60) # Clamp to 23:59 for display purposes if end_h >= 24: end_h, end_m = 23, 59 return f"{start.strftime('%H:%M')}-{end_h:02d}:{end_m:02d}" def check_overlap( db: Session, user_id: int, target_date: date, scheduled_at: time, estimated_duration: int, *, exclude_slot_id: Optional[int] = None, ) -> list[SlotConflict]: """Check for time conflicts on *target_date* for *user_id*. Parameters ---------- db : Active database session. user_id : The user whose calendar is being checked. target_date : The date to check. scheduled_at : Proposed start time. estimated_duration : Proposed duration in minutes. exclude_slot_id : If editing an existing slot, pass its ``id`` so it is not counted as conflicting with itself. Returns ------- list[SlotConflict] Empty list means no conflicts. Non-empty means the proposed slot overlaps with one or more existing slots. """ new_start = _time_to_minutes(scheduled_at) new_end = new_start + estimated_duration conflicts: list[SlotConflict] = [] # ---- 1. Check real (materialized) slots -------------------------------- query = ( db.query(TimeSlot) .filter( TimeSlot.user_id == user_id, TimeSlot.date == target_date, TimeSlot.status.notin_([s.value for s in _INACTIVE_STATUSES]), ) ) if exclude_slot_id is not None: query = query.filter(TimeSlot.id != exclude_slot_id) existing_slots: list[TimeSlot] = query.all() for slot in existing_slots: slot_start = _time_to_minutes(slot.scheduled_at) slot_end = slot_start + slot.estimated_duration if _ranges_overlap(new_start, new_end, slot_start, slot_end): existing_range = _format_time_range(slot.scheduled_at, slot.estimated_duration) proposed_range = _format_time_range(scheduled_at, estimated_duration) conflicts.append(SlotConflict( conflicting_slot_id=slot.id, scheduled_at=slot.scheduled_at, estimated_duration=slot.estimated_duration, slot_type=slot.slot_type.value if hasattr(slot.slot_type, 'value') else str(slot.slot_type), message=( f"Proposed slot {proposed_range} overlaps with existing " f"{slot.slot_type.value if hasattr(slot.slot_type, 'value') else slot.slot_type} " f"slot (id={slot.id}) at {existing_range}" ), )) # ---- 2. Check virtual (plan-generated) slots --------------------------- virtual_slots = get_virtual_slots_for_date(db, user_id, target_date) for vs in virtual_slots: vs_start = _time_to_minutes(vs["scheduled_at"]) vs_end = vs_start + vs["estimated_duration"] if _ranges_overlap(new_start, new_end, vs_start, vs_end): existing_range = _format_time_range(vs["scheduled_at"], vs["estimated_duration"]) proposed_range = _format_time_range(scheduled_at, estimated_duration) slot_type_val = vs["slot_type"].value if hasattr(vs["slot_type"], 'value') else str(vs["slot_type"]) conflicts.append(SlotConflict( conflicting_virtual_id=vs["virtual_id"], scheduled_at=vs["scheduled_at"], estimated_duration=vs["estimated_duration"], slot_type=slot_type_val, message=( f"Proposed slot {proposed_range} overlaps with virtual plan " f"slot ({vs['virtual_id']}) at {existing_range}" ), )) return conflicts # --------------------------------------------------------------------------- # Convenience wrappers for create / edit scenarios # --------------------------------------------------------------------------- def check_overlap_for_create( db: Session, user_id: int, target_date: date, scheduled_at: time, estimated_duration: int, ) -> list[SlotConflict]: """Check overlap when creating a brand-new slot (no exclusion).""" return check_overlap( db, user_id, target_date, scheduled_at, estimated_duration, ) def check_overlap_for_edit( db: Session, user_id: int, slot_id: int, target_date: date, scheduled_at: time, estimated_duration: int, ) -> list[SlotConflict]: """Check overlap when editing an existing slot (exclude itself).""" return check_overlap( db, user_id, target_date, scheduled_at, estimated_duration, exclude_slot_id=slot_id, )