Selenium Crawling

회사에서 진행한 업무로 뉴스 기사 카테고리 분류 모델 및 뉴스 크롤링을 진행했으며,
Selenium 기반에 BeatifulSoup을 일부 활용한 크롤링을 작성하였습니다.
저작권을 침범하지 않는 내에서 작업했음을 밝힙니다.
해당하는 .py file 들을 docker에 supervisor로 띄워 Infinite loof 프로세스 내에서 반복적으로 크롤링하도록 구현하였습니다.



https://github.com/hipster4020/selenium_crawling

GitHub - hipster4020/selenium_crawling

Selenium.py

selenium 메소드를 모듈화하는 파일.

◎ __init__

Chrome Option Arguments
--headless : 실제로 크롬 띄우지 않고 프로세스 실행
--no-sandbox : 샌드박스 사용 해제
--disable-dev-shm-usage : shm memory 사용 해제
--disable-gpu : gpu 사용 해제

ChromeDriverManager()
ChromeDriverManager를 통해 버전, 운영체제 상관없이 구글 크롬 드라이버를 생성

driver.get(url)
해당 url으로 이동

◎ method 설명

get_text_by_xpath : xpath를 통해 text를 리턴
click_by_xpath : 해당 xpath의 객체의 클릭 이벤트 활성
go_to : execute_script("location.href='url'")로 해당 url로 자바스크립트 실행하여 이동
back : driver.back() 이전 페이지 이동
get_attribute_by_xpath : get_attribute(attribute_name)로 attribute text 리턴

import re from webdriver_manager.chrome import ChromeDriverManager from selenium import webdriver from selenium.webdriver.remote.webdriver import WebDriver class Selenium: def __init__(self, url: str, timeout=60, page_timeout=5, headless=True): """ Chrome Option Arguments --headless : 실제로 크롬 띄우지 않고 프로세스 실행 --no-sandbox : 샌드박스 사용 해제 --disable-dev-shm-usage : shm memory 사용 해제 --disable-gpu : gpu 사용 해제 ChromeDriverManager() ChromeDriverManager를 통해 버전, 운영체제 상관없이 구글 크롬 드라이버를 생성 driver.get(url) 해당 url으로 이동 """ self.timeout = timeout self.driver: WebDriver = None chrome_options = webdriver.ChromeOptions() if headless: chrome_options.add_argument("--headless") chrome_options.add_argument("--no-sandbox") chrome_options.add_argument("--disable-dev-shm-usage") chrome_options.add_argument("--disable-gpu") self.driver = webdriver.Chrome( ChromeDriverManager().install(), options=chrome_options ) self.driver.set_page_load_timeout(page_timeout) try: self.driver.get(url) except Exception as e: print(e) def __enter__(self): return self def __exit__(self): if self.driver: self.driver.quit() self.driver = None def get_text_by_xpath(self, xpath: str) -> str: """ xpath를 통해 text를 리턴 Args: xpath (str): html xpath Returns: str: html tag text """ tag_text = None try: self.driver.implicitly_wait(self.timeout) tag_text = self.driver.find_element_by_xpath(xpath).text except Exception as e: print(e) return tag_text def click_by_xpath(self, xpath: str): """ 해당 xpath의 객체의 클릭 이벤트 활성 Args: xpath(str): html xpath Returns: void """ try: self.driver.implicitly_wait(self.timeout) self.driver.find_element_by_xpath(xpath).click() except Exception as e: print(e) def go_to(self, url: str): """ execute_script("location.href='url'")로 해당 url로 자바스크립트 실행하여 이동 Args: url(str): 홈페이지 주소 Returns: void: 없음 """ try: self.driver.execute_script("location.href='{}'".format(url)) except Exception as e: print(e) def back(self): """ driver.back() 이전 페이지 이동 Returns: void: 없음 """ try: self.driver.back() except Exception as e: print(e) def get_attribute_by_xpath(self, xpath: str, attribute_name: str) -> str: """ get_attribute(attribute_name)로 attribute text 리턴 Args: xpath(str): html xpath attribute_name(str): html 속성명 Returns: str: html tag text """ tag_attribute = None try: self.driver.implicitly_wait(self.timeout) tag_attribute = self.driver.find_element_by_xpath(xpath).get_attribute( attribute_name ) except Exception as e: print(e) return tag_attribute

Crawling.py

크롤링 실행부.
전체적인 크롤링은 페이지의 xpath를 가지고 Selenium을 사용하였고, beautifulsoup 사용(line 97)으로 첫 페이지의 모든 href url 값을 리스트화하여 각 본문 기사로 이동 후 스크랩한 뒤 driver.back()으로 첫 페이지 돌아오는 방식으로 반복문 처리.

line 14 : 로그 세팅
line 31 : 기사 본문 전처리
line 42 : 날짜 전처리
line 77 : 네이버 속보 기사 Infinite loof 방식으로 카테고리 별 첫 페이지 무한 크롤링
line 281 : naver_helper.driver.close() - 크롬 드라이버에서 웹 창 탭 종료
naver_helper.driver.quit() - 크롬 드라이버 종료
line 246 : list_con.append - dictionary 형식으로 리스트에 크롤링 텍스트 값 저장
line 262 : totalSql = pd.DataFrame(list_con) - 텍스트 값이 담긴 list를 dataframe으로 전환
line 267 : totalSql.drop_duplicates() - 중복 데이터 제거
line 270, 276 : list_con.clear(), del totalSql - Garbage collection
line 273 : alchemy.DataSource("mysql", "news_scraper").df_to_sql(totalSql, "naver_news")
- dataframe to_sql database insert

# -*- coding: utf-8 -*- import datetime import hashlib import logging import time from logging import handlers import pandas as pd from bs4 import BeautifulSoup from db.alchemy import alchemy from selenium import helper # log setting 로그 세팅 carLogFormatter = logging.Formatter("%(asctime)s,%(message)s") carLogHandler = handlers.TimedRotatingFileHandler( filename="log/scrap.log", when="midnight", interval=1, encoding="utf-8", ) carLogHandler.setFormatter(carLogFormatter) carLogHandler.suffix = "%Y%m%d" scarp_logger = logging.getLogger() scarp_logger.setLevel(logging.INFO) scarp_logger.addHandler(carLogHandler) def content_replacing(text): """ content processing 기사 본문 전처리 """ result = str(text).replace("ⓒPixabay", "").replace("\n\n", "\n").replace("\n", "") return result def date_str_to_date(str_date): """ date processing 날짜 전처리 """ if str_date == "" or str_date == " " or str_date is None: str_date = str(datetime.datetime.now()) str_date = str_date[:19] str_date = datetime.datetime.strptime(str_date, "%Y-%m-%d %H:%M:%S") elif str_date.endswith("일전"): datenum = str_date.replace("일전", "").replace(" ", "") str_date = str(datetime.datetime.now() - datetime.timedelta(days=int(datenum))) str_date = str(str_date[:11]) + "00:00:00" elif str_date.endswith("시간전"): datenum = str_date.replace("시간전", "").replace(" ", "") str_date = str(datetime.datetime.now() - datetime.timedelta(hours=int(datenum))) str_date = str(str_date[:19]) elif str_date.endswith("분전"): datenum = str_date.replace("분전", "").replace(" ", "") str_date = str(datetime.datetime.now() - datetime.timedelta(minutes=int(datenum))) str_date = str(str_date[:19]) else: if str_date.startswith("기사입력"): str_date = str_date.replace("기사입력 ", "") str_date = str_date.replace("오전", "AM").replace("오후", "PM").replace("오 전", "AM").replace("오 후", "PM") str_date = datetime.datetime.strptime(str_date, "%Y.%m.%d. %p %I:%M").strftime("%Y-%m-%d %H:%M:%S") return str(str_date) def naver_crawling(): """ 네이버 속보 기사 Infinite loof 방식으로 카테고리 별 첫 페이지 무한 크롤링 """ # category_dict category_dict = { "200100": "100", "200101": "101", "200102": "102", "200103": "103", "200105": "105", "200106": "106", "200107": "107", } time.sleep(3) naver_helper.driver.implicitly_wait(10) loop = True while loop: try: for key in category_dict: list_con = [] logging.info(str(key) + " naver start") category_url = ( "https://news.naver.com/main/list.naver?mode=LSD&mid=sec&sid1=" + str(category_dict[key]) + "&listType=summary" ) naver_helper.go_to(category_url) # beautifulsoup 사용 soup = BeautifulSoup(naver_helper.driver.page_source, "html.parser") news_url = list( dict.fromkeys( [url["href"] for url in soup.find_all("a", attrs={"class": "nclicks(fls.list)"})] ) ) for url in news_url: naver_helper.go_to(url) if key == "200106": title = naver_helper.get_text_by_xpath("/html/body/div/div[3]/div[1]/div/h2") body = naver_helper.get_text_by_xpath("/html/body/div/div[3]/div[1]/div/div[4]/div") body = content_replacing(body) # create_date dummyDate = naver_helper.get_text_by_xpath( "/html/body/div/div[3]/div[1]/div/div[2]/span/em" ) createDate = date_str_to_date(dummyDate) # image_url image_url = naver_helper.get_attribute_by_xpath( "/html/body/div/div[3]/div[1]/div/div[4]/div/span[1]/span/img", "src", ) if not image_url: image_url = naver_helper.get_attribute_by_xpath( "/html/body/div/div[3]/div[1]/div/div[4]/div/span/span/img", "src", ) else: image_url = image_url if not image_url: image_url = "no image" # urlmd5 urlmd5 = hashlib.md5(url.encode("utf-8")).hexdigest() # media media = naver_helper.get_attribute_by_xpath( "/html/body/div/div[3]/div[1]/div/div[1]/a/img", "alt", ) elif key == "200107": title = naver_helper.get_text_by_xpath( "/html/body/div[2]/div[1]/div/div/div[1]/div/div[1]/h4" ) if not title: title = naver_helper.get_text_by_xpath( "/html/body/div[2]/div[2]/div/div/div[1]/div/div[1]/h4" ) body = naver_helper.get_text_by_xpath( "/html/body/div[2]/div[1]/div/div/div[1]/div/div[3]" ) if not body: body = naver_helper.get_text_by_xpath( "/html/body/div[2]/div[2]/div/div/div[1]/div/div[3]" ) body = content_replacing(body) # create_date dummyDate = naver_helper.get_text_by_xpath( "/html/body/div[2]/div[2]/div/div/div[1]/div/div[1]/div/span[1]" ) createDate = date_str_to_date(dummyDate) # image_url image_url = naver_helper.get_attribute_by_xpath( "/html/body/div[2]/div[2]/div/div/div[1]/div/div[3]/span/img", "src", ) if not image_url: image_url = naver_helper.get_attribute_by_xpath( "/html/body/div[2]/div[1]/div/div/div[1]/div/div[3]/span/img", "src", ) else: image_url = image_url if not image_url: image_url = "no image" # urlmd5 urlmd5 = hashlib.md5(url.encode("utf-8")).hexdigest() # media media = naver_helper.get_attribute_by_xpath( "/html/body/div[2]/div[2]/div/div/div[1]/div/div[1]/span/a/img", "alt", ) else: title = naver_helper.get_text_by_xpath( "/html/body/div[2]/table/tbody/tr/td[1]/div/div[1]/div[3]/h3" ) body = naver_helper.get_text_by_xpath( "/html/body/div[2]/table/tbody/tr/td[1]/div/div[2]/div[1]" ) body = content_replacing(body) # create_date dummyDate = naver_helper.get_text_by_xpath( "/html/body/div[2]/table/tbody/tr/td[1]/div/div[1]/div[3]/div/span[2]" ) createDate = date_str_to_date(dummyDate) # image_url image_url = naver_helper.get_attribute_by_xpath( "/html/body/div[2]/table/tbody/tr/td[1]/div/div[2]/div[1]/span[1]/img", "src", ) if not image_url: image_url = naver_helper.get_attribute_by_xpath( "/html/body/div[2]/table/tbody/tr/td[1]/div/div[2]/div[1]/span[2]/img", "src", ) else: image_url = image_url if not image_url: image_url = "no image" media = naver_helper.get_attribute_by_xpath( "/html/body/div[2]/table/tbody/tr/td[1]/div/div[1]/div[1]/a/img", "alt", ) # urlmd5 urlmd5 = hashlib.md5(url.encode("utf-8")).hexdigest() # dictionary type list append list_con.append( { "url": url, "title": title, "content": body, "create_date": createDate, "kind": key, "url_md5": urlmd5, "image_url": image_url, "portal": "naver", "media": media, } ) naver_helper.back() logging.info(str(key) + " naver end") # list → dataframe totalSql = pd.DataFrame( list_con, ) # duplicate drop totalSql.drop_duplicates() # Garbage Collection list_con.clear() # dataframe to_sql database insert alchemy.DataSource("mysql", "news_scraper").df_to_sql(totalSql, "naver_news") # Garbage Collection del totalSql except Exception as e: logging.info(e) naver_helper.driver.close() # chrome driver tab close naver_helper.driver.quit() # chrome driver close loop = False if __name__ == "__main__": with helper.Helper( "https://news.naver.com/main/list.naver?mode=LSD&mid=sec&listType=paper", timeout=1, ) as naver_helper: # naver crawling totalSql = naver_crawling()

Database Alchemy

config.py

DB 접속 정보를 dictionary화하여 mysql 어댑트 연결 URI를 리턴받는 config 모듈.

from urllib.parse import quote data_source_dict = { "mysql": { "db_id": "id", "password": "password", "host": "localhost", "port": "3306", }, } def make_data_source(db_type, db_name): """ Args: db_type(str): 사용할 db서버 ex) CPU : cpu server mariadb AWS_TEST : aws test mariadb AWS_PROD : aws production mariadb db_name(str): 사용할 db이름 Returns: str: data_source """ db_id = data_source_dict[db_type]["db_id"] password = data_source_dict[db_type]["password"] host = data_source_dict[db_type]["host"] port = data_source_dict[db_type]["port"] data_source = "mysql+pymysql://" + db_id + ":" + quote(password) + "@" + host + ":" + port + "/" + db_name return data_source

alchemy.py

config 모듈의 인스턴스에서 mysql 어댑트 연결 URI 받아 연결 및 sql 관련 메서드를 실행하는 모듈

import pandas as pd import sqlalchemy from config import make_data_source class DataSource: def __init__(self, db_type: str, db_name: str): """ database_uri 생성 및 sqlalchemy engine 생성 Args: db_type(str): 사용할 데이터베이스 서버 db_name(str): 사용할 데이터베이스 이름 """ self.table_dict = {} self.database_uri = make_data_source(db_type, db_name) self.engine = sqlalchemy.create_engine(self.database_uri, pool_pre_ping=True) # pool_pre_ping : testing of connections upon checkout is achievable by using the Pool.pre_ping argument def __enter__(self): return self.engine.begin() def __exit__(self): if self.engine: self.engine.dispose() def __del__(self): if self.engine: self.engine.dispose() def df_to_sql(self, df: pd.DataFrame, table_name: str): """ dataframe to_sql Args: df(DataFrame): 데이터베이스에 저장할 데이터프레임 객체 table_name(str): 테이블 이름 """ df.to_sql(table_name, con=self.engine, if_exists="append", index=False) def execute_query(self, query: str): """ 단순히 query 실행 Args: query(str): 실행할 쿼리 """ try: with self.engine.begin() as con: result = con.execute(query) return result except Exception as e: print(e) def execute_query_to_df(self, query: str): """ 쿼리 결과를 데이터프레임 형태로 반환 Args: query(str): 실행할 쿼리 """ row_list = [] try: with self.engine.begin() as con: result = con.execute(query) for row in result: row_list.append(dict(row)) except Exception as e: print(e) return pd.DataFrame(row_list)

---

Supervisor

해당 무한 루프 스크랩 .py을 docker container 내부에서 꺼지지 않고 띄우는 방식을 사용하였습니다.
몇몇 파일은 crontab에서 매 시간 실행하도록 작업하였습니다.

requirements.py

requests glances bs4 webdriver_manager json5==0.9.5 logging3==0.0.2 PyMySQL==1.0.2

Dockerfile

# Ubuntu 20.04 image use FROM ubuntu:20.04 # file copy COPY requirements.txt requirements.txt RUN mkdir /scrap/portal RUN mkdir log COPY naver_scraper.py /scrap/portal/naver_scraper.py # 키지 설치시에도 상호작용 방지기능이 적용 ENV DEBIAN_FRONTEND noninteractive # 설치 실행 RUN apt-get update && apt-get install -y gcc wget gnupg cron vim procps python3.8 python3-pip tzdata rsyslog supervisor RUN sh -c 'echo "deb [arch=amd64] http://dl.google.com/linux/chrome/deb/ stable main" >> /etc/apt/sources.list.d/google.list' RUN wget -q -O - https://dl.google.com/linux/linux_signing_key.pub | apt-key add - RUN apt-get update && apt-get install -y google-chrome-stable RUN pip3 install -r requirements.txt # supervisor file COPY scrap.conf /etc/supervisor/conf.d/scrap.conf COPY supervisord.conf /etc/supervisor/supervisord.conf # crontab file COPY cron.txt /etc/cron.d/news.cron RUN chmod 777 /etc/cron.d/news.cron RUN crontab /etc/cron.d/news.cron RUN service cron start RUN ln -sf /usr/share/zoneinfo/Asia/Seoul /etc/localtime # docker build 시 supervisor 실행 CMD ["/usr/bin/supervisord"]