AWS を駆使して動画に YouTuber 感をちょい足しするソリューションを作ってみる

{
  "items":[{
    "start_time":"4.06",
    "end_time":"4.21",
    "alternatives":[{
      "confidence":"0.831",
      "content":"みな"
    }],
    "type":"pronunciation"
  },{
    "start_time":"4.21",
    "end_time":"4.36",
    "alternatives":[{
      "confidence":"0.8416",
      "content":"さん"
    }],
    "type":"pronunciation"
  },{
    "start_time":"4.36",
    "end_time":"4.82",
    "alternatives":[{
      "confidence":"1.0",
      "content":"こんにちは"
    }]
  },{
    "start_time":"5.23",
    "end_time":"5.81",
    "alternatives":[{
      "confidence":"1.0",
      "content":"アマゾンウェブサービス"
    }]
  }]
}

sam build

sam deploy --guided

transcribe = boto3.client("transcribe")

def check_transcribe_status(event, context):
    """
    Transcribe のジョブのステータスを確認し、完了まで待機する関数

    Parameter
    ----------
    - event: Step Functions からの入力
        - job_name: Transcribe の実行ジョブ名
        - iterator:
            - index: 何回目のステータス確認かを表すインデックス
    - context: Lambda 関数の実行コンテキスト

    Return
    ----------
    - iterator:
        - index: 何回目のステータス確認かを表すインデックス
        - is_continue: 待機した後にもう一度ステータス確認を行うか示すフラグ
        - is_timeout: ステータス確認回数の上限を超えたかどうかを示すフラグ

    """

    logger.info({"event": event})
    job_name = event["job_name"]

    try:
        response = transcribe.get_transcription_job(
            TranscriptionJobName=job_name
        )
        logger.info({"transcribe_job_status": response})
    except ClientError as e:
        logger.error({"error": e})
        raise
    transcribe_status = response["TranscriptionJob"]["TranscriptionJobStatus"]

    max_count = 200
    index = event["iterator"]["index"]
    index += 1
    is_timeout = False
    is_continue = True
    if transcribe_status == "COMPLETED":
        is_continue = False
        index = 0
    elif index > max_count or transcribe_status == "FAILED":
        is_timeout = True
        is_continue = False

    iterator = {
        "index": index,
        "is_continue": is_continue,
        "is_timeout": is_timeout,
    }
    return iterator

import json
import os
from decimal import Decimal
import boto3
from aws_lambda_powertools.logging import Logger
from botocore.exceptions import ClientError

TABLE_NAME = os.getenv("DYNAMODB_TABLE_NAME")
QUEUE_NAME = os.getenv("SQS_QUEUE_NAME")
NO_SOUND_DURATION = Decimal(os.getenv("NO_SOUND_DURATION", "0.2"))
DIRECTORY = "/mnt/lambda"

dynamo = boto3.resource("dynamodb")
sqs = boto3.resource("sqs")
table = dynamo.Table(TABLE_NAME)
queue = sqs.get_queue_by_name(QueueName=QUEUE_NAME)
logger = Logger()


def analyze_speech(event, context):
    """
    動画の書き起こし結果から発話区間 (無音でない区間) を分析し、
    SQS と DynamoDB へ区間ごとに結果を送信・保存する関数

    Parameter
    ----------
    - event: Step Functions からの入力
        - job_name: Step Functions の実行名
        - file_name: S3 にアップロードされた動画のファイル名
    - context: Lambda 関数の実行コンテキスト

    Return
    ----------
    - file_count: 生成される動画ファイル数

    """

    logger.info({"event": event})
    job_name = event["job_name"]
    file_name = event["file_name"]

    def clip_speech_activity(movie_start, movie_end, index):
        """
        発話区間のデータを SQS と DynamoDB へ書き込む関数

        Parameter
        ----------
        - movie_start: 発話区間の開始時間
        - movie_end: 発話区間の終了時間
        - index: 何番目の発話区間かを表すインデックス

        Return
        ----------
        - index: 次の発話区間のインデックス

        """

        duration = movie_end - movie_start
        if duration != 0:
            message = {
                "job_name": job_name,
                "index": index,
            }
            record = {
                **message,
                "file_name": file_name,
                "start_time": movie_start,
                "duration": duration,
            }
            try:
                queue.send_message(MessageBody=json.dumps(message))
                table.put_item(Item=record)
            except ClientError as e:
                logger.error({"error": e})
                raise
            index = index + 1
        return index

    transcription_path = f"{DIRECTORY}/{job_name}/input/transcription.json"
    try:
        with open(transcription_path, "r") as file:
            transcription = json.load(file)
    except json.JSONDecodeError as e:
        logger.error({"error": e})
        raise
    transcription_items = transcription["results"]["items"]

    last_movie_start = 0
    last_movie_end = 0
    next_index = 0
    for item in transcription_items:
        if item["type"] != "pronunciation":
            continue
        start_time = Decimal(item["start_time"])
        if start_time - last_movie_end >= NO_SOUND_DURATION:
            next_index = clip_speech_activity(
                last_movie_start, last_movie_end, next_index
            )
            last_movie_start = start_time
        last_movie_end = Decimal(item["end_time"])

    file_count = clip_speech_activity(
        last_movie_start, last_movie_end, next_index
    )
    return file_count

{
  "items":[{
    "start_time":"4.06",
    "end_time":"4.21",
    "alternatives":[{
      "confidence":"0.831",
      "content":"みな"
    }],
    "type":"pronunciation"
  },{
    "start_time":"4.21",
    "end_time":"4.36",
    "alternatives":[{
      "confidence":"0.8416",
      "content":"さん"
    }],
    "type":"pronunciation"
  },{
    "start_time":"4.36",
    "end_time":"4.82",
    "alternatives":[{
      "confidence":"1.0",
      "content":"こんにちは"
    }]
  },{
    "start_time":"5.23",
    "end_time":"5.81",
    "alternatives":[{
      "confidence":"1.0",
      "content":"アマゾンウェブサービス"
    }]
  }]
}

def clip_speech(event, context):
    """
    SQS でカットの対象となる発話区間を含んだメッセージを受信すると、
    その区間にもとづき、動画のカットを行う関数
    正常に完了した場合、DynamoDB から該当するレコードを削除する

    Parameter
    ----------
    - event: SQS から受信したメッセージ
        - Records: 受信したメッセージの配列
            - job_name: Step Functions の実行名
            - index: 何番目の発話区間かを表すインデックス
    - context: Lambda 関数の実行コンテキスト

    Return
    ----------
    None

    """

    def spawn_clip_process(start_time, duration, input, output):
        """
        動画のカットを行い EFS 上に保存するプロセス (ffmpeg) を立ち上げる関数

        Parameter
        ----------
        - start_time: 発話区間の開始時間
        - duration: 発話区間の長さ
        - input: カットの対象となる動画ファイルのパス
        - output: カットした動画ファイルを配置するパス

        Return
        ----------
        - is_success: プロセスの実行が成功したかどうかを表すフラグ

        """

        cmd = [
            "./ffmpeg",
            "-y",
            "-ss",
            str(start_time),
            "-i",
            input,
            "-t",
            str(duration),
            output,
        ]
        result = subprocess.run(
            cmd,
            stdin=subprocess.DEVNULL,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )
        stdout = result.stdout.decode("utf8")
        stderr = result.stderr.decode("utf8")
        if len(stdout) != 0:
            logger.info({"ffmpeg_stdout": stdout})
        if len(stderr) != 0:
            logger.info({"ffmpeg_stderr": stderr})

        is_success = True
        if result.returncode != 0:
            is_success = False
        return is_success

    logger.info({"event": event})

    messages = event["Records"]
    for message in messages:
        message_body = json.loads(message["body"])
        job_name = message_body["job_name"]
        index = message_body["index"]

        try:
            record = table.get_item(Key={"job_name": job_name, "index": index})
        except ClientError as e:
            logger.error({"error": e})
        item = record["Item"]
        logger.info({"dynamodb_item": item})

        file_name = item["file_name"]
        input_file = f"{DIRECTORY}/{job_name}/input/{file_name}"
        output_file = f"{DIRECTORY}/{job_name}/clipped/{index}.mp4"
        is_success = spawn_clip_process(
            item["start_time"], item["duration"], input_file, output_file
        )
        if is_success:
            try:
                table.delete_item(Key={"job_name": job_name, "index": index})
            except ClientError as e:
                logger.error({"error": e})
                raise
        else:
            raise MovieClippingError()

    return