Amazon ConnectとWatson Speech To Textの連携ー通話内容をテキスト化しメールで確認ー

# はじめに

コールセンターを新規に構築する際、Amazon Connectを使ってみようというニーズはあると思います。
しかし通話内容のテキスト化や、テキスト化した内容の利用方法については、まだよく分からない部分も多いのではないでしょうか。

例えば、COTOHAを使ったサービスとしては@azuki2iceさんの以下の記事があります。
[Amazon Connectの通話をCOTOHAで音声認識させて通話テキストをSalesforceに自動登録する](https://qiita.com/azuki2ice/items/3a5911666c104d8b9f04)

Amazon Connectは他サービスでも使えるので、今回はWatsonを連携し、通話内容を文字起こしして、その内容をメールで飛ばすサービスを作ってみたいと思います。

本記事の内容を実装することで、以下の図のようなサービスが実現できます。

<img width="647" alt="d171248b-3e7f-d3df-ddf3-c1312107cefa.png" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/413092/a2be7279-9b3d-d4a6-e303-0e92a133f450.png">


なおこのサービスは構築にあたり、@Masaakiさんにご協力いただきました。ありがとうございます。

# 構成

![08c6a5c2-7ffd-8f80-3226-569e0f07028f.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/413092/cd7bb3f0-d18d-feb8-508e-e42131a960c9.png)


1. Amazon ConnectからKinesis Video Streams(KVS)にデータを保存
2. Lambda①によりAmazon ConnectからSQSにキューを保存
3. SQS経由で起動されたLambda②がKVSに保存されたデータをwavに変換、S3に保存
4. S3への保存をトリガーにLambda③がWatson Speech To Textを起動
5. 変換されたテキストファイルをS3に保存
6. S3に保存されたテキストをLambda④を用い、Amazon SES(Simple Email Service)でユーザにメール配信

# Amazon Connectの設定

問い合わせフローは以下のように設定します。

<img width="926" alt="34ff9441-2159-74af-289e-1080536ced96.png" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/413092/ef5032f8-4906-e6ae-c957-12b44fa2f25d.png">

- 「メディアストリーミングの開始」ブロックから「メディアストリーミングの停止」ブロックまでの通録が保存されます。
- 問い合わせフロー上で設定する必要があるLambdaは①のみです。
- 特に説明のない「プロンプトの再生」ブロックは、フローを切り分けられるよう任意のメッセージを流すよう設定しました。
- 「<font color="red">テキスト化</font>」と書いてある「プロンプトの再生」は、次のように設定しています。
  - 「テキスト読み上げ機能(アドホック)」を選択
  - 「テキストの入力」を選択
  - 「解釈する」はSSMLを選択し、テキストメッセージには以下のように記述
  
``` 
    <speak>
      <break time="10s"/>
      メッセージをお預かりしました。
    </speak>
```


# Lambdaの設定

- Lambda①、②はNode.js 10.x、③、④はPython3.7で設定しています。

## Lambda①

- 関数のロールにSQSを追加します。

```index.js
const AWS = require("aws-sdk");

const account  = process.env.account;
const name = process.env.queueName;
const region = process.env.region;
const url = 'https://sqs.'+ region +'.amazonaws.com/' + account + '/' + name;

exports.handler = async (event) => {
    console.log(JSON.stringify(event));

    const sqs = new AWS.SQS();
    const params = {
        MessageBody: JSON.stringify(event),
        QueueUrl: url,
    };
    result = await sqs.sendMessage(params).promise();

    console.log(result);
}
```

## Lambda②

- 関数のトリガーにSQS、ロールにSQS、KVS、S3を設定します。

```index.js
const AWS = require("aws-sdk");

const bucketName = process.env.bucketName;
const region = process.env.region;

const ConnectVoiceMail = require("./ConnectVoiceMail");

exports.handler = async (event) => {

    console.log(JSON.stringify(event));

    for(var i=0; i<event.Records.length; i++) {
        const body = JSON.parse(event.Records[i].body);
        const audio = body.Details.ContactData.MediaStreams.Customer.Audio;
        const streamName = audio.StreamARN.split('stream/')[1].split('/')[0];
        const fragmentNumber = audio.StartFragmentNumber;
        const startTime = new Date(Number(audio.StartTimestamp));

        console.log('streamName:' + streamName);
        console.log('fragmentNumber:' + fragmentNumber);

        const connectVoiceMail = new ConnectVoiceMail();
        const wav = await connectVoiceMail.getWav(region, streamName, fragmentNumber);

        const s3 = new AWS.S3({region:region});
        const key = dateString(startTime) + '.wav';
        console.log('wavFile: ' + key);
        const params = {
            Bucket: bucketName,
            Key: key,
            Body: Buffer.from(wav.buffer),
        };
        await s3.putObject(params).promise();
    }
}

function dateString(date) {
    const year = date.getFullYear();
    const mon = (date.getMonth() + 1);
    const day = date.getDate();
    const hour = date.getHours();
    const min = date.getMinutes();

    const space = (n) => {
        return ('0' + (n)).slice(-2)
    }

    let result = year + '-';
    result += space(mon) + '-';
    result += space(day) + '_';
    result += space(hour) + ':';
    result += space(min);
    return result;
}
```

```ConnectVoiceMail.js
const ebml = require('ebml');
const AWS = require("aws-sdk");

module.exports = class ConnectVoiceMail {

    async getWav(region, streamName, fragmentNumber) {
        const raw = await this._getMedia(region, streamName, fragmentNumber);
        const wav = this._createWav(raw, 8000);
        return wav;
    }

    async _getMedia(region, streamName, fragmentNumber) {
        // Endpointの取得
        const kinesisvideo = new AWS.KinesisVideo({region: region});
        var params = {
            APIName: "GET_MEDIA",
            StreamName: streamName
        };
        const end = await kinesisvideo.getDataEndpoint(params).promise();

        // RAWデータの取得
        const kinesisvideomedia = new AWS.KinesisVideoMedia({endpoint: end.DataEndpoint, region:region});
        var params = {
            StartSelector: { 
                StartSelectorType: "FRAGMENT_NUMBER",
                AfterFragmentNumber:fragmentNumber,
            },
            StreamName: streamName
        };
        const data = await kinesisvideomedia.getMedia(params).promise();
        const decoder = new ebml.Decoder();
        let chunks = [];
        decoder.on('data', chunk => {
            if(chunk[1].name == 'SimpleBlock'){
                chunks.push(chunk[1].data);
            }
        });
        decoder.write(data["Payload"]);

        // chunksの結合
        const margin = 4; // 各chunkの先頭4バイトを破棄する
        var sumLength = 0;
        chunks.forEach( chunk => {
            sumLength += chunk.byteLength - margin;
        })
        var sample = new Uint8Array(sumLength);
        var pos = 0;
        chunks.forEach(chunk => {
            let tmp = new Uint8Array(chunk.byteLength - margin);
            for(var e = 0; e < chunk.byteLength -  margin; e++){
                tmp[e] = chunk[e + margin];
            }
            sample.set(tmp, pos);
            pos += chunk.byteLength - margin;

        })
        return sample.buffer;
    }

    // WAVファイルの生成
    _createWav(samples, sampleRate) {
        const len = samples.byteLength;
        const view = new DataView(new ArrayBuffer(44 + len));
        this._writeString(view, 0, 'RIFF');
        view.setUint32(4, 32 + len, true);
        this._writeString(view, 8, 'WAVE');
        this._writeString(view, 12, 'fmt ');
        view.setUint32(16, 16, true);
        view.setUint16(20, 1, true); // リニアPCM
        view.setUint16(22, 1, true); // モノラル
        view.setUint32(24, sampleRate, true); 
        view.setUint32(28, sampleRate * 2, true);
        view.setUint16(32, 2, true);
        view.setUint16(34, 16, true);
        this._writeString(view, 36, 'data');
        view.setUint32(40, len, true);
        let offset = 44;
        const srcView = new DataView(samples);
        for (var i = 0; i < len; i+=4, offset+=4) {
            view.setInt32(offset, srcView.getUint32(i));
        }
        return view;
    }

    _writeString(view, offset, string) {
        for (var i = 0; i < string.length; i++) {
            view.setUint8(offset + i, string.charCodeAt(i));
        }
    }
}
```

## Lambda③

- 関数のロールにS3を追加します。

```lambda_function.py
def lambda_handler(event, context):
  bucket_name = event['Records'][0]['s3']['bucket']['name']
  file_key = event['Records'][0]['s3']['object']['key']
  cont_type = "audio/wav"
  lang = "ja-JP_NarrowbandModel"
  iam_apikey = "XXXXXXXXXXXXXXXXX"
  url = "https://gateway-tok.watsonplatform.net/speech-to-text/api"

  s3 = boto3.resource('s3')
  local_file_name = '/tmp/'+file_key.replace('%3A', '-')
  s3.Bucket(bucket_name).download_file(file_key.replace('%3A', ':'), local_file_name)
  stt = SpeechToTextV1(iam_apikey=iam_apikey, url=url)
  result_json = stt.recognize(audio=open(local_file_name, 'rb'), content_type=cont_type, model=lang)

  # json file save
  result = json.dumps(result_json.result, indent=2)
  res = s3.Bucket('voice-mail-transcripted-json-file-bucket').put_object(Key=local_file_name+'.json', Body=result)
  return res
```

## Lambda④

```lambda_function.py
import boto3
import json
s3 = boto3.client('s3')
dynamodb = boto3.client('dynamodb')

SRC_MAIL = "your_email_address"
DST_MAIL = "your_email_address"
REGION = "us-west-2"
DINAMO_TABLE_NAME = "your_dinamo_name"

def lock(key):
    try:
        dynamodb.put_item(
        TableName = DINAMO_TABLE_NAME,
        Item = {'filename':{'S':key},'status':{'S':'complete!'}},
        Expected = {'filename':{'Exists':False}}
        )
        return True
    except Exception as e:
        return False

def unlock(key):
    dynamodb.delete_item(
        TableName = 'test-lambda',
        Key = {
            'filename': {'S': key}
        }
    )

def send_email(source, to, subject, body):
    client = boto3.client('ses', region_name=REGION)

    response = client.send_email(
        Source=source,
        Destination={
            'ToAddresses': [
                to,
            ]
        },
        Message={
            'Subject': {
                'Data': subject,
            },
            'Body': {
                'Text': {
                    'Data': body,
                },
            }
        }
    )

    return response

def lambda_handler(event, context):
    bucket_name = event['Records'][0]['s3']['bucket']['name']
    file_key = event['Records'][0]['s3']['object']['key']

    response = s3.get_object(Bucket=bucket_name, Key=file_key)
    transcripted = json.loads(response['Body'].read().decode('utf-8'))

    email_content = ''
    for i in range(len(transcripted['results'])):
      email_content += transcripted['results'][i]["alternatives"][0]["transcript"].rstrip()
    email_title = "Amazon Connectを用いたボイスメモ1"
    message = json.dumps(event, indent = 4)
    print(email_content)
    # print(file_key)
    # if(lock(file_key)):
    #     r = send_email(SRC_MAIL, DST_MAIL, email_title, email_content)
    # else:
    #     print(file_key + " seems to be completed already.")
    r = send_email(SRC_MAIL, DST_MAIL, email_title, email_content)
    return r
```

# おわりに

Speech To Textのサービスは色々ありますが、ポリシーなどの問題でWatsonを使うという需要は一定数あると思います。
IBM Cloudは[無料枠](https://www.ibm.com/jp-ja/cloud/lite-account)が設定されているので、ぜひ試してみてください！