|
|
import os |
|
|
import sys |
|
|
from fastapi import FastAPI, HTTPException, Depends, status |
|
|
from fastapi.responses import PlainTextResponse |
|
|
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials |
|
|
from pydantic import BaseModel, Field |
|
|
from typing import Optional, List, Dict, Any, Union |
|
|
from datetime import datetime, timezone, timedelta, date |
|
|
import zoneinfo |
|
|
import psycopg2 |
|
|
from psycopg2 import pool as psycopg2_pool |
|
|
from jose import JWTError, jwt |
|
|
import uvicorn |
|
|
from dotenv import load_dotenv |
|
|
import time |
|
|
import uuid |
|
|
from src.recommendation_api import ( |
|
|
TourRecommendationRequest, |
|
|
TourRecommendationResponse, |
|
|
get_tour_recommendations |
|
|
) |
|
|
|
|
|
load_dotenv() |
|
|
|
|
|
try: |
|
|
from src.config import DB_USER, DB_PASSWORD, DB_HOST, DB_PORT, DB_NAME, DB_ENDPOINT_ID, GOOGLE_API_KEY, JWT_SECRET_KEY, ALGORITHM |
|
|
from src.database import conn_pool |
|
|
from src.graph_builder import graph_app |
|
|
from src.embedding import embedding_model |
|
|
from langchain_core.messages import HumanMessage, AIMessage, BaseMessage |
|
|
from src.tools import extract_entities_tool, search_tours_tool |
|
|
except ImportError as e: |
|
|
print(f"Error importing from src: {e}. Using placeholders. API will likely fail at runtime until this is fixed.") |
|
|
DB_USER, DB_PASSWORD, DB_HOST, DB_PORT, DB_NAME, DB_ENDPOINT_ID, GOOGLE_API_KEY, JWT_SECRET_KEY, ALGORITHM = [None]*9 |
|
|
conn_pool = None |
|
|
graph_app = None |
|
|
embedding_model = None |
|
|
class HumanMessage: |
|
|
def __init__(self, content): |
|
|
self.content = content |
|
|
class AIMessage: |
|
|
def __init__(self, content): |
|
|
self.content = content |
|
|
BaseMessage = Dict |
|
|
|
|
|
app = FastAPI( |
|
|
title="Travel Chatbot, Tour Search & Recommendation API", |
|
|
version="1.0.0" |
|
|
) |
|
|
|
|
|
reusable_oauth2 = HTTPBearer( |
|
|
scheme_name="Bearer" |
|
|
) |
|
|
|
|
|
class EmbeddingRequest(BaseModel): |
|
|
text: Union[str, List[str]] |
|
|
|
|
|
class EmbeddingResponse(BaseModel): |
|
|
embeddings: List[List[float]] |
|
|
model: str |
|
|
dimensions: int |
|
|
|
|
|
class TourSearchRequest(BaseModel): |
|
|
query: str = Field(..., description="The search query for tours, e.g., 'tôi muốn đi đà nẵng'") |
|
|
page: int = Field(1, ge=1, description="Current page number for pagination") |
|
|
limit: int = Field(10, ge=1, le=100, description="Number of items per page") |
|
|
|
|
|
class TourSummary(BaseModel): |
|
|
tour_id: Any |
|
|
title: str |
|
|
duration: Optional[str] = None |
|
|
departure_location: Optional[str] = None |
|
|
destination: Optional[List[str]] = None |
|
|
region: Optional[str] = None |
|
|
itinerary: Optional[str] = None |
|
|
max_participants: Optional[int] = None |
|
|
departure_id: Optional[Any] = None |
|
|
start_date: Optional[Union[datetime, date]] = None |
|
|
price_adult: Optional[float] = None |
|
|
price_child_120_140: Optional[float] = None |
|
|
price_child_100_120: Optional[float] = None |
|
|
promotion_name: Optional[str] = None |
|
|
promotion_type: Optional[str] = None |
|
|
promotion_discount: Optional[Any] = None |
|
|
|
|
|
class PaginatedTourResponse(BaseModel): |
|
|
currentPage: int |
|
|
itemsPerPage: int |
|
|
totalItems: int |
|
|
totalPages: int |
|
|
hasNextPage: bool |
|
|
hasPrevPage: bool |
|
|
tours: List[TourSummary] |
|
|
|
|
|
class TokenData(BaseModel): |
|
|
user_id: Optional[int] = None |
|
|
|
|
|
async def get_current_user(token: HTTPAuthorizationCredentials = Depends(reusable_oauth2)) -> int: |
|
|
credentials_exception = HTTPException( |
|
|
status_code=status.HTTP_401_UNAUTHORIZED, |
|
|
detail="Could not validate credentials", |
|
|
headers={"WWW-Authenticate": "Bearer"}, |
|
|
) |
|
|
try: |
|
|
payload = jwt.decode(token.credentials, JWT_SECRET_KEY, algorithms=[ALGORITHM]) |
|
|
|
|
|
user_id: Optional[int] = payload.get("id") |
|
|
if user_id is None: |
|
|
user_id = payload.get("userId") |
|
|
|
|
|
if user_id is None: |
|
|
print(f"JWT payload does not contain 'id' or 'userId' field. Payload: {payload}") |
|
|
raise credentials_exception |
|
|
|
|
|
except JWTError as e: |
|
|
print(f"JWTError: {e}") |
|
|
raise credentials_exception |
|
|
except Exception as e: |
|
|
print(f"An unexpected error occurred during JWT decoding: {e}") |
|
|
raise credentials_exception |
|
|
return user_id |
|
|
|
|
|
class ChatMessageInput(BaseModel): |
|
|
message: str = Field(..., description="The text message sent by the user to the chatbot.") |
|
|
session_id: Optional[str] = Field(None, description="An optional identifier for a specific chat session.") |
|
|
|
|
|
class ChatResponseOutput(BaseModel): |
|
|
user_id: int = Field(..., description="The ID of the user (from JWT token).") |
|
|
response: str = Field(..., description="The chatbot's generated textual response.") |
|
|
session_id: Optional[str] = Field(None, description="The session identifier, mirrored if provided in input.") |
|
|
timestamp: datetime = Field(..., description="UTC timestamp of when the response was generated.") |
|
|
|
|
|
def get_db_connection(): |
|
|
if conn_pool is None: |
|
|
print("conn_pool is None in get_db_connection. Database module likely not initialized.") |
|
|
raise HTTPException(status_code=503, detail="Database connection pool not initialized. Check src.database and .env configuration.") |
|
|
try: |
|
|
conn = conn_pool.getconn() |
|
|
yield conn |
|
|
finally: |
|
|
if conn: |
|
|
conn_pool.putconn(conn) |
|
|
|
|
|
def fetch_conversation_history(db_conn, user_id: int, session_id: Optional[str] = None) -> List[BaseMessage]: |
|
|
history: List[BaseMessage] = [] |
|
|
try: |
|
|
with db_conn.cursor(cursor_factory=psycopg2.extras.DictCursor) as cursor: |
|
|
if session_id: |
|
|
cursor.execute( |
|
|
"SELECT message, response FROM ChatbotHistory WHERE user_id = %s AND session_id = %s ORDER BY interaction_time ASC", |
|
|
(user_id, session_id) |
|
|
) |
|
|
else: |
|
|
cursor.execute( |
|
|
"SELECT message, response FROM ChatbotHistory WHERE user_id = %s AND session_id IS NULL ORDER BY interaction_time ASC", |
|
|
(user_id,) |
|
|
) |
|
|
records = cursor.fetchall() |
|
|
for record in records: |
|
|
if record["message"]: |
|
|
history.append(HumanMessage(content=record["message"])) |
|
|
if record["response"]: |
|
|
history.append(AIMessage(content=record["response"])) |
|
|
except Exception as e: |
|
|
print(f"Error fetching conversation history for user_id {user_id}, session_id {session_id}: {e}") |
|
|
return history |
|
|
|
|
|
def save_interaction_to_history(db_conn, user_id: int, user_message: str, chatbot_response: str, session_id: Optional[str] = None): |
|
|
try: |
|
|
with db_conn.cursor() as cursor: |
|
|
cursor.execute( |
|
|
"INSERT INTO ChatbotHistory (user_id, message, response, interaction_time, session_id) VALUES (%s, %s, %s, %s, %s)", |
|
|
(user_id, user_message, chatbot_response, datetime.now(zoneinfo.ZoneInfo("Asia/Bangkok")), session_id) |
|
|
) |
|
|
db_conn.commit() |
|
|
except Exception as e: |
|
|
print(f"Error saving interaction to history for user_id {user_id}, session_id {session_id}: {e}") |
|
|
db_conn.rollback() |
|
|
|
|
|
|
|
|
@app.on_event("startup") |
|
|
async def startup_event(): |
|
|
try: |
|
|
if embedding_model: |
|
|
embedding_model.load_model() |
|
|
except Exception as e: |
|
|
print(f"Failed to load embedding model on startup: {str(e)}") |
|
|
|
|
|
@app.get("/", include_in_schema=False, response_class=PlainTextResponse) |
|
|
async def root(): |
|
|
return "API is running" |
|
|
|
|
|
@app.post("/api/chat/", response_model=ChatResponseOutput, tags=["Chat"], summary="Chat with the Travel Chatbot") |
|
|
async def chat_endpoint(payload: ChatMessageInput, current_user_id: int = Depends(get_current_user), db_conn = Depends(get_db_connection)): |
|
|
if graph_app is None: |
|
|
print("graph_app is None in chat_endpoint. Graph_builder module likely not initialized.") |
|
|
raise HTTPException(status_code=503, detail="Chatbot graph not initialized. Check src.graph_builder.") |
|
|
|
|
|
user_id = current_user_id |
|
|
user_message_content = payload.message |
|
|
session_id = payload.session_id |
|
|
|
|
|
if not session_id: |
|
|
session_id = str(uuid.uuid4()) |
|
|
|
|
|
history = fetch_conversation_history(db_conn, user_id, session_id) |
|
|
|
|
|
current_message = HumanMessage(content=user_message_content) |
|
|
all_messages = history + [current_message] |
|
|
|
|
|
inputs = { |
|
|
"messages": all_messages, |
|
|
"user_query": user_message_content, |
|
|
"current_date": None, |
|
|
"available_locations": None, |
|
|
"extracted_entities": None, |
|
|
"search_results": None, |
|
|
"final_response": None, |
|
|
"error": None, |
|
|
"routing_decision": None |
|
|
} |
|
|
|
|
|
full_response_content = "" |
|
|
try: |
|
|
result = graph_app.invoke(inputs) |
|
|
|
|
|
if isinstance(result, dict) and "final_response" in result: |
|
|
full_response_content = result["final_response"] |
|
|
elif isinstance(result, dict) and "messages" in result and result["messages"]: |
|
|
last_message = result["messages"][-1] |
|
|
if isinstance(last_message, AIMessage): |
|
|
full_response_content = last_message.content |
|
|
|
|
|
if not full_response_content: |
|
|
full_response_content = "Sorry, I could not process your request at this moment." |
|
|
|
|
|
except Exception as e: |
|
|
print(f"Error during graph invocation for user_id {user_id}: {e}") |
|
|
raise HTTPException(status_code=500, detail=f"Error processing message with chatbot: {str(e)}") |
|
|
|
|
|
save_interaction_to_history(db_conn, user_id, user_message_content, full_response_content, session_id) |
|
|
|
|
|
return ChatResponseOutput( |
|
|
user_id=user_id, |
|
|
response=full_response_content, |
|
|
session_id=session_id, |
|
|
timestamp=datetime.now(zoneinfo.ZoneInfo("Asia/Bangkok")) |
|
|
) |
|
|
|
|
|
@app.post("/api/tours/search", response_model=PaginatedTourResponse, tags=["Tours"], summary="Search for tours based on user query") |
|
|
async def search_tours_api(request: TourSearchRequest, current_user_id: int = Depends(get_current_user)): |
|
|
current_date_str = date.today().strftime('%Y-%m-%d') |
|
|
|
|
|
entities = extract_entities_tool(user_query=request.query, current_date_str=current_date_str) |
|
|
|
|
|
if not entities or (isinstance(entities, dict) and entities.get("error")): |
|
|
return PaginatedTourResponse( |
|
|
currentPage=request.page, |
|
|
itemsPerPage=request.limit, |
|
|
totalItems=0, |
|
|
totalPages=0, |
|
|
hasNextPage=False, |
|
|
hasPrevPage=False, |
|
|
tours=[] |
|
|
) |
|
|
|
|
|
all_found_tours = search_tours_tool(entities) |
|
|
|
|
|
if not all_found_tours: |
|
|
return PaginatedTourResponse( |
|
|
currentPage=request.page, |
|
|
itemsPerPage=request.limit, |
|
|
totalItems=0, |
|
|
totalPages=0, |
|
|
hasNextPage=False, |
|
|
hasPrevPage=False, |
|
|
tours=[] |
|
|
) |
|
|
|
|
|
total_items = len(all_found_tours) |
|
|
total_pages = (total_items + request.limit - 1) // request.limit |
|
|
|
|
|
start_index = (request.page - 1) * request.limit |
|
|
end_index = start_index + request.limit |
|
|
paginated_tours_data = all_found_tours[start_index:end_index] |
|
|
|
|
|
tour_summaries = [] |
|
|
for tour_data in paginated_tours_data: |
|
|
region = tour_data.get("region") |
|
|
if region is not None and not isinstance(region, str): |
|
|
region = str(region) |
|
|
|
|
|
destination = tour_data.get("destination") |
|
|
if destination and not isinstance(destination, list): |
|
|
destination = [str(destination)] |
|
|
elif destination: |
|
|
destination = [str(dest) for dest in destination] |
|
|
|
|
|
tour_summaries.append(TourSummary( |
|
|
tour_id=tour_data.get("tour_id"), |
|
|
title=str(tour_data.get("title", "")), |
|
|
duration=str(tour_data.get("duration")) if tour_data.get("duration") is not None else None, |
|
|
departure_location=str(tour_data.get("departure_location")) if tour_data.get("departure_location") is not None else None, |
|
|
destination=destination, |
|
|
region=region, |
|
|
itinerary=str(tour_data.get("itinerary")) if tour_data.get("itinerary") is not None else None, |
|
|
max_participants=int(tour_data.get("max_participants")) if tour_data.get("max_participants") is not None else None, |
|
|
departure_id=tour_data.get("departure_id"), |
|
|
start_date=tour_data.get("start_date"), |
|
|
price_adult=float(tour_data.get("price_adult")) if tour_data.get("price_adult") is not None else None, |
|
|
price_child_120_140=float(tour_data.get("price_child_120_140")) if tour_data.get("price_child_120_140") is not None else None, |
|
|
price_child_100_120=float(tour_data.get("price_child_100_120")) if tour_data.get("price_child_100_120") is not None else None, |
|
|
promotion_name=str(tour_data.get("promotion_name")) if tour_data.get("promotion_name") is not None else None, |
|
|
promotion_type=str(tour_data.get("promotion_type")) if tour_data.get("promotion_type") is not None else None, |
|
|
promotion_discount=tour_data.get("promotion_discount") |
|
|
)) |
|
|
|
|
|
return PaginatedTourResponse( |
|
|
currentPage=request.page, |
|
|
itemsPerPage=request.limit, |
|
|
totalItems=total_items, |
|
|
totalPages=total_pages, |
|
|
hasNextPage=(request.page < total_pages), |
|
|
hasPrevPage=(request.page > 1), |
|
|
tours=tour_summaries |
|
|
) |
|
|
|
|
|
@app.get("/api/recommendations", response_model=TourRecommendationResponse, tags=["Recommendations"], summary="Get tour recommendations") |
|
|
async def get_recommendations( |
|
|
user_id: Optional[int] = None, |
|
|
tour_id: Optional[int] = None, |
|
|
limit: int = 3 |
|
|
): |
|
|
try: |
|
|
if limit < 1 or limit > 10: |
|
|
limit = 3 |
|
|
recommendations = get_tour_recommendations(user_id, tour_id, limit) |
|
|
return recommendations |
|
|
except Exception as e: |
|
|
raise HTTPException(status_code=500, detail=f"Error getting recommendations: {str(e)}") |
|
|
|
|
|
@app.post("/api/embed", response_model=EmbeddingResponse, tags=["Embeddings"], summary="Generate text embeddings") |
|
|
async def get_embedding(request: EmbeddingRequest): |
|
|
if embedding_model is None: |
|
|
raise HTTPException( |
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE, |
|
|
detail="Embedding service not initialized. Check src.embedding module." |
|
|
) |
|
|
|
|
|
try: |
|
|
embeddings = embedding_model.get_embedding(request.text) |
|
|
|
|
|
return { |
|
|
"embeddings": embeddings, |
|
|
"model": embedding_model.model_name, |
|
|
"dimensions": len(embeddings[0]) |
|
|
} |
|
|
except Exception as e: |
|
|
raise HTTPException( |
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, |
|
|
detail=f"Error generating embeddings: {str(e)}" |
|
|
) |
