IRIS Legacy

""" Trusty Personal Assistant - Booking Page Standalone FastAPI app for external users to confirm meeting bookings. This is a thin UI layer - all business logic is handled via MCP server. Architecture: - GET /book/{session_id}: Display booking page - POST /api/confirm: Confirm selected slot - POST /api/find_new_times: Find alternative times - All operations call MCP server (no direct Graph API calls) Branding: Tinexta InfoCert official colors + professional UX URL: trustypa.brainaihub.tech """ import logging import os import httpx from datetime import datetime from typing import Dict, Any, Optional from dotenv import load_dotenv from fastapi import FastAPI, HTTPException, Request from fastapi.responses import HTMLResponse, JSONResponse from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates from pydantic import BaseModel # Load environment variables load_dotenv() # Import database after loading env vars from src.database import create_tables # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) # Initialize database tables try: create_tables() logger.info("✅ Database tables initialized successfully") except Exception as e: logger.error(f"❌ Failed to initialize database tables: {e}") # Create FastAPI app app = FastAPI( title="Trusty Personal Assistant - Booking Page", description="Booking confirmation page for external users", version="1.0.0", root_path="/booking" ) # Setup templates and static files templates = Jinja2Templates(directory=os.path.join(os.path.dirname(__file__), "templates")) app.mount("/static", StaticFiles(directory=os.path.join(os.path.dirname(__file__), "static")), name="static") # MCP Server URL (default to localhost for development) MCP_SERVER_URL = os.getenv("MCP_SERVER_URL", "http://localhost:8001") # Pydantic models class ConfirmBookingRequest(BaseModel): """Request to confirm a booking slot.""" session_id: str selected_slot_index: int class FindNewTimesRequest(BaseModel): """Request to find new available times.""" session_id: str async def call_mcp_operation(action: str, params: Dict[str, Any]) -> Dict[str, Any]: """ Call MCP server bookings operation. Args: action: MCP action to perform params: Action parameters Returns: MCP response data Raises: HTTPException: If MCP call fails """ try: async with httpx.AsyncClient(timeout=30.0) as client: response = await client.post( f"{MCP_SERVER_URL}/mcp/bookings_operations", json={ "action": action, "params": params } ) response.raise_for_status() result = response.json() if not result.get("success"): error_msg = result.get("error", {}).get("message", "Unknown error") raise HTTPException(status_code=400, detail=error_msg) return result.get("data", {}) except httpx.HTTPError as e: logger.error(f"MCP call failed: {e}") raise HTTPException(status_code=500, detail=f"Failed to communicate with server: {str(e)}") @app.get("/") async def root(): """Root endpoint.""" return { "service": "Trusty Personal Assistant - Booking Page", "version": "1.0.0", "branding": "Tinexta InfoCert" } @app.get("/health") async def health_check(): """Health check endpoint.""" return { "status": "healthy", "mcp_server_url": MCP_SERVER_URL } @app.get("/book/{session_id}", response_class=HTMLResponse) async def get_booking_page(session_id: str, request: Request): """ Display booking page for external user. Args: session_id: Booking session ID from email link request: FastAPI request object Returns: HTML page with booking options """ try: # Get session details from MCP server session = await call_mcp_operation("get_hybrid_session", {"session_id": session_id}) # Check if session is expired or already confirmed if session['status'] == 'expired': return templates.TemplateResponse( "error.html", { "request": request, "error_title": "Link Expired", "error_message": "This booking link has expired. Please contact the organizer for a new invitation.", "organizer_name": session.get('organizer_name', 'the organizer') } ) if session['status'] == 'confirmed': return templates.TemplateResponse( "already_booked.html", { "request": request, "session": session, "confirmed_slot": session.get('confirmed_slot', {}) } ) # Check availability of each slot in real-time slots_with_availability = [] for i, slot in enumerate(session['proposed_slots']): # Check slot availability via MCP availability = await call_mcp_operation( "check_hybrid_slot", { "session_id": session_id, "slot_index": i } ) slots_with_availability.append({ "index": i, "start": slot['start'], "end": slot['end'], "available": availability['available'], "confidence": availability.get('confidence', 0) }) # Count available slots available_count = sum(1 for slot in slots_with_availability if slot['available']) # Determine scenario if available_count == 0: scenario = "none_available" elif available_count < len(slots_with_availability): scenario = "some_available" else: scenario = "all_available" # Render booking page return templates.TemplateResponse( "booking.html", { "request": request, "session": session, "slots": slots_with_availability, "scenario": scenario, "available_count": available_count } ) except HTTPException: raise except Exception as e: logger.error(f"Error loading booking page: {e}") return templates.TemplateResponse( "error.html", { "request": request, "error_title": "Error", "error_message": "An error occurred while loading the booking page. Please try again or contact the organizer.", "organizer_name": "the organizer" } ) @app.post("/api/confirm") async def confirm_booking(request: ConfirmBookingRequest): """ Confirm booking and create calendar event. Args: request: Confirmation request with session_id and selected_slot_index Returns: Confirmation details with event_id and meeting link """ try: # Confirm booking via MCP server result = await call_mcp_operation( "confirm_hybrid_booking", { "session_id": request.session_id, "selected_slot_index": request.selected_slot_index } ) logger.info(f"Booking confirmed for session {request.session_id}") return JSONResponse({ "success": True, "message": "Booking confirmed successfully!", "data": result }) except HTTPException: raise except Exception as e: logger.error(f"Error confirming booking: {e}") raise HTTPException(status_code=500, detail=str(e)) @app.post("/api/find_new_times") async def find_new_times(request: FindNewTimesRequest): """ Find new available times if original slots are taken. Args: request: Request with session_id Returns: New list of available slots """ try: # Find new slots via MCP server result = await call_mcp_operation( "find_new_hybrid_slots", {"session_id": request.session_id} ) # Check availability of new slots new_slots_with_availability = [] for i, slot in enumerate(result['new_slots']): availability = await call_mcp_operation( "check_hybrid_slot", { "session_id": request.session_id, "slot_index": i } ) new_slots_with_availability.append({ "index": i, "start": slot['start'], "end": slot['end'], "available": availability['available'], "confidence": availability.get('confidence', 0) }) logger.info(f"Found {len(new_slots_with_availability)} new slots for session {request.session_id}") return JSONResponse({ "success": True, "slots": new_slots_with_availability }) except HTTPException: raise except Exception as e: logger.error(f"Error finding new times: {e}") raise HTTPException(status_code=500, detail=str(e)) if __name__ == "__main__": import uvicorn # Run on port 8081 (production will use Nginx reverse proxy) uvicorn.run( "app:app", host="0.0.0.0", port=8081, reload=True, log_level="info" )