File size: 3,600 Bytes
ebf1310 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 |
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import List, Optional
import os
from dotenv import load_dotenv
from .ai_engine import CultureAI
from .content_db import CulturalDatabase
# Load environment variables
load_dotenv()
# Initialize FastAPI app
app = FastAPI(
title="CultureBot API",
description="AI-powered cultural insights and information",
version="2.1.0"
)
# Configure CORS
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Initialize AI engine and database
culture_ai = CultureAI()
cultural_db = CulturalDatabase()
# Define Pydantic models
class ChatMessage(BaseModel):
message: str
user_id: Optional[str] = "anonymous"
class ChatResponse(BaseModel):
response: str
confidence: float
sources: List[str]
category: Optional[str] = None
class CulturalFact(BaseModel):
country: str
fact: str
category: str
source: str
@app.get("/")
async def root():
return {
"message": "Welcome to CultureBot API",
"version": "2.1.0",
"status": "active"
}
@app.post("/chat", response_model=ChatResponse)
async def chat_with_bot(message: ChatMessage):
"""
Main chat endpoint for interacting with CultureBot
"""
try:
# Get AI response
ai_response = await culture_ai.generate_response(message.message)
# Get relevant cultural facts from database
relevant_facts = cultural_db.search_facts(message.message)
return ChatResponse(
response=ai_response["response"],
confidence=ai_response["confidence"],
sources=ai_response["sources"],
category=ai_response.get("category")
)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Error processing request: {str(e)}")
@app.get("/facts/random", response_model=CulturalFact)
async def get_random_fact():
"""
Get a random cultural fact
"""
try:
fact = cultural_db.get_random_fact()
return CulturalFact(**fact)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Error fetching fact: {str(e)}")
@app.get("/facts/country/{country}", response_model=List[CulturalFact])
async def get_country_facts(country: str):
"""
Get cultural facts for a specific country
"""
try:
facts = cultural_db.get_facts_by_country(country)
return [CulturalFact(**fact) for fact in facts]
except Exception as e:
raise HTTPException(status_code=500, detail=f"Error fetching country facts: {str(e)}")
@app.get("/facts/category/{category}", response_model=List[CulturalFact])
async def get_category_facts(category: str):
"""
Get cultural facts for a specific category
"""
try:
facts = cultural_db.get_facts_by_category(category)
return [CulturalFact(**fact) for fact in facts]
except Exception as e:
raise HTTPException(status_code=500, detail=f"Error fetching category facts: {str(e)}")
@app.get("/health")
async def health_check():
"""
Health check endpoint
"""
return {
"status": "healthy",
"ai_engine": "operational",
"database": "connected"
}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
|