使用 Python 和 Google Meet REST API 觀察會議事件

本教學課程將說明如何使用 Google Meet REST API 搭配 Google Workspace 事件 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

在 OAuth 2.0 流程中，用戶端 ID 會充當應用程式的憑證。由於應用程式會在本機執行，請建立電腦版用戶端 ID。

1. 在 Google Cloud 控制台中，依序前往「Menu」menu >「APIs & Services」>「Credentials」。

   前往「憑證」

2. 依序按一下「建立憑證」>「OAuth 用戶端 ID」。
3. 依序點選「應用程式類型」>「桌面應用程式」。
4. 在「Name」欄位中輸入憑證名稱。這個名稱只會顯示在 Google Cloud 控制台中。
5. 按一下「建立」，系統會顯示「OAuth client created」(已建立 OAuth 用戶端) 畫面，其中顯示新的用戶端 ID 和用戶端密碼。
6. 按一下「確定」。新建立的憑證會顯示在「OAuth 2.0 Client IDs」下方。

安裝 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 在事件可用時將其拉取。各種處理常式會列印相應事件的相關資訊。

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

加入會議

如要為應用程式產生事件，請使用應用程式顯示的網址加入會議。加入後，您可以嘗試以下動作來觸發事件：

• 離開會議再重新加入。
• 邀請其他人或使用手機撥入。
• 啟用錄音和轉錄稿功能。

每項活動都會產生事件，應用程式會接收這些事件並記錄到 Google Cloud 控制台。

完成後，請使用 ctrl-c 中斷程式。

選用：可嘗試執行的其他步驟

應用程式會記錄事件的基本詳細資料。如要繼續探索 Meet REST API，請嘗試修改應用程式，以便執行這些額外動作。

• 使用 People API 擷取已登入參與者的其他資訊。
• 使用 Google Drive API 下載錄音和逐字稿。
• 請使用 Meet REST API 中的結構化轉錄稿方法擷取轉錄稿，而非從 Google 雲端硬碟下載轉錄稿。

選用：清除

如要避免系統向您的 Google Cloud 控制台帳戶收取本教學課程所用資源的費用，建議您清除所有建立的資源和專案。

如要刪除訂閱項目，請按照下列步驟操作：

如要刪除主題，請按照下列步驟操作：

如要刪除專案，請進行以下操作：

相關主題

• 瞭解可訂閱的 Meet 活動類型。
• 進一步瞭解 Google Meet REST API 的功能，並參閱參考說明文件。
• 使用 Google Workspace Events API 建立 Google Workspace 訂閱項目，以便使用 Meet 會議空間或使用者。
• 如要進一步瞭解驗證機制，請參閱「設定 OAuth 同意畫面並選擇範圍」。