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 # 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 lambda_handler(event, context): # Create a Boto3 client for Lambda lambda_client = boto3.client('lambda') # Initialize Supabase client url: str = get_parameter('/supabase/url') key: str = get_parameter('/supabase/anon') supabase: Client = create_client(url, key) supabase.postgrest.rpc("raw_sql", {"query": "SET statement_timeout = '0';"}) upload_response = supabase.table("sjreport_uploads").select("*").eq("processed","FALSE").limit(1).execute() # upload_response = supabase.table("sjreport_uploads").select("*").eq("processed","TRUE").eq("test","TRUE").limit(1).execute() # for testing only will run the testing file muliple times # print(f'Upload respose : {upload_response}') # Extracting details from the response if upload_response.data: # 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 file_name.endswith('.xlsx'): try: sheets = pd.read_excel(BytesIO(file_content), sheet_name=None, dtype=str) # get file and always make columns string as some of the fields are converted incorectly # Change column names to lowercase for each sheet for sheet_name in sheets.keys(): sheets[sheet_name].columns = sheets[sheet_name].columns.str.lower().str.replace(' ', '_') except: # Handle the error if the date format is incorrect print(f"Invalid file") supabase.table("sjreport_uploads").update({"processed": True,"error":"Unsupported file format"}).eq("id", record_id).execute() return { 'statusCode': 400, 'body': json.dumps({'error': 'Unsupported file format'}) } else: return { 'statusCode': 400, 'body': json.dumps({'error': 'Unsupported file format'}) } # Specify the data to add to a row url = file_name insert_amount = 1000 # after 1000 the request to insert starts to time out. last_week_start_date = None # Initialize variable to hold last week's start date # 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(): # 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 # Keep original value for other # replace row_data with modified_row_data row_data = modified_row_data # Calculate the week start date based on the date field try: # Attempt to parse the date # Check if "SPAR" is included in "chanel" and modify "store_code" if "SPAR" in modified_row_data["channel"]: modified_row_data["store_code"] = "S" + modified_row_data["store_code"] date_string = row_data["date"][:10] # only get the date value of the date field. (will get a invalid date error if uisng entire date) date = datetime.strptime(date_string, '%Y-%m-%d').date() week_start_date = date + timedelta(days=(7 - date.weekday())) # Calculate the next Monday last_week_start_date = week_start_date.strftime('%Y-%m-%d') # week start date to send to the create task function row_data['row'] = index + 1 # Adding the row number (1-based index) row_data['week_start_date'] = week_start_date.strftime('%Y-%m-%d') row_data['file'] = url row_data['sheetname'] = sheet_name data_to_insert.append(row_data) except: # Handle the error if the date format is incorrect print(f"Invalid date format") # update the record to set processed to true supabase.table("sjreport_uploads").update({"processed": True,"error":"Invalid date format"}).eq("id", record_id).execute() # Insert in batches try: # Insert in batches if len(data_to_insert) == insert_amount: # print(f'Data to insert = {data_to_insert}') # print(f'insert amount = {insert_amount}') response = supabase.table(table_name).upsert(data_to_insert, on_conflict='row, week_start_date, file, sheetname').execute() # Check for errors in the response if hasattr(response, 'error') and response.error: raise Exception(response.error.message) 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: # Handle the error when inserting data print(f"Error inserting data: {e}") # Print the specific error message # update the record to set processed to true supabase.table("sjreport_uploads").update({"processed": True,"error": "Error inserting data"}).eq("id", record_id).execute() return { 'statusCode': 400, 'body': json.dumps({'error': str(e)}) } # 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, week_start_date, file, sheetname').execute() print(f"Inserted {len(data_to_insert)} records into the table.") print(f"Data : {data_to_insert}") # update the record to set processed to true supabase.table("sjreport_uploads").update({"processed": True,"error":''}).eq("id", record_id).execute() supabase.rpc("update_product_table").execute(); payload = { "week_start_date": last_week_start_date } response = lambda_client.invoke( FunctionName='datafyNew-CreateTasks-ccMR48YAjXLP', InvocationType='Event', # Set to 'Event' for asynchronous invocation Payload=json.dumps(payload) # Send an empty JSON object as payload ) else: return { 'statusCode': 400, 'body': json.dumps({'error': 'Wrong upload folder'}) } return { 'statusCode': 200, 'body': json.dumps({'message': 'Data inserted successfully'}) }