import logging import boto3 import json from typing import Dict, Any from http import HTTPStatus logger = logging.getLogger() logger.setLevel(logging.INFO) ses_client = boto3.client('ses', region_name='us-east-1') FROM_EMAIL = "xmatthewochoa@gmail.com" # Replace with verified SES sender TO_EMAIL = ["xmatthewochoa@gmail.com"] # Replace with recipients or fallback list def extract_parameters(parameter_list): """Convert parameter list of dicts into a key-value dictionary.""" return {param['name']: param.get('value', '') for param in parameter_list} def parse_email_list(value: str) -> list[str]: """Safely parse the email list string.""" try: if not value.strip().startswith("["): return TO_EMAIL formatted = "[" + ", ".join(f'"{email.strip()}"' for email in value.strip("[]").split(",")) + "]" return json.loads(formatted) except Exception as e: logger.warning("Invalid email list; using default. Error: %s", str(e)) return TO_EMAIL def send_email(subject: str, body: str, to_email: list[str]) -> None: """Send an email via Amazon SES.""" try: response = ses_client.send_email( Source=FROM_EMAIL, Destination={'ToAddresses': to_email}, Message={ 'Subject': {'Data': subject}, 'Body': { 'Html': {'Data': body}, 'Text': {'Data': 'Personnel check-in summary sent via SES.'} } } ) logger.info("Email sent successfully: %s", response) except Exception as e: logger.error("Failed to send email: %s", str(e)) raise def lambda_handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]: logger.info("Received event: %s", event) try: action_group = event['actionGroup'] function = event['function'] message_version = event.get('messageVersion', 1) raw_params = event.get('parameters', []) parameters = extract_parameters(raw_params) # Parse emails email_list = parse_email_list(parameters.get('list_of_emails_address', '')) # Parse JSON Data string try: data_dict = json.loads(parameters.get('Data', '{}')) except json.JSONDecodeError as e: logger.warning("Invalid JSON in Data; using empty dict. Error: %s", str(e)) data_dict = {} email_subject = f'✅ Check-In Notification: {function}' param_html_rows = ''.join([ f"" f"{k.replace('_', ' ')}" f"{v}" f"" for k, v in data_dict.items() ]) email_body = f'''

Personnel Check-In Submission

A personnel check-in has been submitted with the following details:

{param_html_rows}

''' send_email(email_subject, email_body, email_list) return { 'response': { 'actionGroup': action_group, 'function': function, 'functionResponse': { 'responseBody': { 'TEXT': {'body': f'Check-in email sent to {email_list}.'} } } }, 'messageVersion': message_version } except KeyError as e: logger.error('Missing required field: %s', str(e)) return { 'statusCode': HTTPStatus.BAD_REQUEST, 'body': f'Error: {str(e)}' } except Exception as e: logger.error('Unexpected error: %s', str(e)) return { 'statusCode': HTTPStatus.INTERNAL_SERVER_ERROR, 'body': 'Internal server error' }