고정 헤더 영역
상세 컨텐츠
본문
반응형
관련 깃허브 링크
streamer-persona-chatbot/src/streaming_api/youtube.py at main · Kiminjo/streamer-persona-chatbot
Streamer Background Knowledge-Based Answer Automatic Generation System - Kiminjo/streamer-persona-chatbot
github.com
포스팅이 작성된 5월 11일 기준, 아직 진행 중인 프로젝트입니다.
프로젝트 경과에 따라 포스팅 내용이 변경될 수 있습니다.
위의 [관련 프로젝트] 링크는 프로젝트 종료 시까지 비공개 포스팅으로 둘 계획입니다.
프로젝트 종료 후 공개 포스팅으로 전환하겠습니다.
본 프로젝트의 핵심을 1문장으로 줄이면 '인터넷 방송 시청자들의 채팅을 파악하여 사전 지식으로 답변 가능한 내용이라면 AI가 자동으로 답변한다' 입니다.
위의 문장에서 핵심이 되는 부분은 크게 3가지인데요.
1. 시청자들의 채팅을 파악한다.
2. AI가 적절한 답변을 생성한다.
3. 생성된 답변을 시청자에게 발송한다.
이 중 2. AI가 적절한 답변을 생성한다. 는 지난 첫번째 글에서 다뤘었죠.
그래서 본 글에서는 시청자들의 채팅을 불러오는 방법과, 생성된 답변을 다시 발송하는 방법에 대해 공유드리겠습니다.
우선, 저희는 유튜브의 실시간 스트리밍을 타겟팅했는데요.
본 프로젝트를 착수하기 직전 트위치가 한국 철수를 선언했으며, 아직 치지직은 개발자 API를 제공하지 않았기 때문에 유튜브만을 타겟팅하였습니다.
1. YouTube API 사용 설정하기
유튜브의 실시간 채팅을 처리하기 위해서는 구글에서 제공하는 YouTube API를 사용해야합니다.
구글에서 제공하는 유튜브 관련 API는 3개가 있는데요.
저는 실시간 채팅 정보를 필요로 하니 Data API에 접근할 계획입니다.
아래 링크로 들어가 먼저 API 사용 등록을 해줍니다.
Google Cloud console
console.cloud.google.com
사용 등록을 하셨으면, 이제 OAuth 인증을 위한 클라이언트 ID를 발급받아야해요.
아무래도 채팅이 다소 민감할 수 있는 개인 정보다보니 보안을 위해 이 절차를 꼭 진행해야합니다.
위의 이미지에서 보이는 것처럼 '사용자 인증 정보' -> '사용자 인증 정보 만들기'로 들어가면 됩니다.
들어가면 애플리케이션 유형을 설정할 수 있는 페이지가 나오는데요.
다른 블로그를 보니 아무거나 선택하면 된다고 하더라고요.
그래서 별 생각없이 '웹 애플리케이션'으로 했는데 자꾸 OAuth 인증이 진행이 안됐습니다 ㅠㅠ
용도에 맞게 '데스크톱 앱'으로 했더니 이제야 되더라고요.
파이썬용 데스크톱 앱을 만드시는거라면 꼭 '데스크톱 앱'으로 설정해주세요.
그리고 '만들기'를 눌러주시면 OAuth 클라이언트 ID가 생성됩니다.
이제 발급받은 OAuth 키를 로컬에서 사용할 수 있도록 json 파일로 다운로드 받으면 됩니다.
다운받은 클라이언트 키는 working directory로 이동시켜줍니다.
2. YouTube API를 이용해서 실시간 채팅 불러오기
이제 파이썬을 이용해서 실시간 채팅을 불러오면 돼요.
해당 API는 list라는 이름으로 구현되어 있으며 아래 링크를 통해서 자세히 확인할 수 있어요.
또한 페이지 우측의 API 탐색기를 클릭하면 간단한 예제 코드도 확인할 수 있어요.
LiveChatMessages: list | YouTube Live Streaming API | Google for Developers
이 페이지는 Cloud Translation API를 통해 번역되었습니다. LiveChatMessages: list 컬렉션을 사용해 정리하기 내 환경설정을 기준으로 콘텐츠를 저장하고 분류하세요. 특정 채팅의 실시간 채팅 메시지를
developers.google.com
import os
from pathlib import Path
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
from google_auth_oauthlib.flow import InstalledAppFlow
class YouTubeAPI:
def __init__(self,
client_secrets_file: str,
scopes: list = ['https://www.googleapis.com/auth/youtube.force-ssl'],
api_service_name: str = 'youtube',
api_version: str = 'v3',
):
self.CLIENT_SECRETS_FILE = client_secrets_file
self.SCOPES = scopes
self.API_SERVICE_NAME = api_service_name
self.API_VERSION = api_version
# Get authenticated service
self.service = self._get_authenticated_service()
# Get livechatid
self.livechatid = self._get_livechatid('active')
def _get_authenticated_service(self,
port: int = 0):
flow = InstalledAppFlow.from_client_secrets_file(self.CLIENT_SECRETS_FILE,
self.SCOPES)
credentials = flow.run_local_server(port=port)
return build(self.API_SERVICE_NAME,
self.API_VERSION,
credentials=credentials)
def _get_livechatid(self, broadcast_status: str):
list_broadcasts_request = self.service.liveBroadcasts().list(
broadcastStatus=broadcast_status,
part='id,snippet',
maxResults=50
)
list_broadcasts_response = list_broadcasts_request.execute()
return list_broadcasts_response['items'][0]['snippet']['liveChatId']
def get_live_chat(self,
part: str = 'id,snippet,authorDetails',
nextpagetoken: str = None
):
if nextpagetoken:
response = self.service.liveChatMessages().list(
liveChatId=self.livechatid,
part=part,
pageToken=nextpagetoken
).execute()
else:
response = self.service.liveChatMessages().list(
liveChatId=self.livechatid,
part=part
).execute()
nextpagetoken = response['nextPageToken']
output_chat = []
for chat in response.get('items', []):
output_chat.append(chat['snippet']['textMessageDetails']['messageText'])
return output_chat, nextpagetoken
if __name__ == "__main__":
yt = YouTubeAPI(<YOUR_OAUTH_JSON_FILE>)
chat, nextpagetoken = yt.get_live_chat()
저는 위와 같이 코드를 구현했습니다.
init method에서 OAuth 인증을 수행 후, 내가 불러오고자 하는 방송의 chat id를 불러옵니다.
그후 get_live_chat method를 호출하면 지금까지 올라왔던 채팅들이 모두 불러올 수 있습니다.
그런데, 매번 지금까지 했던 모든 채팅을 호출한다면 매우 비효율적입니다.
방송이 30분 정도만 지나도 몇천, 몇만개의 채팅을 불러와야할 수도 있죠.
그래서 nextpagetoken을 설정해주었습니다.
nextpagetoken은 내가 마지막으로 불러온 채팅의 token인데요.
이를 통해, 다음번에 호출 시에는 해당 채팅 이후에 작성된 채팅만 불러올 수 있게 구현하였습니다.
일종의 save point를 지정해준거죠.
"""
유저1: 안녕
유저2: 우와 저 생방은 처음이에요.
유저3: 하이하이
유저4: 인하
유저2: 너무 신기해요 <- 이 시점에 get_live_chat method 실행, nextpagetoken 저장
유저5: 반가워요
유저6: ㅎㅇㅎㅇ <- 이 시점에 2번째 get_live_chat method 실행
"""
> 첫번째 실행 결과
"""
안녕
우와 저 생방은 처음이에요.
하이하이
인하
너무 신기해요
"""
> 두번째 실행 결과
"""
반가워요
ㅎㅇㅎㅇ
"""
이런식으로 배치 단위로 채팅을 불러올 수 있도록 설정하였으며, 해당 코드는 30초에 한번씩 실행되도록 설정하였습니다.
3. YouTube API를 이용해 채팅 발송하기
이제 일련의 과정을 거쳐, 시청자 질문에 대한 AI의 답변이 생성되었어요.
그럼 해당 내용을 다시 채팅을 통해서 시청자에게 전달해줘야겠죠.
이에 대한 내용은 insert라는 형식으로 구현되어 있어요.
LiveChatMessages: insert | YouTube Live Streaming API | Google for Developers
이 페이지는 Cloud Translation API를 통해 번역되었습니다. LiveChatMessages: insert 컬렉션을 사용해 정리하기 내 환경설정을 기준으로 콘텐츠를 저장하고 분류하세요. 실시간 채팅에 메시지를 추가합니
developers.google.com
import os
from pathlib import Path
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
from google_auth_oauthlib.flow import InstalledAppFlow
def convert_chat_format(chat: str,
livechatid: str,
display_content: bool = True
):
snippet = {
'snippet': {
'liveChatId': livechatid,
'type': 'textMessageEvent',
'hasDisplayContent': display_content,
'textMessageDetails': {
'messageText': chat
}
}
}
return snippet
class YouTubeAPI:
def __init__(self,
client_secrets_file: str,
scopes: list = ['https://www.googleapis.com/auth/youtube.force-ssl'],
api_service_name: str = 'youtube',
api_version: str = 'v3',
):
self.CLIENT_SECRETS_FILE = client_secrets_file
self.SCOPES = scopes
self.API_SERVICE_NAME = api_service_name
self.API_VERSION = api_version
# Get authenticated service
self.service = self._get_authenticated_service()
# Get livechatid
self.livechatid = self._get_livechatid('active')
def send_live_chat(self,
snippet: dict
):
self.service.liveChatMessages().insert(
part='snippet',
body=snippet
).execute()
if __name__=="__main__":
yt = YouTubeAPI(<YOUR_OAUTH_JSON_FILE>)
snippet = convert_chat_format('안녕하세요', yt.livechatid)
yt.send_live_chat(snippet)
생성된 AI 답변은 string 형식인데요.
이를 YouTube용 포맷에 맞춰 Json 형식으로 변경해줍니다.
해당 기능은 convert_chat_format 함수에 구현해뒀어요.
이렇게 생성된 답변은 send_live_chat method를 이용해 사용자들에게 발송됩니다.
일단 현재 시점에서는 모든 방송 시청자들에게 답변이 발송되도록 구현을 해뒀어요.
4. 전체 코드
import os
from pathlib import Path
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
from google_auth_oauthlib.flow import InstalledAppFlow
def convert_chat_format(chat: str,
livechatid: str,
display_content: bool = True
):
snippet = {
'snippet': {
'liveChatId': livechatid,
'type': 'textMessageEvent',
'hasDisplayContent': display_content,
'textMessageDetails': {
'messageText': chat
}
}
}
return snippet
class YouTubeAPI:
def __init__(self,
client_secrets_file: str,
scopes: list = ['https://www.googleapis.com/auth/youtube.force-ssl'],
api_service_name: str = 'youtube',
api_version: str = 'v3',
):
self.CLIENT_SECRETS_FILE = client_secrets_file
self.SCOPES = scopes
self.API_SERVICE_NAME = api_service_name
self.API_VERSION = api_version
# Get authenticated service
self.service = self._get_authenticated_service()
# Get livechatid
self.livechatid = self._get_livechatid('active')
def _get_authenticated_service(self,
port: int = 0):
flow = InstalledAppFlow.from_client_secrets_file(self.CLIENT_SECRETS_FILE,
self.SCOPES)
credentials = flow.run_local_server(port=port)
return build(self.API_SERVICE_NAME,
self.API_VERSION,
credentials=credentials)
def _get_livechatid(self, broadcast_status: str):
list_broadcasts_request = self.service.liveBroadcasts().list(
broadcastStatus=broadcast_status,
part='id,snippet',
maxResults=50
)
list_broadcasts_response = list_broadcasts_request.execute()
return list_broadcasts_response['items'][0]['snippet']['liveChatId']
def get_live_chat(self,
part: str = 'id,snippet,authorDetails',
nextpagetoken: str = None
):
if nextpagetoken:
response = self.service.liveChatMessages().list(
liveChatId=self.livechatid,
part=part,
pageToken=nextpagetoken
).execute()
else:
response = self.service.liveChatMessages().list(
liveChatId=self.livechatid,
part=part
).execute()
nextpagetoken = response['nextPageToken']
output_chat = []
for chat in response.get('items', []):
output_chat.append(chat['snippet']['textMessageDetails']['messageText'])
return output_chat, nextpagetoken
def send_live_chat(self,
snippet: dict
):
self.service.liveChatMessages().insert(
part='snippet',
body=snippet
).execute()
if __name__=="__main__":
yt = YouTubeAPI(<YOUR_OAUTH_JSON_FILE>)
chat, nextpagetoken = yt.get_live_chat()
print(chat)
snippet = convert_chat_format('안녕하세요', yt.livechatid)
yt.send_live_chat(snippet)반응형
'Not Use > 🎥 Streamer Assist Bot' 카테고리의 다른 글
| [Streamer Assist Bot] 1. LLM 모델 구현 (0) | 2024.05.04 |
|---|
