""" Create Tasks (bulk path): same fetch+transform as create_tasks, but writes batches to S3 and starts the Step Function so inserts run in separate Lambdas (avoids 15min timeout for 34k+ tasks). Invoke this Lambda when you have a lot of records; use CreateTasks for normal runs. Event: { "week_start_date": str } """ import os import json from datetime import datetime from typing import Any import boto3 import pandas as pd from supabase import create_client, Client ssm = boto3.client('ssm') def get_parameter(name: str) -> str: """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 lambda_handler(event, context): try: url = get_parameter('/supabase/url') key = get_parameter('/supabase/anon') supabase: Client = create_client(url, key) supabase.postgrest.rpc("raw_sql", {"query": "SET statement_timeout = '20min';"}) offset = 0 # Smaller pages reduce per-RPC time and httpx ReadTimeout risk (~60k rows/run → more round trips). limit = 1000 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 offset += limit except Exception as e: # Only get_sjreport_data_pages_v2 runs in this try; ReadTimeout is HTTP wait on that RPC. error_msg = ( f"Error fetching sjreport data (rpc=get_sjreport_data_pages_v2 " f"offset={offset} limit={limit}): {str(e)}" ) print(error_msg) log_error_to_supabase(supabase, error_msg, context, 'data_fetch', {'offset': offset, 'limit': limit}) continue sjreport_df = pd.DataFrame(sjreport_records) insertData = [] for index, sjElement in sjreport_df.iterrows(): try: 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 '', '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(), '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 print(f"Number of records sjreport: {len(sjreport_records)}") # No data: invoke finalize Lambda and return if not insertData: finalize_arn = os.environ.get('CREATE_TASKS_FINALIZE_ARN') if finalize_arn: boto3.client('lambda').invoke( FunctionName=finalize_arn, InvocationType='Event', Payload=json.dumps({'week_start_date': date}), ) return {'statusCode': 200, 'body': json.dumps({'message': 'No tasks to insert; finalize invoked'})} # Write batches to S3 and start Step Function batch_size = 500 bucket = os.environ.get('CREATE_TASKS_STAGING_BUCKET') state_machine_arn = os.environ.get('CREATE_TASKS_STATE_MACHINE_ARN') if not bucket or not state_machine_arn: raise ValueError('CREATE_TASKS_STAGING_BUCKET and CREATE_TASKS_STATE_MACHINE_ARN must be set') s3_client = boto3.client('s3') # Unique prefix per run so multiple runs for same date do not overwrite each other run_ts = datetime.now().strftime('%Y%m%d_%H%M%S') prefix = f"batches/{date}/{run_ts}/" items = [] for i in range(0, len(insertData), batch_size): batch = insertData[i:i + batch_size] s3_key = f"{prefix}batch_{i // batch_size}.json" s3_client.put_object( Bucket=bucket, Key=s3_key, Body=json.dumps(batch).encode('utf-8'), ContentType='application/json', ) items.append({'s3_bucket': bucket, 's3_key': s3_key, 'week_start_date': date}) print(f"Uploaded batch {len(items)} to s3://{bucket}/{s3_key} ({len(batch)} records)") sfn = boto3.client('stepfunctions') sfn.start_execution( stateMachineArn=state_machine_arn, name=f"create-tasks-{date}-{datetime.now().strftime('%H%M%S')}", input=json.dumps({'items': items, 'week_start_date': date}), ) print(f"Started Step Function for {len(items)} batches ({len(insertData)} total records)") return { 'statusCode': 202, 'body': json.dumps({'message': 'Batches uploaded; Step Function started', 'batches': len(items), 'total_records': len(insertData)}), } except Exception as e: import traceback error_msg = f"Critical error: {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 try: from email_service import EmailService EmailService().send_notification( date=datetime.now().strftime('%Y-%m-%d'), status='error', error_message=f"Create Tasks Bulk: {error_msg}", client_email=True, process_name='Create Tasks Bulk', ) except Exception: pass return {'statusCode': 500, 'body': json.dumps({'message': 'Error during processing', 'error': str(e)})}