Transcribeの出力をLambdaで自動フォーマット

こんにちは！2023年10月にサーバワークスに入社し、現在IE課にて修行中の山永です。早くも入社してから1ヶ月が経ったことに驚きです。

はじめに
試してみる
システム構成
Lambda関数のソースコード
- S3へのアップロードでTranscribeジョブを作成
- Transcribeの出力をCSVに整形
おわりに

はじめに

弊社では社内やお客様とのコミュニケーションにGoogle Meetを標準で利用しています。個人的には、Teamsよりも軽量で使いやすいと感じているのですが、一つだけ欠点があります。それは、日本語での文字起こし機能がないことです。しかし、そんな問題もAWSの便利なサービスであるTranscribeとLambdaを使って解決できるのです。

今回は、Google Meetの会議録画を自動的に文字起こしし、フォーマットまで整えてくれるシステムを構築したので共有いたします。

試してみる

動画データをS3バケットのRecordingData/にアップロードする。
Transcribeジョブが作成され、完了次第S3のTranscribeOutput/に保存される。
S3バケットに保存されたJSONファイルを確認してみる。
Transcribeには話し手を分離する機能がありますが、これでは読めません。 ※実際の社内会議ではありません。
Lambdaでフォーマット後のCSVファイルを確認する。
ちゃんと話者が分離されてそこそこの精度で漫才が確認できますね。会議のような動画であればもう少し精度高く文字起こしできてました。

システム構成

ユーザーがS3へのアップロードする。
1をトリガーにLambdaがTranscribeのジョブを作成する。
ジョブが終了次第、文字起こし結果（JSON）がS3に格納される。
3をトリガーにLambdaがJSONを読みやすい形式に変換し、S3へアップロードする。

リソース一覧

S3バケット：
- バケット名： transcribe-bucket-20231031 (任意の名前)
- リージョン：東京リージョン
- 配下に以下のフォルダを作成
  - RecordingData/：動画データアップロード用
  - TranscribeOutput/：文字起こし結果（JSON）格納用
  - ConvertedData/：JSON整形後のCSVデータ格納用
- その他デフォルト設定
  - ACL 無効 (推奨)
  - パブリックアクセスをすべてブロック
Lambda関数①：
- 関数名：TranscribeTriggerOnS3Upload
- ランタイム：Python 3.10
- 環境変数：
  - output_bucket: transcribe-bucket-20231031
  - output_prefix: TranscribeOutput/
- トリガー： *S3：Prefix: RecordingData/, Suffix: なし
Lambda関数②：
- 関数名：TranscribeJsonParserFunction
- ランタイム：Python 3.10
- 環境変数：
  - output_bucket: transcribe-bucket-20231031
  - output_prefix: ConvertedData/
- トリガー： *S3：Prefix: TranscribeOutput/, Suffix: .json
  トリガーのPrefixとSuffixは確実に設定すること。Lambdaが意図せず呼び出されてしまいます。
Lambda関数のIAMロール（①、②共通）
以下のAWS管理ポリシーを付与したIAMロールを作成する。
- AmazonS3FullAccess
- AmazonTranscribeFullAccess
- CloudWatchLogsFullAccess

Lambda関数のソースコード

S3へのアップロードでTranscribeジョブを作成

import boto3
import os
from pathlib import Path
from datetime import datetime 



client = boto3.client('transcribe')

def lambda_handler(event, context):
    aws_region = os.environ['AWS_REGION']
    bucket_name = event['Records'][0]['s3']['bucket']['name']
    object_name = event['Records'][0]['s3']['object']['key']

    
    file_uri = f'https://{bucket_name}.s3-{aws_region}.amazonaws.com/{object_name}'
    job_name = Path(object_name).stem + '-' +datetime.now().strftime("%Y%m%d%H%M%S")
    client.start_transcription_job(
        TranscriptionJobName=job_name,
        Media={
            'MediaFileUri':file_uri
        },
        Settings={
            'ShowSpeakerLabels': True,
            'MaxSpeakerLabels': 4,
        },
        LanguageCode='ja-JP',
        OutputBucketName= os.environ.get('output_bucket'),
        OutputKey = os.environ.get('output_prefix')
    )

Transcribeの出力をCSVに整形

import os
import csv
import json
from datetime import datetime
from pathlib import Path
import boto3


def lambda_handler(event, context):
    # AWS Lambdaが実行されているリージョンを取得
    aws_region = os.environ['AWS_REGION']
    input_bucket_name = event['Records'][0]['s3']['bucket']['name']
    input_object_key = event['Records'][0]['s3']['object']['key']

    # S3クライアントを作成
    s3 = boto3.client('s3', region_name=aws_region)

    # JSONファイルをS3からダウンロード
    response = s3.get_object(Bucket=input_bucket_name, Key=input_object_key)    
    # ダウンロードしたJSONデータを読み込む
    json_data = json.loads(response['Body'].read().decode('utf-8'))
        

    ############### JSONファイルの変換処理 ############################
    # 生成したCSVファイルのファイルパス
    output_csv_file = input_bucket_name
    csv_file_path = "/tmp/output.csv"
    speechSegmentList = generate_speech_segments(json_data)
    speech_segments_to_csv(speechSegmentList, csv_file_path)

    ############### s3へのアップロード処理 ############################
    # S3バケット名とアップロードするオブジェクトのキー（ファイルパス）
    output_bucket_name = os.environ["output_bucket"]
    output_object_key = os.environ["output_prefix"] + Path(input_object_key).stem + datetime.now().strftime("%Y%m%d%H%M%S") + '.csv'

    # CSVファイルをS3バケットにアップロード
    s3.upload_file(csv_file_path, output_bucket_name, output_object_key)


    # 生成した一時ファイルを削除
    os.remove(csv_file_path)

    return {
        'statusCode': 200,
        'body': '変換データをS3バケットに保存しました。'
    }



START_NEW_SEGMENT_DELAY = 2.0

class SpeechSegment:
    """ 個々の音声セグメントに関する情報を保持するクラス """
    def __init__(self):
        self.segmentStartTime = 0.0
        self.segmentEndTime = 0.0
        self.segmentSpeaker = ""
        self.segmentText = ""


def generate_speech_segments(json_data):
    """
    TranscribeからのJSON結果データを処理し、ターンごとの音声セグメントを生成.

    :param json_data: TranscribeからのJSON結果データ
    :return: 生成された音声セグメントのリスト
    """

    # 初期化
    speechSegmentList = []  # 音声セグメントのリスト
    lastSpeaker = ""  # 最後のスピーカー
    lastEndTime = 0.0  # 最後の終了時間
    skipLeadingSpace = False  # 先頭のスペースをスキップするかどうかのフラグ
    nextSpeechSegment = None  # 次の音声セグメント

    # セグメントは個々のスピーカーによる発音と句読点の塊です
    for segment in json_data["results"]["speaker_labels"]["segments"]:
        
        # セグメントにコンテンツがある場合
        if len(segment["items"]) > 0:
            # 次のデータを取得
            nextStartTime = float(segment["start_time"])
            nextEndTime = float(segment["end_time"])
            nextSpeaker = str(segment["speaker_label"])

            # スピーカーが変わった場合、またはギャップがある場合、新しいセグメントを作成
            if (nextSpeaker != lastSpeaker) or ((nextStartTime - lastEndTime) >= START_NEW_SEGMENT_DELAY):
                nextSpeechSegment = SpeechSegment()
                speechSegmentList.append(nextSpeechSegment)
                nextSpeechSegment.segmentStartTime = nextStartTime
                nextSpeechSegment.segmentSpeaker = nextSpeaker
                skipLeadingSpace = True

            nextSpeechSegment.segmentEndTime = nextEndTime

            # スピーカーと終了時間を記録
            lastSpeaker = nextSpeaker
            lastEndTime = nextEndTime

            # セグメント内の各単語について処理
            for word in segment["items"]:

                # 最も高い信頼度の単語を取得
                pronunciations = list(filter(lambda x: x["type"] == "pronunciation", json_data["results"]["items"]))
                word_result = list(filter(lambda x: x["start_time"] == word["start_time"] and x["end_time"] == word["end_time"], pronunciations))
                try:
                    result = sorted(word_result[-1]["alternatives"], key=lambda x: x["confidence"])[-1]
                    confidence = float(result["confidence"])
                except:
                    result = word_result[-1]["alternatives"][0]
                    confidence = float(result["redactions"][0]["confidence"])

                # 単語を書き込み、セグメントの先頭でない場合は先頭のスペースを追加
                if skipLeadingSpace:
                    skipLeadingSpace = False
                    wordToAdd = result["content"]
                else:
                    wordToAdd = result["content"]

                # 次のアイテムが句読点である場合、それを現在の単語に追加
                try:
                    word_result_index = json_data["results"]["items"].index(word_result[0])
                    next_item = json_data["results"]["items"][word_result_index + 1]
                    if next_item["type"] == "punctuation":
                        wordToAdd += next_item["alternatives"][0]["content"]
                except IndexError:
                    pass
                nextSpeechSegment.segmentText += wordToAdd

    return speechSegmentList


def speech_segments_to_csv(data_list, output_file):
    """
    SpeechSegmentクラスのリストをUTF-8エンコーディングでCSVファイルに変換

    :param data_list: SpeechSegmentクラスのリスト
    :param output_file: 出力するCSVファイルのファイルパス
    """
    # SpeechSegmentクラスの属性名（ヘッダー）を取得
    fieldnames = list(data_list[0].__dict__.keys())

    # CSVファイルを書き込み
    with open(output_file, 'w', newline='', encoding='cp932') as csv_file:
        writer = csv.DictWriter(csv_file, fieldnames=fieldnames)
        
        # ヘッダーを書き込み
        writer.writeheader()
        
        # 各SpeechSegmentオブジェクトのデータをCSVに書き込み
        for segment in data_list:
            writer.writerow(segment.__dict__)

おわりに

今回、AWSの機械学習サービスであるTranscribeを活用することで、簡単に文字起こしシステムを構築する方法を共有しました。自分でモデルを作成する手間をかけずに、こうしたサービスを利用できるのがAWSの強みですね。
今後もAWSの多彩なサービスを活用して、より便利で効率的なソリューションを開発する様々な方法を探求していきたいと思います。皆さんもAWSの機械学習サービスを活用して、新たなプロジェクトに挑戦してみてください。お読みいただき、ありがとうございました。