[python] 유튜브 댓글 크롤링(api 없이)

유튜브 영상을 검색하여, 

나오는 영상들의 댓글을 모두 수집해야 했다. 

 

여러 참고 자료를 활용하여 아래와 같은 코드를 완성시켰다. 

 

 

import pandas as pd
from bs4 import BeautifulSoup
from selenium import webdriver
from webdriver_manager.chrome import ChromeDriverManager
from tqdm import tqdm
import time
import re
import requests
import random
import datetime

#Queue의 기본적인 기능 구현
class Queue():
    def __init__(self, maxsize):
        self.queue = []
        self.maxsize = maxsize
        
    # Queue에 Data 넣음
    def enqueue(self, data):
        self.queue.append(data)

    # Queue에 가장 먼저 들어온 Data 내보냄
    def dequeue(self):
        dequeue_object = None
        if self.isEmpty():
            print("Queue is Empty")
        else:
            dequeue_object = self.queue[0]
            self.queue = self.queue[1:]
        return dequeue_object
    
    # Queue에 가장 먼저들어온 Data return
    def peek(self):
        peek_object = None
        if self.isEmpty():
            print("Queue is Empty")
        else:
            peek_object = self.queue[0]
        return peek_object
    
    # Queue가 비어있는지 확인
    def isEmpty(self):
        is_empty = False
        if len(self.queue) == 0:
            is_empty = True
        return is_empty
    
    # Queue의 Size가 Max Size를 초과하는지 확인
    def isMaxSizeOver(self):
        queue_size = len(self.queue)
        if (queue_size > self.maxsize):
            return False
        else :
            return True

# 키워드를 넣으면 검색되는 영상의 제목, url, 정보를 수집하여 csv로 저장
def get_urls_from_youtube_with_keyword(keyword):
    options = webdriver.ChromeOptions()
    options.add_argument('window-size=1920x1080')
    options.add_argument('disable-gpu')
    options.add_argument('user')
    options.add_argument("user-agent=Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/61.0.3163.100 Safari/537.36")
    options.add_argument("lang=ko_KR")
    driver = webdriver.Chrome(ChromeDriverManager().install(), chrome_options=options)
    
    key_word = keyword.replace(' ', '+')
    search_keyword_encode = requests.utils.quote(key_word)
    url = "https://www.youtube.com/results?search_query=" + search_keyword_encode
    driver.get(url)
    last_page_height = driver.execute_script("return document.documentElement.scrollHeight")

    #페이지 Open 후 기다리는 시간
    time.sleep(5)

    #down the scroll
    body = driver.find_element_by_tag_name('body')
    last_page_height = driver.execute_script("return document.documentElement.scrollHeight")
    
    # max size 50의 Queue 생성
    # 0.1sec * 50 = 5sec 동안 Scroll 업데이트가 없으면 스크롤 내리기 종료
    szQ = Queue(50)
    enqueue_count = 0
    
    while True:
        # Scroll 내리기
        driver.execute_script("window.scrollTo(0, document.documentElement.scrollHeight);")
        
        # Scroll Height를 가져오는 주기
        time.sleep(0.1)
        new_page_height = driver.execute_script("return document.documentElement.scrollHeight")
        
        # Queue가 꽉 차는 경우 스크롤 내리기 종료
        if(enqueue_count > szQ.maxsize):
            break
        
        # 첫 Loop 수행 (Queue가 비어있는 경우) 예외 처리
        if(szQ.isEmpty()) :
            szQ.enqueue(new_page_height)
            enqueue_count += 1
            
        # Queue에 가장 먼저 들어온 데이터와 새로 업데이트 된 Scroll Height를 비교함
        # 같으면 그대로 Enqueue, 다르면 Queue의 모든 Data를 Dequeue 후 새로운 Scroll Height를 Enqueue 함.    
        else :
            if(szQ.peek() == new_page_height) :
                szQ.enqueue(new_page_height)
                enqueue_count += 1
            else :
                szQ.enqueue(new_page_height)
                for z in range(enqueue_count) :
                    szQ.dequeue()
                enqueue_count = 1
    
    html_source = driver.page_source
    soup = BeautifulSoup(html_source, 'lxml')
    
    driver.close()

    datas = [a.get('href')  # a태그 - href 값 가져오기
            for a in soup.select('a#video-title')  # a태그 - Id: video-title 가져오기
            if not 'shorts' in a.get('href')]  # 'shorts' 제외하기
    
    youtube_url ='https://www.youtube.com'
    url_lists = [youtube_url + a for a in datas]

    print("관련 영상은 총 {0}개 입니다.".format(len(datas)))
    
    page = 1
    data_list = []
    
    for page_url in tqdm(url_lists):
        
        if __name__=="__main__":
            #set option of selenium
            options = webdriver.ChromeOptions()
            options.add_argument('disable-gpu')
            options.add_argument('user')
            options.add_argument("user-agent=Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/61.0.3163.100 Safari/537.36")
            options.add_argument("lang=ko_KR")
            options.add_argument("headless") #창 숨기기 생성 
            driver = webdriver.Chrome(ChromeDriverManager().install(), chrome_options=options)

            #target of crawling
            
            driver.get(page_url)
            driver.set_window_size(800, 600)

            #페이지 Open 후 기다리는 시간
            driver.implicitly_wait(20)


            #down the scroll
            body = driver.find_element_by_tag_name('body')
            last_page_height = driver.execute_script("return document.documentElement.scrollHeight")

            # max size 50의 Queue 생성
            # 0.1sec * 50 = 5sec 동안 Scroll 업데이트가 없으면 스크롤 내리기 종료
            szQ = Queue(50)
            enqueue_count = 0

            while True:
                # Scroll 내리기
                driver.execute_script("window.scrollTo(0, document.documentElement.scrollHeight);")
                time.sleep(0.1)
                new_page_height = driver.execute_script("return document.documentElement.scrollHeight")

                if(enqueue_count > szQ.maxsize):
                    break

                if(szQ.isEmpty()) :
                    szQ.enqueue(new_page_height)
                    enqueue_count += 1
   
                else :
                    if(szQ.peek() == new_page_height) :
                        szQ.enqueue(new_page_height)
                        enqueue_count += 1
                    else :
                        szQ.enqueue(new_page_height)
                        for z in range(enqueue_count) :
                            szQ.dequeue()
                        enqueue_count = 1

            html0 = driver.page_source
            html = BeautifulSoup(html0, 'html.parser')

            yt_title = html.find('h1', {'class':'title style-scope ytd-video-primary-info-renderer'}).text
            uploader = html.find('yt-formatted-string', {'class': 'style-scope ytd-channel-name'}).text
            comments_list = html.findAll('ytd-comment-thread-renderer', {'class':'style-scope ytd-item-section-renderer'})

            print('{0}/{1}페이지 댓글 총 {2}개 크롤링, url: {3}'.format(url_lists.index(page_url)+1,len(url_lists), len(comments_list), page_url))
            driver.close()
            
            for j in range(len(comments_list)):
                
                comment = comments_list[j].find('yt-formatted-string',{'id':'content-text'}).text
                comment = comment.replace('\n', '') 
                comment = comment.replace('\t', '')


                youtube_id = comments_list[j].find('a', {'id': 'author-text'}).span.text
                youtube_id = youtube_id.replace('\n', '') 
                youtube_id = youtube_id.replace('\t', '') 
                youtube_id = youtube_id.strip()

                raw_date = comments_list[j].find('yt-formatted-string', { 'class': 'published-time-text above-comment style-scope ytd-comment-renderer'})
                date = raw_date.a.text

                try:
                    like_num = comments_list[j].find('span', {'id': 'vote-count-middle'}).text
                    like_num = like_num.replace('\n', '') 
                    like_num = like_num.replace('\t', '')
                    like_num = like_num.strip()

                except: like_num = 0

                data = {'yt_title':yt_title,'youtube_id': youtube_id, 'uploader':uploader, 'comment': comment, 'date': date, 'like_num': like_num, 'page_url': page_url}
                data_list.append(data)
                
            page += 1 
            
        df = pd.DataFrame(data_list, columns=['yt_title','uploader','youtube_id','comment','date','like_num','page_url'])    
        
        csv_name = keyword + datetime.datetime.now().strftime('%Y%m%d') +'.csv'
        df.to_csv("./Excel_Data/"+csv_name, mode = 'w', index=False)

get_urls_from_youtube_with_keyword('검색 키워드')

 

 

예를 들어, 드라마 사내맞선을 입력하면, 

사내맞선 검색시 나오는 약 600개의 영상에 대한 모든 댓글을 크롤링해주는 코드이다.