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

このチュートリアルでは、Google Meet REST API と Google Workspace Events API と Google Cloud Pub/Sub を使用してイベントの監視と対応を実施 Meet の会議スペースで会議に参加できます。サンプル アプリケーションは、 会議の開始時刻と終了時刻、参加者が参加または退出したとき、誰かが アーティファクトを使用できます。

特定の会議スペースをサブスクライブする代わりに、 ゲストが参加していない会議スペースのイベントを ユーザーによって所有または編成されます。詳しくは、 Google Meet のイベントに登録する 詳しくは Google Workspace Events API ドキュメントをご覧ください

前提条件

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

• 以下を満たす Google Workspace アカウント Google Meet へのアクセス権。
• 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. [ユーザーの種類] で [内部] を選択し、[作成] をクリックします。 <ph type="x-smartling-placeholder">
3. [アプリ名] に「Meet REST API Tutorial」と入力します。
4. アプリ登録フォームに入力し、[保存して次へ] をクリックします。
5. [スコープを追加または削除] をクリックします。パネルが表示され、スコープのリストが表示されます。 Google Cloud プロジェクトで有効にした各 API の一意の ID です。
6. [スコープを手動で追加] に、次のスコープを貼り付けます。 <ph type="x-smartling-placeholder">
   </ph>
   • https://www.googleapis.com/auth/meetings.space.created
7. [Add to Table] をクリックします。
8. [更新] をクリックします。
9. 必要なスコープを選択したら、 [Save and Continue] をクリックします。
10. ユーザータイプとして [外部] を選択した場合は、テストユーザーを追加します。 <ph type="x-smartling-placeholder">
    </ph>
    1. [テストユーザー] で [ユーザーを追加] をクリックします。
    2. メールアドレスと、承認された他のテストユーザーを入力して、[ [保存して次へ] をクリックします。
11. アプリ登録の概要を確認します。変更するには、[編集] をクリックします。アプリが 問題がなければ、[ダッシュボードに戻る] をクリックします。

クライアント ID の作成

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

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

   [認証情報] に移動

2. [認証情報を作成] &gt; [OAuth クライアント ID] をクリックします。
3. [アプリケーションの種類] &gt; [デスクトップ アプリ] をクリックします。
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 に追加して、サブスクライブ用のメソッドを定義します。 。このコードは、会議のすべての予定をサブスクライブします。 選択します。サブスクライブされると、イベントが 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

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

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

引き続き main.py を編集し、次のサンプルコードを追加します。このコード 受信側を実装し、Google Cloud Pub/Sub API を使用して 新しいイベントが表示されます。さまざまなハンドラ メソッドで、情報を出力 詳細を確認できます。

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: 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 を使用してプログラムを中断します。

任意: 追加で試すことができる手順

このアプリは、イベントに関する基本情報をログに記録します。この動画で詳しく アプリケーションを変更してみてください。 できます。

• People API を使用すると、ユーザーに関する追加情報を取得できます。 できます。
• Google Drive API を使用して録画をダウンロードする 利用できます
• Google ドライブから文字起こしをダウンロードする代わりに、 構造化された文字起こし メソッド Meet REST API でこれを行えます。

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

リソースに対して Google Cloud コンソール アカウントに課金されないようにする リソースやプロジェクトをクリーンアップして 作成されます。

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

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

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

関連トピック

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