selenium을 이용하여 유튜브 댓글 크롤링하기
- 키워드 입력 후 검색되는 영상의 제목, url, 정보를 수집하여 데이터프레임 및 csv로 저장
- 저장된 url을 조회하여 영상의 댓글을 수집하여 데이터 프레임 및 csv로 저장
- 모든 댓글을 1개의 csv로 저장
# -*- coding:utf-8 -*-
from selenium import webdriver as wd
from bs4 import BeautifulSoup
import time
import pandas as pd
import requests
import re
from datetime import datetime
import os
import glob## 키워드를 넣으면 검색되는 영상의 제목, url, 정보를 수집하여 csv로 저장
def get_urls_from_youtube_with_keyword(keyword):
titles = []
urls = []
infos =[]
search_keyword_encode = requests.utils.quote(keyword)
url = "https://www.youtube.com/results?search_query=" + search_keyword_encode
driver = wd.Chrome(executable_path="chromedriver.exe")
driver.get(url)
last_page_height = driver.execute_script("return document.documentElement.scrollHeight")
while True:
driver.execute_script("window.scrollTo(0, document.documentElement.scrollHeight);")
time.sleep(3.0)
new_page_height = driver.execute_script("return document.documentElement.scrollHeight")
if new_page_height == last_page_height:
break
last_page_height = new_page_height
html_source = driver.page_source
driver.quit()
soup = BeautifulSoup(html_source, 'lxml')
datas = soup.select("a#video-title")
for data in datas:
title = data.text.replace('\n', '')
url = "https://www.youtube.com/" + data.get('href')
info = data.get('aria-label')
titles.append(title)
urls.append(url)
infos.append(info)
pd_data_title_url = {"title":titles, "url":urls, "info" : infos}
df_title_url = pd.DataFrame(pd_data_title_url)
# info 컬럼 -> 분해
df_title_url['publisher'] = df_title_url['info'].apply(lambda x : 'YouTube 영화'
if " ".join(x.split('게시자: ')[1].split(' ')[:2])=='YouTube 영화'
else " ".join(x.split('게시자: ')[1].split(' 전 ')[0].split(' ')[:-1]))
df_title_url['datetime_before'] = df_title_url['info'].apply(lambda x : x.split('게시자: ')[1].split(' 전 ')[0].split(' ')[-1])
df_title_url['duration'] = df_title_url.apply(lambda row : row['info'].split('게시자: ')[1].split('영화 ')[1]
if (row['publisher'] == "YouTube 영화" )
else (row['info'].split('게시자: ')[1].split(' 전 ')[1].split(' 조회수 ')[0]), axis=1)
df_title_url['click_cnt'] = df_title_url['info'].apply(lambda x : "".join(x.split(' ')[-1].split('회')[0].split(',')))
# 영상 제목 + url + 정보 테이블 저장하기
file_name = keyword + '_' + datetime.today().strftime("%Y%m%d%H%M%S") # 저장 경로는 코드랑 동일, 파일명에 저장되는 시간 기록
df_title_url.to_csv("{}.csv".format(file_name), index=False)
return titles, urls, file_name
# 수집한 영상 url을 입력하여 영상마다 html 페이지를 수집
def crawl_youtube_page_html_sources(urls):
html_sources = []
# len(urls)로 변경해야함, 테스트시 시간이 오래걸려서 5개의 영상의 댓글만 수집함
for i in range(0, 5):
driver = wd.Chrome(executable_path="chromedriver.exe")
driver.get(urls[i])
last_page_height = driver.execute_script("return document.documentElement.scrollHeight")
while True:
driver.execute_script("window.scrollTo(0, document.documentElement.scrollHeight);")
time.sleep(3.0)
new_page_height = driver.execute_script("return document.documentElement.scrollHeight")
if new_page_height == last_page_height:
break
last_page_height = new_page_height
html_source = driver.page_source
html_sources.append(html_source)
print("OK") # 영상 1개 html 페이지 수집이 완료될 때마다 ok print, 영상의 개수가 많다면 삭제하고 진행해도 무방
driver.quit()
return html_sources
# 수집한 html 페이지에서 ID, 댓글을 수집함
def get_user_IDs_and_comments(html_sources):
my_dataframes = []
for html in html_sources:
soup = BeautifulSoup(html, 'lxml')
youtube_user_IDs = soup.select('div#header-author a#author-text span')
youtube_comments = soup.select('yt-formatted-string#content-text')
str_youtube_userIDs = []
str_youtube_comments = []
for i in range(len(youtube_user_IDs)):
str_tmp = str(youtube_user_IDs[i].text)
# print(str_tmp)
str_tmp = str_tmp.replace('\n', '')
str_tmp = str_tmp.replace('\t', '')
str_tmp = str_tmp.replace(' ','')
str_youtube_userIDs.append(str_tmp)
str_tmp = str(youtube_comments[i].text)
str_tmp = str_tmp.replace('\n', '')
str_tmp = str_tmp.replace('\t', '')
str_tmp = str_tmp.replace(' ', '')
str_youtube_comments.append(str_tmp)
pd_data = {"ID":str_youtube_userIDs, "Comment":str_youtube_comments}
youtube_pd = pd.DataFrame(pd_data)
my_dataframes.append(youtube_pd)
return my_dataframes
# 영상별 댓글을 저장한 폴더를 생성
def createFolder(directory):
try:
if not os.path.exists(directory):
os.makedirs(directory)
except OSError:
print ('Error: Creating directory. ' + directory)
# 영상별 댓글을 영상제목.csv 파일로 저장. 위에서 만든 폴더에
def convert_csv_from_dataframe(file_name ,titles, my_dataframes):
for i in range(len(my_dataframes)):
title = re.sub('[-=+,#/\?:^$.@*\"※~&%ㆍ!』\\‘|\(\)\[\]\<\>`\'…《\》]', '', titles[i])
folder_path = "./" + file_name
createFolder(folder_path)
save_path = "./" + file_name + '/'
my_dataframes[i].to_csv("{}{}.csv".format(save_path, title), index=False)
# 영상제목.csv 파일들을 1개의 csv 파일로 만듬
def make_one_csv_from_folder(file_name):
folder_path = "./" + file_name
all_data = []
for f in glob.glob( folder_path + "/*.csv"):
all_data.append(pd.read_csv(f))
df = pd.concat(all_data, ignore_index=True)
df.to_csv(file_name + '_commnet_total'+ '.csv', index = False)keyword ="목포여행"
titles, urls, file_name = get_urls_from_youtube_with_keyword(keyword)
html_sorces = crawl_youtube_page_html_sources(urls)
my_dataframes = get_user_IDs_and_comments(html_sorces)
convert_csv_from_dataframe(file_name, titles, my_dataframes)
make_one_csv_from_folder(file_name)