Movie Ticket Booking

  1  from __future__ import annotations
  2  from enum import Enum
  3  from dataclasses import dataclass, field
  4  from datetime import datetime, timedelta
  5  from threading import Lock
  6  from typing import Protocol
  7  import uuid
  8  
  9  
 10  class SeatType(Enum):
 11      REGULAR = "regular"
 12      PREMIUM = "premium"
 13      VIP = "vip"
 14  
 15  
 16  class SeatStatus(Enum):
 17      AVAILABLE = "available"
 18      LOCKED = "locked"
 19      BOOKED = "booked"
 20  
 21  SEAT_TRANSITIONS: dict[SeatStatus, set[SeatStatus]] = {
 22      SeatStatus.AVAILABLE: {SeatStatus.LOCKED},
 23      SeatStatus.LOCKED: {SeatStatus.BOOKED, SeatStatus.AVAILABLE},
 24      SeatStatus.BOOKED: {SeatStatus.AVAILABLE},
 25  }
 26  
 27  
 28  class BookingStatus(Enum):
 29      PENDING = "pending"
 30      CONFIRMED = "confirmed"
 31      CANCELLED = "cancelled"
 32  
 33  
 34  class Seat:
 35      def __init__(self, seat_id: str, seat_type: SeatType, row: int, number: int):
 36          self.seat_id = seat_id
 37          self.type = seat_type
 38          self.row = row
 39          self.number = number
 40          self.status = SeatStatus.AVAILABLE
 41  
 42      def _transition(self, target: SeatStatus) -> None:
 43          allowed = SEAT_TRANSITIONS.get(self.status, set())
 44          if target not in allowed:
 45              raise RuntimeError(
 46                  f"Seat {self.seat_id}: cannot go from {self.status.value} to {target.value}"
 47              )
 48          self.status = target
 49  
 50      def lock(self) -> None:
 51          self._transition(SeatStatus.LOCKED)
 52  
 53      def book(self) -> None:
 54          self._transition(SeatStatus.BOOKED)
 55  
 56      def release(self) -> None:
 57          self._transition(SeatStatus.AVAILABLE)
 58  
 59      def __repr__(self) -> str:
 60          return f"Seat({self.seat_id}, {self.type.value}, {self.status.value})"
 61  
 62  
 63  class PricingStrategy(Protocol):
 64      def calculate_price(self, seat_type: SeatType, show_time: datetime) -> float: ...
 65  
 66  
 67  class DynamicPricing:
 68      """Adjusts price based on seat type, time of day, and day of week."""
 69      BASE_RATES = {
 70          SeatType.REGULAR: 150.0,
 71          SeatType.PREMIUM: 250.0,
 72          SeatType.VIP: 400.0,
 73      }
 74  
 75      def calculate_price(self, seat_type: SeatType, show_time: datetime) -> float:
 76          base = self.BASE_RATES[seat_type]
 77          multiplier = 1.0
 78  
 79          # Weekend surcharge
 80          if show_time.weekday() >= 5:
 81              multiplier *= 1.3
 82  
 83          # Evening shows cost more than matinees
 84          hour = show_time.hour
 85          if hour >= 18:
 86              multiplier *= 1.2
 87          elif hour < 12:
 88              multiplier *= 0.8
 89  
 90          return round(base * multiplier, 2)
 91  
 92  
 93  class FlatPricing:
 94      """Simple flat rate per seat type. No time-based adjustments."""
 95      RATES = {
 96          SeatType.REGULAR: 150.0,
 97          SeatType.PREMIUM: 250.0,
 98          SeatType.VIP: 400.0,
 99      }
100  
101      def calculate_price(self, seat_type: SeatType, show_time: datetime) -> float:
102          return self.RATES[seat_type]
103  
104  
105  class Show:
106      def __init__(self, show_id: str, movie: str, screen: "Screen",
107                   start_time: datetime, duration_minutes: int,
108                   pricing: PricingStrategy | None = None):
109          self.show_id = show_id
110          self.movie = movie
111          self.screen = screen
112          self.start_time = start_time
113          self.end_time = start_time + timedelta(minutes=duration_minutes)
114          self.pricing = pricing or DynamicPricing()
115          self.seats: dict[str, Seat] = {}
116          self._build_seats()
117  
118      def _build_seats(self) -> None:
119          for seat_type, count in self.screen.seat_layout.items():
120              type_prefix = seat_type.value[0].upper()
121              for i in range(1, count + 1):
122                  row = (i - 1) // 10 + 1
123                  number = (i - 1) % 10 + 1
124                  seat_id = f"{type_prefix}-R{row}-S{number}"
125                  self.seats[seat_id] = Seat(seat_id, seat_type, row, number)
126  
127      def available_seats(self, seat_type: SeatType | None = None) -> list[Seat]:
128          seats = list(self.seats.values())
129          if seat_type:
130              seats = [s for s in seats if s.type == seat_type]
131          return [s for s in seats if s.status == SeatStatus.AVAILABLE]
132  
133      def get_seat(self, seat_id: str) -> Seat:
134          seat = self.seats.get(seat_id)
135          if seat is None:
136              raise ValueError(f"Seat {seat_id} not found in show {self.show_id}")
137          return seat
138  
139      def get_price(self, seat_type: SeatType) -> float:
140          return self.pricing.calculate_price(seat_type, self.start_time)
141  
142      def __repr__(self) -> str:
143          return f"Show({self.movie}, {self.start_time.strftime('%H:%M')}, Screen {self.screen.name})"
144  
145  
146  class Screen:
147      def __init__(self, screen_id: str, name: str, seat_layout: dict[SeatType, int]):
148          self.screen_id = screen_id
149          self.name = name
150          self.seat_layout = seat_layout
151          self._shows: list[Show] = []
152  
153      def has_conflict(self, start: datetime, end: datetime) -> bool:
154          for show in self._shows:
155              if start < show.end_time and show.start_time < end:
156                  return True
157          return False
158  
159      def add_show(self, show: Show) -> None:
160          if self.has_conflict(show.start_time, show.end_time):
161              raise RuntimeError(
162                  f"Screen {self.name} has a time conflict for "
163                  f"{show.start_time} to {show.end_time}"
164              )
165          self._shows.append(show)
166  
167      @property
168      def shows(self) -> list[Show]:
169          return list(self._shows)
170  
171      @property
172      def total_capacity(self) -> int:
173          return sum(self.seat_layout.values())
174  
175  
176  @dataclass
177  class Cinema:
178      cinema_id: str
179      name: str
180      city: str
181      screens: list[Screen] = field(default_factory=list)
182  
183      def get_shows(self, movie: str | None = None) -> list[Show]:
184          all_shows = []
185          for screen in self.screens:
186              for show in screen.shows:
187                  if movie is None or show.movie.lower() == movie.lower():
188                      all_shows.append(show)
189          return all_shows
190  
191  
192  @dataclass
193  class SeatLock:
194      lock_id: str
195      user_id: str
196      show: Show
197      seats: list[Seat]
198      expires_at: datetime
199  
200      def is_expired(self) -> bool:
201          return datetime.now() > self.expires_at
202  
203  
204  @dataclass
205  class Booking:
206      booking_id: str
207      user_id: str
208      show: Show
209      seats: list[Seat]
210      total_amount: float
211      status: BookingStatus = BookingStatus.CONFIRMED
212  
213  
214  class WaitlistObserver(Protocol):
215      def on_seats_released(self, show_id: str, count: int) -> None: ...
216  
217  class WaitlistNotifier:
218      def on_seats_released(self, show_id: str, count: int) -> None:
219          print(f"  [Waitlist] {count} seat(s) available for show {show_id}")
220  
221  
222  class BookMyShow:
223      LOCK_TTL = timedelta(minutes=5)
224  
225      def __init__(self):
226          self._cinemas: list[Cinema] = []
227          self._bookings: dict[str, Booking] = {}
228          self._locks: dict[str, SeatLock] = {}
229          self._observers: list[WaitlistObserver] = []
230          self._mutex = Lock()
231  
232      def add_cinema(self, cinema: Cinema) -> None:
233          self._cinemas.append(cinema)
234  
235      def add_observer(self, observer: WaitlistObserver) -> None:
236          self._observers.append(observer)
237  
238      def search_shows(self, city: str, movie: str | None = None) -> list[Show]:
239          results = []
240          for cinema in self._cinemas:
241              if cinema.city.lower() != city.lower():
242                  continue
243              results.extend(cinema.get_shows(movie))
244          return results
245  
246      def lock_seats(self, show: Show, user_id: str,
247                     seat_ids: list[str]) -> SeatLock:
248          with self._mutex:
249              self._cleanup_expired()
250              seats = [show.get_seat(sid) for sid in seat_ids]
251              for seat in seats:
252                  seat.lock()
253              seat_lock = SeatLock(
254                  lock_id=str(uuid.uuid4())[:8],
255                  user_id=user_id,
256                  show=show,
257                  seats=seats,
258                  expires_at=datetime.now() + self.LOCK_TTL,
259              )
260              self._locks[seat_lock.lock_id] = seat_lock
261              return seat_lock
262  
263      def confirm_booking(self, lock_id: str) -> Booking:
264          with self._mutex:
265              seat_lock = self._locks.pop(lock_id, None)
266              if seat_lock is None:
267                  raise ValueError(f"Lock {lock_id} not found or already expired")
268              if seat_lock.is_expired():
269                  for seat in seat_lock.seats:
270                      if seat.status == SeatStatus.LOCKED:
271                          seat.release()
272                  raise RuntimeError("Lock expired. Seats have been released.")
273  
274              total = sum(
275                  seat_lock.show.get_price(seat.type) for seat in seat_lock.seats
276              )
277              for seat in seat_lock.seats:
278                  seat.book()
279  
280              booking = Booking(
281                  booking_id=str(uuid.uuid4())[:8],
282                  user_id=seat_lock.user_id,
283                  show=seat_lock.show,
284                  seats=seat_lock.seats,
285                  total_amount=total,
286              )
287              self._bookings[booking.booking_id] = booking
288              return booking
289  
290      def cancel_booking(self, booking_id: str) -> None:
291          with self._mutex:
292              booking = self._bookings.get(booking_id)
293              if booking is None:
294                  raise ValueError(f"Booking {booking_id} not found")
295              if booking.status == BookingStatus.CANCELLED:
296                  raise RuntimeError("Booking already cancelled")
297              booking.status = BookingStatus.CANCELLED
298              released = 0
299              for seat in booking.seats:
300                  seat.release()
301                  released += 1
302              for obs in self._observers:
303                  obs.on_seats_released(booking.show.show_id, released)
304  
305      def cancel_lock(self, lock_id: str) -> None:
306          with self._mutex:
307              seat_lock = self._locks.pop(lock_id, None)
308              if seat_lock is None:
309                  return
310              for seat in seat_lock.seats:
311                  if seat.status == SeatStatus.LOCKED:
312                      seat.release()
313  
314      def _cleanup_expired(self) -> None:
315          expired = [lid for lid, lk in self._locks.items() if lk.is_expired()]
316          for lid in expired:
317              lk = self._locks.pop(lid)
318              for seat in lk.seats:
319                  if seat.status == SeatStatus.LOCKED:
320                      seat.release()
321  
322  
323  if __name__ == "__main__":
324      # Set up cinema with two screens
325      screen1 = Screen("S1", "Screen 1", {
326          SeatType.REGULAR: 40,
327          SeatType.PREMIUM: 20,
328          SeatType.VIP: 10,
329      })
330      screen2 = Screen("S2", "Screen 2", {
331          SeatType.REGULAR: 30,
332          SeatType.PREMIUM: 15,
333      })
334  
335      cinema = Cinema("C1", "PVR Phoenix", "Mumbai", [screen1, screen2])
336  
337      # Create shows with different times
338      friday_evening = datetime(2025, 8, 15, 21, 0)  # Friday 9 PM
339      tuesday_matinee = datetime(2025, 8, 19, 10, 0)  # Tuesday 10 AM
340  
341      show1 = Show("SH1", "Inception", screen1, friday_evening, 150, DynamicPricing())
342      screen1.add_show(show1)
343  
344      show2 = Show("SH2", "Inception", screen2, tuesday_matinee, 150, DynamicPricing())
345      screen2.add_show(show2)
346  
347      # Set up the system
348      bms = BookMyShow()
349      bms.add_cinema(cinema)
350      bms.add_observer(WaitlistNotifier())
351  
352      print("=== Movie Ticket Booking ===\n")
353  
354      # Search shows
355      shows = bms.search_shows("Mumbai", "Inception")
356      for s in shows:
357          print(f"  {s}")
358          print(f"    Regular: ${s.get_price(SeatType.REGULAR):.2f}")
359          print(f"    Premium: ${s.get_price(SeatType.PREMIUM):.2f}")
360          if SeatType.VIP in s.screen.seat_layout:
361              print(f"    VIP:     ${s.get_price(SeatType.VIP):.2f}")
362  
363      # Book Friday evening show with premium seats
364      print(f"\nBooking premium seats for Friday evening show:")
365      lock1 = bms.lock_seats(show1, "user-A", ["P-R1-S1", "P-R1-S2"])
366      for s in lock1.seats:
367          print(f"  Locked: {s}")
368      print(f"  Available premium after lock: {len(show1.available_seats(SeatType.PREMIUM))}")
369  
370      booking1 = bms.confirm_booking(lock1.lock_id)
371      print(f"  Booking {booking1.booking_id} confirmed | Total: ${booking1.total_amount:.2f}")
372  
373      # Book Tuesday matinee with regular seats (much cheaper)
374      print(f"\nBooking regular seats for Tuesday matinee:")
375      lock2 = bms.lock_seats(show2, "user-B", ["R-R1-S1", "R-R1-S2", "R-R1-S3"])
376      booking2 = bms.confirm_booking(lock2.lock_id)
377      print(f"  Booking {booking2.booking_id} confirmed | Total: ${booking2.total_amount:.2f}")
378  
379      # Show the price difference
380      print(f"\nPrice comparison:")
381      print(f"  Friday 9PM Premium (2 seats): ${booking1.total_amount:.2f}")
382      print(f"  Tuesday 10AM Regular (3 seats): ${booking2.total_amount:.2f}")
383  
384      # Cancel and trigger waitlist notification
385      print(f"\nCancelling Friday booking:")
386      bms.cancel_booking(booking1.booking_id)
387      print(f"  Premium available: {len(show1.available_seats(SeatType.PREMIUM))}")