使用 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

在 OAuth 2.0 流程中，客户端 ID 充当应用的凭据。由于应用在本地运行，因此请创建一个桌面客户端 ID。

1. 在 Google Cloud 控制台中，依次选择“菜单”图标 menu > API 和服务 > 凭据。

   进入“凭据”页面

2. 依次点击创建凭据 > OAuth 客户端 ID。
3. 依次点击应用类型 > 桌面应用。
4. 在名称字段中，输入凭据的名称。此名称仅在 Google Cloud 控制台中显示。
5. 点击创建。系统随即会显示“OAuth 客户端已创建”屏幕，其中会显示您的新客户端 ID 和客户端密钥。
6. 点击确定。新创建的凭据会显示在 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 在有事件可用时拉取事件。各种处理脚本方法会输出与相应事件相关的信息。

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 权限请求页面并选择范围。