# services/inventory/app/api/classification.py """ Product Classification API Endpoints AI-powered product classification for onboarding automation """ from fastapi import APIRouter, Depends, HTTPException, Path from typing import List, Dict, Any from uuid import UUID from pydantic import BaseModel, Field import structlog from app.services.product_classifier import ProductClassifierService, get_product_classifier from shared.auth.decorators import get_current_user_dep, get_current_tenant_id_dep router = APIRouter(tags=["classification"]) logger = structlog.get_logger() class ProductClassificationRequest(BaseModel): """Request for single product classification""" product_name: str = Field(..., description="Product name to classify") sales_volume: float = Field(None, description="Total sales volume for context") sales_data: Dict[str, Any] = Field(default_factory=dict, description="Additional sales context") class BatchClassificationRequest(BaseModel): """Request for batch product classification""" products: List[ProductClassificationRequest] = Field(..., description="Products to classify") class ProductSuggestionResponse(BaseModel): """Response with product classification suggestion""" suggestion_id: str original_name: str suggested_name: str product_type: str category: str unit_of_measure: str confidence_score: float estimated_shelf_life_days: int = None requires_refrigeration: bool = False requires_freezing: bool = False is_seasonal: bool = False suggested_supplier: str = None notes: str = None class BusinessModelAnalysisResponse(BaseModel): """Response with business model analysis""" model: str # production, retail, hybrid confidence: float ingredient_count: int finished_product_count: int ingredient_ratio: float recommendations: List[str] class BatchClassificationResponse(BaseModel): """Response for batch classification""" suggestions: List[ProductSuggestionResponse] business_model_analysis: BusinessModelAnalysisResponse total_products: int high_confidence_count: int low_confidence_count: int @router.post("/tenants/{tenant_id}/inventory/classify-product", response_model=ProductSuggestionResponse) async def classify_single_product( request: ProductClassificationRequest, tenant_id: UUID = Path(..., description="Tenant ID"), current_tenant: str = Depends(get_current_tenant_id_dep), current_user: Dict[str, Any] = Depends(get_current_user_dep), classifier: ProductClassifierService = Depends(get_product_classifier) ): """Classify a single product for inventory creation""" try: # Verify tenant access if str(tenant_id) != current_tenant: raise HTTPException(status_code=403, detail="Access denied to this tenant") # Classify the product suggestion = classifier.classify_product( request.product_name, request.sales_volume ) # Convert to response format response = ProductSuggestionResponse( suggestion_id=str(UUID.uuid4()), # Generate unique ID for tracking original_name=suggestion.original_name, suggested_name=suggestion.suggested_name, product_type=suggestion.product_type.value, category=suggestion.category, unit_of_measure=suggestion.unit_of_measure.value, confidence_score=suggestion.confidence_score, estimated_shelf_life_days=suggestion.estimated_shelf_life_days, requires_refrigeration=suggestion.requires_refrigeration, requires_freezing=suggestion.requires_freezing, is_seasonal=suggestion.is_seasonal, suggested_supplier=suggestion.suggested_supplier, notes=suggestion.notes ) logger.info("Classified single product", product=request.product_name, classification=suggestion.product_type.value, confidence=suggestion.confidence_score, tenant_id=tenant_id) return response except Exception as e: logger.error("Failed to classify product", error=str(e), product=request.product_name, tenant_id=tenant_id) raise HTTPException(status_code=500, detail=f"Classification failed: {str(e)}") @router.post("/tenants/{tenant_id}/inventory/classify-products-batch", response_model=BatchClassificationResponse) async def classify_products_batch( request: BatchClassificationRequest, tenant_id: UUID = Path(..., description="Tenant ID"), current_tenant: str = Depends(get_current_tenant_id_dep), current_user: Dict[str, Any] = Depends(get_current_user_dep), classifier: ProductClassifierService = Depends(get_product_classifier) ): """Classify multiple products for onboarding automation""" try: # Verify tenant access if str(tenant_id) != current_tenant: raise HTTPException(status_code=403, detail="Access denied to this tenant") if not request.products: raise HTTPException(status_code=400, detail="No products provided for classification") # Extract product names and volumes product_names = [p.product_name for p in request.products] sales_volumes = {p.product_name: p.sales_volume for p in request.products if p.sales_volume} # Classify products in batch suggestions = classifier.classify_products_batch(product_names, sales_volumes) # Convert suggestions to response format suggestion_responses = [] for suggestion in suggestions: suggestion_responses.append(ProductSuggestionResponse( suggestion_id=str(UUID.uuid4()), original_name=suggestion.original_name, suggested_name=suggestion.suggested_name, product_type=suggestion.product_type.value, category=suggestion.category, unit_of_measure=suggestion.unit_of_measure.value, confidence_score=suggestion.confidence_score, estimated_shelf_life_days=suggestion.estimated_shelf_life_days, requires_refrigeration=suggestion.requires_refrigeration, requires_freezing=suggestion.requires_freezing, is_seasonal=suggestion.is_seasonal, suggested_supplier=suggestion.suggested_supplier, notes=suggestion.notes )) # Analyze business model ingredient_count = sum(1 for s in suggestions if s.product_type.value == 'ingredient') finished_count = sum(1 for s in suggestions if s.product_type.value == 'finished_product') total = len(suggestions) ingredient_ratio = ingredient_count / total if total > 0 else 0 # Determine business model if ingredient_ratio >= 0.7: model = 'production' elif ingredient_ratio <= 0.3: model = 'retail' else: model = 'hybrid' confidence = max(abs(ingredient_ratio - 0.5) * 2, 0.1) recommendations = { 'production': [ 'Focus on ingredient inventory management', 'Set up recipe cost calculation', 'Configure supplier relationships', 'Enable production planning features' ], 'retail': [ 'Configure central baker relationships', 'Set up delivery schedule tracking', 'Enable finished product freshness monitoring', 'Focus on sales forecasting' ], 'hybrid': [ 'Configure both ingredient and finished product management', 'Set up flexible inventory categories', 'Enable both production and retail features' ] } business_model_analysis = BusinessModelAnalysisResponse( model=model, confidence=confidence, ingredient_count=ingredient_count, finished_product_count=finished_count, ingredient_ratio=ingredient_ratio, recommendations=recommendations.get(model, []) ) # Count confidence levels high_confidence_count = sum(1 for s in suggestions if s.confidence_score >= 0.7) low_confidence_count = sum(1 for s in suggestions if s.confidence_score < 0.6) response = BatchClassificationResponse( suggestions=suggestion_responses, business_model_analysis=business_model_analysis, total_products=len(suggestions), high_confidence_count=high_confidence_count, low_confidence_count=low_confidence_count ) logger.info("Batch classification complete", total_products=len(suggestions), business_model=model, high_confidence=high_confidence_count, low_confidence=low_confidence_count, tenant_id=tenant_id) return response except Exception as e: logger.error("Failed batch classification", error=str(e), products_count=len(request.products), tenant_id=tenant_id) raise HTTPException(status_code=500, detail=f"Batch classification failed: {str(e)}")