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 import traceback from typing import Any from email_service import email_service import traceback # Initialize AWS clients ssm = boto3.client('ssm') ses = boto3.client('ses') 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 log_error_to_supabase(supabase: Client, error_message: str, context: Any, error_type: str = 'general', additional_data: dict = None): """Log error to Supabase tasks_create_logs table""" try: error_data = { 'error_message': error_message, 'error_type': error_type, 'function_name': context.function_name if context else None, 'timestamp': datetime.now().isoformat(), 'additional_data': json.dumps(additional_data) if additional_data else None } supabase.table('tasks_create_logs').insert(error_data).execute() print(f"Error logged to Supabase: {error_message}") except Exception as e: print(f"Failed to log error to Supabase: {str(e)}") def send_error_notification(error_message, context): """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 Datafy Processing\n\nError Message: {error_message}\n\nTime: {datetime.now().isoformat()}", client_email=True, process_name='Create Tasks Processing' ) print(f"Error notification email sent") except Exception as e: print(f"Failed to send error notification email: {str(e)}") def send_success_notification(context): """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 imported tasks from SJREPORT\n\nTime: {datetime.now().isoformat()}", client_email=True, process_name='SJREPORT Import' ) print(f"Success notification email sent") except Exception as e: print(f"Failed to send success notification email: {str(e)}") def lambda_handler(event, context): try: # 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 = '20min';"}) offset = 0 limit = 2000 sjreport_records = [] date = event['week_start_date'] while True: try: sjreport = supabase.rpc("get_sjreport_data_pages_v2",{ "target_date": date, "page_offset": offset, "limit_count": limit }).execute() if sjreport.data is None: print("No data returned from sjreport.") break print(f"Number of records sj read : {len(sjreport.data)} + offset : {offset}") sjreport_records.extend(sjreport.data) if len(sjreport.data) < limit: break # Exit if fewer records than limit are returned offset += limit except Exception as e: error_msg = f"Error fetching sjreport data: {str(e)}" print(error_msg) log_error_to_supabase(supabase, error_msg, context, 'data_fetch', {'offset': offset, 'limit': limit}) continue # Convert sjreport data to DataFrame sjreport_df = pd.DataFrame(sjreport_records) # Prepare to store the combined records insertData = [] # Iterate through sjreport and callsheet DataFrames to create the combined records for index, sjElement in sjreport_df.iterrows(): try: # Create the new combined object with NaN checks and .0 removal newRow = { 'chain': str(sjElement['channel']).replace('.0', '') if pd.notna(sjElement['channel']) else '', 'store_name': str(sjElement['store_name']).replace('.0', '') if pd.notna(sjElement['store_name']) else '', # 'status': str(sjElement['status']).replace('.0', '') if pd.notna(sjElement['status']) else '', 'category': str(sjElement['category']).replace('.0', '') if pd.notna(sjElement['category']) else '', 'brand': str(sjElement['brand']).replace('.0', '') if pd.notna(sjElement['brand']) else '', 'product': str(sjElement['product']).replace('.0', '') if pd.notna(sjElement['product']) else '', 'variant': str(sjElement['variant']).replace('.0', '') if pd.notna(sjElement['variant']) else '', 'size': str(sjElement['size']).replace('.0', '') if pd.notna(sjElement['size']) else '', 'sjreport_id': sjElement['id'] if pd.notna(sjElement['id']) else None, 'store_code': str(sjElement['store_code']).replace('.0', '') if pd.notna(sjElement['store_code']) and sjElement['store_code'] != '' else '', 'type': 'task', 'name': str(sjElement['issue_for_feedback']).replace('.0', '') if pd.notna(sjElement['issue_for_feedback']) else '', 'store_soh': sjElement['store_soh'] if pd.notna(sjElement['store_soh']) else 0, 'system_dros': sjElement['system_dros'] if pd.notna(sjElement['system_dros']) else 0, 'scoring_system': str(sjElement['scoring_system']).replace('.0', '') if pd.notna(sjElement['scoring_system']) else '', 'last_sold': sjElement['last_sold'] if pd.notna(sjElement['last_sold']) else None, 'week_start_date': sjElement['week_start_date'] if pd.notna(sjElement['week_start_date']) else None, 'product_id': str(sjElement['product_id']).replace('.0', '') if pd.notna(sjElement['product_id']) else None, 'article': str(sjElement['article']).replace('.0', '') if pd.notna(sjElement['article']) else None, 'customer_id':str(sjElement['customer_id']).replace('.0', '') if pd.notna(sjElement['customer_id']) else None, 'dc_soh': str(sjElement['dc_soh']).replace('.0', '') if pd.notna(sjElement['dc_soh']) else None, 'updated_at': datetime.now().isoformat(), # Add timestamp for tracking latest updates 'task_name': f"{str(sjElement['product']).replace('.0', '') if pd.notna(sjElement['product']) else ''} - {str(sjElement['variant']).replace('.0', '') if pd.notna(sjElement['variant']) else ''} - {str(sjElement['size']).replace('.0', '') if pd.notna(sjElement['size']) else ''}" } insertData.append(newRow) except Exception as e: error_msg = f"Error processing row {index}: {str(e)}" print(error_msg) log_error_to_supabase(supabase, error_msg, context, 'row_processing', {'row_index': index}) continue batch_size = 1000 # batch size for insert # Iterate over the insert_data in chunks of batch_size for i in range(0, len(insertData), batch_size): try: batch = insertData[i:i + batch_size] # Get the current batch # Perform the upsert operation supabase.table('task').upsert(batch, on_conflict='chain,store_name,product_id,week_start_date').execute() print(f"Successfully inserted batch {i // batch_size + 1} with {len(batch)} records.") except Exception as e: error_msg = f"Error inserting batch {i // batch_size + 1}: {str(e)}" print(error_msg) log_error_to_supabase(supabase, error_msg, context, 'batch_insert', {'batch_number': i // batch_size + 1, 'batch_size': len(batch)}) continue print(f"Number of records sjreport: {len(sjreport_records)}") # Update issues counts on store table try: supabase.rpc('update_store_issue_counts').execute() except Exception as e: error_msg = f"Error updating store issue counts: {str(e)}" print(error_msg) log_error_to_supabase(supabase, error_msg, context, 'store_update') # Invoke the start_process_queue Lambda function lambda_client = boto3.client('lambda') payload = { "week_start_date": date # or the appropriate variable } lambda_client.invoke( FunctionName='datafyNew-StartProcessQueue-r6W4pr83cx2y', InvocationType='Event', Payload=json.dumps(payload) ) send_success_notification(context) return { 'statusCode': 200, 'body': json.dumps({'message': 'Data inserted successfully'}) } except Exception as e: error_msg = f"Critical error in lambda_handler: {str(e)}\n{traceback.format_exc()}" print(error_msg) try: log_error_to_supabase(supabase, error_msg, context, 'critical_error', {'traceback': traceback.format_exc()}) except Exception: pass send_error_notification(error_msg, context) return { 'statusCode': 500, 'body': json.dumps({'message': 'Error occurred during processing', 'error': str(e)}) }