Python
AWS
ffmpeg
lambda

Amazon Lambdaを使って音声読み上げLINE Botを作りました。

TL;DR

テキストメッセージを送信すると、音声に変換して返してくれるBotを作りました。

DeZ3_JJUQAAUAU8.jpg
OIoaB_FTXr.png
友だち追加

ソースコードは以下
https://github.com/lboavde1121/lambda_linebot

使用サービス

ソースコード

少し長くなるので先にソースコードをどん。

lambda_function.py
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""This module is Text to Speech."""

import logging
import os
import json
import base64
import hashlib
import hmac
import boto3
from contextlib import closing
import subprocess
from dotenv import load_dotenv
import requests

dotenv_path = os.path.join(os.path.dirname(__file__), '.env')
load_dotenv(dotenv_path)

logger = logging.getLogger()
logger.setLevel(logging.INFO)
# Sessionを作成
client = boto3.client('polly')

# リプライ用URL
reply_url = 'https://api.line.me/v2/bot/message/reply'
# トークン
token = os.getenv("LINE_TOKEN")
# リクエストヘッダ
headers = {
    'Content-Type': 'application/json',
    'Authorization': 'Bearer %s' % token,
}


def shorten_url(url):
    """Shorten the URL."""
    bitly_url_base = "https://api-ssl.bitly.com/v3/shorten"
    token = os.getenv("BITLY_TOKEN")

    bitly_url = "%s?access_token=%s&longUrl=%s" % (bitly_url_base,
                                                   token, url)
    print(bitly_url)
    bitly_req = requests.get(bitly_url)
    url = bitly_req.json()['data']['url'].replace("http://", "https://")
    return url


def sent_message(event, text):
    """Sent message."""
    body = {
        'replyToken': event['replyToken'],
        'messages': [
                {
                    'type': 'text',
                    'text': text,
                }
            ]
    }
    req = requests.post(reply_url, json=body, headers=headers)
    logger.info(req.text)


def put_s3_object(bucketname, keyname, filename, acl="public-read"):
    """Put S3 Object."""
    s3 = boto3.resource('s3')
    # バケットを取得
    bucket = s3.Bucket(bucketname)
    with open(filename, 'rb') as f:
        # ファイル出力
        bucket.put_object(
            ACL=acl,
            Body=f.read(),
            Key=keyname
        )
    return "Success"


def text_to_speech(event):
    """Text to speech."""
    text = event['message']['text']
    charnum = int(len(text))
    if charnum > 200:
        error_txt = "テキストが長すぎます。200文字以内でお願いします。"
        sent_message(event, error_txt)
        return

    response = client.synthesize_speech(
        OutputFormat='mp3',
        Text=text,
        TextType='text',
        VoiceId='Mizuki'
    )
    logger.info(response)
    mp3_path = '/tmp/' + event['message']['id'] + '.mp3'
    with open(mp3_path, "wb") as f:
        logger.info("Start Writing")
        with closing(response["AudioStream"]) as stream:
            f.write(stream.read())

    m4a_name = event['message']['id'] + '.m4a'
    m4a_path = '/tmp/' + m4a_name

    # ffmpeg実行
    ffmpeg_cmd = './ffmpeg_build/bin/ffmpeg -i %s %s' % (mp3_path, m4a_path)
    subprocess.call(ffmpeg_cmd, shell=True)

    s3_bucket_name = os.getenv["S3_BUCKET_NAME"]
    put_s3_object(s3_bucket_name, m4a_name, m4a_path)
    os.remove(mp3_path)
    os.remove(m4a_path)

    # LINE 投稿
    speech_url = ("https://s3-ap-northeast-1.amazonaws.com/%s/%s" %
                  (s3_bucket_name, m4a_name))
    # URLを短縮
    shoot_url = shorten_url(speech_url)
    req_json = {
        'replyToken': event['replyToken'],
        'messages': [
            {
                "type": "audio",
                "originalContentUrl": shoot_url,
                "duration": charnum * 166,
            }
        ]
    }
    req = requests.post(reply_url, json=req_json, headers=headers)
    logger.info(req.text)


def lambda_handler(request, context):
    """AWS lambda function."""
    # リクエスト検証
    channel_secret = os.getenv("CHANNNEL_SERCRET")
    logger.info(channel_secret)
    body = request.get('body', '')
    hash = hmac.new(channel_secret.encode('utf-8'),
                    body.encode('utf-8'), hashlib.sha256).digest()
    signature = base64.b64encode(hash).decode('utf-8')

    # # LINE 以外からのアクセスだった場合は処理を終了させる
    if signature != request.get('headers').get('X-Line-Signature', ''):
        logger.info(f'LINE 以外からのアクセス request={request}')
        return {'statusCode': 200, 'body': '{}'}
    logger.info(request)

    # for event in request['events']:
    for event in json.loads(body).get('events', []):
        logger.info(json.dumps(event))
        msg_type = event['message']['type']
        if msg_type == "text":
            # Amaon Pollyで音声作成
            text_to_speech(event)

        if msg_type == "audio":
            # TODO Audio Recognition
            pass

    return {'statusCode': 200, 'body': '{}'}

作成手順

事前準備

requirements.txt
python-dotenv
requests

処理の流れ

  1. 送られてきたテキストを取得
  2. 取得したテキストをPollyを使用して音声へ変換
  3. 音声をm4aに変換
  4. S3にアップロード
  5. MessegingAPIを使用して音声を返却

送られてきたテキストを取得

LINEから送られてきたメッセージは以下のようになってAPIに送られます。

{
    "type": "message",
    "replyToken": "リプライトークン",
    "source": {
        "userId": "ユーザーID",
        "type": "user"
    },
    "timestamp": 1528209887362,
    "message": {
        "type": "text",
        "id": "8070789813877",
        "text": "こんにちは!"
    }
}

これを、Lambdaで受け取って、メッセージのテキストを抽出します。

lambda_function.py
import json

def lambda_handler(request, context):
    body = request.get('body', '')

    for event in json.loads(body).get('events', []):
        text = event['message']['text']

簡単

取得したテキストをPollyを使用して音声へ変換

LambdaからPollyを使用するには、Lambda関数に権限を付与する必要があります。

ロール作成→ https://console.aws.amazon.com/iam/home?region=ap-northeast-1#/roles

スクリーンショット 2018-06-06 1.06.16.png
スクリーンショット 2018-06-06 1.08.37.png

「ロールの作成」→Lambdaを選択して「次のステップ:アクセス権限」

  • CloudWatchFullAccess
  • AmazonPollyFullAccess
  • AmazonS3FullAccess

を選択して「次のステップ:確認」
本当はフルアクセス付与するのはよろしくないので、必要な権限だけ付与しましょう。

ロール名と説明を入力し作成完了。

スクリーンショット 2018-06-06 1.16.51.png

Lambda関数コンソールの実行ロールの項目から作成したロールを選択。

実行権限が付与できたらテキストを音声に変換

lambda_function.py
import boto3

# 色々省略

client = boto3.client('polly')
response = client.synthesize_speech(
        OutputFormat='mp3',
        Text=text,
        TextType='text',
        VoiceId='Mizuki'
    )

上記実行すると、

{
  "ResponseMetadata": {
    "RequestId": "",
    "HTTPStatusCode": 200,
    "HTTPHeaders": {
      "content-type": "audio/mpeg",
      "date": "Tue, 05 Jun 2018 14:45:37 GMT",
      "x-amzn-requestcharacters": "6",
      "x-amzn-requestid": "",
      "transfer-encoding": "chunked",
      "connection": "keep-alive"
    },
    "RetryAttempts": 0
  },
  "ContentType": "audio/mpeg",
  "RequestCharacters": "6",
  "AudioStream": <botocore.response.StreamingBody object at 0x7f9e3cc2bda0>
}

のようなJSONが帰ってくる。
"AudioStream" 内のオブジェクトに音声のバイナリデータが含まれている。
このバイナリデータをファイルにバイナリモードで書き込めば、音声ファイルになる。

lambda_function.py
from contextlib import closing

# 色々省略

with open("/tmp/test.mp3", "wb") as f:
    logger.info("Start Writing")
    with closing(response["AudioStream"]) as stream:
        f.write(stream.read())

Lambdaでファイルを書き込む場合は/tmp配下でないと、エラーが発生する。

音声をm4aに変換

Lineに音声を送信する場合、M4A以外のフォーマットだと再生できません。
なので、ffmpegを使用して音声フォーマットを変更します。

Lambdaで使用するためには、AmazonLinux上でffmpegをビルドする必要があります。

AmazonLinux Pull

docker run -v /Path/to/workdir:/usr/local/src/ -i -t amazonlinux:2016.09 /bin/bash

ワーキングディレクトリをマウントしてAmazonLinuxイメージを起動し、bashログイン。

yum install -y git
mkdir /usr/local/src/ffmpeg_sources
cd /usr/local/src/ffmpeg_sources
git clone --depth 1 git://git.code.sf.net/p/opencore-amr/fdk-aac

cd fdk-aac
autoreconf -fiv
./configure --prefix="/usr/local/src/ffmpeg_build" --disable-shared
make
make install

cd /usr/local/src/ffmpeg_sources
git clone --depth 1 git://source.ffmpeg.org/ffmpeg
cd ffmpeg
PKG_CONFIG_PATH="/usr/local/src/ffmpeg_build/lib/pkgconfig" ./configure --enable-static --disable-shared --prefix="/usr/local/src/ffmpeg_build" --extra-cflags="-I/usr/local/src/ffmpeg_build/include" --extra-ldflags="-L/usr/local/src/ffmpeg_build/lib" --bindir="/usr/local/src/bin" --enable-gpl --enable-nonfree --enable-libfdk_aac
make
make install

参考:https://qiita.com/junya108/items/c65096b0be68039a9817

スクリーンショット 2018-06-06 23.33.44.png

ファイルサイズが大きいと、lambdaにアップロードした際にエラーになるので不要なファイルは削除してしまいましょう(bin/配下はffmpeg以外使用しないので削除してあります)。

そして、subprocessモジュールを使用してPythonからffmpegを実行します。

lambda_function.py
import subprocess

# 色々省略

# ffmpeg実行
ffmpeg_cmd = './ffmpeg_build/bin/ffmpeg -i %s %s' % ('/tmp/test.mp3', 'tmp/test.m4a')
subprocess.call(ffmpeg_cmd, shell=True)

S3にアップロード

S3バケットに突っ込みます。

lambda_function.py
s3 = boto3.resource('s3')
# バケットを取得
bucket = s3.Bucket("S3_BUCKET_NAME")
with open("/tmp/test.m4a", 'rb') as f:
    # ファイル出力
    bucket.put_object(
        ACL='public-read',
        Body=f.read(),
        Key=keyname
    )

MessegingAPIを使用して音声を返却

requestsモジュールを使います。
pip install requests -t .

lambda_function.py
import repuests
# 色々省略
headers = {
    'Content-Type': 'application/json',
    'Authorization': 'Bearer %s' % token,
}

req_json = {
    'replyToken': event['replyToken'],
    'messages': [
        {
            "type": "audio",
            "originalContentUrl": ,
            "duration": charnum * 166, # テキスト文字数から、音声のおおよその時間を推定
        }
    ]
}
req = requests.post(reply_url, json=req_json, headers=headers)

zip化してアップロード

S3にアップロード。
Lambdaで「S3からのファイルのアップロード」でzipファイルのURLを指定してアップロード。

LambdaのトリガーにAPIGatewayを追加

スクリーンショット 2018-06-07 0.20.57.png

ドラッグ&ドロップでいけます。

APIGatewayをクリックすると、下に詳細が表示されるので、エンドポイント(URLの呼び出し)を確認
スクリーンショット 2018-06-07 0.22.59.png

LINEBotのWebhookでLambdaを実行

https://developers.line.me/

スクリーンショット 2018-06-07 0.14.10.png

ここにLambdaのエンドポイントを設定

動作確認

テキストを送って音声が返ってくることを確認。

おしまい

使ってみた感想、質問等フィードバックいただければ喜びます。
https://twitter.com/rt_s2t