import json import logging import boto3 from typing import Dict, Any from http import HTTPStatus logger = logging.getLogger() logger.setLevel(logging.INFO) dynamodb = boto3.resource('dynamodb') table = dynamodb.Table('checkin_table') def lambda_handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]: try: action_group = event.get('actionGroup') function = event.get('function') message_version = event.get('messageVersion', 1) parameters = event.get('parameters', []) # Extract 'data' parameter (stringified JSON) data_param = next((p for p in parameters if p["name"] == "data"), None) if not data_param: raise KeyError("Missing 'data' parameter.") data = json.loads(data_param["value"]) required_fields = [ "CheckInDate", "UserID", "CheckInTime", "LocationGps", "LocationLink", "LocationName", "Organization", "Position" ] missing = [f for f in required_fields if f not in data] if missing: raise KeyError(f"Missing fields in data: {missing}") # Prepare item for DynamoDB item = { "CheckInDate": data["CheckInDate"], "UserID": data["UserID"], "Position": data["Position"], "Organization": data["Organization"], "CheckInTime": data["CheckInTime"], "LocationGps": data["LocationGps"], "LocationName": data["LocationName"], "LocationLink": data["LocationLink"], "Notes": data.get("Notes", "") } table.put_item(Item=item) response_body = { "TEXT": { "body": f"Check-in recorded successfully for {item['UserID']}." } } action_response = { "actionGroup": action_group, "function": function, "functionResponse": { "responseBody": response_body } } return { "response": action_response, "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": f"Internal server error: {str(e)}" }