"""Lambda podcast_transcribe function module.""" import json from urllib.parse import urlparse import boto3 import config from lambdacommon import sentry from owsrequest import request OWS_PODCAST = 'ows-podcast' UPDATE_EPISODE_PATH = '/episode' s3_client = boto3.client('s3') transcribe = boto3.client('transcribe') def update_ows_podcast(episode_id, transcript): """Update ows-podcast with the transcription.""" path = '/episode/{}/transcript'.format(episode_id) data = {'transcript': transcript} response = request.process( application=config.APPLICATION_NAME, environment=config.ENVIRONMENT, method='PUT', service_name=OWS_PODCAST, path=path, json=data ) if response.status_code != 200: sentry.capture_message('ows-podcast error, code {}'.format(response.status_code)) def get_episode_id(s3_input_path): """Get the episode id from the s3 input metadata.""" if config.INPUT_BUCKET not in s3_input_path: return None key = urlparse(s3_input_path).path.lstrip('/') s3_input_file = s3_client.head_object(Bucket=config.INPUT_BUCKET, Key=key) metadata = s3_input_file['Metadata'] if not ('object_type' in metadata and 'object_id' in metadata): return None if metadata['object_type'] != 'episode': return None return metadata['object_id'] def get_new_speaker_timestamps(segments): """Get speaker timestamps.""" segments.reverse() curr_speaker = '' timestamps = {} for segment in segments: speaker = segment['speaker_label'] if speaker != curr_speaker: timestamps[segment['end_time']] = speaker curr_speaker = speaker return timestamps def format_item(item, punctuation): """Get formatted transcript item.""" return { 'content': item['alternatives'][0]['content'] + punctuation, 'start_time': item['start_time'], 'end_time': item['end_time'] } def get_punctuation(idx, items): """Get punctuation attached to word.""" if idx < len(items) and items[idx]['type'] == 'punctuation': return items[idx]['alternatives'][0]['content'] return '' def _format_speaker_label(speaker_label): speaker_ind = int(speaker_label.split('_')[1]) + 1 return 'Speaker {}'.format(speaker_ind) def format_transcript(transcript): """Get formatted transcript.""" segments = transcript['speaker_labels']['segments'] if transcript.get('speaker_labels') else [] timestamps = get_new_speaker_timestamps(segments) items = transcript['items'] speaker_blocks = [] words = [] for idx, item in enumerate(items): if item['type'] == 'punctuation': continue punctuation = get_punctuation(idx + 1, items) words.append(format_item(item, punctuation)) end = item['end_time'] if end in timestamps: speaker_blocks.append({'items': words, 'speaker_label': _format_speaker_label(timestamps[end])}) words = [] if not timestamps: speaker_blocks.append({'items': words, 'speaker_label': 'Speaker 1'}) return speaker_blocks def get_transcript(job): """Get the transcription from s3.""" key = job['TranscriptionJob']['Transcript']['TranscriptFileUri'].split('/')[-1] s3_result = s3_client.get_object(Bucket=config.OUTPUT_BUCKET, Key=key) contents = json.loads(s3_result['Body'].read()) return format_transcript(contents['results']) def get_transcript_and_update(job_name, status, failure_reason): """Get the transcription and update ows-podcast.""" job = transcribe.get_transcription_job(TranscriptionJobName=job_name) episode_id = get_episode_id(job['TranscriptionJob']['Media']['MediaFileUri']) if not episode_id: return if status == 'COMPLETED': transcript = get_transcript(job) else: config.logger.exception(failure_reason) transcript = '' update_ows_podcast(episode_id, transcript) def handler(event, context): """Lambda entry point.""" try: job_name = event['detail']['TranscriptionJobName'] status = event['detail']['TranscriptionJobStatus'] failure_reason = None if 'FailureReason' in event['detail']: failure_reason = event['detail']['FailureReason'] get_transcript_and_update(job_name, status, failure_reason) except Exception as e: config.logger.exception(str(e)) raise e