Python 및 Google Meet REST API로 회의 이벤트 관찰

이 가이드에서는 Google Meet REST API를 Google Workspace Events API 및 Google Cloud Pub/Sub와 함께 사용하여 회의 이벤트를 관찰하고 대응하는 방법을 보여줍니다. 샘플 애플리케이션은 회의 시작 및 종료 시점, 참여자가 참여 또는 나가는 시점, 생성된 회의 아티팩트를 사용할 수 있는 시점을 기록합니다.

이벤트 작업에 대한 자세한 내용은 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. Add or Remove Scopes(범위 추가 또는 삭제)를 클릭합니다. Google Cloud 프로젝트에서 사용 설정한 각 API의 범위 목록이 포함된 패널이 나타납니다.
6. 수동으로 범위 추가 아래에 다음 범위를 붙여넣습니다.
   • https://www.googleapis.com/auth/meetings.space.created
7. Add to Table(테이블에 추가)을 클릭합니다.
8. Update를 클릭합니다.
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를 사용하여 이벤트가 제공될 때 이를 가져옵니다. 다양한 핸들러 메서드가 해당 이벤트에 대한 정보를 출력합니다.

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 Drive에서 스크립트를 다운로드하는 대신 Meet REST API의 구조화된 스크립트 작성 메서드를 사용하여 가져옵니다.

선택사항: 삭제

이 튜토리얼에서 사용한 리소스 비용이 Google Cloud 콘솔 계정에 청구되지 않도록 하려면 생성된 모든 리소스와 프로젝트를 삭제하는 것이 좋습니다.

구독을 삭제하려면 다음 안내를 따르세요.

주제를 삭제하는 방법은 다음과 같습니다.

프로젝트를 삭제하는 방법은 다음과 같습니다.

관련 주제

• Google Meet REST API로 수행할 수 있는 작업에 대해 자세히 알아보고 참조 문서를 확인하세요.
• Google Workspace Events API를 사용하여 이벤트를 구독하는 방법을 알아보세요.
• 인증에 대해 자세히 알아보려면 OAuth 동의 화면 구성 및 범위 선택을 참고하세요.