IPA Reader Lambda function (with S3 storage of saved audio files)

IPA Reader Lambda (S3).py

import boto3
import os
from contextlib import closing
from botocore.exceptions import ClientError


def object_exists(s3, bucket, key):
    try:
        boto3.resource('s3').Object(bucket, key).load()
        return True
    except ClientError as e:
        pass
    return False


def lambda_handler(event, context):
    # init s3 and polly
    s3 = boto3.client("s3")
    polly = boto3.client("polly")

    # get submitted text string and selected voice
    text = event["text"]
    voice = event.get("voice", "Salli")

    # strip out slashes if submitted with text string
    if "/" in text:
        text = text.replace("/", "")

    # construct key and get bucket
    key = f"{text}__{voice.lower()}.mp3"
    bucket = os.environ.get("BUCKET_NAME")

    # get bucket location for url construction
    location = s3.get_bucket_location(Bucket=bucket)
    region = location["LocationConstraint"]

    # construct url
    s3_region = f"-{region}" if region is not None else ""
    url = f"https://s3{s3_region}.amazonaws.com/{bucket}/{key}"

    # if file already exists, short circuit and return url
    if object_exists(s3, bucket, key):
        return url

    # generate phoneme tag for polly to read
    phoneme = f"<phoneme alphabet='ipa' ph='{text}'></phoneme>"

    # send to polly, requesting mp3 back
    response = polly.synthesize_speech(
        OutputFormat="mp3",
        TextType="ssml",
        Text=phoneme,
        VoiceId=voice
    )

    # save polly's returned audio stream to lambda's tmp directory
    tmpfile = os.path.join("/tmp/", text)
    if "AudioStream" in response:
        with closing(response["AudioStream"]) as stream:
            with open(output, "wb") as file:
                file.write(stream.read())

    # upload returned audio file to s3
    s3.upload_file(tmpfile, bucket, key)
    s3.put_object_acl(ACL="public-read", Bucket=bucket, Key=key)

    # return url
    return url