"""
AI Model API Routes for FastAPI
"""
from fastapi import APIRouter, HTTPException, Query
from typing import List, Optional
from datetime import datetime
from app.models.data_model import DataModel
from app.ai.local_model import ai_model
from app.schemas.item import ItemResponse
router = APIRouter()
data_model = DataModel()
@router.get("/ai/model-info")
async def get_model_info():
"""Get information about the loaded AI model"""
return ai_model.get_model_info()
@router.post("/ai/change-model")
async def change_model(model_type: str, model_name: str):
"""Change the AI model"""
try:
result = await ai_model.change_model(model_type, model_name)
return {
"success": True,
"data": result,
"message": f"Model changed to {model_type}: {model_name}"
}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.post("/ai/analyze-text")
async def analyze_text(text: str):
"""Analyze text using the AI model"""
try:
result = await ai_model.analyze_text(text)
if "error" in result:
raise HTTPException(status_code=400, detail=result["error"])
return {
"success": True,
"data": result,
"text": text
}
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.get("/ai/analyze-items")
async def analyze_all_items():
"""Analyze all items using the AI model"""
try:
items = await data_model.get_all_items()
if not items:
return {
"success": True,
"data": {"message": "No items to analyze"},
"item_count": 0
}
result = await ai_model.analyze_items(items)
if "error" in result:
raise HTTPException(status_code=400, detail=result["error"])
return {
"success": True,
"data": result,
"item_count": len(items)
}
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.get("/ai/similar-items")
async def find_similar_items(
query: str = Query(..., description="Search query"),
top_k: int = Query(5, description="Number of similar items to return")
):
"""Find items similar to a query using AI embeddings"""
try:
items = await data_model.get_all_items()
if not items:
return {
"success": True,
"data": {"message": "No items to search"},
"item_count": 0
}
result = await ai_model.find_similar_items(query, items, top_k)
if "error" in result:
raise HTTPException(status_code=400, detail=result["error"])
return {
"success": True,
"data": result,
"query": query,
"top_k": top_k
}
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.get("/ai/analyze-item/{item_id}")
async def analyze_single_item(item_id: str):
"""Analyze a specific item using the AI model"""
try:
item = await data_model.getItemById(item_id)
if not item:
raise HTTPException(status_code=404, detail="Item not found")
# Extract text from item
text_parts = []
if item.get('name'):
text_parts.append(item['name'])
if item.get('description'):
text_parts.append(item['description'])
if item.get('category'):
text_parts.append(item['category'])
text = ' '.join(text_parts)
if not text.strip():
return {
"success": True,
"data": {"message": "Item has no text content to analyze"},
"item": item
}
result = await ai_model.analyze_text(text)
if "error" in result:
raise HTTPException(status_code=400, detail=result["error"])
return {
"success": True,
"data": result,
"item": item,
"analyzed_text": text
}
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.get("/ai/smart-search")
async def smart_search(
query: str = Query(..., description="Search query"),
top_k: int = Query(10, description="Number of results to return")
):
"""Smart search combining traditional search with AI similarity"""
try:
# Get traditional search results
traditional_results = await data_model.search_items(query)
# Get AI similarity results
all_items = await data_model.get_all_items()
ai_results = await ai_model.find_similar_items(query, all_items, top_k)
# Combine and rank results
combined_results = []
# Add traditional search results with high weight
for item in traditional_results:
combined_results.append({
"item": item,
"score": 1.0,
"source": "traditional_search"
})
# Add AI similarity results
if "similar_items" in ai_results.get("data", {}):
for ai_item in ai_results["data"]["similar_items"]:
# Check if item already exists
exists = any(r["item"]["id"] == ai_item["item"]["id"] for r in combined_results)
if not exists:
combined_results.append({
"item": ai_item["item"],
"score": ai_item["similarity_score"],
"source": "ai_similarity"
})
# Sort by score (descending)
combined_results.sort(key=lambda x: x["score"], reverse=True)
return {
"success": True,
"data": {
"results": combined_results[:top_k],
"traditional_count": len(traditional_results),
"ai_model": ai_model.get_model_info(),
"query": query
},
"total_results": len(combined_results)
}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
