トップ > AWS > Selenium(Python)でAmazonConnectの問い合わせフロー(デフォルトフロー)のセットアップを自動化する

Selenium(Python)でAmazonConnectの問い合わせフロー(デフォルトフロー)のセットアップを自動化する

AWS Amazon Connect
記事タイトルとURLをコピーする

エンジニアの小林です。

Amazon Connectは2020/7現在、CloudFormation非対応＆APIはCreateUser以外公開されていないんです。
https://docs.aws.amazon.com/connect/latest/APIReference/Welcome.html

自動化できない...マネコンからポチポチするしかないのか... ...いや、Seleniumでブラウザ操作を自動化することで、自動化できるのでは...？

と、弊社のエンジニアが考え、社内でスクリプトを書いてみて、動くものができたので、紹介します。

自動化するデフォルトフローって何？

今回は、Connectを使用する上で、フロー関係なく、共通して使用する可能性が高い一部デフォルトフローの更新のみ、Seleniumでブラウザ自動化をしました。

デフォルトフローとは、Connectインスタンスを立ち上げると標準で作成されている問い合わせフローのことです。以下のblogで詳細に解説されていますので、是非チェックしてみてください。

blog.serverworks.co.jp

このデフォルトフローのみを自動化する理由としては、今後Connectで各種Create系のAPIがサポートされてもっと楽に開発・保守できるようになる可能性を考慮したからです。そのため、あくまでAPIサポートまでの繋ぎツールとして開発しました。

今回はデフォルトフローの中でも、Default agent transferとDefault outboundの二つのフローをセットアップ自動化対象とします。Default agent transfer（エージェントへの転送）は、エージェントからエージェントに電話を転送する場合に転送されたエージェントのフローのことです。Default outbound（アウトバウンドウィスパー）は、エージェントからお客さんに電話した場合のエージェントと繋がる直前に顧客のフローのことです。

以下の表ではどのようにフローを変えるかについて整理しています。Default agent transferは標準の設定だと音声案内が英語で流れてしまうので日本語に変更、Default outboundは標準の設定だと通話記録しない＆音声案内が英語で流れてしまうので通話記録をする（音声案内なし）ように変更します。

名前 タイプ 更新前動作
（20200528時点）		 更新後動作
Default agent transfer エージェントへの転送 1. 「Transferring now...」を再生 1. 音声の設定追加（日本語：Mizuki）
2. 「ただいま転送しています」を再生
Default outbound アウトバウンドウィスパー 1. 通話記録動作の設定（なし）
2. 「This call is not being recorded.」を再生		 1. 通話記録動作の設定
（エージェント AND 顧客）

自動化更新前動作の状態のフローは、Connectの問い合わせフロー一覧画面から対象のフローを選択し、json形式でExportを選ぶと取得することができます。そのjsonファイルを更新後動作に変更してください。フローごとにjsonファイルを作成し、全てのファイルを同一のディレクトリに格納します。
（後述の/Users/xxx/setup-automation/file/）

なお、今回作成したスクリプトは、問い合わせフローのInport/Export機能を使用します。
こちらは、2020/07現在ベータ版の機能になります。AWS公式ドキュメントには以下の通り記載がございますので、ご使用の際はご注意ください。

コンタクトのインポート/エクスポート機能は現在ベータ版です。更新と改善により、ベータ段階でエクスポートされた問い合わせフローをインポートする将来のリリースで問題が発生する可能性があります。

https://docs.aws.amazon.com/ja_jp/connect/latest/adminguide/contact-flow-import-export.html

どんなものができるの？

結論を先にいうと、以下のようにができました！（設定内容については後ほどつらつらと書いています）

youtu.be

自動化をすることで、設定漏れ、設定忘れを防止することができます。間違えて設定していたり、保存はしたけど公開し忘れていたり...。そのような人的ミスを防止するのに、自動化は打ってつけですよね。（自動化ならダブルチェックも必要なし！）

また、上記表の通り、手動で変更すると、私は3分程度かかったのですが、スクリプトを流せば数秒で実行完了しました（上記動画は設定内容が分かりやすいように、作成したスクリプトを意図的にゆっくりに設定していますが、これ以降紹介するスクリプトは一瞬で設定するようになってます）。このblogでは、Default agent transferとDefault outboundを自動化対象としましたが、その他デフォルトフローも設定を変更する必要があります。試しに全てのデフォルトフローを手動で設定したら、私は10分以上かかってしまいました。しかし、自動化すると、1分程度で終わっちゃうんです。

デフォルトフローはユーザーが本来注力したいフローではないサブ的な立ち位置のフローだと思います。なので、注力したい問い合わせフローに時間を割くため、デフォルトフローのようなある程度定型化できるフローは自動化してみましょう！

面白そうだな〜私も自動化したいな〜、と思ってきた頃合いだと思うので、これ以降はどのような設定をしたか紹介します。

手順

前提

  • 動作するブラウザはGoogleChrome
  • ローカルで実施
  • Python3系

Connectインスタンスおよびユーザーの作成

事前にConnectインスタンスとフローの更新権限を持つユーザーを作成しましょう。（ここはSeleniumで自動化することもできます）

モジュールのインストール

以下モジュールをインストールしてください。

  • chromedriver-binary
  • PyYAML
  • selenium

ここで重要なのがchromedriver-binaryです。モジュールのインストールをする前にChromeのバージョンの確認をします。Chromeのバージョンの確認は以下の [アップデートの有無と現在のブラウザ バージョンを確認する]に記載してます。
 https://support.google.com/chrome/answer/95414?co=GENIE.Platform%3DDesktop&hl=ja

chromedriver-binaryのバージョンはスクリプト実行上重要となるため ご自身のChromeのバージョンと合ったものを利用してください。chromeのバージョンによっては 完全一致のchromedriver-binaryのバージョンがない可能性があります その場合一番近いバージョンで試してみてください。

configの作成

設定ファイル（config.yaml）は以下の通り作成しました。

DriverGet:
  ConnectConsole: "https://fugafugafuga.awsapps.com/connect/login"
  ContactFlowSetting: "https://fugafugafuga.awsapps.com/connect/contact-flows"
SendKeys:
  InstanceName:
    Key: "input[data-test-id='onboarding-new-directory-alias-tbx']" #css-selector
    Value: "fugafugafuga"
  ConnectLoginUserName:
    Key: "wdc_username"
    Value: "admin"
  ConnectLoginPassword:
    Key: "wdc_password"
    Value: "sashisuseso"
  FilePath:
    Key: "import-cf-file"
    Value: "/Users/xxxxx/annex/connect-setup-automation/file/"
Click:
  ConnectLogin: "wdc_login_button" #id
  SaveNext: "awsui-button[class='sub-right dropdown-toggle ng-scope']" #css-selector
  SaveSelect: "a[ng-click='verifyImport()']" #css-selector
  FileImport: "awsui-button[click='importContactFlow(selectedFile)']" #css-selector
  FlowPublish: "button[ng-click='verifyPublish()']" #css-selector
  PublishTrue: "awsui-button[click='publishContactFlow(true)']" #css-selector
FlowNames:
  - "Default agent transfer"
  - "Default outbound"
  • DriverGet
    •  ConnectConsole
      • fugafugafugaのところを構築するConnectインスタンス名
    • ContactFlowSetting
      • fugafugafugaのところを構築するConnectインスタンス名
  • SendKeys（Valueの値を変更）
    • InstanceName
      • fugafugafugaのところを構築するConnectインスタンス名
    • ConnectLoginUserName
      • 前述で作成したConnectユーザーのユーザー名
    • ConnectLoginPassword
      • 前述で作成したConnectユーザーのパスワード
    • UploadFile
      • インポートするjsonファイルが存在するディレクトリのパス（/Users/xxx/setup-automation/file/）
        • ファイル名は記載しない
        • 絶対パスで指定

スクリプトの実行

ディレクトリ構成は以下の通りです。

connect-setup
├─file
├─Default agent hold.json
│     ├─Default agent transfer.json
│     └─Default outbound.json
├─config.yaml
└─update_flows.py

スクリプト（update_flows.py）は以下の通り作成しました。

import os
import signal
import time

import chromedriver_binary
import yaml
from selenium import webdriver
from selenium.webdriver.common.keys import Keys


def main():
    try:
        driver = webdriver.Chrome()
        driver.maximize_window()
        element_dict = get_element_dict()
        connect_domain = 'https://%s.awsapps.com/connect/' \
                         % element_dict['SendKeys']['InstanceName']

        login_contact_center(driver, element_dict)
        time.sleep(1)  # ログイン後、connect画面を出すまでのリダイレクト時間が必要なため1秒sleepしてます。

        for flow_name in element_dict['FlowNames']:
            file_path = element_dict['SendKeys']['FilePath']['Value'] \
                + flow_name + ".json"
            update_flow(driver, element_dict, flow_name, file_path)

    finally:
        os.kill(driver.service.process.pid, signal.SIGTERM)


# Login Connect Instance
def login_contact_center(driver, element_dict):
    driver.implicitly_wait(10)
    driver.get(element_dict['DriverGet']['ConnectConsole'])
    input_data(driver, element_dict['SendKeys']['ConnectLoginUserName'], 'id')
    input_data(driver, element_dict['SendKeys']['ConnectLoginPassword'], 'id')
    click_button(driver, element_dict['Click']['ConnectLogin'], 'id')


# Update Flow
def update_flow(driver, element_dict, flow_name, file_path):
    driver.implicitly_wait(30)
    driver.get(element_dict['DriverGet']['ContactFlowSetting'])
    click_button(driver, flow_name, 'link')
    click_button(driver, element_dict['Click']['SaveNext'], 'css')
    click_button(driver, element_dict['Click']['SaveSelect'], 'css')
    upload_file(driver, element_dict['SendKeys']['FilePath'], file_path)
    time.sleep(2)
    click_button(driver, element_dict['Click']['FileImport'], 'css')
    click_button(driver, element_dict['Click']['FlowPublish'], 'css')
    click_button(driver, element_dict['Click']['PublishTrue'], 'css')


def get_element_dict():
    with open('./config.yaml') as file:
        element_dict = yaml.safe_load(file)
        return element_dict


def input_data(driver, data_dict, element_type):
    value = data_dict['Value']
    if element_type == 'id':
        field = driver.find_element_by_id(data_dict['Key'])
    elif element_type == 'css':
        field = driver.find_element_by_css_selector(data_dict['Key'])

    if value == 'SelectRadioButton':
        field.send_keys(Keys.SPACE)
    else:
        field.send_keys(value)


def upload_file(driver, data_dict, value):
    field = driver.find_element_by_id(data_dict['Key'])
    field.send_keys(value)


def click_button(driver, key, element_type):
    if element_type == 'id':
        button = driver.find_element_by_id(key)
    elif element_type == 'css':
        button = driver.find_element_by_css_selector(key)
    elif element_type == 'link':
        button = driver.find_element_by_link_text(key)
    button.click()


main()

上記スクリプトを実行すると、前半にご紹介した動画のように、自動で対象のデフォルトフローを更新していきます。スクリプト実行中は、勝手に設定が実行されるので、特に何もする必要はありません。

自動テスト基盤の作成

上記コード作成と同じタイミングで自動テスト基盤の作成も実施しました。Connectのコンソールにアップデートがあり、HTMLが現在と変更があったときに対応できるようにするためです。

ServerlessFrameworkを利用し、日次でLambdaを実行させ、問題なくデフォルトフローの更新を実施できているかを確認するLambdaになってます。Lambdaの処理がエラーになった場合は、Slackに通知するようになっています。

ディレクトリ構成は以下の通りです。

connect-setup-headless
├─ bin
│     ├─ chromedriver
│     └─ headless-chromium
├─file
│     ├─Default agent transfer.json
│     └─Default outbound.json
├─handler.py
├─config.yaml
├─serverless.yaml
└─requirements.txt

serverless.yaml

  • selenium 、PyYAMLといった pip でインストールしたパッケージは serverless-python-requirements を利用してzip化しています
  • binフォルダのバイナリファイル（/bin）、更新するデフォルトフローの設定内容の入ったjsonファイル（/file）は packageの include でzip化しています
    • 必要ないファイル除外のために exclude で一旦全ファイル除外して、 include で必要分追加する形になっています
service: connect-defaultflow-update

provider:
  name: aws
  runtime: python3.6
  stage: ${opt:stage, self:custom.defaultStage}
  region: ap-northeast-1
  timeout: 400
  tags:
    owner: connect


custom:
  defaultStage: dev

plugins:
  - serverless-python-requirements

package:
  include:
    - handler.py
    - config.yaml
    - bin/chromedriver
    - bin/headless-chromium
    - file/**
  exclude:
    - ./**

functions:
  connect-defaultflow-update:
    handler: handler.main
    name: ${self:provider.stage}-connect-defaultflow-update
    events:
      - schedule: cron(0 0 ? * MON-FRI *)

 

handler.py

import os
import signal
import time
import yaml
import urllib.request
import subprocess
import logging
import json
from selenium import webdriver
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.chrome.options import Options


def main(event, context):
    try:
        options = Options()
        options.binary_location = './bin/headless-chromium'
        options.add_argument('--headless')
        options.add_argument('--no-sandbox')
        options.add_argument('--disable-gpu')
        options.add_argument('--single-process')
        options.add_argument('--disable-dev-shm-usage')
        driver = webdriver.Chrome(executable_path='./bin/chromedriver',chrome_options=options)
        driver.maximize_window()
        element_dict = get_element_dict()
        connect_domain = 'https://%s.awsapps.com/connect/' % element_dict['SendKeys']['InstanceName']

        login_contact_center(driver, element_dict)

        for flow_name in element_dict['FlowNames']:
            file_path = element_dict['SendKeys']['FilePath']['Value'] + flow_name + ".json"
            update_flow(driver, element_dict, flow_name, file_path)
            print(str(flow_name) + "完了")

        print("全フロー変更完了")

    except Exception:
        import traceback

        logger = logging.getLogger()
        logger.error(traceback.format_exc())
        notice_in_failure(traceback.format_exc())
    finally:
        os.kill(driver.service.process.pid, signal.SIGTERM)


# Login Connect Instance
def login_contact_center(driver,element_dict):
    driver.get(element_dict['DriverGet']['ConnectConsole'])
    time.sleep(3)
    input_data(driver, element_dict['SendKeys']['ConnectLoginUserName'], 'id')
    input_data(driver, element_dict['SendKeys']['ConnectLoginPassword'], 'id')
    time.sleep(3)
    click_button(driver, element_dict['Click']['ConnectLogin'], 'id')
    time.sleep(3)

# Update Flow
def update_flow(driver, element_dict, flow_name, file_path):
    driver.get(element_dict['DriverGet']['ContactFlowSetting'])
    time.sleep(3)
    click_button(driver, flow_name, 'link')
    click_button(driver, element_dict['Click']['SaveNext'], 'css')
    click_button(driver, element_dict['Click']['SaveSelect'], 'css')
    time.sleep(3)
    upload_file(driver, element_dict['SendKeys']['FilePath'], file_path)
    time.sleep(3)
    click_button(driver, element_dict['Click']['FileImport'], 'css')
    time.sleep(3)
    click_button(driver, element_dict['Click']['FlowPublish'], 'css')
    click_button(driver, element_dict['Click']['PublishTrue'], 'css')
    time.sleep(3)



def get_element_dict():
    with open('./config.yaml') as file:
        element_dict = yaml.safe_load(file)
        return element_dict

def input_data(driver, data_dict, element_type):
    value = data_dict['Value']

    if element_type == 'id':
        if len(driver.find_elements_by_id(data_dict['Key'])) > 0:
            field = driver.find_element_by_id(data_dict['Key'])
        else:
            print(json.dumps(data_dict['Key']) + "のHTMLタグが存在していません")
            field = driver.find_element_by_id(data_dict['Key'])
    elif element_type == 'css':
        if len(driver.find_elements_by_css_selector(data_dict['Key'])) > 0:
            field = driver.find_element_by_css_selector(data_dict['Key'])
        else:
            print(json.dumps(data_dict['Key']) + "のHTMLタグが存在していません")
            field = driver.find_elements_by_css_selector(data_dict['Key'])

    if value == 'SelectRadioButton':
        field.send_keys(Keys.SPACE)
    else:
        field.send_keys(value)
    time.sleep(1)

def upload_file(driver, data_dict, value):
    if len(driver.find_elements_by_id(data_dict['Key'])) > 0:
        field = driver.find_element_by_id(data_dict['Key'])
    else:
        print(json.dumps(data_dict['Key']) + "のHTMLタグが存在していません")
        field = driver.find_element_by_id(data_dict['Key'])

    field.send_keys(value)
    time.sleep(3)

def click_button(driver, key, element_type):
    if element_type == 'id':
        if len(driver.find_elements_by_id(key)) > 0:
            button = driver.find_element_by_id(key)
        else:
            print(json.dumps(key) + "のHTMLタグが存在していません")
            button = driver.find_element_by_id(key)
    elif element_type == 'css':
        if len(driver.find_elements_by_css_selector(key)) > 0:
            button = driver.find_element_by_css_selector(key)
        else:
            print(json.dumps(key) + "のHTMLタグが存在していません")
            button = driver.find_element_by_css_selector(key)
    elif element_type == 'link':
        if len(driver.find_elements_by_link_text(key)) > 0:
            button = driver.find_element_by_link_text(key)
        else:
            print(json.dumps(key) + "のHTMLタグが存在していません")
            button = driver.find_element_by_link_text(key)

    button.click()
    time.sleep(1)


def notice_in_failure(err_str):

    message_template = """
    <@xxxxxx> <@xxxxxx> <@xxxxxx>
    Connectの初期セットアップ自動化処理用Lambda関数が失敗しました。
    エラー内容は以下の通りです。
    """

    send_data = {
        "text": message_template + "```" + err_str + "```"
    }
    send_text = "payload=" + json.dumps(send_data)
    request = urllib.request.Request(
        'https://hooks.slack.com/services/xxxxxx/xxxxxx/xxxxxx',
        data = send_text.encode('utf-8'),
        method="POST"
        )

    with urllib.request.urlopen(request) as response:
        response_body = response.read().decode("utf-8")service: connect-defaultflow-update

 

エラーの場合は以下のようにSlackに通知されます。

最後に

Selenium(Python)で問い合わせフローを自動で更新する方法とその定期テスト基盤の作成方法をご紹介しました。

ConnectがAPI公開することを願いつつ、もしこのスクリプトのいい活用方法が思いつきましたら使ってみてください。