Amazon Connect でコンタクトフローとキューやモジュールの依存関係を一覧で出力してみた

こんにちは、リサリサです。最近は Amazon Connect でコンタクトフローを作っているのですが、どのフローでどのキューを使っているのか、どのモジュールを使っているのかなどを1フローずつ目で見るのが大変すぎてスクリプトを書いたので記事にしてみます。

やりたい事

コンタクトフローで使われているキュー、フロー、プロンプト、モジュールを一覧で出力したい

やってみた

import boto3
import json

client = boto3.client('connect')

def list_contact_flows(instance_id):

    # 1度に100フローしか取得できないので、NEXT_TOKEN で複数回に分けて全フロー取得
    list_contact_flows = []
    next_token = ''
    done = False
    while not done:
        if next_token:
            response = client.list_contact_flows(
                InstanceId=instance_id,
                NextToken=next_token
                )
        else:
            response = client.list_contact_flows(
                InstanceId=instance_id,
                )
        list_contact_flows.extend(response.get('ContactFlowSummaryList', []))
        next_token = response.get('NextToken', None)
        done = next_token is None
    return list_contact_flows

def get_flow_list(instance_id,instance_name):
    # あとで顧客キューフローなどのタイプの確認をしたいので、全フローのlistを取得
    all_flows = list_contact_flows(instance_id)

    # コンタクトフローのみ絞り込み
    contact_flows = [ contact_flow for contact_flow in all_flows if contact_flow['ContactFlowType'] == 'CONTACT_FLOW']

    output = []
    for contact_flow in contact_flows:
            try:
                describe_contact_flow = client.describe_contact_flow(
                    InstanceId=instance_id,
                    ContactFlowId=contact_flow['Id']
                )
                action_metadata=json.loads(describe_contact_flow['ContactFlow']['Content'])['Metadata']['ActionMetadata']
                for block_id in action_metadata:
                    module_type = ''
                    module_name = ''
                    if 'contactFlowModuleName' in action_metadata[block_id]:
                        # モジュール
                        module_type = 'MODULE'
                        module_name = action_metadata[block_id]['contactFlowModuleName']
                        arn = ''
                    elif 'promptName' in action_metadata[block_id]:
                        # プロンプト
                        module_type = 'PROMPT'
                        module_name = action_metadata[block_id]['promptName']
                        arn = ''
                    elif 'queue' in action_metadata[block_id]:
                        # キュー
                        module_type = 'QUEUE'
                        if type(action_metadata[block_id]['queue']) is dict:
                            module_name = action_metadata[block_id]['queue']['text']
                            arn = action_metadata[block_id]['queue'].get('id', '')
                        else:
                            # queueがdictになっていない場合がある
                            module_name = action_metadata[block_id]['queue']
                            arn = ''
                    elif 'contactFlow' in action_metadata[block_id]:
                        # 顧客キューフローや顧客保留フローなど
                        # describe_contact_flowではフローのタイプ(顧客キューフローや顧客保留フローなど)が分からないので、list_contact_flowsから対象フローのContactFlowTypesを検索
                        module_type = [ contact_flow for contact_flow in all_flows if contact_flow['Arn'] == action_metadata[block_id]['contactFlow']['id'] ][0]['ContactFlowType']
                        arn = action_metadata[block_id]['contactFlow']['id']
                        module_name=action_metadata[block_id]['contactFlow']['text']
                    if module_type:
                        output.append({
                            'flow_name':contact_flow['Name'],
                            'type':module_type,
                            'name':module_name,
                            'arn':arn
                        })
            except client.exceptions.ContactFlowNotPublishedException:
                # 公開されていないフローを取得しようとすると、botocore.errorfactory.ContactFlowNotPublishedException が発生する
                # list_contact_flowsでは判別出来ないので、exceptで対応
                output.append({
                    'flow_name':contact_flow['Name'],
                    'type':'未公開のため取得不可',
                    'name':'未公開のため取得不可',
                    'arn':'未公開のため取得不可'
                })
                print(f"{contact_flow['Name']}は未公開")

    import csv
    output.sort(key=lambda x: x['flow_name'])

    # CSVのヘッダー
    field_name = ['flow_name','type','name','arn']
    with open(f"{instance_name}_flow_dependence_list.csv",'w',encoding='utf-8') as csvfile:
        writer = csv.DictWriter(csvfile, fieldnames = field_name)
        writer.writeheader()
        writer.writerows(output)

# 全connectインスタンスを取得
instances = client.list_instances()['InstanceSummaryList']
for instance in instances:
    get_flow_list(instance['Id'],instance['InstanceAlias'])

内容

実行するとconnectインスタンスごとに、{instance名}_flow_dependence_list.csv が出力されます。

コンタクトフローで使われているキュー、フロー、プロンプト、モジュールの一覧が取得できます。

フローとキューは名前と合わせて arn も出力します。(キューは設定した時期によってarnが取得できたりできなかったりするので、取得できた場合のみ出力します。)

type には MODULE,PLOMPT,QUEUE とフロータイプ(CUSTOMER_QUEUE,COUSOMER_HOLD など)を出力します。

保存済みで公開していないフローは、APIの特性上、詳細の取得ができないので、「未公開のため取得不可」として出力します。

最後に

どなたかのお役に立てば幸いです。