import os import json from supabase import create_client, Client import boto3 import pandas as pd from io import BytesIO from datetime import datetime, timedelta from typing import Dict, List, Any, Optional import logging from email_service import email_service import traceback from rapidfuzz import process, fuzz # Configure logging logger = logging.getLogger() logger.setLevel(logging.INFO) # Initialize AWS SSM client ssm = boto3.client('ssm') def get_parameter(name: str) -> str: """Get a parameter from AWS SSM Parameter Store""" try: response = ssm.get_parameter( Name=name, WithDecryption=True ) return response['Parameter']['Value'] except Exception as e: logger.error(f"Error getting parameter {name}: {str(e)}") raise def get_supabase_client() -> Client: """Initialize and return Supabase client""" try: url: str = get_parameter('/supabase/url') key: str = get_parameter('/supabase/anon') # Initialize Supabase client (use library defaults for HTTP configuration) # Note: Older versions of supabase-py do not support ClientOptions(http_client=...) supabase: Client = create_client(url, key) supabase.postgrest.rpc("raw_sql", {"query": "SET statement_timeout = '0';"}) return supabase except Exception as e: logger.error(f"Error initializing Supabase client: {str(e)}") raise def get_unprocessed_upload(supabase: Client) -> Optional[Dict]: """Get the next unprocessed upload record with times_ran < 5""" try: upload_response = supabase.table("sjreport_uploads").select("*").eq("processed", "FALSE").lt("times_ran", 5).limit(1).execute() return upload_response.data[0] if upload_response.data else None except Exception as e: logger.error(f"Error getting unprocessed upload: {str(e)}") raise def download_file(supabase: Client, bucket_name: str, file_name: str) -> bytes: """Download file from Supabase storage""" try: response = supabase.storage.from_(bucket_name).download(file_name) return response except Exception as e: logger.error(f"Error downloading file {file_name}: {str(e)}") raise def delete_storage_file(supabase: Optional[Client], bucket_name: str, file_name: str) -> None: """Best-effort delete of a file from Supabase storage (used on success and error).""" try: if not supabase or not bucket_name or not file_name: return supabase.storage.from_(bucket_name).remove([file_name]) logger.info(f"Deleted file '{file_name}' from bucket '{bucket_name}'") except Exception as e: # We never fail the Lambda on delete errors, only log them logger.error(f"Error deleting file '{file_name}' from bucket '{bucket_name}': {str(e)}") def process_excel_file(file_content: bytes) -> Dict[str, pd.DataFrame]: """Process Excel file and return dictionary of DataFrames with fuzzy-matched column names""" try: # List of correct column names as in your DB correct_columns = [ 'date', 'region', 'channel', 'store_code', 'banner', 'store_name', 'category', 'brand', 'product', 'variant', 'size', 'article', 'barcode', 'store_soh', 'system_dros', 'days_cover', 'last_sold', 'last_receive', 'last_order', 'rsp_value_loss', 'issue_for_feedback', 'status', 'dc_soh','customer_id' ] # Load only visible sheets; ignore hidden/veryHidden sheets # Read as string first to preserve exact format, then we'll convert dates manually try: from openpyxl import load_workbook # Lazy import to avoid cold-start overhead wb = load_workbook(BytesIO(file_content), read_only=True, data_only=True) visible_sheet_names = [ ws.title for ws in wb.worksheets if getattr(ws, "sheet_state", "visible") == "visible" ] if visible_sheet_names: # Read as string to preserve date formats exactly as they appear sheets = pd.read_excel(BytesIO(file_content), sheet_name=visible_sheet_names, dtype=str, keep_default_na=False) else: sheets = {} except Exception: # Fallback: load all sheets if workbook inspection fails sheets = pd.read_excel(BytesIO(file_content), sheet_name=None, dtype=str, keep_default_na=False) for sheet_name in list(sheets.keys()): # Standardize column names cols = sheets[sheet_name].columns.str.lower().str.replace(' ', '_') new_cols = [] for col in cols: match, score, _ = process.extractOne( col, correct_columns, scorer=fuzz.ratio ) # If the match is good enough, use it; otherwise, keep the original if score > 80: new_cols.append(match) else: new_cols.append(col) sheets[sheet_name].columns = new_cols # Convert date column from Excel serial number to ISO date format if 'date' in sheets[sheet_name].columns: # Counter to track logging - only log first occurrence of each type log_counters = {'debug': 0, 'info': 0, 'warning': 0, 'error': 0} def convert_excel_date(value): """Convert Excel serial date or date string to ISO date format""" if pd.isna(value) or value == '' or str(value).strip() == '': return '' value_str = str(value).strip() # Try to parse as Excel serial date (numeric string) # Excel serial dates are typically integers or floats (e.g., 44927, 44927.5) try: # Try to convert to float to check if it's numeric excel_serial = float(value_str) # Only treat as Excel serial if it's in a reasonable range (Excel dates are typically 1-100000+) # Dates before 1900 would be < 1, dates in 2023-2024 are around 45000-46000 if 1 <= excel_serial <= 1000000: # Excel epoch is 1899-12-30, but Excel incorrectly treats 1900 as a leap year # So we subtract 1 day to account for this base_date = pd.Timestamp('1899-12-30') date = base_date + pd.Timedelta(days=excel_serial - 1) result = date.strftime('%Y-%m-%d') logger.debug(f"Converted Excel serial {excel_serial} to {result}") return result except (ValueError, OverflowError): # Not a numeric value, try parsing as date string below pass # Try to parse as existing date string # Try multiple explicit formats in order of specificity try: # 1. ISO datetime format "YYYY-MM-DD HH:MM:SS" (e.g., "2025-12-08 00:00:00") # This handles dates that Excel/pandas already converted to datetime strings # Check if it looks like a datetime string first if ' ' in value_str and ':' in value_str: # Try just the date part first (simpler and more reliable) date_part = value_str.split(' ')[0] if len(date_part) == 10 and date_part.count('-') == 2: try: date = pd.to_datetime(date_part, format='%Y-%m-%d', errors='raise') result = date.strftime('%Y-%m-%d') if log_counters['debug'] == 0: logger.debug(f"Parsed '{value_str}' as {result} by extracting date part '{date_part}'") log_counters['debug'] += 1 return result except (ValueError, TypeError): pass # Try full datetime format as fallback try: date = pd.to_datetime(value_str, format='%Y-%m-%d %H:%M:%S', errors='raise') result = date.strftime('%Y-%m-%d') if log_counters['debug'] == 0: logger.debug(f"Parsed '{value_str}' as {result} using 'YYYY-MM-DD HH:MM:SS' format") log_counters['debug'] += 1 return result except (ValueError, TypeError): pass # 2. ISO format "YYYY-MM-DD" (e.g., "2025-12-08") try: date = pd.to_datetime(value_str, format='%Y-%m-%d', errors='raise') result = date.strftime('%Y-%m-%d') if log_counters['debug'] == 0: logger.debug(f"Parsed '{value_str}' as {result} using 'YYYY-MM-DD' format") log_counters['debug'] += 1 return result except (ValueError, TypeError): # 3. "DayOfWeek, DD Month YYYY" format (e.g., "Monday, 08 December 2025") try: date = pd.to_datetime(value_str, format='%A, %d %B %Y', errors='raise') return date.strftime('%Y-%m-%d') except (ValueError, TypeError): # 4. "DayOfWeek, DD Mon YYYY" format (e.g., "Mon, 08 Dec 2025") try: date = pd.to_datetime(value_str, format='%a, %d %b %Y', errors='raise') return date.strftime('%Y-%m-%d') except (ValueError, TypeError): # 5. "DD Mon YYYY" format (e.g., "08 Dec 2025") - most common in spreadsheets # This MUST be tried before any slash formats to avoid ambiguity try: date = pd.to_datetime(value_str, format='%d %b %Y', errors='raise') result = date.strftime('%Y-%m-%d') if log_counters['debug'] == 0: logger.debug(f"Parsed '{value_str}' as {result} using 'DD Mon YYYY' format") log_counters['debug'] += 1 return result except (ValueError, TypeError): # 6. "DD Month YYYY" format (e.g., "08 December 2025") - full month name try: date = pd.to_datetime(value_str, format='%d %B %Y', errors='raise') result = date.strftime('%Y-%m-%d') if log_counters['debug'] == 0: logger.debug(f"Parsed '{value_str}' as {result} using 'DD Month YYYY' format") log_counters['debug'] += 1 return result except (ValueError, TypeError): # 7. Slash format "YYYY/MM/DD" (e.g., "2025/12/08") try: date = pd.to_datetime(value_str, format='%Y/%m/%d', errors='raise') return date.strftime('%Y-%m-%d') except (ValueError, TypeError): # 8. Slash format "DD/MM/YYYY" (e.g., "08/12/2025") - South African format (day first) try: date = pd.to_datetime(value_str, format='%d/%m/%Y', errors='raise') return date.strftime('%Y-%m-%d') except (ValueError, TypeError): # 9. If all explicit formats fail, log error - DO NOT use fallback inference if log_counters['error'] == 0: logger.error(f"Could not parse date with any known format: '{value_str}' (original type: {type(value)})") log_counters['error'] += 1 raise ValueError(f"Could not parse date with any known format: '{value_str}'") except (ValueError, TypeError) as e: # If all parsing fails, log and return original value logger.warning(f"Could not parse date value '{value_str}' in sheet '{sheet_name}': {str(e)}") return value_str # Log raw date values BEFORE conversion for debugging raw_dates = sheets[sheet_name]['date'].head(5).tolist() logger.info(f"Sheet '{sheet_name}': Sample RAW date values (before conversion): {raw_dates}") # Apply conversion to date column sheets[sheet_name]['date'] = sheets[sheet_name]['date'].apply(convert_excel_date) # Log sample of converted dates for debugging (first 3 non-empty values) sample_dates = sheets[sheet_name]['date'].head(10).tolist() non_empty_samples = [d for d in sample_dates if d and str(d).strip() and str(d).lower() != 'nan'][:3] if non_empty_samples: logger.info(f"Sheet '{sheet_name}': Sample converted dates (after conversion): {non_empty_samples}") return sheets except Exception as e: logger.error(f"Error processing Excel file: {str(e)}") raise def process_row(row: pd.Series, url: str, sheet_name: str) -> Dict: """Process a single row of data""" try: row_data = row.to_dict() modified_row_data = {} # Handle null values for col, value in row_data.items(): modified_row_data[col] = "" if pd.isna(value) else value # # Process SPAR store codes # if "SPAR" in modified_row_data.get("channel", ""): # modified_row_data["store_code"] = "S" + modified_row_data["store_code"] # Process date date_string = modified_row_data["date"][:10] date = datetime.strptime(date_string, '%Y-%m-%d').date() week_start_date = date + timedelta(days=(7 - date.weekday())) # Add metadata modified_row_data.update({ 'row': row.name + 1, 'week_start_date': week_start_date.strftime('%Y-%m-%d'), 'file': url, 'sheetname': sheet_name }) return modified_row_data except Exception as e: logger.error(f"Error processing row: {str(e)}") raise def insert_batch(supabase: Client, table_name: str, batch: List[Dict]) -> None: """Insert a batch of records into the database""" try: response = supabase.table(table_name).upsert(batch, on_conflict='row, week_start_date, file, sheetname').execute() if hasattr(response, 'error') and response.error: raise Exception(response.error.message) logger.info(f"Successfully inserted {len(batch)} records") except Exception as e: logger.error(f"Error inserting batch: {str(e)}") raise def increment_times_ran(supabase: Client, record_id: str) -> None: try: # Fetch current value response = supabase.table("sjreport_uploads").select("times_ran").eq("id", record_id).execute() if response.data and len(response.data) > 0: current = response.data[0].get("times_ran", 0) or 0 new_value = current + 1 supabase.table("sjreport_uploads").update({"times_ran": new_value}).eq("id", record_id).execute() except Exception as e: logger.error(f"Error incrementing times_ran: {str(e)}") raise def update_upload_status(supabase: Client, record_id: str, error: str = '', processed: bool = True) -> None: """Update the upload record status""" try: update_data = { "error": error, "processed": processed } supabase.table("sjreport_uploads").update(update_data).eq("id", record_id).execute() except Exception as e: logger.error(f"Error updating upload status: {str(e)}") raise def invoke_lambda_function(lambda_client: boto3.client, week_start_date: str, record_id: str) -> None: """Invoke the create tasks Lambda function""" try: payload = { "week_start_date": week_start_date, "record_id": record_id } lambda_client.invoke( FunctionName='datafyNew-BatchUpdateSjreport-utMCtdWkdNo8', InvocationType='Event', Payload=json.dumps(payload) ) except Exception as e: logger.error(f"Error invoking create tasks: {str(e)}") raise def lambda_handler(event, context): """Main Lambda handler function""" try: # Defaults to prevent UnboundLocalError in broad exception paths file_name: str = "" bucket_name: str = "" record_id: Optional[str] = None supabase: Optional[Client] = None # Initialize clients lambda_client = boto3.client('lambda') supabase = get_supabase_client() # Get unprocessed upload record = get_unprocessed_upload(supabase) if not record: return { 'statusCode': 400, 'body': json.dumps({'error': 'No unprocessed uploads found'}) } record_id = record['id'] bucket_name = "public/" + record['payload']['bucket_name'] file_name = record['payload']['file_name'] table_name = record['payload']['path_tokens'][0] # Increment times_ran increment_times_ran(supabase, record_id) # Download and process file file_content = download_file(supabase, bucket_name, file_name) sheets = process_excel_file(file_content) # Validate required columns exist in at least one sheet required_columns = ['date', 'store_code'] sheets_with_required_cols = [] for sheet_name, sheet_df in sheets.items(): missing_cols = [col for col in required_columns if col not in sheet_df.columns] if missing_cols: logger.warning(f"Sheet '{sheet_name}': Missing required columns: {missing_cols}. Available: {list(sheet_df.columns)}") else: sheets_with_required_cols.append(sheet_name) if not sheets_with_required_cols: error_msg = f"No sheets found with required columns {required_columns}. Cannot process file." logger.error(error_msg) # Always mark as processed so we don't retry indefinitely; then log and email update_upload_status(supabase, record_id, error_msg, processed=True) email_service.send_notification( date=datetime.now().strftime('%Y-%m-%d'), status='error', error_message=f"File: {file_name}\n{error_msg}", client_email=False, process_name='Read Files Processing' ) email_service.send_notification( date=datetime.now().strftime('%Y-%m-%d'), status='error', error_message=f"An error occurred during the SJREPORT import.\nFile: {file_name}\n{error_msg}", client_email=True, process_name='SJREPORT Import' ) # Always delete the uploaded file so the user can fix and re-upload delete_storage_file(supabase, bucket_name, file_name) return { 'statusCode': 400, 'body': json.dumps({'error': error_msg}) } # Process each sheet last_week_start_date = None BATCH_SIZE = 500 # Reduced batch size for better memory management total_rows_processed = 0 total_rows_skipped = 0 for sheet_name, sheet_df in sheets.items(): logger.info(f"Processing sheet: {sheet_name} with {len(sheet_df)} rows") data_to_insert = [] # Track statistics rows_processed = 0 rows_skipped_empty_date = 0 rows_skipped_invalid_date = 0 rows_skipped_missing_date_col = 0 # Counter to track logging - only log first occurrence of each type parse_log_counters = {'info': 0, 'warning': 0} # Cache for date parsing – most files have the same date on every row, # so we parse once and reuse for the rest of the sheet. cached_date_value = None cached_date_obj = None cached_week_start = None for index, row in sheet_df.iterrows(): # Convert each row to a dictionary and add additional fields row_data = row.to_dict() modified_row_data = {} for col, value in row_data.items(): if pd.isna(value): modified_row_data[col] = "" else: modified_row_data[col] = value # replace row_data with modified_row_data row_data = modified_row_data # Check if date column exists if 'date' not in row_data or not row_data.get('date'): rows_skipped_missing_date_col += 1 if rows_skipped_missing_date_col <= 5: # Log first 5 only logger.warning(f"Row {index + 1}: Missing 'date' column. Available columns: {list(row_data.keys())}") continue # Get the date value - it should already be converted to ISO format, but handle edge cases date_value = str(row_data["date"]).strip() if not date_value or date_value == "" or date_value.lower() == "nan": rows_skipped_empty_date += 1 if rows_skipped_empty_date <= 5: # Log first 5 only logger.warning(f"Row {index + 1}: Empty date value") continue try: # If we have already parsed this exact date for this sheet, reuse it if cached_date_value is not None and date_value == cached_date_value: date = cached_date_obj week_start_date = cached_week_start else: # Try to parse the date - handle multiple formats date = None # Try multiple date formats in order of specificity # First, try ISO format (YYYY-MM-DD) - this is what convert_excel_date should return try: date_string = date_value[:10] if len(date_value) >= 10 else date_value date = datetime.strptime(date_string, '%Y-%m-%d').date() except ValueError: # If ISO format fails, try explicit formats try: # 1. "DayOfWeek, DD Month YYYY" format (e.g., "Monday, 08 December 2025") date = pd.to_datetime(date_value, format='%A, %d %B %Y', errors='raise').date() if parse_log_counters['info'] == 0: logger.info(f"Row {index + 1}: Parsed date '{date_value}' as {date} using 'DayOfWeek, DD Month YYYY' format") parse_log_counters['info'] += 1 except (ValueError, TypeError): # 2. "DayOfWeek, DD Mon YYYY" format (e.g., "Mon, 08 Dec 2025") try: date = pd.to_datetime(date_value, format='%a, %d %b %Y', errors='raise').date() if parse_log_counters['info'] == 0: logger.info(f"Row {index + 1}: Parsed date '{date_value}' as {date} using 'DayOfWeek, DD Mon YYYY' format") parse_log_counters['info'] += 1 except (ValueError, TypeError): # 3. "DD Mon YYYY" format (e.g., "08 Dec 2025") - most common in spreadsheets try: date = pd.to_datetime(date_value, format='%d %b %Y', errors='raise').date() if parse_log_counters['info'] == 0: logger.info(f"Row {index + 1}: Parsed date '{date_value}' as {date} using 'DD Mon YYYY' format") parse_log_counters['info'] += 1 except (ValueError, TypeError): # 4. "DD Month YYYY" format (e.g., "08 December 2025") - full month name try: date = pd.to_datetime(date_value, format='%d %B %Y', errors='raise').date() if parse_log_counters['info'] == 0: logger.info(f"Row {index + 1}: Parsed date '{date_value}' as {date} using 'DD Month YYYY' format") parse_log_counters['info'] += 1 except (ValueError, TypeError): # 5. Slash format "YYYY/MM/DD" (e.g., "2025/12/08") try: date = pd.to_datetime(date_value, format='%Y/%m/%d', errors='raise').date() if parse_log_counters['info'] == 0: logger.info(f"Row {index + 1}: Parsed date '{date_value}' as {date} using 'YYYY/MM/DD' format") parse_log_counters['info'] += 1 except (ValueError, TypeError): # 6. Slash format "DD/MM/YYYY" (e.g., "08/12/2025") - South African format (day first) try: date = pd.to_datetime(date_value, format='%d/%m/%Y', errors='raise').date() if parse_log_counters['info'] == 0: logger.info(f"Row {index + 1}: Parsed date '{date_value}' as {date} using 'DD/MM/YYYY' format") parse_log_counters['info'] += 1 except (ValueError, TypeError): # 7. Last resort: try without format but with dayfirst=True (no inference) # This handles edge cases but is less reliable than explicit formats try: date = pd.to_datetime(date_value, errors='raise', dayfirst=True).date() if parse_log_counters['warning'] == 0: logger.warning(f"Row {index + 1}: Parsed date '{date_value}' as {date} using fallback parsing (dayfirst=True). Consider using explicit format.") parse_log_counters['warning'] += 1 except (ValueError, TypeError) as parse_error: raise ValueError(f"Could not parse date '{date_value}': {str(parse_error)}") # After successful parse, cache for subsequent rows with the same date value cached_date_value = date_value cached_date_obj = date # week_start_date is computed just below, so we cache after that # # Check if "SPAR" is included in "channel" and modify "store_code" # if "SPAR" in modified_row_data["channel"]: # modified_row_data["store_code"] = "S" + modified_row_data["store_code"] week_start_date = date + timedelta(days=(7 - date.weekday())) last_week_start_date = week_start_date.strftime('%Y-%m-%d') cached_week_start = week_start_date # Ensure date is stored in ISO format row_data['date'] = date.strftime('%Y-%m-%d') row_data['row'] = index + 1 row_data['week_start_date'] = week_start_date.strftime('%Y-%m-%d') row_data['file'] = file_name row_data['sheetname'] = sheet_name data_to_insert.append(row_data) rows_processed += 1 # Insert in smaller batches while iterating to avoid huge payloads if len(data_to_insert) >= BATCH_SIZE: insert_batch(supabase, table_name, data_to_insert) data_to_insert = [] except Exception as e: rows_skipped_invalid_date += 1 error_details = f"Row {index + 1}: Invalid date format or row error: {str(e)} | Date value: '{date_value}' | Store code: {row_data.get('store_code', 'N/A')}" logger.error(error_details) # Only send email for first few errors to avoid spam if rows_skipped_invalid_date <= 3: email_service.send_notification( date=datetime.now().strftime('%Y-%m-%d'), status='error', error_message=f"Error processing row:\n{error_details}\n\nStack Trace:\n{traceback.format_exc()}", client_email=False, process_name='Read Files Row Processing' ) continue # Log statistics for this sheet sheet_skipped = rows_skipped_empty_date + rows_skipped_invalid_date + rows_skipped_missing_date_col total_rows_processed += rows_processed total_rows_skipped += sheet_skipped logger.info(f"Sheet '{sheet_name}' statistics: {rows_processed} processed, {sheet_skipped} skipped " f"(empty: {rows_skipped_empty_date}, invalid: {rows_skipped_invalid_date}, missing col: {rows_skipped_missing_date_col})") # Insert remaining records if data_to_insert: try: insert_batch(supabase, table_name, data_to_insert) logger.info(f"Inserted {len(data_to_insert)} records into the table.") except Exception as e: error_details = f"Error inserting data: {str(e)} | Batch: {data_to_insert}" logger.error(error_details) # Send error email for batch insertion error # User-friendly error message for client user_friendly_message = "An error occurred during the SJREPORT import." if "Could not find the" in str(e) and "column" in str(e): import re match = re.search(r"Could not find the '([^']+)' column", str(e)) if match: missing_col = match.group(1).replace('_', ' ').upper() user_friendly_message = f'Could not find column "{missing_col}" in the database. Please fix your file and try again.' # Add file name to the user-friendly message file_info = f'File: {file_name}\n' user_friendly_message = file_info + user_friendly_message email_service.send_notification( date=datetime.now().strftime('%Y-%m-%d'), status='error', error_message=user_friendly_message, client_email=True, process_name='SJREPORT Import' ) # Mark processed=True so we don't retry; error already logged and emailed above update_upload_status(supabase, record_id, error_details, processed=True) # Delete the file so the client can correct and re-upload delete_storage_file(supabase, bucket_name, file_name) return { 'statusCode': 400, 'body': json.dumps({'error': error_details}) } # Update product table and create tasks supabase.postgrest.rpc("raw_sql", {"query": "SET statement_timeout = '0';"}) # Some runs may reprocess weeks already inserted; ignore duplicate-key violations from the RPC try: supabase.rpc("insert_distinct_products").execute() except Exception as e: # Ignore duplicates (Postgres code 23505) and continue; re-raises anything else if "23505" in str(e) or "duplicate key value violates unique constraint" in str(e): logger.warning("insert_distinct_products: duplicate products detected; continuing without failing the run") else: raise # Run post-processing RPCs independently; log and continue on failure try: # Retry the batch update until the RPC reports no more work. BATCH_SIZE = 1000 MAX_LOOPS = 200 # safety guard against infinite loops total_updated = 0 loops = 0 while True: loops += 1 resp = supabase.rpc("batch_update_sjreport_product_id", {"p_batch_size": BATCH_SIZE, "p_hours_back": 1}).execute() updated = 0 # supabase-py returns .data; different RPCs may return int or object if hasattr(resp, "data"): if isinstance(resp.data, int): updated = resp.data elif isinstance(resp.data, dict): updated = resp.data.get("rows_updated", 0) or 0 logger.info(f"batch_update_sjreport_product_id: loop={loops}, updated={updated}, total={total_updated + (updated if isinstance(updated, int) else 0)}") if not updated or (isinstance(updated, int) and updated < BATCH_SIZE): break total_updated += updated if isinstance(updated, int) else 0 if loops >= MAX_LOOPS: logger.warning("batch_update_sjreport_product_id: reached MAX_LOOPS guard; stopping further retries") break except Exception as e: err = f"RPC batch_update_sjreport_product_id failed: {str(e)}" logger.error(err) email_service.send_notification( date=datetime.now().strftime('%Y-%m-%d'), status='error', error_message=err, client_email=False, process_name='Read Files Processing' ) update_upload_status(supabase, record_id, err, processed=True) try: # Loop until no more customer IDs to update BATCH_SIZE = 1000 MAX_LOOPS = 200 total_updated = 0 loops = 0 while True: loops += 1 resp = supabase.rpc("update_sjreport_customer_ids", {"p_batch_size": BATCH_SIZE, "p_hours_back": 1}).execute() updated = 0 if hasattr(resp, "data"): if isinstance(resp.data, int): updated = resp.data elif isinstance(resp.data, dict): updated = resp.data.get("rows_updated", 0) or 0 logger.info(f"update_sjreport_customer_ids: loop={loops}, updated={updated}, total={total_updated + (updated if isinstance(updated, int) else 0)}") if not updated or (isinstance(updated, int) and updated < BATCH_SIZE): break total_updated += updated if isinstance(updated, int) else 0 if loops >= MAX_LOOPS: logger.warning("update_sjreport_customer_ids: reached MAX_LOOPS guard; stopping further retries") break except Exception as e: err = f"RPC update_sjreport_customer_ids failed: {str(e)}" logger.error(err) email_service.send_notification( date=datetime.now().strftime('%Y-%m-%d'), status='error', error_message=err, client_email=False, process_name='Read Files Processing' ) update_upload_status(supabase, record_id, err, processed=True) try: # Loop until no more product->customer IDs to update BATCH_SIZE = 1000 MAX_LOOPS = 200 total_updated = 0 loops = 0 while True: loops += 1 resp = supabase.rpc("update_product_customer_ids", {"p_batch_size": BATCH_SIZE, "p_hours_back": 1}).execute() updated = 0 if hasattr(resp, "data"): if isinstance(resp.data, int): updated = resp.data elif isinstance(resp.data, dict): updated = resp.data.get("rows_updated", 0) or 0 logger.info(f"update_product_customer_ids: loop={loops}, updated={updated}, total={total_updated + (updated if isinstance(updated, int) else 0)}") if not updated or (isinstance(updated, int) and updated < BATCH_SIZE): break total_updated += updated if isinstance(updated, int) else 0 if loops >= MAX_LOOPS: logger.warning("update_product_customer_ids: reached MAX_LOOPS guard; stopping further retries") break except Exception as e: err = f"RPC update_product_customer_ids failed: {str(e)}" logger.error(err) email_service.send_notification( date=datetime.now().strftime('%Y-%m-%d'), status='error', error_message=err, client_email=False, process_name='Read Files Processing' ) update_upload_status(supabase, record_id, err, processed=True) if last_week_start_date: invoke_lambda_function(lambda_client, last_week_start_date, record_id) # Mark upload as processed update_upload_status(supabase, record_id, processed=True) # Delete the source file after successful processing to keep storage clean delete_storage_file(supabase, bucket_name, file_name) # Log final statistics logger.info(f"Import complete for file '{file_name}': {total_rows_processed} rows processed, {total_rows_skipped} rows skipped") return { 'statusCode': 200, 'body': json.dumps({ 'message': 'Data processed successfully', 'rows_processed': total_rows_processed, 'rows_skipped': total_rows_skipped }) } except Exception as e: logger.error(f"Error in lambda_handler: {str(e)}") error_message = f"Error Type: {type(e).__name__}\nError Message: {str(e)}\n\nStack Trace:\n{traceback.format_exc()}" # Send error email to admin email_service.send_notification( date=datetime.now().strftime('%Y-%m-%d'), status='error', error_message=error_message, client_email=False, process_name='Read Files Processing' ) # Send error email to client user_friendly_message = f"An error occurred during the SJREPORT import.\nFile: {file_name}\nError: {str(e)}" email_service.send_notification( date=datetime.now().strftime('%Y-%m-%d'), status='error', error_message=user_friendly_message, client_email=True, process_name='SJREPORT Import' ) # Always mark as processed so we don't retry; error already logged and emailed above if supabase and record_id: update_upload_status(supabase, record_id, f"Error in lambda_handler: {str(e)}", processed=True) # Always attempt to delete the source file on unhandled errors so the user can re-upload delete_storage_file(supabase, bucket_name, file_name) return { 'statusCode': 500, 'body': json.dumps({'error': str(e)}) }