Spaces:
Sleeping
Sleeping
File size: 5,724 Bytes
61d9463 |
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 156 157 158 159 160 161 162 163 164 |
from fastapi import APIRouter, HTTPException
from typing import List
from backend.app.services.user_items_recommendations import ItemsRecommendationService
from backend.app.schemas.users import (
UserBase,
RecommendationResponse,
QueryRecommendationResponse,
QueryRecommendationRequest
)
from backend.app.helper.dependencies import events_vector_db, travels_vector_db , user_db, event_db, travel_db
from backend.app.helper.dependencies import embedding_model
router = APIRouter()
@router.post("/get/")
async def get_users(user_ids: List[str]):
try:
users = user_db.get_users(user_ids)
return {
"message": f"Found {len(users)} users",
"status_code": 200,
"data": users,
}
except Exception as e:
raise HTTPException(status_code=400, detail=str(e))
@router.post("/add/")
async def add_users(users: List[UserBase]):
try:
created_users = user_db.create_users(users)
input_ids = [user.id for user in users]
created_ids = [user['id'] for user in created_users]
existing_users = list(set(input_ids) - set(created_ids))
if not created_users:
raise HTTPException(
status_code=400,
detail=f"All users already exist. Existing IDs: {existing_users}"
)
return {
"message": f"Users created successfully",
"status_code": 201,
"created_users": created_ids,
}
except Exception as e:
raise HTTPException(status_code=400, detail=str(e))
@router.put("/update/")
async def update_users(users: List[UserBase]):
try:
updated_users = user_db.update_users(users)
if not updated_users:
raise HTTPException(status_code=404, detail="No users found to update")
return {
"message": f"Users updated successfully {len(updated_users)}",
"status_code": 200,
"updated_users": [updated_user["id"] for updated_user in updated_users],
}
except Exception as e:
raise HTTPException(status_code=400, detail=str(e))
@router.delete("/delete/")
async def delete_users(user_ids: List[str]):
try:
deleted_ids = user_db.delete_users(user_ids)
if not deleted_ids:
existing_users = user_db.get_users(user_ids)
non_existent = list(set(user_ids) - {user['id'] for user in existing_users})
raise HTTPException(
status_code=404,
detail=f"No existing users found to delete. Non-existent IDs: {non_existent}"
)
return {
"message": f"Successfully deleted users",
"status_code": 200,
"deleted_ids": deleted_ids,
}
except Exception as e:
raise HTTPException(status_code=400, detail=str(e))
@router.get("/recommend_events/{user_id}", response_model=RecommendationResponse)
async def get_event_recommendations(user_id: str, Num_recommendations: int = 7, Num_highest_intercations: int = 4):
try:
return ItemsRecommendationService.get_recommendations(
user_id=user_id,
interaction_type="event",
user_db=user_db,
vector_db=events_vector_db,
content_db=event_db,
top_k=Num_recommendations,
based_on_n=Num_highest_intercations
)
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=400, detail=str(e))
@router.get("/recommend_travels/{user_id}", response_model=RecommendationResponse)
async def get_travel_recommendations(user_id: str, Num_recommendations: int = 7, Num_highest_intercations: int = 4):
try:
return ItemsRecommendationService.get_recommendations(
user_id=user_id,
interaction_type="travel",
user_db=user_db,
vector_db=travels_vector_db,
content_db=travel_db,
top_k=Num_recommendations,
based_on_n=Num_highest_intercations
)
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=400, detail=str(e))
@router.post("/recommend_by_query/", response_model=QueryRecommendationResponse)
async def get_recommendations_by_query(
request : QueryRecommendationRequest
):
try:
response = {
"message": "Recommendations generated successfully",
"status_code": 200,
"query": request.query,
"event_recommendations": [],
"travel_recommendations": []
}
if request.type == "event" or request.type == None:
event_recommendations = events_vector_db.search_by_query(
query=request.query,
model = embedding_model,
top_k= request.top_k // 2 if (request.type == "event" or request.type is None) else request.top_k,
)
response["event_recommendations"] = event_recommendations
if request.type == "travel" or request.type == None:
travel_recommendations = travels_vector_db.search_by_query(
query=request.query,
model = embedding_model,
top_k= request.top_k - request.top_k // 2 if (request.type == "travel" or request.type is None) else request.top_k,
)
response["travel_recommendations"] = travel_recommendations
return response
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=400, detail=str(e))
|