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 email_service import email_service # Initialize AWS SSM client ssm = boto3.client('ssm') def get_parameter(name): """Get a parameter from AWS SSM Parameter Store""" response = ssm.get_parameter( Name=name, WithDecryption=True ) return response['Parameter']['Value'] def send_error_notification(error_message): """Send error notification email using email service""" try: email_service.send_notification( date=datetime.now().strftime('%Y-%m-%d'), status='error', error_message=f"Error in Promos or Kaching Processing\n\nError Message: {error_message}\n\nTime: {datetime.now().isoformat()}", client_email=True, process_name='Promos or Kaching Import' ) print(f"Error notification email sent") except Exception as e: print(f"Failed to send error notification email: {str(e)}") def send_success_notification(): """Send success notification email using email service""" try: email_service.send_notification( date=datetime.now().strftime('%Y-%m-%d'), status='success', error_message=f"Successfully processed Promos or Kaching import\n\nTime: {datetime.now().isoformat()}", client_email=True, process_name='Promos or Kaching Import' ) print(f"Success notification email sent") except Exception as e: print(f"Failed to send success notification email: {str(e)}") def delete_storage_file(supabase, bucket_name, file_name): """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]) print(f"Deleted file '{file_name}' from bucket '{bucket_name}'") except Exception as e: # Do not fail the Lambda on delete errors print(f"Failed to delete file '{file_name}' from bucket '{bucket_name}': {str(e)}") def lambda_handler(event, context): try: # Defaults to allow cleanup in error paths supabase = None bucket_name = "" file_name = "" # Initialize Supabase client url: str = get_parameter('/supabase/url') key: str = get_parameter('/supabase/anon') supabase: Client = create_client(url, key) upload_response = supabase.table("promo_kaching_uploads").select("*").eq("processed","FALSE").limit(1).execute() # upload_response = supabase.table("promo_kaching_uploads").select("*").eq("processed","TRUE").eq("test","TRUE").limit(1).execute() # for testing only will run the testing file muliple times # Extracting details from the response if not upload_response.data: error_msg = 'No unprocessed uploads found' send_error_notification(error_msg) return { 'statusCode': 400, 'body': json.dumps({'error': error_msg}) } # Assuming you want to process the first item in the data list record = upload_response.data[0] # Extract the ID of the record record_id = record['id'] bucket_name = "public/" + record['payload']['bucket_name'] # Prepend "public/" file_name = record['payload']['file_name'] table_name = record['payload']['path_tokens'][0] # First element of path_tokens print(f'Bucket: {bucket_name}, File Name: {file_name}, Table Name: {table_name}') # Read a file from Supabase storage response = supabase.storage.from_(bucket_name).download(file_name) file_content = response # Load the file content into a Pandas DataFrame if not file_name.endswith('.xlsx'): error_msg = 'Unsupported file format' send_error_notification(error_msg) supabase.table("promo_kaching_uploads").update({"processed": True,"error": error_msg}).eq("id", record_id).execute() # 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}) } try: # Load only visible sheets; ignore hidden/veryHidden sheets 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: sheets = pd.read_excel(BytesIO(file_content), sheet_name=visible_sheet_names, dtype=str) else: sheets = {} except Exception: # Fallback: load all sheets if workbook inspection fails sheets = pd.read_excel(BytesIO(file_content), sheet_name=None, dtype=str) # Change column names to lowercase for each sheet for sheet_name in list(sheets.keys()): sheets[sheet_name].columns = sheets[sheet_name].columns.str.lower().str.replace(' ', '_') except Exception as e: error_msg = f"Invalid file format: {str(e)}" send_error_notification(error_msg) supabase.table("promo_kaching_uploads").update({"processed": True,"error": error_msg}).eq("id", record_id).execute() # 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}) } # Specify the data to add to a row url = file_name insert_amount = 1000 # after 1000 the request to insert starts to time out. # Iterate through each sheet and its rows for sheet_name, sheet_df in sheets.items(): print(f"Processing sheet: {sheet_name}") # Prepare data for insertion data_to_insert = [] for index, row in sheet_df.iterrows(): try: # Convert each row to a dictionary and add additional fields row_data = row.to_dict() modified_row_data = {} # Remove unwanted columns row_data.pop('id', None) # Remove 'id' if it exists row_data.pop('created_at', None) # Remove 'created_at' if it exists for col, value in row_data.items(): if pd.isna(value): modified_row_data[col] = "" else: modified_row_data[col] = value # Keep original value for other modified_row_data['row'] = index + 1 # Adding the row number (1-based index) modified_row_data['file'] = url modified_row_data['sheetname'] = sheet_name # replace row_data with modified_row_data row_data = modified_row_data data_to_insert.append(row_data) # Insert in batches if len(data_to_insert) == insert_amount: supabase.table(table_name).upsert(data_to_insert, on_conflict='row,file, sheetname').execute() print(f"Inserted {len(data_to_insert)} records into the table.") data_to_insert = [] # Reset the list for the next batch except Exception as e: error_msg = f"Error inserting data: {str(e)}" send_error_notification(error_msg) supabase.table("promo_kaching_uploads").update({"processed": True,"error": error_msg}).eq("id", record_id).execute() # 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}) } # Insert any remaining records that didn't make a full batch if data_to_insert: supabase.table(table_name).upsert(data_to_insert, on_conflict='row, file, sheetname').execute() print(f"Inserted {len(data_to_insert)} records into the table.") # Update the record to set processed to true supabase.table("promo_kaching_uploads").update({"processed": True,"error":''}).eq("id", record_id).execute() # Delete the source file after successful processing delete_storage_file(supabase, bucket_name, file_name) # Send success notification send_success_notification() return { 'statusCode': 200, 'body': json.dumps({'message': 'Data inserted successfully'}) } except Exception as e: error_msg = f"Critical error in processing: {str(e)}" send_error_notification(error_msg) if 'record_id' in locals(): supabase.table("promo_kaching_uploads").update({"processed": True,"error": error_msg}).eq("id", record_id).execute() # Always attempt to delete the source file on critical errors so the user can re-upload delete_storage_file(supabase, bucket_name, file_name) return { 'statusCode': 500, 'body': json.dumps({'error': error_msg}) }