Fix new services implementation 5
This commit is contained in:
Urtzi Alfaro
@@ -428,7 +428,7 @@ class DataImportService:
|
||||
repository: SalesRepository,
|
||||
filename: Optional[str] = None
|
||||
) -> Dict[str, Any]:
|
||||
"""Enhanced CSV processing with better data handling"""
|
||||
"""Enhanced CSV processing with batch product resolution for better reliability"""
|
||||
try:
|
||||
reader = csv.DictReader(io.StringIO(csv_content))
|
||||
rows = list(reader)
|
||||
@@ -445,22 +445,41 @@ class DataImportService:
|
||||
# Enhanced column mapping
|
||||
column_mapping = self._detect_columns(list(rows[0].keys()))
|
||||
|
||||
records_created = 0
|
||||
errors = []
|
||||
warnings = []
|
||||
# Pre-process to extract unique products for batch creation
|
||||
unique_products = set()
|
||||
parsed_rows = []
|
||||
|
||||
logger.info(f"Processing {len(rows)} records from CSV with enhanced mapping")
|
||||
logger.info(f"Pre-processing {len(rows)} records to identify unique products")
|
||||
|
||||
for index, row in enumerate(rows):
|
||||
try:
|
||||
# Enhanced data parsing and validation
|
||||
parsed_data = await self._parse_row_data(row, column_mapping, index + 1)
|
||||
if parsed_data.get("skip"):
|
||||
errors.extend(parsed_data.get("errors", []))
|
||||
warnings.extend(parsed_data.get("warnings", []))
|
||||
continue
|
||||
|
||||
# Resolve product name to inventory_product_id
|
||||
if not parsed_data.get("skip"):
|
||||
unique_products.add((
|
||||
parsed_data["product_name"],
|
||||
parsed_data.get("product_category", "general")
|
||||
))
|
||||
parsed_rows.append((index, parsed_data))
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to parse row {index + 1}: {e}")
|
||||
continue
|
||||
|
||||
logger.info(f"Found {len(unique_products)} unique products, attempting batch resolution")
|
||||
|
||||
# Try to resolve/create all unique products in batch
|
||||
await self._batch_resolve_products(unique_products, tenant_id)
|
||||
|
||||
# Now process the actual sales records
|
||||
records_created = 0
|
||||
errors = []
|
||||
warnings = []
|
||||
|
||||
logger.info(f"Processing {len(parsed_rows)} validated records for sales creation")
|
||||
|
||||
for index, parsed_data in parsed_rows:
|
||||
try:
|
||||
# Resolve product name to inventory_product_id (should be cached now)
|
||||
inventory_product_id = await self._resolve_product_to_inventory_id(
|
||||
parsed_data["product_name"],
|
||||
parsed_data.get("product_category"),
|
||||
@@ -914,47 +933,57 @@ class DataImportService:
|
||||
logger.info("Import cache cleared for new session")
|
||||
|
||||
async def _resolve_product_to_inventory_id(self, product_name: str, product_category: Optional[str], tenant_id: UUID) -> Optional[UUID]:
|
||||
"""Resolve a product name to an inventory_product_id via the inventory service with caching and rate limiting"""
|
||||
"""Resolve a product name to an inventory_product_id via the inventory service with improved error handling and fallback"""
|
||||
|
||||
# Check cache first
|
||||
if product_name in self.product_cache:
|
||||
logger.debug("Product resolved from cache", product_name=product_name, tenant_id=tenant_id)
|
||||
return self.product_cache[product_name]
|
||||
|
||||
# Skip if this product already failed to resolve
|
||||
# Skip if this product already failed to resolve after all attempts
|
||||
if product_name in self.failed_products:
|
||||
logger.debug("Skipping previously failed product", product_name=product_name, tenant_id=tenant_id)
|
||||
return None
|
||||
|
||||
max_retries = 3
|
||||
base_delay = 1.0 # Start with 1 second delay
|
||||
max_retries = 5 # Increased retries
|
||||
base_delay = 2.0 # Increased base delay
|
||||
fallback_retry_delay = 10.0 # Longer delay for fallback attempts
|
||||
|
||||
for attempt in range(max_retries):
|
||||
try:
|
||||
# Add delay before API calls to avoid rate limiting
|
||||
# Add progressive delay to avoid rate limiting
|
||||
if attempt > 0:
|
||||
delay = base_delay * (2 ** (attempt - 1)) # Exponential backoff
|
||||
# Use longer delays for later attempts
|
||||
if attempt >= 3:
|
||||
delay = fallback_retry_delay # Use fallback delay for later attempts
|
||||
else:
|
||||
delay = base_delay * (2 ** (attempt - 1)) # Exponential backoff
|
||||
|
||||
logger.info(f"Retrying product resolution after {delay}s delay",
|
||||
product_name=product_name, attempt=attempt, tenant_id=tenant_id)
|
||||
await asyncio.sleep(delay)
|
||||
|
||||
# First try to search for existing product by name
|
||||
products = await self.inventory_client.search_products(product_name, tenant_id)
|
||||
try:
|
||||
products = await self.inventory_client.search_products(product_name, tenant_id)
|
||||
|
||||
if products:
|
||||
# Return the first matching product's ID
|
||||
product_id = products[0].get('id')
|
||||
if product_id:
|
||||
uuid_id = UUID(str(product_id))
|
||||
self.product_cache[product_name] = uuid_id # Cache for future use
|
||||
logger.info("Resolved product to existing inventory ID",
|
||||
product_name=product_name, product_id=product_id, tenant_id=tenant_id)
|
||||
return uuid_id
|
||||
except Exception as search_error:
|
||||
logger.warning("Product search failed, trying direct creation",
|
||||
product_name=product_name, error=str(search_error), tenant_id=tenant_id)
|
||||
|
||||
if products:
|
||||
# Return the first matching product's ID
|
||||
product_id = products[0].get('id')
|
||||
if product_id:
|
||||
uuid_id = UUID(str(product_id))
|
||||
self.product_cache[product_name] = uuid_id # Cache for future use
|
||||
logger.info("Resolved product to existing inventory ID",
|
||||
product_name=product_name, product_id=product_id, tenant_id=tenant_id)
|
||||
return uuid_id
|
||||
# Add delay before creation attempt to avoid hitting rate limits
|
||||
await asyncio.sleep(1.0)
|
||||
|
||||
# Add small delay before creation attempt to avoid hitting rate limits
|
||||
await asyncio.sleep(0.5)
|
||||
|
||||
# If not found, create a new ingredient/product in inventory
|
||||
# If not found or search failed, create a new ingredient/product in inventory
|
||||
ingredient_data = {
|
||||
'name': product_name,
|
||||
'type': 'finished_product', # Assuming sales are of finished products
|
||||
@@ -965,36 +994,133 @@ class DataImportService:
|
||||
'category': product_category or 'general'
|
||||
}
|
||||
|
||||
created_product = await self.inventory_client.create_ingredient(ingredient_data, str(tenant_id))
|
||||
if created_product and created_product.get('id'):
|
||||
product_id = created_product['id']
|
||||
uuid_id = UUID(str(product_id))
|
||||
self.product_cache[product_name] = uuid_id # Cache for future use
|
||||
logger.info("Created new inventory product for sales data",
|
||||
product_name=product_name, product_id=product_id, tenant_id=tenant_id)
|
||||
return uuid_id
|
||||
try:
|
||||
created_product = await self.inventory_client.create_ingredient(ingredient_data, str(tenant_id))
|
||||
if created_product and created_product.get('id'):
|
||||
product_id = created_product['id']
|
||||
uuid_id = UUID(str(product_id))
|
||||
self.product_cache[product_name] = uuid_id # Cache for future use
|
||||
logger.info("Created new inventory product for sales data",
|
||||
product_name=product_name, product_id=product_id, tenant_id=tenant_id)
|
||||
return uuid_id
|
||||
except Exception as creation_error:
|
||||
logger.warning("Product creation failed",
|
||||
product_name=product_name, error=str(creation_error), tenant_id=tenant_id)
|
||||
|
||||
logger.warning("Failed to resolve or create product in inventory",
|
||||
product_name=product_name, tenant_id=tenant_id, attempt=attempt)
|
||||
|
||||
except Exception as e:
|
||||
error_str = str(e)
|
||||
if "429" in error_str or "rate limit" in error_str.lower():
|
||||
logger.warning("Rate limit hit, retrying",
|
||||
if "429" in error_str or "rate limit" in error_str.lower() or "too many requests" in error_str.lower():
|
||||
logger.warning("Rate limit or service overload detected, retrying with longer delay",
|
||||
product_name=product_name, attempt=attempt, error=error_str, tenant_id=tenant_id)
|
||||
if attempt < max_retries - 1:
|
||||
continue # Retry with exponential backoff
|
||||
elif "503" in error_str or "502" in error_str or "service unavailable" in error_str.lower():
|
||||
logger.warning("Service unavailable, retrying with backoff",
|
||||
product_name=product_name, attempt=attempt, error=error_str, tenant_id=tenant_id)
|
||||
if attempt < max_retries - 1:
|
||||
continue # Retry for service unavailable errors
|
||||
elif "timeout" in error_str.lower() or "connection" in error_str.lower():
|
||||
logger.warning("Network issue detected, retrying",
|
||||
product_name=product_name, attempt=attempt, error=error_str, tenant_id=tenant_id)
|
||||
if attempt < max_retries - 1:
|
||||
continue # Retry for network issues
|
||||
else:
|
||||
logger.error("Error resolving product to inventory ID",
|
||||
logger.error("Non-retryable error resolving product to inventory ID",
|
||||
error=error_str, product_name=product_name, tenant_id=tenant_id)
|
||||
break # Don't retry for non-rate-limit errors
|
||||
if attempt < max_retries - 1:
|
||||
# Still retry even for other errors, in case it's transient
|
||||
continue
|
||||
else:
|
||||
break # Don't retry on final attempt
|
||||
|
||||
# If all retries failed, mark as failed and return None
|
||||
# If all retries failed, log detailed error but don't mark as permanently failed yet
|
||||
# Instead, we'll implement a fallback mechanism
|
||||
logger.error("Failed to resolve product after all retries, attempting fallback",
|
||||
product_name=product_name, tenant_id=tenant_id)
|
||||
|
||||
# FALLBACK: Try to create a temporary product with minimal data
|
||||
try:
|
||||
# Use a simplified approach with minimal data
|
||||
fallback_data = {
|
||||
'name': product_name,
|
||||
'type': 'finished_product',
|
||||
'unit': 'unit',
|
||||
'current_stock': 0,
|
||||
'cost_per_unit': 0
|
||||
}
|
||||
|
||||
logger.info("Attempting fallback product creation with minimal data",
|
||||
product_name=product_name, tenant_id=tenant_id)
|
||||
|
||||
created_product = await self.inventory_client.create_ingredient(fallback_data, str(tenant_id))
|
||||
if created_product and created_product.get('id'):
|
||||
product_id = created_product['id']
|
||||
uuid_id = UUID(str(product_id))
|
||||
self.product_cache[product_name] = uuid_id
|
||||
logger.info("SUCCESS: Fallback product creation succeeded",
|
||||
product_name=product_name, product_id=product_id, tenant_id=tenant_id)
|
||||
return uuid_id
|
||||
except Exception as fallback_error:
|
||||
logger.error("Fallback product creation also failed",
|
||||
product_name=product_name, error=str(fallback_error), tenant_id=tenant_id)
|
||||
|
||||
# Only mark as permanently failed after all attempts including fallback
|
||||
self.failed_products.add(product_name)
|
||||
logger.error("Failed to resolve product after all retries",
|
||||
logger.error("CRITICAL: Permanently failed to resolve product - this will result in missing training data",
|
||||
product_name=product_name, tenant_id=tenant_id)
|
||||
return None
|
||||
|
||||
async def _batch_resolve_products(self, unique_products: set, tenant_id: str) -> None:
|
||||
"""Batch resolve/create products to reduce API calls and improve success rate"""
|
||||
|
||||
if not unique_products:
|
||||
return
|
||||
|
||||
logger.info(f"Starting batch product resolution for {len(unique_products)} unique products")
|
||||
|
||||
# Convert set to list for easier handling
|
||||
products_list = list(unique_products)
|
||||
batch_size = 5 # Process in smaller batches to avoid overwhelming the inventory service
|
||||
|
||||
for i in range(0, len(products_list), batch_size):
|
||||
batch = products_list[i:i + batch_size]
|
||||
logger.info(f"Processing batch {i//batch_size + 1}/{(len(products_list) + batch_size - 1)//batch_size}")
|
||||
|
||||
# Process each product in the batch with retry logic
|
||||
for product_name, product_category in batch:
|
||||
try:
|
||||
# Skip if already in cache or failed list
|
||||
if product_name in self.product_cache or product_name in self.failed_products:
|
||||
continue
|
||||
|
||||
# Try to resolve the product
|
||||
await self._resolve_product_to_inventory_id(product_name, product_category, tenant_id)
|
||||
|
||||
# Add small delay between products to be gentle on the API
|
||||
await asyncio.sleep(0.5)
|
||||
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to batch process product {product_name}: {e}")
|
||||
continue
|
||||
|
||||
# Add delay between batches
|
||||
if i + batch_size < len(products_list):
|
||||
logger.info("Waiting between batches to avoid rate limiting...")
|
||||
await asyncio.sleep(2.0)
|
||||
|
||||
successful_resolutions = len([p for p, _ in products_list if p in self.product_cache])
|
||||
failed_resolutions = len([p for p, _ in products_list if p in self.failed_products])
|
||||
|
||||
logger.info(f"Batch product resolution completed: {successful_resolutions} successful, {failed_resolutions} failed")
|
||||
|
||||
if failed_resolutions > 0:
|
||||
logger.warning(f"ATTENTION: {failed_resolutions} products failed to resolve - these will be missing from training data")
|
||||
|
||||
return
|
||||
|
||||
def _structure_messages(self, messages: List[Union[str, Dict]]) -> List[Dict[str, Any]]:
|
||||
"""Convert string messages to structured format"""
|
||||
structured = []
|
||||
|
||||
Reference in New Issue
Block a user