Upload app/api/ai_search.py

app/api/ai_search.py

@@ -0,0 +1,141 @@
+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,
+            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
+    )