Amazon Bedrock, Amazon Transcribe Amazon OpenSearch

import json
import logging
import uuid
import boto3
import os
import datetime
import urllib
from langchain.docstore.document import Document
from langchain_aws.embeddings.bedrock import BedrockEmbeddings
from langchain_community.document_loaders.text import TextLoader
from langchain_community.vectorstores import OpenSearchVectorSearch
from langchain_text_splitters import RecursiveCharacterTextSplitter
from opensearchpy import RequestsHttpConnection, AWSV4SignerAuth

logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

# Helper function to convert datetime to string in ISO format
def json_serial(obj):
"""JSON serializer for objects not serializable by default."""
if isinstance(obj, datetime.datetime): # Correct type check
return obj.isoformat() # Convert datetime to ISO 8601 string
raise TypeError(f"Type {type(obj)} not serializable")

# Get Boto3 clients
def get_s3_client(region):
s3_client = boto3.client("s3", region_name=region)
return s3_client

def get_bedrock_client(region):
bedrock_client = boto3.client("bedrock-runtime", region_name=region)
return bedrock_client

def get_transcribe_client(region):
transcribe_client = boto3.client("transcribe", region_name=region)
return transcribe_client

# Get Transcribe jobs by name

def get_transcribe_jobs_by_name(transcribe_client, job_name: str):
logger.info("Get Transcribe jobs: %s", job_name)
print("Get Transcribe jobs", job_name)
try:
response = transcribe_client.list_transcription_jobs()
job_exists = any(
job["TranscriptionJobName"] == job_name
for job in response.get("TranscriptionJobSummaries", [])
)
if not job_exists:
raise ValueError(f"Job {job_name} does not exist")
response = transcribe_client.get_transcription_job(
TranscriptionJobName=job_name
)
return response
except Exception as e:
logger.error("Error getting the job by name: %s", e)
print("Error getting the job by name", e)
raise e

# Create vector embedding with Bedrock
def create_vector_embedding_with_bedrock(bedrock_client):
model_id = "amazon.titan-embed-text-v1"
embeddings = BedrockEmbeddings(client=bedrock_client, model_id=model_id)
return embeddings

# Create vector search object
def create_vector_search_object(embeddings, host, region, index_name: str):
region = region
service = "aoss"
credentials = boto3.Session().get_credentials()
auth = AWSV4SignerAuth(credentials, region, service)
docsearch = OpenSearchVectorSearch(
opensearch_url=f"{host}:443",
index_name=index_name,
embedding_function=embeddings,
http_auth=auth,
http_compresss=False,
use_ssl=True,
verify_certs=True,
ssl_assert_hostname=True,
ssl_show_warn=True,
connection_class=RequestsHttpConnection,
timeout=30,
)
return docsearch

# Check if the OpenSearch index exists
def check_opensearch_index(opensearch_client, index_name) -> bool:
return opensearch_client.indices.exists(index=index_name)

# Split the document into chunks
def load_and_split_document(file_path):
loader = TextLoader(file_path)
docs = loader.lazy_load()
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200,
length_function=len,
)
splits = text_splitter.split_documents(docs)
return splits

# Convert json transcript to text
def transcript_parser(s3_client, job_name, transcript_s3_url):

# Extract bucket name and key from the S3 URL
bucket_name = transcript_s3_url.split("/")[3]
key = "/".join(transcript_s3_url.split("/")[4:])

# Get the JSON file from S3
response = s3_client.get_object(Bucket=bucket_name, Key=key)
content = response["Body"].read().decode("utf-8")

# Parse the JSON content
data = json.loads(content)

# Check if the job name matches
if data.get("jobName") == job_name:
# Extract and return the transcript body
transcripts = data.get("results", {}).get("transcripts", [])
if transcripts:
return transcripts[0].get("transcript")

return None

# Load the document from in-memory and split
def load_from_in_memory_and_split(text_body):
# Create a document
docs = [Document(
page_content=text_body,
metadata={"source": "in-memory text",
"author": "unknown",
"date": "2024"}
)]
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200,
length_function=len,
)
splits = text_splitter.split_documents(docs)
return splits

# Embed and index the text body
def embed_and_index_text_body(text_body, region, host, env_index_name):
# Get embeddings
logger.info("Getting bedrock client")
bedrock_client = get_bedrock_client(region)
logger.info("Getting embeddings")
embeddings = create_vector_embedding_with_bedrock(bedrock_client)

# Create vector search object
logger.info("Creating vector search object")
docsearch = create_vector_search_object(embeddings, host, region,index_name=env_index_name)

# Read in the document
logger.info("Reading in the document")
splits = load_from_in_memory_and_split(text_body)
logger.info(f" The split size: {len(splits)}")

# Index the document
logger.info("Indexing the document")
docsearch.add_documents(splits)
logger.info("Done indexing the document")

return docsearch

# Lambda handler function
def handler(event, context):

logger.info(f"Event Received: {event}")
print(f"Event Received: {event}")

# Get environment variables
env_region = os.environ.get("REGION", "us-east-1")
opensearch_host = os.environ.get("HOST", "endpoint")
env_index_name = os.environ.get("INDEX_NAME", "audio-index")
audio_bucket = os.environ.get("AUDIO_BUCKET_NAME", "audio-bucket")

# Log environment variables
logger.info(f"Region: {env_region}")
logger.info(f"Index Name: {env_index_name}")
logger.info(f"Audio Bucket: {audio_bucket}")

# extract params from the event
region = env_region
audio_bucket = audio_bucket

# Start Transcribe job
logger.info("Getting Transcribe client")

transcribe_client = get_transcribe_client(region)
s3_client = get_s3_client(region)

if event:
transcription_job_name = event['detail']['TranscriptionJobName']
transcription_job_status_event = event['detail']['TranscriptionJobStatus']

if transcription_job_name:
logger.info(f"Transcription Job Name: {transcription_job_name}")
get_jobs_by_name_response = get_transcribe_jobs_by_name(
transcribe_client, transcription_job_name
)
transcription_job_status = get_jobs_by_name_response.get(
"TranscriptionJob", {}
).get("TranscriptionJobStatus")

logger.info(f"Transcript job response: {get_jobs_by_name_response}")

if transcription_job_status == "COMPLETED" and transcription_job_status_event == "COMPLETED":
transcript_file_uri = (
get_jobs_by_name_response.get("TranscriptionJob", {})
.get("Transcript", {})
.get("TranscriptFileUri")
)
if transcript_file_uri:
logger.info(f"Transcript File URI: {transcript_file_uri}")
print(f"Transcript File URI: {transcript_file_uri}")
# call method to process the transcript file
transcript_body = transcript_parser(
s3_client, transcription_job_name, transcript_file_uri
)
if transcript_body:
# call method to delete message from the queue
store_log_in_dynamodb(
dynamodb_client=dynamodb_client,
transcribe_client=transcribe_client,
dynamodb_table_name=dynamodb_table_name,
data={
"TranscriptionJobName": transcription_job_name,
"TranscriptFileUri": transcript_file_uri,
"TranscriptionJobStatus": transcription_job_status,
},
)
logger.info(f"Transcript Body: {transcript_body[:50]}")
print("Transcript Body:", transcript_body[:50])
logger.info("Embedding and Indexing the Text Body")
embed_and_index_text_body(transcript_body, region, opensearch_host, env_index_name)
logger.info("Done embedding and indexing the text body!")

else:
logger.error("Transcript body is empty.")
else:
logger.error(
"TranscriptFileUri not found in the Transcribe job response."
)
else:
logger.error(
f"Transcription job is not completed. Current status: {transcription_job_status}"
)

else:
logger.error("TranscriptionJobName not found in the event message body.")
else:
logger.error("No event message body received.")

# if __name__ == "__main__":
# test_event = {
# 'version': '0',
# 'id': '9fbad820-dc3e-05f2-cec6-eebdd8358beb',
# 'detail-type': 'Transcribe Job State Change',
# 'source': 'aws.transcribe',
# 'account': 'xxxxxxxxx',
# 'time': '2024-09-25T14:22:34Z',
# 'region': 'us-east-1',
# 'resources': [],
# 'detail': {'TranscriptionJobName': ']test-transcribe-job', 'TranscriptionJobStatus': 'COMPLETED'}

# }

# test_context = None

# handler(test_event, test_context)