Python と Google Meet REST API を使用して会議イベントを監視する

このチュートリアルでは、Google Meet REST API と Google Workspace Events API、Google Cloud Pub/Sub を使用して、Meet のミーティング スペースでイベントをモニタリングし、イベントに応答する方法について説明します。サンプル アプリケーションは、会議の開始と終了、参加者の参加と退出、生成された会議アーティファクトが利用可能になったタイミングを記録します。

特定の会議スペースに登録する代わりに、Meet ユーザーに登録して、そのユーザーが所有または主催するすべての会議スペースのイベントを受け取ることができます。詳しくは、Google Workspace Events API ドキュメントの Google Meet のイベントに登録するをご覧ください。

前提条件

組織で次の前提条件を有効にする必要がある場合は、Google Workspace 管理者に有効にするよう依頼してください。

• Google Meet へのアクセス権を持つ Google Workspace アカウント。
• Google Cloud プロジェクトを作成するためのアクセス権。
• Python 3 がインストールされている。
• gcloud CLI がインストールされていること。

環境を準備する

このセクションでは、このチュートリアルのローカル環境と Google Cloud プロジェクトを作成して構成する方法について説明します。

作業ディレクトリと Python 仮想環境を作成する

新しい仮想環境を作成して有効にするには、ターミナルで次のコマンドを実行します。

mkdir meet-tutorial
cd meet-tutorial
python3 -mvenv env
source env/bin/activate

mkdir meet-tutorial
cd meet-tutorial
python3 -mvenv env
env/bin/activate.bat

mkdir meet-tutorial
cd meet-tutorial
python3 -mvenv env
env/bin/activate.ps1

Google Cloud プロジェクトを作成する

Google Cloud プロジェクトの課金を有効にする

認証と権限付与の設定

認証と認可により、アプリは Meet REST API リソースにアクセスできます。Meet REST API を呼び出すには、ユーザーの承認が必要です。このセクションでは、ユーザー認証情報を構成して認可をリクエストする方法について説明します。

OAuth 同意画面を設定し、スコープを選択する

次の手順では、アプリの OAuth 同意画面を構成するためのプレースホルダ情報を示します。アプリを外部に公開する前に、この情報を更新してください。

1. Google Cloud コンソールで、メニュー menu > [API とサービス] > [OAuth 同意画面] に移動します。

   OAuth 同意画面に移動

2. [ユーザーの種類] で [内部] を選択し、[作成] をクリックします。
3. [アプリ名] に「Meet REST API Tutorial」と入力します。
4. アプリ登録フォームに入力し、[保存して続行] をクリックします。
5. [スコープを追加または削除] をクリックします。Google Cloud プロジェクトで有効にした各 API のスコープのリストを含むパネルが表示されます。
6. [スコープを手動で追加] に、次のスコープを貼り付けます。
   • https://www.googleapis.com/auth/meetings.space.created
7. [表に追加] をクリックします。
8. [更新] をクリックします。
9. アプリに必要なスコープを選択したら、[保存して次へ] をクリックします。
10. ユーザータイプに [外部] を選択した場合は、テストユーザーを追加します。
    1. [テストユーザー] で [ユーザーを追加] をクリックします。
    2. メールアドレスと他の承認済みテストユーザーを入力し、[保存して次へ] をクリックします。
11. アプリ登録の概要を確認します。変更するには、[編集] をクリックします。アプリの登録に問題がなければ、[ダッシュボードに戻る] をクリックします。

クライアント ID の作成

クライアント ID は、OAuth 2.0 フロー中にアプリケーションの認証情報として機能します。アプリはローカルで実行されるため、デスクトップ クライアント ID を作成します。

1. Google Cloud コンソールで、メニュー menu > [API とサービス] > [認証情報] に移動します。

   [認証情報] に移動

2. [認証情報を作成] > [OAuth クライアント ID] をクリックします。
3. [アプリケーションの種類] > [デスクトップ アプリ] をクリックします。
4. [名前] フィールドに、認証情報の名前を入力します。この名前は Google Cloud コンソールにのみ表示されます。
5. [作成] をクリックします。[OAuth クライアントを作成しました] 画面が表示され、新しいクライアント ID とクライアント シークレットが表示されます。
6. [OK] をクリックします。新しく作成された認証情報が [OAuth 2.0 クライアント ID] の下に表示されます。

Google 認証ライブラリをインストールする

Google 認証ライブラリをインストールします。

pip install google-auth google-auth-oauthlib

承認を実行する

Meet REST API では、OAuth 2.0 アクセス トークンの形式のユーザー認証情報が必要です。このセクションでは、OAuth 2.0 フローを実装して、ユーザーのアクセス トークンと更新トークンをリクエストします。

1. 作業ディレクトリに main.py ファイルを作成し、次の内容を追加します。

   import os
   import json
   
   from google.auth.transport import requests
   from google.oauth2.credentials import Credentials
   from google_auth_oauthlib.flow import InstalledAppFlow
   
   def authorize() -> Credentials:
       """Ensure valid credentials for calling the Meet REST API."""
       CLIENT_SECRET_FILE = "./client_secret.json"
       credentials = None
   
       if os.path.exists('token.json'):
           credentials = Credentials.from_authorized_user_file('token.json')
   
       if credentials is None:
           flow = InstalledAppFlow.from_client_secrets_file(
               CLIENT_SECRET_FILE,
               scopes=[
                   'https://www.googleapis.com/auth/meetings.space.created',
               ])
           flow.run_local_server(port=0)
           credentials = flow.credentials
   
       if credentials and credentials.expired:
           credentials.refresh(requests.Request())
   
       if credentials is not None:
           with open("token.json", "w") as f:
               f.write(credentials.to_json())
   
       return credentials
   
   USER_CREDENTIALS = authorize()
   

2. コードを実行するには、先ほど作成したクライアント ID とシークレットの両方が必要です。ダウンロードしたクライアント シークレット ファイルをプロジェクトの作業ディレクトリにコピーし、名前を client_secret.json に変更します。

3. 認可の仕組みをテストするには、次のコマンドを実行します。アプリは認可を求め、リクエストが承認されると、プロジェクトの作業ディレクトリに token.json ファイルを作成します。

   python3 main.py
   

Meet REST API を追加する

認可コードが完成したので、Meet REST API を有効にして呼び出します。

API を有効にする

このセクションでは Meet REST API に焦点を当てていますが、このチュートリアルでは Google Cloud Pub/Sub と Google Workspace Events API も使用します。

Meet REST API クライアント ライブラリをインストールする

Meet REST API クライアント ライブラリをインストールする手順は次のとおりです。

1. 次のコマンドを実行します。

   pip install google-apps-meet
   

2. main.py ファイルを編集してクライアントをインポートします。

   from google.apps import meet_v2 as meet
   

スペースを作成

Meet REST API が利用可能になったので、サブスクライブできる会議スペースを作成する関数を定義します。

main.py を編集して、次のように追加します。

def create_space() -> meet.Space:
    """Create a meeting space."""
    client = meet.SpacesServiceClient(credentials=USER_CREDENTIALS)
    request = meet.CreateSpaceRequest()
    return client.create_space(request=request)

イベントに登録する

会議スペースに関するイベントを受信するには、Google Workspace Events API を使用してサブスクリプションを作成します。また、アプリがイベントを受信する通知エンドポイントとして機能する Google Cloud Pub/Sub トピックを作成してサブスクライブする必要があります。

Google Cloud Pub/Sub を構成する

Pub/Sub トピックを作成してサブスクライブする手順は次のとおりです。

トピック名をメモし、{project} の値がアプリの Cloud プロジェクト ID であることを確認します。トピック名は、後で Google Workspace サブスクリプションを作成する際に使用します。

サービス アカウントを作成する

サービス アカウントを使用する

サービス アカウントを作成したら、サービス アカウントの権限借用にアクセス権を付与します。

Pub/Sub クライアント ライブラリをインストールする

1. pip を使用して、Pub/Sub のクライアント ライブラリをインストールします。

   pip install google-cloud-pubsub
   

2. 次に、main.py を編集してクライアントをインポートします。

   from google.cloud import pubsub_v1
   

Google Workspace サブスクリプションを作成する

次のコードを main.py に追加して、Meet イベントをサブスクライブするメソッドを定義します。このコードは、会議スペースのすべてのイベントをサブスクライブします。サブスクライブすると、イベントが Pub/Sub トピックに投稿されます。

def subscribe_to_space(space_name: str = None, topic_name: str = None):
    """Subscribe to events for a meeting space."""
    session = requests.AuthorizedSession(USER_CREDENTIALS)
    body = {
        'targetResource': f"//meet.googleapis.com/{space_name}",
        "eventTypes": [
            "google.workspace.meet.conference.v2.started",
            "google.workspace.meet.conference.v2.ended",
            "google.workspace.meet.participant.v2.joined",
            "google.workspace.meet.participant.v2.left",
            "google.workspace.meet.recording.v2.fileGenerated",
            "google.workspace.meet.transcript.v2.fileGenerated",
        ],
        "payloadOptions": {
            "includeResource": False,
        },
        "notificationEndpoint": {
            "pubsubTopic": topic_name
        },
        "ttl": "86400s",
    }
    response = session.post("https://workspaceevents.googleapis.com/v1/subscriptions", json=body)
    return response

次に、イベントを取得して処理する対応するコードを追加します。

イベントをリッスンして処理する

main.py の編集を続けて、次のサンプルコードを追加します。このコードは受信側を実装し、Google Cloud Pub/Sub API を使用して、利用可能になったイベントを pull します。さまざまなハンドラ メソッドは、対応するイベントに関する情報を出力します。

def format_participant(participant: meet.Participant) -> str:
    """Formats a participant for display on the console."""
    if participant.anonymous_user:
        return f"{participant.anonymous_user.display_name} (Anonymous)"

    if participant.signedin_user:
        return f"{participant.signedin_user.display_name} (ID: {participant.signedin_user.user})"

    if participant.phone_user:
        return f"{participant.phone_user.display_name} (Phone)"

    return "Unknown participant"


def fetch_participant_from_session(session_name: str) -> meet.Participant:
    """Fetches the participant for a session."""
    client = meet.ConferenceRecordsServiceClient(credentials=USER_CREDENTIALS)
    # Use the parent path of the session to fetch the participant details
    parsed_session_path = client.parse_participant_session_path(session_name)
    participant_resource_name = client.participant_path(
        parsed_session_path["conference_record"],
        parsed_session_path["participant"])
    return client.get_participant(name=participant_resource_name)


def on_conference_started(message: pubsub_v1.subscriber.message.Message):
    """Display information about a conference when started."""
    payload = json.loads(message.data)
    resource_name = payload.get("conferenceRecord").get("name")
    client = meet.ConferenceRecordsServiceClient(credentials=USER_CREDENTIALS)
    conference = client.get_conference_record(name=resource_name)
    print(f"Conference (ID {conference.name}) started at {conference.start_time.rfc3339()}")


def on_conference_ended(message: pubsub_v1.subscriber.message.Message):
    """Display information about a conference when ended."""
    payload = json.loads(message.data)
    resource_name = payload.get("conferenceRecord").get("name")
    client = meet.ConferenceRecordsServiceClient(credentials=USER_CREDENTIALS)
    conference = client.get_conference_record(name=resource_name)
    print(f"Conference (ID {conference.name}) ended at {conference.end_time.rfc3339()}")


def on_participant_joined(message: pubsub_v1.subscriber.message.Message):
    """Display information about a participant when they join a meeting."""
    payload = json.loads(message.data)
    resource_name = payload.get("participantSession").get("name")
    client = meet.ConferenceRecordsServiceClient(credentials=USER_CREDENTIALS)
    session = client.get_participant_session(name=resource_name)
    participant = fetch_participant_from_session(resource_name)
    display_name = format_participant(participant)
    print(f"{display_name} joined at {session.start_time.rfc3339()}")


def on_participant_left(message: pubsub_v1.subscriber.message.Message):
    """Display information about a participant when they leave a meeting."""
    payload = json.loads(message.data)
    resource_name = payload.get("participantSession").get("name")
    client = meet.ConferenceRecordsServiceClient(credentials=USER_CREDENTIALS)
    session = client.get_participant_session(name=resource_name)
    participant = fetch_participant_from_session(resource_name)
    display_name = format_participant(participant)
    print(f"{display_name} left at {session.end_time.rfc3339()}")


def on_recording_ready(message: pubsub_v1.subscriber.message.Message):
    """Display information about a recorded meeting when artifact is ready."""
    payload = json.loads(message.data)
    resource_name = payload.get("recording").get("name")
    client = meet.ConferenceRecordsServiceClient(credentials=USER_CREDENTIALS)
    recording = client.get_recording(name=resource_name)
    print(f"Recording available at {recording.drive_destination.export_uri}")


def on_transcript_ready(message: pubsub_v1.subscriber.message.Message):
    """Display information about a meeting transcript when artifact is ready."""
    payload = json.loads(message.data)
    resource_name = payload.get("transcript").get("name")
    client = meet.ConferenceRecordsServiceClient(credentials=USER_CREDENTIALS)
    transcript = client.get_transcript(name=resource_name)
    print(f"Transcript available at {transcript.docs_destination.export_uri}")


def on_message(message: pubsub_v1.subscriber.message.Message) -> None:
    """Handles an incoming event from the Google Cloud Pub/Sub API."""
    event_type = message.attributes.get("ce-type")
    handler = {
        "google.workspace.meet.conference.v2.started": on_conference_started,
        "google.workspace.meet.conference.v2.ended": on_conference_ended,
        "google.workspace.meet.participant.v2.joined": on_participant_joined,
        "google.workspace.meet.participant.v2.left": on_participant_left,
        "google.workspace.meet.recording.v2.fileGenerated": on_recording_ready,
        "google.workspace.meet.transcript.v2.fileGenerated": on_transcript_ready,
    }.get(event_type)

    try:
        if handler is not None:
            handler(message)
        message.ack()
    except Exception as error:
        print("Unable to process event")
        print(error)


def listen_for_events(subscription_name: str = None):
    """Subscribe to events on the subscription."""
    subscriber = pubsub_v1.SubscriberClient()
    with subscriber:
        future = subscriber.subscribe(subscription_name, callback=on_message)
        print("Listening for events")
        try:
            future.result()
        except KeyboardInterrupt:
            future.cancel()
    print("Done")

コードを完成させる

次のコードを main.py に追加して、メソッドを呼び出してスペースを作成、イベントに登録、リッスンします。TOPIC_NAME 定数と SUBSCRIPTION_NAME 定数を、前に作成した独自のトピック名とサブスクリプション名に更新します。

1. main.py にコードを追加します。

   space = create_space()
   print(f"Join the meeting at {space.meeting_uri}")
   
   TOPIC_NAME = "projects/PROJECT_ID/topics/TOPIC_ID"
   SUBSCRIPTION_NAME = "projects/PROJECT_ID/subscriptions/SUBSCRIPTION_ID"
   
   subscription = subscribe_to_space(topic_name=TOPIC_NAME, space_name=space.name)
   listen_for_events(subscription_name=SUBSCRIPTION_NAME)
   

   次のように置き換えます。

   • PROJECT_ID: アプリの一意の Cloud プロジェクト ID（my-sample-project-191923 など）。

   • TOPIC_ID: Cloud プロジェクトで作成した Pub/Sub トピックの名前。

   • SUBSCRIPTION_ID: サブスクリプションの名前（workspace-events-sub など）。

2. プログラムを実行します。

   python3 main.py
   

プログラムを初めて実行する場合は、承認を求めるメッセージが表示されます。Meet REST API を呼び出すためのアプリケーションへのアクセス権を付与します。プログラムが正常に実行されると、次のような出力が表示されます。

Join the meeting at https://meet.google.com/abc-mnop-xyz

会議に参加する

アプリケーションのイベントを生成するには、アプリケーションに表示された URL を使用して会議に参加します。参加したら、次のアクションを試してイベントをトリガーできます。

• 会議から退出して再び参加します。
• 他のユーザーを招待するか、スマートフォンからダイヤルインします。
• 録音と音声文字変換を有効にします。

これらのアクティビティはそれぞれ、アプリケーションが受信して Google Cloud コンソールにログに記録するイベントを生成します。

完了したら、ctrl-c を使用してプログラムを中断します。

省略可: その他の解決方法

アプリは、イベントに関する基本的な詳細をログに記録します。Meet REST API の詳細を確認するには、これらの追加アクションを実行するようにアプリケーションを変更してみてください。

• People API を使用して、ログインした参加者に関する追加情報を取得します。
• Google Drive API を使用して、録音と文字起こしをダウンロードします。
• Google ドライブから文字起こしをダウンロードする代わりに、Meet REST API の構造化文字起こしメソッドを使用して文字起こしを取得します。

オプション: クリーンアップ

このチュートリアルで使用したリソースに対して Google Cloud コンソール アカウントで課金されないようにするには、作成したリソースとプロジェクトをクリーンアップすることをおすすめします。

サブスクリプションを削除するには:

トピックを削除するには:

プロジェクトを削除するには、次の操作を行います。

関連トピック

• 定期購入できる Meet イベントの種類について学習する。
• Google Meet REST API の機能の詳細とリファレンス ドキュメントを確認する。
• Google Workspace Events API を使用して、Meet の会議スペースまたはユーザーの Google Workspace サブスクリプションを作成します。
• 認証の詳細については、OAuth 同意画面を設定し、スコープを選択するをご覧ください。