Badminton Court Booking Agent

from fastapi import FastAPI, HTTPException from pydantic import BaseModel import google.generativeai as genai from tinydb import TinyDB, Query from datetime import datetime, timedelta import re app = FastAPI(title="Badminton Court Booking Agent") # Initialize Gemini 2.5 Flash for code generation genai.configure(api_key=os.getenv("GEMINI_API_KEY")) model = genai.GenerativeModel('gemini-2.5-flash') # Initialize TinyDB for persistent storage db = TinyDB('bookings.json') bookings_table = db.table('bookings') class BookingRequest(BaseModel): user_message: str ic_number: str = None class AgenticBookingAgent: """ M5 Agentic Workflow implementation for badminton court bookings. The LLM generates Python code to execute booking operations. """ def __init__(self): self.model = model self.db = bookings_table # System context for code generation self.system_context = """ You are an AI agent that generates Python code to manage badminton court bookings. AVAILABLE FUNCTIONS: - get_bookings() -> List[Dict]: Returns all bookings from database - create_booking(court, date, start_time, end_time, ic, name) -> Dict - cancel_booking(ic, court, date) -> Dict - check_availability(court, date, start_time, end_time) -> bool - calculate_price(start_time, end_time, date) -> float PRICING RULES: - Daytime (6am-6pm): Weekday RM80/hr, Weekend RM90/hr - Nighttime (6pm-6am): Weekday RM100/hr, Weekend RM120/hr IC NUMBER FORMAT: YYMMDD-PB-###G (e.g., 850715-08-1234) COURTS: 1-12 available Generate executable Python code to handle the user's request. Return only the code, no explanations. """ async def process_request(self, user_message: str, ic_number: str = None) -> Dict: """ Main agentic workflow: Analyze request -> Generate code -> Execute -> Return result """ # Step 1: Analyze user intent and generate executable code generated_code = await self._generate_action_code(user_message, ic_number) # Step 2: Validate generated code for security if not self._validate_code_safety(generated_code): raise HTTPException(status_code=400, detail="Generated code failed security validation") # Step 3: Execute code in controlled environment result = await self._execute_code(generated_code) # Step 4: Format and return response return { "success": True, "result": result, "generated_code": generated_code if os.getenv("DEBUG") else None } async def _generate_action_code(self, user_message: str, ic_number: str) -> str: """ Use Gemini 2.5 Flash to generate executable Python code based on user request. This is the core of the Code-as-Action pattern. """ prompt = f""" {self.system_context} USER REQUEST: {user_message} IC NUMBER: {ic_number if ic_number else "Not provided"} CURRENT DATE: {datetime.now().strftime('%Y-%m-%d')} CURRENT TIME: {datetime.now().strftime('%H:%M')} Generate Python code to handle this request. The code should: 1. Parse the user's intent (book, cancel, check availability, query) 2. Validate inputs (IC format, court number, date/time) 3. Check for conflicts if booking 4. Calculate pricing if booking 5. Execute the database operation 6. Return a clear result message Code must use only the provided functions and follow all pricing rules. """ response = self.model.generate_content(prompt) code = self._extract_code_from_response(response.text) return code def _extract_code_from_response(self, response_text: str) -> str: """Extract Python code from LLM response, handling markdown code blocks.""" # Remove markdown code block markers code = re.sub(r'```python\n', '', response_text) code = re.sub(r'```\n?', '', code) code = code.strip() return code def _validate_code_safety(self, code: str) -> bool: """ Security validation to prevent malicious code execution. Checks for dangerous operations and unauthorized imports. """ # Blacklist dangerous operations dangerous_patterns = [ r'import\s+os', r'import\s+sys', r'import\s+subprocess', r'__import__', r'eval\(', r'exec\(', r'open\(', r'file\(', r'compile\(' ] for pattern in dangerous_patterns: if re.search(pattern, code): return False # Whitelist allowed operations allowed_imports = ['datetime', 'timedelta', 're'] return True async def _execute_code(self, code: str) -> Dict: """ Execute generated code in a controlled namespace with access to booking functions. This is where the AI-generated code actually runs. """ # Create execution namespace with helper functions namespace = { 'get_bookings': self._get_bookings, 'create_booking': self._create_booking, 'cancel_booking': self._cancel_booking, 'check_availability': self._check_availability, 'calculate_price': self._calculate_price, 'datetime': datetime, 'timedelta': timedelta, 're': re, 'result': None } try: # Execute generated code exec(code, namespace) # Return the result (code should set 'result' variable) return namespace.get('result', {"message": "Operation completed"}) except Exception as e: return { "error": True, "message": f"Code execution failed: {str(e)}" }

class BookingOperations: def _create_booking(self, court: int, date: str, start_time: str, end_time: str, ic: str, name: str) -> Dict: """ Create a new booking with conflict detection and pricing calculation. """ # Validate IC number format (Malaysian) if not self._validate_ic_number(ic): return { "error": True, "message": "Invalid IC number format. Use: YYMMDD-PB-###G" } # Validate court number if court < 1 or court > 12: return {"error": True, "message": "Court must be between 1-12"} # Check for conflicts if not self._check_availability(court, date, start_time, end_time): return { "error": True, "message": f"Court {court} is already booked for this time slot" } # Calculate total price total_price = self._calculate_price(start_time, end_time, date) # Create booking record booking = { "booking_id": self._generate_booking_id(), "court": court, "date": date, "start_time": start_time, "end_time": end_time, "ic_number": ic, "customer_name": name, "total_price": total_price, "created_at": datetime.now().isoformat(), "status": "confirmed" } # Save to database self.db.insert(booking) return { "success": True, "message": f"Booking confirmed for Court {court} on {date}", "booking_id": booking["booking_id"], "total_price": f"RM {total_price:.2f}", "details": booking } def _calculate_price(self, start_time: str, end_time: str, date: str) -> float: """ Calculate price based on time of day and day of week. Implements four-tier dynamic pricing system. """ start = datetime.strptime(f"{date} {start_time}", "%Y-%m-%d %H:%M") end = datetime.strptime(f"{date} {end_time}", "%Y-%m-%d %H:%M") # Calculate duration in hours duration_hours = (end - start).total_seconds() / 3600 # Determine if weekend (Saturday=5, Sunday=6) is_weekend = start.weekday() >= 5 # Determine if nighttime (6pm - 6am) is_nighttime = start.hour >= 18 or start.hour < 6 # Four-tier pricing logic if is_weekend and is_nighttime: hourly_rate = 120 # Weekend nighttime elif is_weekend: hourly_rate = 90 # Weekend daytime elif is_nighttime: hourly_rate = 100 # Weekday nighttime else: hourly_rate = 80 # Weekday daytime return hourly_rate * duration_hours def _check_availability(self, court: int, date: str, start_time: str, end_time: str) -> bool: """ Check if court is available for the requested time slot. Returns True if available, False if conflict exists. """ Booking = Query() existing_bookings = self.db.search( (Booking.court == court) & (Booking.date == date) & (Booking.status == "confirmed") ) requested_start = datetime.strptime(f"{date} {start_time}", "%Y-%m-%d %H:%M") requested_end = datetime.strptime(f"{date} {end_time}", "%Y-%m-%d %H:%M") for booking in existing_bookings: existing_start = datetime.strptime( f"{booking['date']} {booking['start_time']}", "%Y-%m-%d %H:%M" ) existing_end = datetime.strptime( f"{booking['date']} {booking['end_time']}", "%Y-%m-%d %H:%M" ) # Check for time overlap if (requested_start < existing_end and requested_end > existing_start): return False # Conflict found return True # No conflicts def _validate_ic_number(self, ic: str) -> bool: """ Validate Malaysian IC number format: YYMMDD-PB-###G Example: 850715-08-1234 (born July 15, 1985, from Johor) """ pattern = r'^[0-9]{6}-[0-9]{2}-[0-9]{4}$' return bool(re.match(pattern, ic)) def _cancel_booking(self, ic: str, court: int, date: str) -> Dict: """Cancel an existing booking and process refund.""" Booking = Query() booking = self.db.get( (Booking.ic_number == ic) & (Booking.court == court) & (Booking.date == date) & (Booking.status == "confirmed") ) if not booking: return { "error": True, "message": "No booking found with these details" } # Update booking status self.db.update( {"status": "cancelled", "cancelled_at": datetime.now().isoformat()}, (Booking.ic_number == ic) & (Booking.court == court) & (Booking.date == date) ) return { "success": True, "message": f"Booking cancelled for Court {court} on {date}", "refund_amount": f"RM {booking['total_price']:.2f}" }

# User Request: "Book court 5 for tomorrow at 7 PM for 2 hours. IC: 850715-08-1234, Name: Ahmad" # This code was GENERATED by Gemini 2.5 Flash: from datetime import datetime, timedelta # Parse booking details booking_date = (datetime.now() + timedelta(days=1)).strftime('%Y-%m-%d') start_time = "19:00" end_time = "21:00" court = 5 ic_number = "850715-08-1234" customer_name = "Ahmad" # Validate IC format if not re.match(r'^[0-9]{6}-[0-9]{2}-[0-9]{4}$', ic_number): result = {"error": True, "message": "Invalid IC format"} else: # Check availability is_available = check_availability(court, booking_date, start_time, end_time) if not is_available: result = { "error": True, "message": f"Court {court} is not available at this time" } else: # Calculate price (nighttime weekday: RM100/hr) total_price = calculate_price(start_time, end_time, booking_date) # Create booking result = create_booking( court=court, date=booking_date, start_time=start_time, end_time=end_time, ic=ic_number, name=customer_name )

# Initialize the agentic booking agent agent = AgenticBookingAgent() @app.post("/book") async def book_court(request: BookingRequest): """ Process natural language booking request through AI agent. The LLM generates code to handle the booking. """ try: result = await agent.process_request( user_message=request.user_message, ic_number=request.ic_number ) return result except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.get("/availability") async def check_availability(court: int, date: str, start: str, end: str): """Check if a specific court is available for booking.""" is_available = agent._check_availability(court, date, start, end) price = agent._calculate_price(start, end, date) return { "court": court, "date": date, "time_slot": f"{start} - {end}", "available": is_available, "price": f"RM {price:.2f}" if is_available else None } @app.get("/bookings") async def get_all_bookings(ic_number: str = None): """Retrieve all bookings, optionally filtered by IC number.""" if ic_number: Booking = Query() bookings = agent.db.search(Booking.ic_number == ic_number) else: bookings = agent.db.all() return {"bookings": bookings, "count": len(bookings)}