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

#Amazon Connect

#AWS

リサリサ

2022.09.22

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

こんにちは、リサリサです。最近は 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'])