""" Main FastAPI application for Silver Table Assistant backend. Provides REST API endpoints and Gradio interface for AI-powered nutrition consultation. """ import os import logging from typing import List, Optional, Dict, Any from uuid import UUID import asyncio # Load environment variables from .env file from dotenv import load_dotenv load_dotenv() # FastAPI imports from fastapi import FastAPI, Request, HTTPException, Depends, status from fastapi.exceptions import RequestValidationError from fastapi.responses import JSONResponse from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import StreamingResponse, JSONResponse from fastapi.staticfiles import StaticFiles # Database imports from sqlalchemy.ext.asyncio import AsyncSession # Import local modules from database import create_db_and_tables, get_session from models import Profile, Order, Donation, MenuItem, ChatConversation from schemas import ( ProfileCreate, ProfileUpdate, ProfileRead, OrderCreate, OrderUpdate, OrderRead, DonationCreate, DonationUpdate, DonationRead, ChatRequest, ChatResponse, MenuItemRead, APIResponse, HealthCheck ) from dependencies import get_current_user, get_optional_user, require_roles, User, get_or_create_user_profile from exceptions import SilverTableException, PaymentException from exceptions import handle_payment_error from crud import ( get_profile, get_profiles_by_user, create_profile, update_profile, delete_profile, create_order, get_orders_by_profile, create_donation, update_donation_status, get_menu_items, get_dashboard_stats ) from menu_data import get_menu_items, get_menu_item_by_id from config import settings # Import service modules from chat_service import chat_stream, get_chat_service from stripe_service import create_checkout_session_for_order, create_checkout_session_for_donation, handle_webhook # Gradio imports import gradio as gr # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # ====== Lifespan Event Handler ====== from contextlib import asynccontextmanager @asynccontextmanager async def lifespan(app: FastAPI): """Handle application startup and shutdown events.""" # Startup logic logger.info("Starting up Silver Table Assistant backend...") # Try to create database tables, but continue if it fails (tables might already exist) try: await create_db_and_tables() logger.info("Database tables created/verified successfully") except Exception as e: logger.warning(f"Database initialization warning (continuing anyway): {str(e)}") # Yield control to the application yield # Shutdown logic logger.info("Shutting down Silver Table Assistant backend...") # Initialize FastAPI app with lifespan handler app = FastAPI( title="銀髮餐桌助手 API", description="專為台灣銀髮族設計的AI營養飲食顧問服務", version=settings.api_version, docs_url="/api/docs", redoc_url="/api/redoc", lifespan=lifespan ) # Custom exception handlers for consistent API responses @app.exception_handler(RequestValidationError) async def validation_exception_handler(request: Request, exc: RequestValidationError): return JSONResponse( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, content={"success": False, "message": "Request validation error", "details": exc.errors()} ) @app.exception_handler(SilverTableException) async def silvertable_exception_handler(request: Request, exc: SilverTableException): # Return 400 for known application errors with friendly message return JSONResponse( status_code=status.HTTP_400_BAD_REQUEST, content={"success": False, "message": exc.message, "details": exc.details} ) @app.exception_handler(PaymentException) async def payment_exception_handler(request: Request, exc: PaymentException): # Return 400 for payment-related errors with friendly message return JSONResponse( status_code=status.HTTP_400_BAD_REQUEST, content={"success": False, "message": exc.message, "details": exc.details} ) # CORS middleware configuration app.add_middleware( CORSMiddleware, allow_origins=settings.cors_origins, allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # ====== Health Check Endpoint ====== @app.get("/health", response_model=HealthCheck) async def health_check(): """Health check endpoint for monitoring service status.""" return HealthCheck( status="healthy", timestamp=settings.get_current_timestamp(), version=settings.api_version, database="connected" ) # ====== Profile Management Endpoints ====== @app.get("/api/profiles", response_model=List[ProfileRead]) async def get_profiles( current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_session) ): """Get all profiles for the current authenticated user.""" try: profiles = await get_profiles_by_user(db, current_user.user_id) return profiles except Exception as e: logger.error(f"Error fetching profiles: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to fetch profiles" ) @app.get("/api/profiles/{profile_id}", response_model=ProfileRead) async def get_profile_by_id( profile_id: str, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_session) ): """ Get a specific profile by ID. Requires authentication. """ try: # Validate profile_id format try: profile_uuid = UUID(profile_id) except ValueError: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid profile ID format" ) # Get the profile profile = await get_profile(db, profile_uuid) if not profile: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Profile not found" ) return profile except HTTPException: raise except Exception as e: logger.error(f"Error fetching profile {profile_id}: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to fetch profile" ) @app.post("/api/profiles", response_model=ProfileRead, status_code=status.HTTP_201_CREATED) async def create_user_profile( profile_data: ProfileCreate, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_session) ): """Create or update a profile for the current authenticated user.""" try: if profile_data.id: # Update specific profile profile = await update_profile(db, profile_data.id, profile_data) if not profile: # If ID was provided but not found, maybe it was deleted, create new profile = await create_profile(db, profile_data, current_user.user_id) else: # Create new profile profile = await create_profile(db, profile_data, current_user.user_id) return profile except Exception as e: logger.error(f"Error creating/updating profile: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to create/update profile" ) @app.put("/api/profiles/{profile_id}", response_model=ProfileRead) async def update_user_profile( profile_id: str, profile_data: ProfileUpdate, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_session) ): """ Update a specific profile by ID. Requires authentication. """ try: # Validate profile_id format try: profile_uuid = UUID(profile_id) except ValueError: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid profile ID format" ) # Update the profile profile = await update_profile(db, profile_uuid, profile_data) if not profile: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Profile not found" ) return profile except HTTPException: raise except Exception as e: logger.error(f"Error updating profile {profile_id}: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to update profile" ) @app.delete("/api/profiles/{profile_id}", response_model=APIResponse) async def delete_user_profile( profile_id: str, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_session) ): """ Delete a specific profile by ID. Requires authentication. """ try: # Validate profile_id format try: profile_uuid = UUID(profile_id) except ValueError: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid profile ID format" ) # Delete the profile deleted = await delete_profile(db, profile_uuid) if not deleted: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Profile not found" ) return APIResponse( success=True, message="Profile deleted successfully", data={"profile_id": profile_id} ) except HTTPException: raise except Exception as e: logger.error(f"Error deleting profile {profile_id}: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to delete profile" ) # ====== Chat Examples Configuration ====== CHAT_EXAMPLES = [ "請問銀髮族應該如何補充蛋白質？", "我爸爸有糖尿病，飲食上有什麼需要注意的？", "推薦一些適合銀髮族的早餐選項", "什麼食物對骨骼健康有好處？", "如何製作軟嫩的料理？" ] # ====== Chat Endpoints ====== class ChatRequestWithProfile(ChatRequest): """Extended chat request schema with profile_id.""" profile_id: Optional[str] = None @app.get("/api/chat/examples", response_model=List[str]) async def get_chat_examples(): """ Get the list of default example questions for the chat interface. These examples are also used in the Gradio interface. """ return CHAT_EXAMPLES @app.post("/api/chat") async def chat_with_assistant( request: ChatRequestWithProfile, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_session) ): """ Chat with the AI nutrition assistant. Requires authentication and can use profile_id for personalized responses. """ try: # Get or use provided profile_id profile_id = request.profile_id if not profile_id: # Get user's first profile for personalization profiles = await get_profiles_by_user(db, current_user.user_id) if profiles: profile_id = str(profiles[0].id) # Create streaming response return StreamingResponse( chat_stream( message=request.message, profile_id=profile_id, history=[{"role": "user", "content": request.message}] # Simplified history ), media_type="text/plain", headers={ "Cache-Control": "no-cache", "Connection": "keep-alive", "Transfer-Encoding": "chunked" } ) except Exception as e: logger.error(f"Error in chat endpoint: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Chat service temporarily unavailable" ) # ====== Menu Endpoints ====== @app.get("/api/menu", response_model=List[Dict[str, Any]]) async def get_menu(): """Get the complete menu of available food items.""" try: menu_items = get_menu_items() return menu_items except Exception as e: logger.error(f"Error fetching menu: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to fetch menu" ) @app.get("/api/menu/{item_id}", response_model=Dict[str, Any]) async def get_menu_item(item_id: int): """ Get a specific menu item by ID. Returns the menu item with all details including nutrition information. """ try: menu_item = get_menu_item_by_id(item_id) return menu_item except ValueError: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Menu item with ID {item_id} not found" ) except Exception as e: logger.error(f"Error fetching menu item {item_id}: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to fetch menu item" ) # ====== Order Management Endpoints ====== @app.post("/api/orders", response_model=OrderRead) async def create_order_endpoint( order_data: OrderCreate, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_session) ): """Create a new food order and initiate Stripe checkout.""" logger.info(f"[ORDER_DEBUG] Received order request from user: {current_user.user_id}") logger.info(f"[ORDER_DEBUG] Order data: {order_data.dict()}") try: # Get or create user profile automatically profile = await get_or_create_user_profile(current_user, db) logger.info(f"[ORDER_DEBUG] Using profile: {profile.id}") # Create order in database logger.info(f"[ORDER_DEBUG] Creating order with profile_id: {profile.id}") order = await create_order(db, order_data, profile.id) logger.info(f"[ORDER_DEBUG] Order created with ID: {order.id}") # Get user's email - prefer frontend-provided email, fall back to authenticated user's email user_email = order_data.customer_email or current_user.email logger.info(f"[ORDER_DEBUG] User email: {user_email}") # Create Stripe checkout session with user's email logger.info(f"[ORDER_DEBUG] Creating Stripe checkout session for order: {order.id}") checkout_url = create_checkout_session_for_order(order, customer_email=user_email) logger.info(f"[ORDER_DEBUG] Stripe checkout URL created: {checkout_url[:50]}...") # Update order with Stripe session ID from crud import update_order_status session_id = checkout_url.split('/')[-1] logger.info(f"[ORDER_DEBUG] Updating order with session_id: {session_id}") await update_order_status(db, order.id, "pending", session_id) return { "id": order.id, "profile_id": order.profile_id, "items": order.items, "total_amount": order.total_amount, "status": "pending", "stripe_session_id": session_id, "created_at": order.created_at, "updated_at": order.updated_at, "checkout_url": checkout_url } except HTTPException: raise except Exception as e: logger.error(f"[ORDER_DEBUG] Error creating order: {str(e)}") logger.error(f"[ORDER_DEBUG] Error type: {type(e).__name__}") import traceback logger.error(f"[ORDER_DEBUG] Traceback: {traceback.format_exc()}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Failed to create order: {str(e)}" ) # ====== Donation Endpoints ====== @app.post("/api/donations", response_model=DonationRead) async def create_donation_endpoint( donation_data: DonationCreate, current_user: Optional[User] = Depends(get_optional_user), db: AsyncSession = Depends(get_session) ): """ Create a new donation and initiate Stripe checkout. Supports both authenticated and anonymous donations. """ try: # Validate donation minimum amount (donation_data.amount is in cents) min_amount_cents = int(settings.MIN_DONATION_AMOUNT * 100) if donation_data.amount < min_amount_cents: raise PaymentException(f"捐款金額必須至少為 NT${settings.MIN_DONATION_AMOUNT} 元") # Create donation in database donation = await create_donation(db, donation_data) # Create Stripe checkout session checkout_url = create_checkout_session_for_donation(donation) # Update donation with Stripe session ID await update_donation_status(db, donation.id, "pending", checkout_url.split('/')[-1]) # Return the donation object with checkout URL added return { **donation.__dict__, "checkout_url": checkout_url } except Exception as e: logger.error(f"Error creating donation: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to create donation" ) # ====== Stripe Webhook Endpoints ====== @app.post("/api/webhook") async def stripe_webhook(request: Request): """Handle Stripe webhook events for payment confirmation.""" try: payload = await request.body() sig_header = request.headers.get("stripe-signature") if not sig_header: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Missing Stripe signature header" ) # Handle webhook result = handle_webhook(payload, sig_header) return JSONResponse( status_code=200, content={"status": "success", "result": result} ) except Exception as e: logger.error(f"Webhook error: {str(e)}") raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Webhook processing failed" ) @app.post("/api/stripe/webhook") async def stripe_webhook_endpoint(request: Request): """Handle Stripe webhook events for payment confirmation. This endpoint specifically handles Stripe webhook events with proper signature verification and supports events like checkout.session.completed, payment_intent.succeeded, etc. Expected webhook signing secret: STRIPE_WEBHOOK_SECRET environment variable """ try: # Get the raw payload and signature header payload = await request.body() sig_header = request.headers.get("stripe-signature") if not sig_header: logger.error("Missing Stripe signature header in webhook request") raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Missing Stripe signature header" ) logger.info(f"Processing Stripe webhook with signature: {sig_header[:20]}...") # Use the existing webhook handler from stripe_service result = handle_webhook(payload, sig_header) logger.info(f"Webhook processed successfully: {result.get('status', 'unknown')}") return JSONResponse( status_code=200, content={ "status": "success", "message": "Webhook processed successfully", "result": result } ) except Exception as e: logger.error(f"Stripe webhook error: {str(e)}") # Return appropriate error response based on exception type if "Invalid webhook signature" in str(e): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid webhook signature" ) else: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Webhook processing failed: {str(e)}" ) # ====== Dashboard Endpoints ====== @app.get("/api/dashboard/{profile_id}") async def get_user_dashboard( profile_id: str, current_user: User = Depends(require_roles(["family", "admin"])), db: AsyncSession = Depends(get_session) ): """ Get nutrition dashboard data for a specific profile. Requires family role or admin permissions. """ try: # Validate profile_id format try: profile_uuid = UUID(profile_id) except ValueError: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid profile ID format" ) # Get dashboard statistics stats = await get_dashboard_stats(db, profile_uuid) return stats except Exception as e: logger.error(f"Error fetching dashboard data: {str(e)}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to fetch dashboard data" ) # ====== Gradio Chat Interface ====== async def gradio_chat(message: str, history: List[Dict[str, str]]): """ Adapt chat_service.chat_stream to Gradio's expected format. """ try: chat_service = get_chat_service() response = "" # In Gradio 6.x, history is already a list of dictionaries with 'role' and 'content' async for chunk in chat_service.chat_stream(message, history=history): response += chunk yield response except Exception as e: logger.error(f"Error in Gradio chat: {str(e)}") yield "抱歉，系統暫時無法回應。請稍後再試。" # Create Gradio interface with gr.Blocks(title="銀髮餐桌助手") as gradio_demo: gr.Markdown( """ # 銀髮餐桌助手 🥄 專為台灣銀髮族設計的AI營養飲食顧問 **功能特色：** - 個人化營養建議 - 健康飲食指導 - 在地食材推薦 - 專業營養諮詢 **使用說明：** 無需登入即可開始對話，系統會根據您的健康狀況提供個人化建議。 """ ) chatbot = gr.ChatInterface( fn=gradio_chat, title="營養飲食諮詢", description="請輸入您的問題，例如：", examples=CHAT_EXAMPLES ) gr.Markdown( """ --- **重要提醒：** - 本系統僅提供營養建議，無法替代專業醫療諮詢 - 如有健康問題，請諮詢專業醫師 - 建議遵循台灣衛福部的營養指導原則 """ ) # Mount Gradio app to FastAPI app = gr.mount_gradio_app(app, gradio_demo, path="/") # ====== Main Entry Point ====== if __name__ == "__main__": import uvicorn logger.info(f"Starting Silver Table Assistant backend on {settings.host}:{settings.port}") # Run the application uvicorn.run( "app:app", host=settings.host, port=settings.port, reload=settings.is_development(), log_level="info" )