231 lines
9.3 KiB
Python
231 lines
9.3 KiB
Python
# 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)}") |