import re import unicodedata from itertools import combinations from dataclasses import dataclass from datetime import date, datetime, time, timedelta from sqlalchemy import select from sqlalchemy.orm import Session, joinedload from app.models.reservation import Reservation, ReservationStatus from app.models.room import Room from app.models.table import Table, TableCombination NON_BLOCKING_STATUSES = {ReservationStatus.cancelled, ReservationStatus.no_show} MAX_DYNAMIC_COMBINATION_SIZE = 15 PHYSICAL_JOIN_GAP_UNITS = 35 MIN_TABLE_EDGE_OVERLAP_UNITS = 20 INTERIOR_PREFERENCE_KEYWORDS = ("intern", "indoor", "dentro", "interno", "interna") EXTERIOR_PREFERENCE_KEYWORDS = ("estern", "outdoor", "fuori", "dehor", "dehors", "terraz", "giardin", "patio") IGNORED_ROOM_TOKENS = {"sala", "area", "zona"} @dataclass class AssignmentCandidate: kind: str label: str table_ids: list[int] assigned_table_id: int | None assigned_combination_id: int | None min_seats: int max_seats: int @property def wasted_seats(self) -> int: return self.max_seats @dataclass class RoomAssignmentCandidate: room_id: int room_name: str candidate: AssignmentCandidate def reservation_window(reservation: Reservation) -> tuple[datetime, datetime]: start_at = datetime.combine(reservation.reservation_date, reservation.start_time) end_at = start_at + timedelta(minutes=reservation.duration_minutes) return start_at, end_at def combine_date_time(day: date, value: time) -> datetime: return datetime.combine(day, value) def overlap( first_start: datetime, first_end: datetime, second_start: datetime, second_end: datetime, ) -> bool: return first_start < second_end and second_start < first_end def get_room_tables(room_id: int, db: Session) -> list[Table]: stmt = select(Table).where(Table.room_id == room_id, Table.is_active.is_(True)).order_by(Table.max_seats, Table.name) return list(db.scalars(stmt)) def get_room_combinations(room_id: int, db: Session) -> list[TableCombination]: stmt = ( select(TableCombination) .where(TableCombination.room_id == room_id, TableCombination.is_active.is_(True)) .order_by(TableCombination.max_seats, TableCombination.name) ) return list(db.scalars(stmt)) def get_venue_rooms(venue_id: int, db: Session) -> list[Room]: stmt = select(Room).where(Room.venue_id == venue_id).order_by(Room.name, Room.id) return list(db.scalars(stmt)) def normalize_room_text(value: str | None) -> str: normalized = unicodedata.normalize("NFKD", value or "").encode("ascii", "ignore").decode("ascii") normalized = normalized.casefold().strip() return re.sub(r"\s+", " ", normalized) def room_name_matches_preference(room_name: str, preference: str) -> bool: normalized_room_name = normalize_room_text(room_name) normalized_preference = normalize_room_text(preference) if not normalized_room_name or not normalized_preference: return False if normalized_preference == normalized_room_name: return True if normalized_preference in normalized_room_name or normalized_room_name in normalized_preference: return True room_tokens = {token for token in normalized_room_name.split() if token not in IGNORED_ROOM_TOKENS} preference_tokens = {token for token in normalized_preference.split() if token not in IGNORED_ROOM_TOKENS} if room_tokens and preference_tokens and room_tokens.intersection(preference_tokens): return True if any(keyword in normalized_preference for keyword in INTERIOR_PREFERENCE_KEYWORDS): return any(keyword in normalized_room_name for keyword in INTERIOR_PREFERENCE_KEYWORDS) if any(keyword in normalized_preference for keyword in EXTERIOR_PREFERENCE_KEYWORDS): return any(keyword in normalized_room_name for keyword in EXTERIOR_PREFERENCE_KEYWORDS) return False def resolve_preferred_room_ids(venue_id: int, area_preference: str | None, db: Session) -> set[int] | None: normalized_preference = normalize_room_text(area_preference) if not normalized_preference: return None rooms = get_venue_rooms(venue_id, db) matching_ids = { room.id for room in rooms if room_name_matches_preference(room.name, normalized_preference) } return matching_ids or None def table_sort_value(table: Table) -> tuple[int, str]: if table.name.isdigit(): return int(table.name), table.name return 10_000, table.name def order_table_ids(table_ids: list[int], tables_by_id: dict[int, Table]) -> list[int]: return sorted(table_ids, key=lambda table_id: table_sort_value(tables_by_id[table_id])) def range_overlap(start_a: int, end_a: int, start_b: int, end_b: int) -> int: return min(end_a, end_b) - max(start_a, start_b) def range_gap(start_a: int, end_a: int, start_b: int, end_b: int) -> int: if end_a < start_b: return start_b - end_a if end_b < start_a: return start_a - end_b return 0 def tables_are_physically_joinable(first: Table, second: Table) -> bool: first_left = first.x first_right = first.x + first.width first_top = first.y first_bottom = first.y + first.height second_left = second.x second_right = second.x + second.width second_top = second.y second_bottom = second.y + second.height vertical_overlap = range_overlap(first_top, first_bottom, second_top, second_bottom) horizontal_overlap = range_overlap(first_left, first_right, second_left, second_right) horizontal_gap = range_gap(first_left, first_right, second_left, second_right) vertical_gap = range_gap(first_top, first_bottom, second_top, second_bottom) required_vertical_overlap = max(MIN_TABLE_EDGE_OVERLAP_UNITS, int(min(first.height, second.height) * 0.25)) required_horizontal_overlap = max(MIN_TABLE_EDGE_OVERLAP_UNITS, int(min(first.width, second.width) * 0.25)) if horizontal_gap <= PHYSICAL_JOIN_GAP_UNITS and vertical_overlap >= required_vertical_overlap: return True if vertical_gap <= PHYSICAL_JOIN_GAP_UNITS and horizontal_overlap >= required_horizontal_overlap: return True return False def build_join_graph( tables: list[Table], combinations_list: list[TableCombination], tables_by_id: dict[int, Table], ) -> dict[int, set[int]]: graph = {table.id: set() for table in tables} join_groups: dict[str, list[int]] = {} for table in tables: if table.join_group: join_groups.setdefault(table.join_group, []).append(table.id) for grouped_table_ids in join_groups.values(): for first_id, second_id in combinations(grouped_table_ids, 2): graph[first_id].add(second_id) graph[second_id].add(first_id) for first_table, second_table in combinations(tables, 2): if not tables_are_physically_joinable(first_table, second_table): continue graph[first_table.id].add(second_table.id) graph[second_table.id].add(first_table.id) for combination_item in combinations_list: ordered_ids = [table_id for table_id in order_table_ids(combination_item.table_ids, tables_by_id) if table_id in graph] for index in range(len(ordered_ids) - 1): first_id = ordered_ids[index] second_id = ordered_ids[index + 1] graph[first_id].add(second_id) graph[second_id].add(first_id) return graph def subset_is_connected(table_ids: tuple[int, ...], join_graph: dict[int, set[int]]) -> bool: subset = set(table_ids) to_visit = [table_ids[0]] visited: set[int] = set() while to_visit: current = to_visit.pop() if current in visited: continue visited.add(current) to_visit.extend(neighbor for neighbor in join_graph[current] if neighbor in subset and neighbor not in visited) return visited == subset def build_dynamic_combination_candidates( tables: list[Table], explicit_combinations: list[TableCombination], guests: int, ) -> list[AssignmentCandidate]: if not tables or guests <= 0: return [] tables_by_id = {table.id: table for table in tables} ordered_table_ids = [table.id for table in sorted(tables, key=table_sort_value)] join_graph = build_join_graph(tables, explicit_combinations, tables_by_id) explicit_sets = {tuple(sorted(combination_item.table_ids)) for combination_item in explicit_combinations} dynamic_candidates: list[AssignmentCandidate] = [] visited_subsets: set[tuple[int, ...]] = set() for start_table_id in ordered_table_ids: frontier: list[tuple[int, ...]] = [(start_table_id,)] while frontier: subset = frontier.pop() normalized_subset = tuple(sorted(subset)) if normalized_subset in visited_subsets: continue visited_subsets.add(normalized_subset) if len(normalized_subset) > MAX_DYNAMIC_COMBINATION_SIZE or not subset_is_connected(normalized_subset, join_graph): continue ordered_subset = order_table_ids(list(normalized_subset), tables_by_id) subset_tables = [tables_by_id[table_id] for table_id in ordered_subset] max_seats = sum(table.max_seats for table in subset_tables) if len(ordered_subset) >= 2 and normalized_subset not in explicit_sets and max_seats >= guests: dynamic_candidates.append( AssignmentCandidate( kind="combination", label=" + ".join(table.name for table in subset_tables), table_ids=ordered_subset, assigned_table_id=None, assigned_combination_id=None, min_seats=sum(table.min_seats for table in subset_tables), max_seats=max_seats, ) ) continue neighbor_ids = { neighbor_id for table_id in normalized_subset for neighbor_id in join_graph[table_id] if neighbor_id not in normalized_subset } for neighbor_id in sorted(neighbor_ids, key=lambda table_id: table_sort_value(tables_by_id[table_id]), reverse=True): frontier.append(tuple(sorted((*normalized_subset, neighbor_id)))) return dynamic_candidates def build_candidates(room_id: int, guests: int, db: Session) -> list[AssignmentCandidate]: tables = get_room_tables(room_id, db) tables_by_id = {table.id: table for table in tables} active_table_ids = {table.id for table in tables} explicit_combinations = get_room_combinations(room_id, db) candidates = [ AssignmentCandidate( kind="table", label=table.name, table_ids=[table.id], assigned_table_id=table.id, assigned_combination_id=None, min_seats=table.min_seats, max_seats=table.max_seats, ) for table in tables ] for combination in explicit_combinations: if not set(combination.table_ids).issubset(active_table_ids): continue ordered_ids = order_table_ids(combination.table_ids, tables_by_id) candidates.append( AssignmentCandidate( kind="combination", label=combination.name, table_ids=ordered_ids, assigned_table_id=None, assigned_combination_id=combination.id, min_seats=combination.min_seats, max_seats=combination.max_seats, ) ) candidates.extend(build_dynamic_combination_candidates(tables, explicit_combinations, guests)) return candidates def reservations_for_assignment(day: date, venue_id: int, db: Session) -> list[Reservation]: stmt = ( select(Reservation) .options( joinedload(Reservation.customer), joinedload(Reservation.assigned_table), joinedload(Reservation.assigned_combination), joinedload(Reservation.status_history), ) .where(Reservation.venue_id == venue_id, Reservation.reservation_date == day) .order_by(Reservation.start_time, Reservation.guests.desc(), Reservation.created_at) ) return list(db.scalars(stmt).unique()) def candidate_is_compatible(candidate: AssignmentCandidate, guests: int) -> bool: return candidate.max_seats >= guests def candidate_conflicts( candidate: AssignmentCandidate, reservation: Reservation, scheduled_reservations: list[Reservation], ) -> bool: start_at, end_at = reservation_window(reservation) for scheduled in scheduled_reservations: if scheduled.id == reservation.id or scheduled.status in NON_BLOCKING_STATUSES: continue if scheduled.assigned_table_id is None and scheduled.assigned_combination_id is None: continue scheduled_table_ids = [] if scheduled.assigned_table_id is not None: scheduled_table_ids = [scheduled.assigned_table_id] elif scheduled.assigned_combination is not None: scheduled_table_ids = scheduled.assigned_combination.table_ids if not set(candidate.table_ids).intersection(scheduled_table_ids): continue scheduled_start, scheduled_end = reservation_window(scheduled) if overlap(start_at, end_at, scheduled_start, scheduled_end): return True return False def select_best_candidate( reservation: Reservation, room_id: int, scheduled_reservations: list[Reservation], db: Session, ) -> AssignmentCandidate | None: candidates = build_candidates(room_id, reservation.guests, db) compatible = [ candidate for candidate in candidates if candidate_is_compatible(candidate, reservation.guests) and not candidate_conflicts(candidate, reservation, scheduled_reservations) ] if not compatible: return None compatible.sort( key=lambda candidate: ( 0 if candidate.kind == "table" else 1, candidate.max_seats - reservation.guests, len(candidate.table_ids), candidate.label, ) ) return compatible[0] def list_available_room_candidates( reservation: Reservation, venue_id: int, scheduled_reservations: list[Reservation], db: Session, *, preferred_room_ids: set[int] | None = None, ) -> list[RoomAssignmentCandidate]: rooms = get_venue_rooms(venue_id, db) if preferred_room_ids: rooms = [room for room in rooms if room.id in preferred_room_ids] room_candidates: list[RoomAssignmentCandidate] = [] for room in rooms: candidate = select_best_candidate(reservation, room.id, scheduled_reservations, db) if candidate is None: continue room_candidates.append( RoomAssignmentCandidate( room_id=room.id, room_name=room.name, candidate=candidate, ) ) room_candidates.sort( key=lambda option: ( 0 if option.candidate.kind == "table" else 1, option.candidate.max_seats - reservation.guests, len(option.candidate.table_ids), option.room_name, option.candidate.label, ) ) return room_candidates def select_best_candidate_across_rooms( reservation: Reservation, venue_id: int, scheduled_reservations: list[Reservation], db: Session, *, preferred_room_ids: set[int] | None = None, ) -> RoomAssignmentCandidate | None: candidates = list_available_room_candidates( reservation, venue_id, scheduled_reservations, db, preferred_room_ids=preferred_room_ids, ) if not candidates: return None return candidates[0] def ensure_candidate_combination(candidate: AssignmentCandidate, room_id: int, db: Session) -> AssignmentCandidate: if candidate.assigned_combination_id is not None or candidate.kind == "table": return candidate combinations_list = get_room_combinations(room_id, db) normalized_target = tuple(sorted(candidate.table_ids)) for combination_item in combinations_list: if tuple(sorted(combination_item.table_ids)) == normalized_target: candidate.assigned_combination_id = combination_item.id candidate.label = combination_item.name return candidate auto_combination = TableCombination( room_id=room_id, name=candidate.label, table_ids=candidate.table_ids, min_seats=candidate.min_seats, max_seats=candidate.max_seats, is_active=True, ) db.add(auto_combination) db.flush() candidate.assigned_combination_id = auto_combination.id return candidate def assign_candidate( reservation: Reservation, candidate: AssignmentCandidate | None, room_id: int, db: Session, ) -> None: if candidate is None: reservation.assigned_table_id = None reservation.assigned_combination_id = None return candidate = ensure_candidate_combination(candidate, room_id, db) reservation.assigned_table_id = candidate.assigned_table_id reservation.assigned_combination_id = candidate.assigned_combination_id def assign_room_candidate( reservation: Reservation, room_candidate: RoomAssignmentCandidate | None, db: Session, ) -> None: if room_candidate is None: reservation.assigned_table_id = None reservation.assigned_combination_id = None return assign_candidate(reservation, room_candidate.candidate, room_candidate.room_id, db) def reassign_single_reservation(reservation: Reservation, db: Session) -> RoomAssignmentCandidate | None: scheduled = reservations_for_assignment(reservation.reservation_date, reservation.venue_id, db) assign_room_candidate(reservation, None, db) db.flush() preferred_room_ids = resolve_preferred_room_ids(reservation.venue_id, reservation.area_preference, db) candidate = select_best_candidate_across_rooms( reservation, reservation.venue_id, scheduled, db, preferred_room_ids=preferred_room_ids, ) assign_room_candidate(reservation, candidate, db) db.flush() db.refresh(reservation) return candidate def recalculate_day_assignments(day: date, venue_id: int, db: Session) -> list[Reservation]: reservations = reservations_for_assignment(day, venue_id, db) for reservation in reservations: reservation.assigned_table_id = None reservation.assigned_combination_id = None db.flush() for reservation in reservations: if reservation.status in NON_BLOCKING_STATUSES: continue preferred_room_ids = resolve_preferred_room_ids(venue_id, reservation.area_preference, db) candidate = select_best_candidate_across_rooms( reservation, venue_id, reservations, db, preferred_room_ids=preferred_room_ids, ) assign_room_candidate(reservation, candidate, db) db.commit() return reservations_for_assignment(day, venue_id, db)