구글에서 제공해주는 API와 client 라이브러리를 활용하여 유튜브 라이브 스트리밍의 채팅들을 수집하고자 합니다.
스트리머들의 화면에서 채팅 내용이 별도로 보이는 것에서 궁금하여 진행해 보았고, 이후 어떤 것을 할지는 차차 고민해보도록 하죠
우선 유튜브 개발자 사이트를 보면, 유튜브 관련 API는 크게 아래 3가지가 있습니다.
- YouTube Data API (v3)
- 동영상 업로드, 재생목록 생성, 관리 등 YouTube 기능을 애플리케이션에 추가할 수 있습니다.
- YouTube 분석 API
- YouTube 동영상 및 채널의 시청 통계, 인기도 측정항목 등을 검색하세요.
- YouTube Live Streaming API
- 실시간 YouTube 방송을 예약하고 방송 동영상 스트림을 관리합니다.
여기서 Data API(V3)와 Live Streaming API를 이용하여, 유튜브 라이브 스트리밍에서 URL 내 video id를 가지고 실시간 채팅을 주기적으로 가져와 볼 계획입니다.
10초 주기로 가져오게끔 해볼 예정인데, 이는 하루에 1만건 제한이 있는 것으로 보이기에 제한사항에 따라 바꾸면 될 것 같습니다. (참고 : https://developers.google.com/youtube/v3/determine_quota_cost?hl=ko)
2가지를 활용하는 이유가, 실시간 채팅에 대한 ID가 필요한데 이 실시간 채팅 ID에 대한 정보를 가져오기 위해 Data API(v3)를 이용하여 실시간 채팅 ID를 가져온 후 Live Streaming API로 채팅 내용을 가져오는 형태로 진행했습니다.
실시간 채팅 리스트에 대한 API 가이드를 보면 요청 시 liveChatId가 필요한데, 이는 video에 대한 API 가이드에서 응답 중 'items'부분에 video 리소스 부분을 보면 "liveStreamingDetails"에 "activeLiveChatId"가 있더라구요. 실시간 채팅 ID에 대해 가장 얻기 쉬운 방법이 유튜브 URL에 있는 video id인 점에서 착안하여 해당 방법으로 진행했습니다.
코딩에 앞서 api key를 획득하기 위한 방법을 먼저 설명하자면, google developer console 에서 youtube data api 사용 설정 후 키 발급을 진행하면 되며 아래와 같습니다. (프로젝트 생성은 생략)
     |
|---|
자, 그럼 이제 python으로 라이브 스트리밍에 대한 video id를 기반으로 채팅을 가져오는 코드를 작성해보죠.
pip를 이용하여 별도로 사용한 python 패키지는 google api client 패키지(https://developers.google.com/explorer-help/code-samples#python)만 사용했습니다.
import argparse
import configparser
import csv
import time
# import google api client
import googleapiclient.discovery
import googleapiclient.errors
# api 사용을 위한 기본 세팅(api_key는 config.ini 파일에 따로 작성)
config = configparser.ConfigParser()
config.read('./config.ini')
api_key = config['KEY']['ApiKey']
api_service_name = "youtube"
api_version = "v3"
youtube = googleapiclient.discovery.build(
api_service_name, api_version, developerKey=api_key)
# 채팅 내용 CSV로 저장(유튜브 채널 ID_비디오 ID.csv 파일로 생성)
def convertDicToCSV(dicList, video_id, channel_id)->None:
fieldnames = ['time','message','user']
file_name = f"{channel_id}_{video_id}.csv"
with open(file_name, "a", encoding='utf-8-sig', newline="") as csvfile:
writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
# writer.writeheader()
for item in dicList:
_time = item["snippet"]["publishedAt"]
_message = item["snippet"]["textMessageDetails"]["messageText"]
_user = item["authorDetails"]["displayName"]
writer.writerow({'time':_time, 'message':_message, 'user':_user})
csvfile.close()
return
# 비디오 정보 가져오기
def get_livechat_id(video_id):
request = youtube.videos().list(
part="liveStreamingDetails,snippet,status",
id=video_id
)
response = request.execute()
channel_id = response["items"][0]["snippet"]["channelId"]
chat_id = response["items"][0]["liveStreamingDetails"]["activeLiveChatId"]
result = {"channel_id":channel_id, "chat_id":chat_id}
return result
# main문
def main(video_id):
_get_video_info = get_livechat_id(video_id) # 유튜브 비디오 아이디로 비디오 관련 정보 획득
_chat_id = _get_video_info["chat_id"] # 실시간 채팅 ID 획득
_channel_id = _get_video_info["channel_id"] # CSV 파일 이름에 채널 ID 추가를 위한 채널 ID 획득
page_token = ""
polling_time = 10000
while True:
request = youtube.liveChatMessages().list(
part="snippet,authorDetails",
liveChatId=_chat_id,
pageToken=page_token
)
response = request.execute()
page_token = response["nextPageToken"] # 다음 채팅 정보 획득을 위한 토큰 획득
convertDicToCSV(response["items"], video_id, _channel_id) # 실시간 채팅 획득 내용을 CSV에 작성
#
interval_polling_time = response["pollingIntervalMillis"]
if interval_polling_time > polling_time:
time.sleep(interval_polling_time / 1000)
else:
time.sleep(polling_time / 1000)
# python 실행 시 스트리밍 video id를 argument로 받아서 실행
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument('-v', help=' : Please set youtube video id (https://www.youtube.com/watch?v=[video_id])')
args = parser.parse_args()
main(args.v)
실행 결과
유튜브에서 스트리밍 중인 뉴스로 테스트를 진행했고, 아래와 같이 유튜브 url 내 video id를 argument로 실행하면 10초 주기로 가져와서 csv파일에 저장합니다. (현재는 while true라 ctrl+c로 python 강제종료 ㅎㅎ)
유튜브 라이브 스트리밍 url('https://www.youtube.com/watch?v=[video id]')에서 video id를 가지고
python에 argument로 실행하면
csv파일이 생기고
시간과 채팅 내용, 채팅작성자를 지속적으로 작성합니다.
