유튜브 라이브 실시간채팅 크롤링

들어가며

김치빌리아드, 큐찾사 유튜브채널 배돌이의 당구생활의 컨텐츠인 라이브 개인큐 랜덤 추첨에서 실시간 시청자들의 댓글을 긁어와서 추첨하는게 필요했습니다.

이벤트는 매우 매우 간단한 방법으로 채팅에 참여한 사람들을 추출해서 돌림판에 넣고 추첨합니다. 초반에는 유튜브에서 제공하는 채팅방 실시간 참여자를 드래그해서 추첨판에 복붙하였습니다.

저의 부캐 채PD로 제가 직접 방송에 참여하면서 컴퓨터를 조작하면 괜찮지만 방송에 참여하지 못하는 날엔 진행에 어려운 부분이 있어서 클릭 한 번만 하면 자동으로 추첨까지 해주는 실행파일 프로그램을 만들기로 했습니다.

유튜브 라이브 실시간 댓글 크롤링

구상한 순서는 다음과 같습니다.

배돌이가 채팅 참여해주세요 라고 말하고 크롤링 프로그램을 킨다.
배돌이가 마감합니다 라고 말하고 아무 키를 누르면 크롤링이 끝난다
채팅에 참여한 모든 사람들을 추출한다.
셀레늄으로 돌림판 제공하는 사이트를 띄우고
추출한 사람들을 자동으로 넣고 돌린다.

라이브러리

유튜브 라이브의 경우 서드파티 라이브러리가 많이 있습니다. Don't reinvent the wheel이란 말처럼 (귀차니즘) 라이브러리를 사용하기로 했고 여기서 사용한 라이브러리는 pychat 입니다.

Thread

추첨을 시작하고 추첨을 끝내기 위해선 어떠한 이벤트가 발생해야 하는데 키보드의 아무 키가 입력되면 멈추게끔 프로그램을 작동시켜야합니다. 그러기 위해서는 멀티쓰레드로 키보드 이벤트를 감지하는 쓰레드가 하나 계속 돌고있어야 합니다.

코드

youtube 크롤링

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105


import pyperclip
import pytchat
import time
import threading as th
import os
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.service import Service
from webdriver_manager.chrome import ChromeDriverManager

# 키보드 이벤트 감지
CAPTURING = True
def stop_capture():
    global CAPTURING
    input()
    CAPTURING = False

def start_capture(video_id):
    # 키보드 이벤트 감지를 쓰레드로 실행
    th.Thread(target=stop_capture, args=(), name='stop_capture', daemon=True).start()
    chat = pytchat.create(video_id=video_id)
    author_name = []
    start = time.strftime('%Y-%m-%d %H-%M-%S', time.localtime(time.time()))
    today = time.strftime('%Y-%m-%d', time.localtime(time.time()))

    # 결과값을 저장할 위치
    current = os.getcwd()
    path = f'{current}/{today}'
    try:
        if not os.path.exists(path):
            os.makedirs(path)
    except OSError:
        print("Error: Cannot create the directory {}".format(path))
    fp = open(f'{path}/{start}.txt', 'w')
    cnt = 0

    # 크롤링 시작
    while CAPTURING and chat.is_alive():
        try:
            for c in chat.get().sync_items():
                if c.author.name not in author_name:
                    author_name.append(f'{c.author.name} | @{c.author.channelId}')
                print(f"[{c.datetime[11:]}] 아이디: {c.author.name} | @{c.author.channelId}\n\t   내용: {c.message}")
                fp.write(f"[{c.datetime[11:]}] 아이디: {c.author.name} | @{c.author.channelId}\n\t   내용: {c.message}\n")
                cnt += 1
        except KeyboardInterrupt:
            break

    chat.terminate()
    end = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
    author_name.sort()
    author_name = (list(set(author_name)))
    print('\n====================================================================')
    print(f"배돌이의당구생활 라이브 추첨 댓글 추출\n시작 {start} | 종료 {end}")
    print(f"채팅에 참여한 사람 {len(list(set(author_name)))}명 (중복제거) | 총 채팅수 : {format(cnt, ',')}개")
    print(author_name)
    print('====================================================================\n\n')
    print('추출을 종료합니다.')

    fp.write('\n====================================================================\n')
    fp.write(f"배돌이의당구생활 라이브 추첨 댓글 추출\n시작 {start} | 종료 {end}\n")
    fp.write(f"채팅에 참여한 사람 {len(list(set(author_name)))}명 (중복제거) | 총 채팅수 : {cnt}개\n")
    fp.write(f"{author_name}\n")
    fp.write('====================================================================\n')
    fp.close()
    time.sleep(0.5)
    picker(author_name)
    return author_name

def list_chuck(arr, n):
    return [arr[i: i + n] for i in range(0, len(arr), n)]

def picker(author_name):
    try:
        driver = webdriver.Chrome(service=Service(ChromeDriverManager().install()))
        url = 'https://spinnerwheel.ahaslides.com/?entries=&action=true'
        driver.get(url)
        driver.implicitly_wait(3)
        input_box = driver.find_element(By.XPATH, '//*[@id="aha-spinner-wheel"]/div[2]/div/div[3]/div[2]/form/div/div/input')
        add_button = driver.find_element(By.XPATH, '//*[@id="aha-spinner-wheel"]/div[2]/div/div[3]/div[2]/form/button')
        result = list_chuck(author_name, 30)

        for r in result:
            text = ','.join(r)
            JS_ADD_TEXT_TO_INPUT = """
            var elm = arguments[0], txt = arguments[1];
            elm.value += txt;
            elm.dispatchEvent(new Event('change'));
            """
            driver.execute_script(JS_ADD_TEXT_TO_INPUT, input_box, text)
            time.sleep(1)
            input_box.send_keys(' ')
            add_button.click()
        while(1):
            pass
    except:
        pass


print('\n========================================')
print(f"배돌이의당구생활 라이브 추첨 댓글 추출")
video_id = input("영상 URL을 입력해주세요 : ")
print(f"확인되었습니다. 댓글 추출을 시작합니다.")
print('========================================')
author_name = start_capture(video_id)

코드설명

pychat 라이브러리를 사용하기 떄문에 코드가 매우 간단합니다. start_capture로 캡쳐가 시작되고 함께 th.Thread(target=stop_capture, args=(), name='stop_capture', daemon=True).start()로 stop_capture 함수를 쓰레드로 시작합니다.

stop_capture에서 input()을 기다리고 있는데 이는 실행중에 키보드값이 입력되면 global로 선언된 CAPTURING 변수를 false로 변경하여 크롤링을 멈추게 됩니다. 쓰레드를 활용하는 환경에서 전체적으로 참조해야할 변수를 선언하려면 꼭 global을 사용해야합니다.

키보드 이벤트가 감지되어 추출이 멈추게되면 picker 함수를 실행시킵니다. 해당 함수는 selenium을 이용하여 결과값을 돌림판에 입력하는것 까지 자동으로 진행되게 합니다. 닉네임 중복이 가능하므로 유저의 고유번호까지 추출되어야하며 나중에 비교하기 위해서 텍스트 파일로 결과를 저장합니다. 추출 결과값을 브라우저의 인풋에 넣고 확인을 눌러줘야 하는데 이때 어쩔 수 없이 자바스크립트 문법이 필요하였습니다.

실행파일

실행파일로 만들고 어떠한 환경에서도 실행되어야 해서 기존의 웹드라이버를 수동으로 다운받아서 옮기지 않고 자동으로 다운로드 받아야하여 다음 명령어로 웹드라이버를 로드하였습니다. webdriver.Chrome(service=Service(ChromeDriverManager().install()))

실행파일로 만들어주기 위해 pyinstaller 라이브러리를 사용했습니다. 이렇게 하면 최신 드라이버를 받아주고 실행파일 용량압박에도 살아남을 수 있게됩니다.

결과

실제로 어떻게 작동되는지 궁금하시면 다음 링크나 아래 영상을 확인해주세요. 도움이 되셨다면 구독과.. 좋..아…요…

들어가며#

유튜브 라이브 실시간 댓글 크롤링#

라이브러리#

Thread#

코드#

코드설명#

실행파일#

결과#