Webex会議のレコーディングをダウンロードまたは再生した人々をリストアップする方法

turatake · ‎2023-04-19

Webex Control Hubで会議の録画に関するレポートを閲覧およびエクスポートすることができます。

Webex Control Hubで利用可能なレコーディングのレポートのデータ：

録画ファイルが生成された日付
会議のタイトル（トピック）
会議の主催者（ユーザ名および表示名）
ファイルサイズ
録画方式（ネットワークベースまたはローカル録画）

上記とは別に各レコーディングのファイルを誰がいつ再生（リプレイ）またはダウンロードしたのかの詳細な一覧を取得したいというご要望を頂戴することがありました。現在はその情報をWebex APIを利用して取得することが可能です。

利用可能なWebex API：

これらのAPIにパラメータとして主催者（ホスト）のメールアドレスとレコーディングを識別するIDを与えれば再生またはダウンロードした人のメールアドレスなどの一覧を取得することができます。

今回はこのAPIを利用したサンプルコード（Python）を例示しつつ、APIの利用方法をご説明します。皆様が使いやすいツールの開発の参考になれば幸いです。

まず最初にサマリーの取得を行います。

url_report_summary = 'https://webexapis.com/v1/recordingReport/accessSummary'

# ページネーション処理のための初期設定
max_results = 100
start = 0
has_more = True

# すべてのレコーディングを取得するために、ページネーション処理を繰り返す
while has_more:
    # リクエストパラメータを設定
    params = {
        'max': max_results,
        'hostEmail': 'cholland@xxx.xxxsandbox.co',
        'from': '2023-04-16T00:00:00Z',
        'to': '2023-04-18T00:00:00Z',
        'start': start
    }

    # Webex APIにリクエストを送信
    response = requests.get(url_report_summary, 
                            headers={'Authorization': f'Bearer {access_token}'}, 
                            params=params)
    if response.status_code == 200:
        data = response.json()

    # ページネーション処理のために、次のページがあるかどうかをチェック
    has_more = response.json().get('hasNext', False)
    start = response.json().get('next', 0)

上記のようなコードで /v1/recordingReport/accessSummary エンドポイントをコールすることで以下のような結果が返ります。（なお、管理者はパラメータhostEmailの指定が必須であることを忘れないようにしてください。）

{
  "items": [
    {
      "recordingId": "701113dfbf0f103b9bff0050xxxxxxxx",
      "topic": "Charles Holland's Personal Room-20230417 0534-1",
      "hostEmail": "cholland@xxx.xxxsandbox.co",
      "viewCount": 0,
      "downloadCount": 1,
      "siteUrl": "xxx-sandbox-x.webex.com",
      "timeRecorded": "2023-04-17T05:34:25Z"
    },
    {
      "recordingId": "44f6d6dabf0f103bbffd0050xxxxxxxx",
      "topic": "Charles Holland's Personal Room-20230417 0533-1",
      "hostEmail": "cholland@xxx.xxxsandbox.co",
      "viewCount": 1,
      "downloadCount": 0,
      "siteUrl": "xxx-sandbox-x.webex.com",
      "timeRecorded": "2023-04-17T05:33:12Z"
    }
  ]
}

これはサマリーを返すエンドポイントですのでその結果にはレコーディングを再生した回数 (viewCount) とダウンロードした回数 (downloadCount) があるものの、誰がいつという情報は含まれていません。

次のステップではここで得られたrecordingIdとhostEmailを利用して /v1/recordingReport/accessSummary エンドポイントにアクセスします。

url_report_detail = 'https://webexapis.com/v1/recordingReport/accessDetail'

# リクエストパラメータを設定
params = {
    'max': max_results,
    'recordingId': recordingId,
    'hostEmail': hostEmail,
    'start': start
}

# Webex APIにリクエストを送信
response = requests.get(url_report_detail, 
                        headers={'Authorization': f'Bearer {access_token}'}, 
                        params=params)
if response.status_code == 200:
    # レスポンスから詳細のリストを取得
    data = response.json()

レスポンスは以下のようになります

{
  "items": [
    {
      "recordingId": "701113dfbf0f103b9bff0050xxxxxxxx",
      "topic": "Charles Holland's Personal Room-20230417 0534-1",
      "name": "Charles Holland",
      "email": "cholland@xxx.xxxsandbox.co",
      "accessTime": "2023-04-17T05:37:39Z",
      "downloaded": true,
      "viewed": false
    }
  ]
}

再生またはダウンロードがいつ誰によって、どちらを行なったのかの情報が得られました。

主催者のメールアドレスの情報があればこの情報を引き出すことができますが、このメールアドレスを得る方法としましてはAPIで検索する方法と、Control Hubの前述のレポート中に含まれるものを利用するなどがあります。

今回はControl Hubからエクスポートしたレポートを入力とし、Webex APIにアクセスして再生およびダウンロードに関する詳細なレポートを取得するデモを作ってみました。

このデモ用のサンプルを起動し、Control HubからダウンロードしたCSV形式のレコーディングのレポートをファイルのアップロードエリア ([Drag and drop file here] と表示された部分) から選択します。アップロードされますとファイルの中をパースし、ホストのメールアドレスの一覧を取得します。また日付もフィルター用に調整できるようになっています。

そして [Go]ボタンを押すとサマリーを取得する /v1/recordingReport/accessSummary エンドポイントへPOSTリクエストを送信します。パラメータはホストのメールアドレス（と日付）です。レポートの中に複数のホストが存在する場合は含まれるすべてのホストに関して順次再起的にコールし続けます。

結果はレコーディングのサマリーという部分に表形式で表示しています。JSON形式のレスポンスをシンプルに表 (Pandas DataFrame) に変換しているだけです。今回は何もしていませんが、WebexサイトのURLやホストのドメイン名などでフィルターできるようにするなどすりともっと良いかもしれません。

このアプリでは、アプリのユーザがレコーディングのサマリーにある表のいずれかの行をクリックして選択すると、レコーディングの詳細を取得する /v1/recordingReport/accessSummary エンドポイントにアクセスして再生やダウンロードの詳細なタイムスタンプなどについて取得し、その結果をレコーディングの詳細にある表に表示します。もしそれをCSVでダウンロードしたい場合はこのデモのように [CSVダウンロード]ボタンを追加するのも良いかもしれません。

データはいずれもPandas DataFrameですので、日付や時間でグルーピングしてダウンロード件数の推移をグラフ化したり、再生した方々のドメイン名で集計したり、あるいはサマリーと詳細の双方を一回のアクションですべて取得してしまうなどさまざまな応用が可能だと思います。

さらに、再生した方々のメールアドレスにフォローアップのアンケートを送ってみるなどの活用方法もあるかもしれません。

このソースコードは以下の通りです。（実用には耐えないと思いますので、ぜひただの参考としてご覧になるだけにとどめてより良いものをお作りください。）

import datetime

import pandas as pd
import requests
import streamlit as st
from st_aggrid import AgGrid
from st_aggrid.grid_options_builder import GridOptionsBuilder
from st_aggrid.shared import GridUpdateMode

# Webex APIの認証情報を設定 (developer.webex.comのTry ItまたはMy Appsで取得します)
access_token = 'YOUR_ACCESS_TOKEN_HERE'

# リクエストを送信するための基本URLを設定
url_report_summary = 'https://webexapis.com/v1/recordingReport/accessSummary'
url_report_detail = 'https://webexapis.com/v1/recordingReport/accessDetail'

def validate_format(fields) -> bool:
    fields_required = ['Creation Date', 'Host User Name']
    for f in fields_required:
        if f not in fields:
            return False
    return True

def set_warning(message) -> None:
    st.session_state['warning'] = message

def clear_warning() -> None:
    if 'warning' in st.session_state:
        del st.session_state['warning']

def check_search_conditions() -> bool:
    # Streamlit UIで日付と日時の選択が正しいかをチェックします
    dt_from = datetime.datetime.combine(date_from, time_from)
    dt_to = datetime.datetime.combine(date_to, time_to)
    td = int(dt_to.timestamp()) - int(dt_from.timestamp())
    if td > 0:
        if td > 3600:
            st.session_state['from'] = dt_from.strftime('%Y-%m-%dT%H:%M:%SZ')
            st.session_state['to'] = dt_to.strftime('%Y-%m-%dT%H:%M:%SZ')
            clear_warning()
            # 日時の選択に問題がなければTrueを返します
            return True
        else:
            set_warning('From日時とTo日時の間隔が狭すぎます')
    else:
        set_warning('ToはFromより前の日時を指定できません')
    return False

def get_recording_report_summary() -> None:
    if not check_search_conditions():
        return

    if 'host_list' not in st.session_state:
        return

    if 'df_details' in st.session_state:
        del st.session_state['df_details']

    if 'selected_rows' in st.session_state:
        del st.session_state['selected_rows']

    if 'df_recordings' in st.session_state:
        del st.session_state['df_recordings']

    # APIから取得したレコーディングのサマリーを一時的に格納するリスト
    recording_list = []

    for host in st.session_state['host_list']:
        # ページネーション処理のための初期設定
        max_results = 100
        start = 0
        has_more = True

        # すべてのレコーディングを取得するために、ページネーション処理を繰り返す
        while has_more:
            # リクエストパラメータを設定
            params = {
                'max': max_results,
                'hostEmail': host,
                'from': st.session_state['from'],
                'to': st.session_state['to'],
                'start': start
            }

            # Webex APIにリクエストを送信
            response = requests.get(url_report_summary, 
                                    headers={'Authorization': f'Bearer {access_token}'}, 
                                    params=params)
            if response.status_code == 200:
                print(f'{url_report_summary} -> {response.status_code} {response.reason}')
            else:
                print(f'{url_report_summary} -> {response.status_code} {response.reason} : {response.text}')

            # レスポンスからレコーディングのリストを取得
            recordings = response.json().get('items', [])

            # 取得したレコーディングのリストを処理
            for recording in recordings:
                # ここに取得したレコーディングの情報を処理するコードを記述
                recording_list.append({
                    'recordingId': recording.get('recordingId'),
                    'topic': recording.get('topic'),
                    'siteUrl': recording.get('siteUrl'),
                    'hostEmail': recording.get('hostEmail'),
                    'viewCount': recording.get('viewCount'),
                    'downloadCount': recording.get('downloadCount'),
                })

            # ページネーション処理のために、次のページがあるかどうかをチェック
            has_more = response.json().get('hasNext', False)
            start = response.json().get('next', 0)

    # レコーディングのリストをPandasのDataFrameに変換
    df_recordings = pd.DataFrame(recording_list)
    st.session_state['df_recordings'] = df_recordings


def get_recording_report_detail() -> None:
    if 'recordingId' not in st.session_state:
        print('recordingId does not exist in st.session_state')
        return

    # APIから取得したレコーディングの詳細を一時的に格納するリスト
    item_list = []

    # ページネーション処理のための初期設定
    max_results = 100
    start = 0
    has_more = True

    # すべての詳細を取得するために、ページネーション処理を繰り返す
    while has_more:
        # リクエストパラメータを設定
        params = {
            'max': max_results,
            'recordingId': st.session_state['recordingId'],
            'hostEmail': st.session_state['hostEmail'],
            'start': start
        }

        # Webex APIにリクエストを送信
        response = requests.get(url_report_detail, 
                                headers={'Authorization': f'Bearer {access_token}'}, 
                                params=params)
        if response.status_code == 200:
            print(f'{url_report_detail} -> {response.status_code} {response.reason}')
        else:
            print(f'{url_report_detail} -> {response.status_code} {response.reason} : {response.text}')

        # レスポンスから詳細のリストを取得
        items = response.json().get('items', [])

        # 取得した詳細のリストを処理
        for item in items:
            # ここに取得した詳細の情報を処理するコードを記述
            item_list.append({
                'email': item.get('email'),
                'accessTime': item.get('accessTime'),
                'viewed': item.get('viewed'),
                'downloaded': item.get('downloaded')
            })

        # ページネーション処理のために、次のページがあるかどうかをチェック
        has_more = response.json().get('hasNext', False)
        start = response.json().get('next', 0)

    # 詳細のリストをPandasのDataFrameに変換
    st.session_state['df_details'] = pd.DataFrame(item_list)
    del st.session_state['hostEmail']
    del st.session_state['recordingId']


st.set_page_config(page_title='Webex API - Recording Report Sample', layout="wide")

# Streamlit ページの描画
# -------------------------------------------------------------------------------------------
if 'warning' in st.session_state:
    # 警告メッセージがある場合は表示
    st.warning(st.session_state['warning'], icon="")

st.header('Webex API - Recording Report サンプル')
with st.container():
    st.subheader('レコーディング検索オプション')

    file = st.file_uploader("CSV形式の録音の主催者レポートファイルを選択",
                            type=['csv'],
                            accept_multiple_files=False)
    if file is not None:
        try:
            df_report = pd.read_csv(file, encoding='UTF-16LE', sep='\t')
            if not validate_format(df_report.columns):
                set_warning('CSVファイルの書式はサポートされません')
            else:
                clear_warning()
                row_count = len(df_report)
                st.session_state['df_report'] = df_report
        except EmptyDataError:
            print("空のファイルでした")
            set_warning('ファイルの中身が空でした')

    if 'df_report' in st.session_state:
        df = st.session_state['df_report']
        st.write(f"ファイル中のレコード件数: {len(df)}")

        host_list = pd.unique(df['Host User Name'])
        st.write(f"ユニークな主催者数: {len(host_list)}")
        st.session_state['host_list'] = host_list

        dates = pd.unique(df['Creation Date'])
        date_list = []
        for d in dates:
            [month, day, year, *_] = d.split('/')
            date_list.append(datetime.datetime(int(year), int(month), int(day)))
        sorted_date_list = sorted(date_list)
        date_oldest = sorted_date_list[0]
        date_latest = sorted_date_list[-1]

        # 検索条件の開始および終了の日時を選択するUI
        st.empty()
        col1, col2, col3, col4 = st.columns(4)
        with col1:
            date_from = st.date_input('From (日付)', value=date_oldest)
        with col2:
            time_from = st.time_input('From (時間)', value=date_oldest, step=3600)
        with col3:
            date_to = st.date_input('To (日付)', value=date_latest)
        with col4:
            time_to = st.time_input('To (時間)', value=date_latest, step=3600)

        # 検索実行のための [Go] ボタン -> サマリを取得
        # サマリの一覧を取得するAPI (/v1/recordingReport/accessDetail) をコール
        st.button('Go', on_click=get_recording_report_summary)

# APIでサマリの取得を完了していた場合は表形式で結果を表示
if 'df_recordings' in st.session_state:
    with st.container():
        st.empty()
        st.subheader('レコーディングのサマリー')
        record_count = len(st.session_state['df_recordings'])
        st.write(f'取得件数: {record_count}')

        gb = GridOptionsBuilder.from_dataframe(st.session_state['df_recordings'])
        gb.configure_selection(selection_mode='single')
        gridOptions = gb.build()
        # tail(20)はデータ件数が多い場合を想定して末尾20件だけに制限するため。Paginationの実装が理想的。
        grid = AgGrid(st.session_state['df_recordings'].tail(20),
                      gridOptions=gridOptions,
                      update_mode=GridUpdateMode.SELECTION_CHANGED)
        st.session_state['selected_rows'] = grid['selected_rows']

# サマリ一覧表上で特定の列が選択された場合に選択されたサマリの詳細を取得するAPI (/v1/recordingReport/accessDetail) をコール
if 'selected_rows' in st.session_state:
    df_selected_rows = pd.DataFrame(st.session_state['selected_rows'])
    if len(df_selected_rows) == 1:
        # 詳細を取得する対象のレコーディングについてIDを保持
        st.session_state['hostEmail'] = df_selected_rows['hostEmail']
        st.session_state['recordingId'] = df_selected_rows['recordingId']
        get_recording_report_detail()

# サマリの詳細を取得することに成功した場合はその結果を一覧表示する表を描画
if 'df_details' in st.session_state:
    with st.container():
        st.empty()
        st.subheader('レコーディングの詳細')
        st.write(f"取得件数: {len(st.session_state['df_details'])}")
        st.table(st.session_state['df_details'])
        df_detail_report = st.session_state['df_details']
        csv = df_detail_report.to_csv(index=False).encode('utf-8')

        # CSVでダウンロードをしたい時用のボタン (おまけ)
        st.download_button(
            "CSVダウンロード",
            csv,
            "webex_recordings_report_details.csv",
            "text/csv",
            key='download-csv'
        )

今回のサンプルは Streamlit を利用していますので、以下のようにして実行し、Webブラウザ上で操作しています。

streamlit run /home/turatake/webex_recording_download_history.py

余談ですが、Webexでは会議のレコーディングを組織内の従業員に限定したり（サインインを必須とするオプションをControl Hubで有効化できます）、さらに、従業員であっても会社が支給するデバイスからのみレコーディングにアクセスできるようにより厳しい制限を課す（これ自体はWebexサービスの機能ではございませんが、WebexがIdPとSAML SSO連携し、IdPでデバイスの電子証明書などの有無をサインインの際にチェックすることができます）ことも可能ですので、ご興味があればご確認ください。