Spaces:
Running
Running
File size: 4,889 Bytes
dcefa44 |
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 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 |
from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from backend.utils import generate_completions
from backend import config
from backend.database import get_db_connection
import psycopg2
from psycopg2.extras import RealDictCursor
from typing import Union, List, Literal, Optional
import logging
import json
logging.basicConfig(level=logging.INFO)
app = FastAPI()
# Add CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # Allows all origins
allow_credentials=True,
allow_methods=["*"], # Allows all methods
allow_headers=["*"], # Allows all headers
)
# Dependency to get database connection
async def get_db():
conn = await get_db_connection()
try:
yield conn
finally:
conn.close()
# class GenerationRequest(BaseModel):
# user_id: int
# query: str
class Message(BaseModel):
role: Literal["user", "assistant"]
content: str
class GenerationRequest(BaseModel):
user_id: int
query: Union[str, List[Message]]
class MetadataRequest(BaseModel):
query: str
# Global metadata variables
native_language: Optional[str] = None
target_language: Optional[str] = None
proficiency: Optional[str] = None
@app.get("/")
async def root():
return {"message": "Welcome to the AI Learning Assistant API!"}
@app.post("/extract/metadata")
async def extract_metadata(data: MetadataRequest):
try:
response_str = await generate_completions.get_completions(
data.query,
config.language_metadata_extraction_prompt
)
metadata_dict = json.loads(response_str)
# Update globals for other endpoints
globals()['native_language'] = metadata_dict.get('native_language', 'unknown')
globals()['target_language'] = metadata_dict.get('target_language', 'unknown')
globals()['proficiency'] = metadata_dict.get('proficiency_level', 'unknown')
return JSONResponse(
content={
"data": metadata_dict,
"type": "language_metadata",
"status": "success"
},
status_code=200
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/generate/flashcards")
async def generate_flashcards(data: GenerationRequest):
try:
# Use previously extracted metadata
instructions = (
config.flashcard_mode_instructions
.replace("{native_language}", native_language or "unknown")
.replace("{target_language}", target_language or "unknown")
.replace("{proficiency}", proficiency or "unknown")
)
response = await generate_completions.get_completions(
data.query,
instructions
)
return JSONResponse(
content={
"data": response,
"type": "flashcards",
"status": "success"
},
status_code=200
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/generate/exercises")
async def generate_exercises(data: GenerationRequest):
try:
# Use previously extracted metadata
instructions = (
config.exercise_mode_instructions
.replace("{native_language}", native_language or "unknown")
.replace("{target_language}", target_language or "unknown")
.replace("{proficiency}", proficiency or "unknown")
)
response = await generate_completions.get_completions(
data.query,
instructions
)
return JSONResponse(
content={
"data": response,
"type": "exercises",
"status": "success"
},
status_code=200
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/generate/simulation")
async def generate_simulation(data: GenerationRequest):
try:
# Use previously extracted metadata
instructions = (
config.simulation_mode_instructions
.replace("{native_language}", native_language or "unknown")
.replace("{target_language}", target_language or "unknown")
.replace("{proficiency}", proficiency or "unknown")
)
response = await generate_completions.get_completions(
data.query,
instructions
)
return JSONResponse(
content={
"data": response,
"type": "simulation",
"status": "success"
},
status_code=200
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e)) |