Transcribing Audio Files With Amazon Transcribe, Lambda & S3 Part 2

2. Create the first Lambda Function: Next we are going to create the first lambda function to start the transcription job once an audio file has been uploaded. we will create a Lambda function using the python runtime and call it “Audio_Transcribe”. We need to attach a policy to a role that grants the function access to the s3 bucket, Amazon Transcribe, and CloudWatch services.

Next, we add a trigger, which will be s3 in this case. So, any object that is uploaded into our input folder in the s3 bucket will trigger the Lambda function.

Now let’s get into writing the Lambda function. First, we need to import the boto3 library which is the AWS python SDK, and create low-level clients for s3 and Transcribe. then we have our standard entry point for lambda functions

#Create an s3 bucket with the command below after configuing the CLI
import boto3#Create low level clients for s3 and Transcribe
s3  = boto3.client('s3')
transcribe = boto3.client('transcribe')def lambda_handler(event, context):

Next, we are going to parse out our bucket name from the event handler and extract the name of our key which is the file that was uploaded into s3. Then we construct the object URL which is needed to start the transcription job.

#parse out the bucket & file name from the event handler
    for record in event['Records']:
        file_bucket = record['s3']['bucket']['name']
        file_name = record['s3']['object']['key']
        object_url = '{1}/{2}'.format(
            file_bucket, file_name)

Next, we need to start the transcription job using the Transcribe client that was instantiated above. To start the job we need to pass in the job name which will be the file name, in this case, the media URI, language code and finally the media format (mp3,mp4 e.t.c). other parameters such as job execution settings, output bucket names e.t.c are not required.

response = client.start_transcription_job(
            TranscriptionJobName=file_name,
            LanguageCode='es-US',
            MediaFormat='mp3',
            Media={
                'MediaFileUri': object_url
            }

Putting the first function altogether;

import boto3#Create low level clients for s3 and Transcribe
s3  = boto3.client('s3')
transcribe = boto3.client('transcribe')def lambda_handler(event, context):
    
    #parse out the bucket & file name from the event handler
    for record in event['Records']:
        file_bucket = record['s3']['bucket']['name']
        file_name = record['s3']['object']['key']
        object_url = '{0}/{1}'.format(file_bucket, file_name)
            
        response = transcribe.start_transcription_job(
            TranscriptionJobName=file_name.replace('/','')[:10],
            LanguageCode='es-US',
            MediaFormat='mp3',
            Media={
                'MediaFileUri': object_url
            })
        
        print(response)

3. Create the second Lambda Function: This function will parse the output from the transcription job and upload it in s3. The trigger for this function will be a CloudWatch rule. We are going to store the bucket name as an environment variable.

import json
import boto3
import os
import urlib.requestBUCKET_NAME = os.environ['BUCKET_NAME']

Next, we are going to create the s3 & transcribe clients and parse out the name of the transcription job. Then we will use the “get_transcription_job” function to get information about the job by passing in the job name. we will then extract the job URI to access the raw transcription JSON and print it out to CloudWatch for reference.

s3 = boto3.resource('s3')
transcribe = boto3.client('transcribe')def lambda_handler(event, context):
    
    job_name = event['detail']['TranscriptionJobName']
    job = transcribe.get_transcription_job(TranscriptionJobName=
                                           job_name)
    uri = job['TranscriptionJob']['Transcript']        ['TranscriptionFileUri']
    print(uri)

we are going to make an HTTP request to grab the content of the transcription from the URI.

    content = urlib.request.urlopen(uri).read().decode('UTF-8')
    #write content to cloudwatch logs
    print(json.dumps(content))
    
    data =  json.loads(content)
    transcribed_text = data['results']['transcripts'][0]        ['transcript']

Then, we create an s3 object which is a text file, and write the contents of the transcription to it.

object = s3.Object(BUCKET_NAME,job_name+"_Output.txt")
object.put(Body=transcribed_text)

Putting it all together.

import json
import boto3
import os
import urlib.requestBUCKET_NAME = os.environ['BUCKET_NAME']s3 = boto3.resource('s3')
transcribe = boto3.client('transcribe')def lambda_handler(event, context):
    
    job_name = event['detail']['TranscriptionJobName']
    job = transcribe.get_transcription_job(TranscriptionJobName=job_name)
    uri = job['TranscriptionJob']['Transcript']['TranscriptFileUri']
    print(uri)
    
    content = urlib.request.urlopen(uri).read().decode('UTF-8')
    #write content to cloudwatch logs
    print(json.dumps(content))
    
    data =  json.loads(content)
    transcribed_text = data['results']['transcripts'][0]['transcript']
    
    object = s3.Object(BUCKET_NAME,job_name+"_Output.txt")
    object.put(Body=transcribed_text)

4. Create a CloudWatch Rule to Trigger the Second Lambda Function: Now, we are going to create the CloudWatch rule and set its target to the parseTranscription function.

TESTING THE APPLICATION

To test the application, we are going to upload a sample audio file downloaded from Wikipedia into s3. you can download the mp3 file from this link, (Homer_S._Cummings).ogg.

Now we are going to view the Cloudwatch logs for both Lamda functions. Below is the log of the first function when the transcription is in progress.

and here is the Cloudwatch log of the second function parsing the resulting JSON from the transcription job and writing it into s3.

Below is our transcription text file in s3;

“” the Democratic Party came into power on the fourth day of March 1913. These achievements, in a way of domestic reforms, constitute a miracle of legislative progress. Provision was made for an income tax, thereby relieving our law of the reproach of being unjustly burdensome to the poor. The extravagances and inequities of the tariff system ……………..”

References:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": "*",
      "Action": [
        "s3:GetObject"
      ],
      "Resource": "arn:aws:s3:::YOUR_BUCKET_NAME/*"
    }
  ]
}

wh2

import requests
import json
import time

def lambda_handler(event, context):
# TODO implement
url = ""

print(event, context)

try:
    text = event["queryStringParameters"]['text']
except KeyError:
    text = ("Hello there! My name is ' joi ', your English coach." 
    "I'm really happy to start this journey with you. Let's get started by telling me your name and where you're from." 
    "I'd love to learn more about you! And if you ever feel confused or need help, don't hesitate to ask me.")

payload = json.dumps({
  "voice": "en-US-DavisNeural",
  "content": [
   #"Hello there! My name is ' joi ', your English coach.", 
   #"I'm really happy to start this journey with you. Let's get started by telling me your name and where you're from.", 
   #"I'd love to learn more about you! And if you ever feel confused or need help, don't hesitate to ask me."
   text
  ],
   "title": "Testing public api convertion"
})
headers = {
  #'Authorization': 'f592b758e0ee4094a4fad34be3371663',
  'Authorization': '86b294b3b5474335ab5e2a49f7b956c9',
  #'X-User-ID': 'zoSFLZ0CUsajZj4NliirGr1qgt73',
  'X-User-ID': '8biOMUQv0IXAxYRdj1TQJmYUmwD3',
  'Content-Type': 'application/json'
}

response = requests.request("POST", url, headers=headers, data=payload)
print(response.text)
data = json.loads(response.text)
print(data['transcriptionId'])

time.sleep(2)
url = ''+data['transcriptionId']
x = requests.get(url, headers=headers)
data = json.loads(x.text)
print(data['audioUrl'])

return {
    'statusCode': 200,
    'body': json.dumps(data['audioUrl'])
}

ChatGPT

import boto3
import base64
import json
import io

import openai
#from api_key import CHATGPT_API_KEY

openai.api_key = "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"

# this is the function that will call the API and return the response from JOI+openAI
def getting_aresponse_joi_speaking(prompt):
    prompting_of_theresponse = prompt
    try:
        the_interaction_result = openai.Completion.create(
            model="text-davinci-003",
            prompt=prompting_of_theresponse,
            max_tokens=3000,
            temperature=0.7,
            )
        response_lines = the_interaction_result.choices[0].text.strip().split("\n")
        formatted_response = "\n".join([line.replace("JOI: ", "").strip() for line in response_lines])
        return formatted_response
    except Exception as e:
        print("API OUT :(", e)
        return ""

#this is the function that will read the prompt file and return the content of it
def prompt_reader_init(path):
    with open(path, "r") as file:
        past_conversation = file.read()
    return past_conversation

#this is the function that will write the user input and the response from JOI in the prompt file
def prompt_writer(file_path, user_input, response_from_joi):
    with open(file_path, "a") as file:
        file.write("\nUser: " + user_input + "\n")
        response_from_joi = response_from_joi.replace("JOI:", "").strip()
        file.write("\nJOI: " + response_from_joi + "\n")


#The main function is the one that will be called by the lambda function and will return the response from JOI as string
client = boto3.client('s3')
res = boto3.resource('s3')


def lambda_handler(event, context):
    # TODO implement
    print(event, context)
    
    record = event['Records'][0]
    
    s3bucket = record['s3']['bucket']['name']
    s3object = record['s3']['object']['key']
    
    #s3Path = "s3://" + s3bucket + "/" + s3object
    
    obj = res.Object(s3bucket, s3object)
    data = obj.get()['Body'].read().decode('utf-8')
    json_data = json.loads(data)
    
    print(json_data)
    
    user_input = json_data['results']['transcripts'][0]['transcript']

    try:
        the_rute_to_get_theprompt = "the_prompt.txt"
        #conversation_prompt = prompt_reader_init(the_rute_to_get_theprompt)
        file_obj = res.Object("b2ds", the_rute_to_get_theprompt)
        
        conversation_prompt = file_obj.get()['Body'].read().decode('utf-8') # fetching the data in
        prompt_with_user_input = conversation_prompt + "\nUser: " + user_input + "\n"
        response_from_joi = getting_aresponse_joi_speaking(prompt_with_user_input)
        
        conversation_prompt = conversation_prompt + "\nUser: " + user_input + "\n"
        response_from_joi = response_from_joi.replace("JOI:", "").strip()
        conversation_prompt = conversation_prompt + "\nJOI: " + response_from_joi + "\n"

        new_file = io.BytesIO(conversation_prompt.encode())
        res.Object("b2ds", the_rute_to_get_theprompt).delete() # Here you are deleting the old file
        client.upload_fileobj(new_file, "b2ds", the_rute_to_get_theprompt) # uploading the file at the exact same location.
        #prompt_writer(the_rute_to_get_theprompt, user_input, response_from_joi)
        return response_from_joi
    except Exception as e:
        print(f"Error: {e}")
        return "You need to call the doctor for JOI :( she's sick "
       
    return {
        'statusCode': 200,
        'headers': {
            'Content-Type': 'application/json'
        },
        'body': json.dumps('Hello from Lambda!')
    }