bakery-admin/bakery-ia
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Start integrating the onboarding flow with backend 2

Browse Source
This commit is contained in:
Urtzi Alfaro
2025-09-04 18:59:56 +02:00
parent a11fdfba24
commit 9eedc2e5f2
30 changed files with 3432 additions and 4735 deletions
Download Patch File Download Diff File Expand all files Collapse all files

865
services/sales/app/services/ai_onboarding_service.py
View File

@@ -1,865 +0,0 @@
# services/sales/app/services/ai_onboarding_service.py
"""
AI-Powered Onboarding Service
Handles the complete onboarding flow: File validation -> Product extraction -> Inventory suggestions -> Data processing
"""
import pandas as pd
import structlog
from typing import List, Dict, Any, Optional
from uuid import UUID, uuid4
from dataclasses import dataclass
import asyncio
from app.services.data_import_service import DataImportService, SalesValidationResult, SalesImportResult
from app.services.inventory_client import InventoryServiceClient
from app.core.database import get_db_transaction
logger = structlog.get_logger()
@dataclass
class ProductSuggestion:
"""Single product suggestion from AI classification"""
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: Optional[int] = None
requires_refrigeration: bool = False
requires_freezing: bool = False
is_seasonal: bool = False
suggested_supplier: Optional[str] = None
notes: Optional[str] = None
sales_data: Optional[Dict[str, Any]] = None
@dataclass
class BusinessModelAnalysis:
"""Business model analysis results"""
model: str # production, retail, hybrid
confidence: float
ingredient_count: int
finished_product_count: int
ingredient_ratio: float
recommendations: List[str]
@dataclass
class OnboardingValidationResult:
"""Result of onboarding file validation step"""
is_valid: bool
total_records: int
unique_products: int
validation_details: SalesValidationResult
product_list: List[str]
summary: Dict[str, Any]
@dataclass
class ProductSuggestionsResult:
"""Result of AI product classification step"""
suggestions: List[ProductSuggestion]
business_model_analysis: BusinessModelAnalysis
total_products: int
high_confidence_count: int
low_confidence_count: int
processing_time_seconds: float
@dataclass
class OnboardingImportResult:
"""Result of final data import step"""
success: bool
import_details: SalesImportResult
inventory_items_created: int
inventory_creation_errors: List[str]
final_summary: Dict[str, Any]
class AIOnboardingService:
"""
Unified AI-powered onboarding service that orchestrates the complete flow:
1. File validation and product extraction
2. AI-powered inventory suggestions
3. User confirmation and inventory creation
4. Final sales data import
"""
def __init__(self):
self.data_import_service = DataImportService()
self.inventory_client = InventoryServiceClient()
# ================================================================
# STEP 1: FILE VALIDATION AND PRODUCT EXTRACTION
# ================================================================
async def validate_and_extract_products(
self,
file_data: str,
file_format: str,
tenant_id: UUID
) -> OnboardingValidationResult:
"""
Step 1: Validate uploaded file and extract unique products
This uses the detailed validation from data_import_service
"""
try:
logger.info("Starting onboarding validation and product extraction",
file_format=file_format, tenant_id=tenant_id)
# Use data_import_service for detailed validation
validation_data = {
"tenant_id": str(tenant_id),
"data": file_data,
"data_format": file_format,
"validate_only": True,
"source": "ai_onboarding"
}
validation_result = await self.data_import_service.validate_import_data(validation_data)
# Extract unique products if validation passes
product_list = []
unique_products = 0
if validation_result.is_valid and file_format.lower() == "csv":
try:
# Parse CSV to extract unique products
import csv
import io
reader = csv.DictReader(io.StringIO(file_data))
rows = list(reader)
# Use data_import_service column detection
column_mapping = self.data_import_service._detect_columns(list(rows[0].keys()) if rows else [])
if column_mapping.get('product'):
product_column = column_mapping['product']
# Extract and clean unique products
products_raw = [row.get(product_column, '').strip() for row in rows if row.get(product_column, '').strip()]
# Clean product names using data_import_service method
products_cleaned = [
self.data_import_service._clean_product_name(product)
for product in products_raw
]
# Get unique products
product_list = list(set([p for p in products_cleaned if p and p != "Producto sin nombre"]))
unique_products = len(product_list)
logger.info("Extracted unique products",
total_rows=len(rows), unique_products=unique_products)
except Exception as e:
logger.error("Failed to extract products", error=str(e))
# Don't fail validation just because product extraction failed
pass
result = OnboardingValidationResult(
is_valid=validation_result.is_valid,
total_records=validation_result.total_records,
unique_products=unique_products,
validation_details=validation_result,
product_list=product_list,
summary={
"status": "valid" if validation_result.is_valid else "invalid",
"file_format": file_format,
"total_records": validation_result.total_records,
"unique_products": unique_products,
"ready_for_ai_classification": validation_result.is_valid and unique_products > 0,
"next_step": "ai_classification" if validation_result.is_valid and unique_products > 0 else "fix_validation_errors"
}
)
logger.info("Onboarding validation completed",
is_valid=result.is_valid,
unique_products=unique_products,
tenant_id=tenant_id)
return result
except Exception as e:
logger.error("Onboarding validation failed", error=str(e), tenant_id=tenant_id)
return OnboardingValidationResult(
is_valid=False,
total_records=0,
unique_products=0,
validation_details=SalesValidationResult(
is_valid=False,
total_records=0,
valid_records=0,
invalid_records=0,
errors=[{
"type": "system_error",
"message": f"Onboarding validation error: {str(e)}",
"field": None,
"row": None,
"code": "ONBOARDING_VALIDATION_ERROR"
}],
warnings=[],
summary={}
),
product_list=[],
summary={
"status": "error",
"error_message": str(e),
"next_step": "retry_upload"
}
)
# ================================================================
# STEP 2: AI PRODUCT CLASSIFICATION
# ================================================================
async def generate_inventory_suggestions(
self,
product_list: List[str],
file_data: str,
file_format: str,
tenant_id: UUID
) -> ProductSuggestionsResult:
"""
Step 2: Generate AI-powered inventory suggestions for products
"""
import time
start_time = time.time()
try:
logger.info("Starting AI inventory suggestions",
product_count=len(product_list), tenant_id=tenant_id)
if not product_list:
raise ValueError("No products provided for classification")
# Analyze sales data for each product to provide context
product_analysis = await self._analyze_product_sales_data(
product_list, file_data, file_format
)
# Prepare products for classification
products_for_classification = []
for product_name in product_list:
sales_data = product_analysis.get(product_name, {})
products_for_classification.append({
"product_name": product_name,
"sales_volume": sales_data.get("total_quantity"),
"sales_data": sales_data
})
# Call inventory service for AI classification
classification_result = await self.inventory_client.classify_products_batch(
products_for_classification, tenant_id
)
if not classification_result or "suggestions" not in classification_result:
raise ValueError("Invalid classification response from inventory service")
suggestions_raw = classification_result["suggestions"]
business_model_raw = classification_result.get("business_model_analysis", {})
# Convert to dataclass objects
suggestions = []
for suggestion_data in suggestions_raw:
suggestion = ProductSuggestion(
suggestion_id=suggestion_data.get("suggestion_id", str(uuid4())),
original_name=suggestion_data["original_name"],
suggested_name=suggestion_data["suggested_name"],
product_type=suggestion_data["product_type"],
category=suggestion_data["category"],
unit_of_measure=suggestion_data["unit_of_measure"],
confidence_score=suggestion_data["confidence_score"],
estimated_shelf_life_days=suggestion_data.get("estimated_shelf_life_days"),
requires_refrigeration=suggestion_data.get("requires_refrigeration", False),
requires_freezing=suggestion_data.get("requires_freezing", False),
is_seasonal=suggestion_data.get("is_seasonal", False),
suggested_supplier=suggestion_data.get("suggested_supplier"),
notes=suggestion_data.get("notes"),
sales_data=product_analysis.get(suggestion_data["original_name"])
)
suggestions.append(suggestion)
# Check if enhanced business intelligence data is available
bi_data = product_analysis.get('__business_intelligence__')
if bi_data and bi_data.get('confidence_score', 0) > 0.6:
# Use enhanced business intelligence analysis
business_type = bi_data.get('business_type', 'bakery')
business_model_detected = bi_data.get('business_model', 'individual')
# Map business intelligence results to existing model format
model_mapping = {
'individual': 'individual_bakery',
'central_distribution': 'central_baker_satellite',
'central_bakery': 'central_baker_satellite',
'hybrid': 'hybrid_bakery'
}
mapped_model = model_mapping.get(business_model_detected, 'individual_bakery')
# Count ingredients vs finished products from suggestions
ingredient_count = sum(1 for s in suggestions if s.product_type == 'ingredient')
finished_product_count = sum(1 for s in suggestions if s.product_type == 'finished_product')
total_products = len(suggestions)
ingredient_ratio = ingredient_count / total_products if total_products > 0 else 0.0
# Enhanced recommendations based on BI analysis
enhanced_recommendations = bi_data.get('recommendations', [])
# Add business type specific recommendations
if business_type == 'coffee_shop':
enhanced_recommendations.extend([
"Configure beverage inventory management",
"Set up quick-service item tracking",
"Enable all-day service optimization"
])
business_model = BusinessModelAnalysis(
model=mapped_model,
confidence=bi_data.get('confidence_score', 0.0),
ingredient_count=ingredient_count,
finished_product_count=finished_product_count,
ingredient_ratio=ingredient_ratio,
recommendations=enhanced_recommendations[:6] # Limit to top 6 recommendations
)
logger.info("Using enhanced business intelligence for model analysis",
detected_type=business_type,
detected_model=business_model_detected,
mapped_model=mapped_model,
confidence=bi_data.get('confidence_score'))
else:
# Fallback to basic inventory service analysis
business_model = BusinessModelAnalysis(
model=business_model_raw.get("model", "unknown"),
confidence=business_model_raw.get("confidence", 0.0),
ingredient_count=business_model_raw.get("ingredient_count", 0),
finished_product_count=business_model_raw.get("finished_product_count", 0),
ingredient_ratio=business_model_raw.get("ingredient_ratio", 0.0),
recommendations=business_model_raw.get("recommendations", [])
)
logger.info("Using basic inventory service business model analysis")
# Calculate confidence metrics
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)
processing_time = time.time() - start_time
result = ProductSuggestionsResult(
suggestions=suggestions,
business_model_analysis=business_model,
total_products=len(suggestions),
high_confidence_count=high_confidence_count,
low_confidence_count=low_confidence_count,
processing_time_seconds=processing_time
)
# Update tenant's business model based on AI analysis
if business_model.model != "unknown" and business_model.confidence >= 0.6:
try:
await self._update_tenant_business_model(tenant_id, business_model.model)
logger.info("Updated tenant business model",
tenant_id=tenant_id,
business_model=business_model.model,
confidence=business_model.confidence)
except Exception as e:
logger.warning("Failed to update tenant business model",
error=str(e), tenant_id=tenant_id)
# Don't fail the entire process if tenant update fails
logger.info("AI inventory suggestions completed",
total_suggestions=len(suggestions),
business_model=business_model.model,
high_confidence=high_confidence_count,
processing_time=processing_time,
tenant_id=tenant_id)
return result
except Exception as e:
processing_time = time.time() - start_time
logger.error("AI inventory suggestions failed",
error=str(e), tenant_id=tenant_id)
# Return fallback suggestions
fallback_suggestions = [
ProductSuggestion(
suggestion_id=str(uuid4()),
original_name=product_name,
suggested_name=product_name.title(),
product_type="finished_product",
category="other_products",
unit_of_measure="units",
confidence_score=0.3,
notes="Fallback suggestion - requires manual review"
)
for product_name in product_list
]
return ProductSuggestionsResult(
suggestions=fallback_suggestions,
business_model_analysis=BusinessModelAnalysis(
model="unknown",
confidence=0.0,
ingredient_count=0,
finished_product_count=len(fallback_suggestions),
ingredient_ratio=0.0,
recommendations=["Manual review required for all products"]
),
total_products=len(fallback_suggestions),
high_confidence_count=0,
low_confidence_count=len(fallback_suggestions),
processing_time_seconds=processing_time
)
# ================================================================
# STEP 3: INVENTORY CREATION (after user confirmation)
# ================================================================
async def create_inventory_from_suggestions(
self,
approved_suggestions: List[Dict[str, Any]],
tenant_id: UUID,
user_id: UUID
) -> Dict[str, Any]:
"""
Step 3: Create inventory items from user-approved suggestions
"""
try:
logger.info("Creating inventory from approved suggestions",
approved_count=len(approved_suggestions), tenant_id=tenant_id)
created_items = []
failed_items = []
for approval in approved_suggestions:
suggestion_id = approval.get("suggestion_id")
is_approved = approval.get("approved", False)
modifications = approval.get("modifications", {})
if not is_approved:
continue
try:
# Build inventory item data from suggestion and modifications
# Map to inventory service expected format
raw_category = modifications.get("category") or approval.get("category", "other")
raw_unit = modifications.get("unit_of_measure") or approval.get("unit_of_measure", "units")
# Map categories to inventory service enum values
category_mapping = {
"flour": "flour",
"yeast": "yeast",
"dairy": "dairy",
"eggs": "eggs",
"sugar": "sugar",
"fats": "fats",
"salt": "salt",
"spices": "spices",
"additives": "additives",
"packaging": "packaging",
"cleaning": "cleaning",
"grains": "flour", # Map common variations
"bread": "other",
"pastries": "other",
"croissants": "other",
"cakes": "other",
"other_products": "other"
}
# Map units to inventory service enum values
unit_mapping = {
"kg": "kg",
"kilograms": "kg",
"g": "g",
"grams": "g",
"l": "l",
"liters": "l",
"ml": "ml",
"milliliters": "ml",
"units": "units",
"pieces": "pcs",
"pcs": "pcs",
"packages": "pkg",
"pkg": "pkg",
"bags": "bags",
"boxes": "boxes"
}
mapped_category = category_mapping.get(raw_category.lower(), "other")
mapped_unit = unit_mapping.get(raw_unit.lower(), "units")
inventory_data = {
"name": modifications.get("name") or approval.get("suggested_name"),
"category": mapped_category,
"unit_of_measure": mapped_unit,
"product_type": approval.get("product_type"),
"description": modifications.get("description") or approval.get("notes", ""),
# Optional fields
"brand": modifications.get("brand") or approval.get("suggested_supplier"),
"is_active": True,
# Explicitly set boolean fields to ensure they're not NULL
"requires_refrigeration": modifications.get("requires_refrigeration", approval.get("requires_refrigeration", False)),
"requires_freezing": modifications.get("requires_freezing", approval.get("requires_freezing", False)),
"is_perishable": modifications.get("is_perishable", approval.get("is_perishable", False))
}
# Add optional numeric fields only if they exist
shelf_life = modifications.get("estimated_shelf_life_days") or approval.get("estimated_shelf_life_days")
if shelf_life:
inventory_data["shelf_life_days"] = shelf_life
# Create inventory item via inventory service
created_item = await self.inventory_client.create_ingredient(
inventory_data, str(tenant_id)
)
if created_item:
created_items.append({
"suggestion_id": suggestion_id,
"inventory_item": created_item,
"original_name": approval.get("original_name")
})
logger.info("Created inventory item",
item_name=inventory_data["name"],
suggestion_id=suggestion_id)
else:
failed_items.append({
"suggestion_id": suggestion_id,
"error": "Failed to create inventory item - no response"
})
except Exception as e:
logger.error("Failed to create inventory item",
error=str(e), suggestion_id=suggestion_id)
failed_items.append({
"suggestion_id": suggestion_id,
"error": str(e)
})
success_rate = len(created_items) / max(1, len(approved_suggestions)) * 100
result = {
"created_items": created_items,
"failed_items": failed_items,
"total_approved": len(approved_suggestions),
"successful_creations": len(created_items),
"failed_creations": len(failed_items),
"success_rate": success_rate,
"ready_for_import": len(created_items) > 0
}
logger.info("Inventory creation completed",
created=len(created_items),
failed=len(failed_items),
success_rate=success_rate,
tenant_id=tenant_id)
return result
except Exception as e:
logger.error("Inventory creation failed", error=str(e), tenant_id=tenant_id)
raise
# ================================================================
# STEP 4: FINAL DATA IMPORT
# ================================================================
async def import_sales_data_with_inventory(
self,
file_data: str,
file_format: str,
inventory_mapping: Dict[str, str], # original_product_name -> inventory_item_id
tenant_id: UUID,
filename: Optional[str] = None
) -> OnboardingImportResult:
"""
Step 4: Import sales data using the detailed processing from data_import_service
"""
try:
logger.info("Starting final sales data import with inventory mapping",
mappings_count=len(inventory_mapping), tenant_id=tenant_id)
# Use data_import_service for the actual import processing
import_result = await self.data_import_service.process_import(
str(tenant_id), file_data, file_format, filename
)
result = OnboardingImportResult(
success=import_result.success,
import_details=import_result,
inventory_items_created=len(inventory_mapping),
inventory_creation_errors=[],
final_summary={
"status": "completed" if import_result.success else "failed",
"total_records": import_result.records_processed,
"successful_imports": import_result.records_created,
"failed_imports": import_result.records_failed,
"inventory_items": len(inventory_mapping),
"processing_time": import_result.processing_time_seconds,
"onboarding_complete": import_result.success
}
)
logger.info("Final sales data import completed",
success=import_result.success,
records_created=import_result.records_created,
tenant_id=tenant_id)
return result
except Exception as e:
logger.error("Final sales data import failed", error=str(e), tenant_id=tenant_id)
return OnboardingImportResult(
success=False,
import_details=SalesImportResult(
success=False,
records_processed=0,
records_created=0,
records_updated=0,
records_failed=0,
errors=[{
"type": "import_error",
"message": f"Import failed: {str(e)}",
"field": None,
"row": None,
"code": "FINAL_IMPORT_ERROR"
}],
warnings=[],
processing_time_seconds=0.0
),
inventory_items_created=len(inventory_mapping),
inventory_creation_errors=[str(e)],
final_summary={
"status": "failed",
"error_message": str(e),
"onboarding_complete": False
}
)
# ================================================================
# HELPER METHODS
# ================================================================
async def _analyze_product_sales_data(
self,
product_list: List[str],
file_data: str,
file_format: str
) -> Dict[str, Dict[str, Any]]:
"""Analyze sales data for each product to provide context for AI classification"""
try:
if file_format.lower() != "csv":
return {}
import csv
import io
reader = csv.DictReader(io.StringIO(file_data))
rows = list(reader)
if not rows:
return {}
# Use data_import_service column detection
column_mapping = self.data_import_service._detect_columns(list(rows[0].keys()))
if not column_mapping.get('product'):
return {}
product_column = column_mapping['product']
quantity_column = column_mapping.get('quantity')
revenue_column = column_mapping.get('revenue')
date_column = column_mapping.get('date')
# Analyze each product
product_analysis = {}
for product_name in product_list:
# Find all rows for this product
product_rows = [
row for row in rows
if self.data_import_service._clean_product_name(row.get(product_column, '')) == product_name
]
if not product_rows:
continue
# Calculate metrics
total_quantity = 0
total_revenue = 0
sales_count = len(product_rows)
for row in product_rows:
try:
# Quantity
qty_raw = row.get(quantity_column, 1)
if qty_raw and str(qty_raw).strip():
qty = int(float(str(qty_raw).replace(',', '.')))
total_quantity += qty
else:
total_quantity += 1
# Revenue
if revenue_column:
rev_raw = row.get(revenue_column)
if rev_raw and str(rev_raw).strip():
rev = float(str(rev_raw).replace(',', '.').replace('€', '').replace('$', '').strip())
total_revenue += rev
except:
continue
avg_quantity = total_quantity / sales_count if sales_count > 0 else 0
avg_revenue = total_revenue / sales_count if sales_count > 0 else 0
avg_unit_price = total_revenue / total_quantity if total_quantity > 0 else 0
product_analysis[product_name] = {
"total_quantity": total_quantity,
"total_revenue": total_revenue,
"sales_count": sales_count,
"avg_quantity_per_sale": avg_quantity,
"avg_revenue_per_sale": avg_revenue,
"avg_unit_price": avg_unit_price
}
# Add enhanced business intelligence analysis
try:
from app.services.business_intelligence_service import BusinessIntelligenceService
bi_service = BusinessIntelligenceService()
# Convert parsed data to format expected by BI service
sales_data = []
product_data = []
for row in rows:
# Create sales record from CSV row
sales_record = {
'date': row.get(date_column, ''),
'product_name': row.get(product_column, ''),
'name': row.get(product_column, ''),
'quantity_sold': 0,
'revenue': 0,
'location_id': row.get('location', 'main'),
'sales_channel': row.get('channel', 'in_store'),
'supplier_name': row.get('supplier', ''),
'brand': row.get('brand', '')
}
# Parse quantity
if quantity_column:
try:
qty_raw = row.get(quantity_column, 1)
if qty_raw and str(qty_raw).strip():
sales_record['quantity_sold'] = int(float(str(qty_raw).replace(',', '.')))
except:
sales_record['quantity_sold'] = 1
# Parse revenue
if revenue_column:
try:
rev_raw = row.get(revenue_column)
if rev_raw and str(rev_raw).strip():
sales_record['revenue'] = float(str(rev_raw).replace(',', '.').replace('€', '').replace('$', '').strip())
except:
pass
sales_data.append(sales_record)
# Create product data entry
product_data.append({
'name': sales_record['product_name'],
'supplier_name': sales_record.get('supplier_name', ''),
'brand': sales_record.get('brand', '')
})
# Run business intelligence analysis
if sales_data:
detection_result = await bi_service.analyze_business_from_sales_data(
sales_data=sales_data,
product_data=product_data
)
# Store business intelligence results in product_analysis
product_analysis['__business_intelligence__'] = {
"business_type": detection_result.business_type,
"business_model": detection_result.business_model,
"confidence_score": detection_result.confidence_score,
"indicators": detection_result.indicators,
"recommendations": detection_result.recommendations,
"analysis_summary": f"{detection_result.business_type.title()} - {detection_result.business_model.replace('_', ' ').title()}"
}
logger.info("Enhanced business intelligence analysis completed",
business_type=detection_result.business_type,
business_model=detection_result.business_model,
confidence=detection_result.confidence_score)
else:
logger.warning("No sales data available for business intelligence analysis")
except Exception as bi_error:
logger.warning("Business intelligence analysis failed", error=str(bi_error))
# Continue with basic analysis even if BI fails
return product_analysis
except Exception as e:
logger.warning("Failed to analyze product sales data", error=str(e))
return {}
async def _update_tenant_business_model(self, tenant_id: UUID, business_model: str) -> None:
"""Update tenant's business model based on AI analysis"""
try:
# Use the gateway URL for all inter-service communication
from app.core.config import settings
import httpx
gateway_url = settings.GATEWAY_URL
url = f"{gateway_url}/api/v1/tenants/{tenant_id}"
# Prepare update data
update_data = {
"business_model": business_model
}
# Make request through gateway
timeout_config = httpx.Timeout(connect=10.0, read=30.0, write=10.0, pool=10.0)
async with httpx.AsyncClient(timeout=timeout_config) as client:
response = await client.put(
url,
json=update_data,
headers={"Content-Type": "application/json"}
)
if response.status_code == 200:
logger.info("Successfully updated tenant business model via gateway",
tenant_id=tenant_id, business_model=business_model)
else:
logger.warning("Failed to update tenant business model via gateway",
tenant_id=tenant_id,
status_code=response.status_code,
response=response.text)
except Exception as e:
logger.error("Error updating tenant business model via gateway",
tenant_id=tenant_id,
business_model=business_model,
error=str(e))
raise
# Factory function for dependency injection
def get_ai_onboarding_service() -> AIOnboardingService:
"""Get AI onboarding service instance"""
return AIOnboardingService()

24
services/sales/app/services/data_import_service.py
View File

@@ -26,7 +26,7 @@ logger = structlog.get_logger()
# Import result schemas (dataclass definitions)
from dataclasses import dataclass
from dataclasses import dataclass, field
from typing import List, Dict, Any
@dataclass
@@ -38,6 +38,8 @@ class SalesValidationResult:
errors: List[Dict[str, Any]]
warnings: List[Dict[str, Any]]
summary: Dict[str, Any]
unique_products: int = 0
product_list: List[str] = field(default_factory=list)
@dataclass
class SalesImportResult:
@@ -99,7 +101,9 @@ class DataImportService:
invalid_records=0,
errors=[],
warnings=[],
summary={}
summary={},
unique_products=0,
product_list=[]
)
errors = []
@@ -216,6 +220,22 @@ class DataImportService:
"code": "MISSING_PRODUCT_COLUMN"
})
# Extract unique products for AI suggestions
if column_mapping.get('product') and not errors:
product_column = column_mapping['product']
unique_products_set = set()
for row in rows:
product_name = row.get(product_column, '').strip()
if product_name and len(product_name) > 0:
unique_products_set.add(product_name)
validation_result.product_list = list(unique_products_set)
validation_result.unique_products = len(unique_products_set)
logger.info(f"Extracted {validation_result.unique_products} unique products from CSV",
tenant_id=data.get("tenant_id"))
if not column_mapping.get('quantity'):
warnings.append({
"type": "missing_column",