Spaces:
Sleeping
Sleeping
File size: 5,124 Bytes
b2be963 | 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 | from fastapi import APIRouter, Depends, Query, HTTPException, status
from sqlalchemy.orm import Session
from typing import List, Optional
from app.db.base import get_db
from app.models.product import Product
from app.schemas.product import ProductListItem, PaginatedResponse
from app.api.auth import get_current_user
from app.services.ai_service import ai_service
from app.schemas.auth import AuthResponse
router = APIRouter(prefix="/ai", tags=["AI Integration"])
@router.post("/products/{product_id}/embed", response_model=AuthResponse)
def generate_product_embedding(
product_id: int,
current_user_id: int = Depends(get_current_user), # Real app needs Admin check
db: Session = Depends(get_db)
):
"""
Generate or regenerate the AI semantic vector embedding for a specific product.
Typically called after a product is created or updated.
"""
product = db.query(Product).filter(Product.id == product_id).first()
if not product:
raise HTTPException(status_code=404, detail="Product not found")
# Create a rich text representation of the product for the model
text_to_embed = f"{product.name_en} {product.name_ar} {product.description_en} {product.description_ar}"
embedding = ai_service.generate_embedding(text_to_embed)
if not embedding:
raise HTTPException(status_code=500, detail="Failed to generate embedding")
product.embedding = embedding
db.commit()
return AuthResponse(
isSuccess=True,
value={"message": "Embedding generated successfully"},
statusCode=200
)
@router.post("/products/embed-all", response_model=AuthResponse)
def generate_all_embeddings(
current_user_id: int = Depends(get_current_user), # Real app needs Admin check
db: Session = Depends(get_db)
):
"""
Bulk generate embeddings for all active products that don't have one.
"""
products = db.query(Product).filter(Product.is_active == True, Product.embedding == None).all()
count = 0
for product in products:
text_to_embed = f"{product.name_en} {product.name_ar} {product.description_en} {product.description_ar}"
embedding = ai_service.generate_embedding(text_to_embed)
if embedding:
product.embedding = embedding
count += 1
db.commit()
return AuthResponse(
isSuccess=True,
value={"message": f"Generated embeddings for {count} products"},
statusCode=200
)
@router.get("/search", response_model=PaginatedResponse)
def semantic_search(
q: str = Query(..., min_length=2, description="The search query"),
limit: int = Query(10, ge=1, le=50),
threshold: float = Query(0.3, ge=0.0, le=1.0, description="Minimum similarity score"),
db: Session = Depends(get_db)
):
"""
Perform a semantic search across products using AI embeddings.
"""
query_embedding = ai_service.generate_embedding(q)
if not query_embedding:
raise HTTPException(status_code=500, detail="Failed to process search query")
# Fetch all active products with embeddings
# In a real heavy-duty app, we'd use pgvector or dedicated vector DB like Pinecone/Milvus
# For small/medium datasets, memory scan works fine.
products = db.query(Product).filter(Product.is_active == True, Product.embedding != None).all()
results = []
for p in products:
similarity = ai_service.calculate_similarity(query_embedding, p.embedding)
if similarity >= threshold:
results.append((similarity, p))
# Sort by similarity descending
results.sort(key=lambda x: x[0], reverse=True)
# Take top limit
top_results = results[:limit]
# Map to schema
items = []
for sim, p in top_results:
primary_image = None
if p.images:
sorted_imgs = sorted(p.images, key=lambda x: x.sort_order)
primary_image = sorted_imgs[0].image_url if sorted_imgs else None
item_dict = ProductListItem(
id=p.id,
slug=p.slug,
name_ar=p.name_ar,
name_en=p.name_en,
price=p.price,
compare_price=p.compare_price,
stock=p.stock,
category_id=p.category_id,
rating=p.rating,
rating_count=p.rating_count,
is_featured=p.is_featured,
image_url=primary_image,
created_at=p.created_at,
).model_dump()
item_dict["created_at"] = item_dict["created_at"].isoformat()
item_dict["similarity_score"] = round(sim, 4) # Add match score
items.append(item_dict)
return PaginatedResponse(
isSuccess=True,
value={
"items": items,
"total": len(items),
"page": 1,
"page_size": limit,
"total_pages": 1
},
statusCode=200
)
|