[python / colab] 증권사 리포트 pdf 파일 크롤링하기

 대표적으로 증권사 리포트를 확인할 수 있는 사이트인 한경컨센서스와 네이버금융리서치를 크롤링하여 리포트 pdf를 구글 드라이브에 저장하는 작업을 코랩에서 수행해 봤다.

 

 공통 단계 

• 사전 준비

# 구글 드라이브 마운트 (구글 드라이브와 코드 연결)
from google.colab import drive
drive.mount('/content/drive')

# selenium 설치
!pip install selenium
!apt-get update

# 구글 드라이브에 choremdriver 설치 (최초 1회)
!apt install chromium-chromedriver
!cp /usr/bin/chromedriver "/" #chorme driver 설치할 dir
!pip install chromedriver-autoinstaller

# 그 외 필요한 패키지들 다운
!pip install requests
!pip install PyPDF2

 

• 실행 준비

# 라이브러리 임포트
from selenium import webdriver
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.common.by import By
import sys
from selenium.webdriver.common.keys import Keys
import urllib.request
import os
from urllib.request import urlretrieve

import time
import pandas as pd
import chromedriver_autoinstaller  # setup chrome options

# chrome_options 설정 (크롬드라이버 실행 옵션)
## chromedriver 경로 명시
chrome_path = "" # chrome driver dir
sys.path.insert(0,chrome_path)

chrome_options = webdriver.ChromeOptions()
## colab은 새 창을 지원하지 않기 때문에 --headless 옵션 필요 (브라우저 창을 띄우지 않고 백그라운드에서 실행)
chrome_options.add_argument('--headless')
## 보안 샌드박스 없이 실행 (제한된 환경인 colab에서 필요한 설정)
chrome_options.add_argument('--no-sandbox')
## colab의 용량 제한으로 인한 오류 방지
chrome_options.add_argument('--disable-dev-shm-usage')
## chrome 언어 한국어로 설정
chrome_options.add_argument('lang=ko_KR')

## set the target URL
chromedriver_autoinstaller.install()

  위 옵션 설정에서 중요한 부분은 argument 중 하나인 --headless는 colab이 새 창을 지원하지 않기 때문에 필요한 옵션으로, 만약 로컬 환경을 사용한다면 추가하지 않아도 된다. 이러한 특성 때문에 실제로 크롤링이 돌아가는 창을 확인하며 내가 의도한 대로 실행되는지 확인할 수 없다는 단점이 있다. 나는 그래서 디버깅을 할 때 아래 코드를 이용해 문제가 된다고 예상되는 파트에서 백그라운드로 실행 중인 화면의 스크린숏을 남겨 확인했다.

driver.save_screenshot("screen.png")
from IPython.display import Image
Image("screen.png")

---

 

 크롤링 - 한경 컨센서스 

https://markets.hankyung.com/consensus

한경 컨센서스 | 한국경제

 

# url과 driver set
main_url = "https://markets.hankyung.com/consensus"
login_url = "https://id.hankyung.com/login/login.do"
driver =  webdriver.Chrome(options=chrome_options)

 

 한경 컨센서스의 경우 로그인을 해야 리포트 파일에 접근 가능하기 때문에 본격적인 크롤링 전 꼭 로그인 단계를 원격으로 실행해줘야 한다.

# 한경 컨센서스 로그인
driver.get(login_url)
user_id = driver.find_element(By.ID, 'userId')
user_id.send_keys("") #한경컨센서스 아이디 입력

user_pw = driver.find_element(By.ID, 'userPass')
user_pw.send_keys("") #한경컨센서스 비밀번호 입력
user_pw.send_keys(Keys.RETURN)

 

 리포트를 순회하면서 가져오는 방식은 현재 페이지에서 가장 첫 번째 리포트의 id를 가져오고 맨 끝 페이지로 이동해서 맨 마지막 리포트의 id를 가져와서 순회하는 식이다. 가장 첫 리포트부터 마지막 리포트까지 하나씩 감소하면서 id가 부여되는 방식이며 리포트 pdf의 url이 base_url + id의 형식을 지켰기에 가능한 방식이었다. 

 또한, 마지막 리포트에 접근할 때는 검색 탭에서 기간을 설정하여 접근 가능하도록 코드를 작성했다.

# report 시작 id 가져오기
import re

driver.get(main_url)
time.sleep(3)

a_tags = driver.find_elements(By.CSS_SELECTOR, 'a.ellip')
href = a_tags[0].get_attribute('href')
start_id = re.search(r'/consensus/view/(\d+)', href).group(1)
start_id = int(start_id)

# report 끝 id 가져오기
from selenium.webdriver.support.ui import Select

driver.get(main_url)
time.sleep(3)

## 기간 선택 (시작 id에서는 불필요)
dropdown_arrow = driver.find_element(By.CLASS_NAME, 'select2-selection__arrow')
dropdown_arrow.click()
time.sleep(1)

# "최근 7일", "1개월", "6개월", "1년"
recent_7 = driver.find_element(By.XPATH, '//li[contains(text(), "최근 7일")]')
recent_7.click()
time.sleep(1)

search_button = driver.find_element(By.CLASS_NAME, "btn-primary")
search_button.click() # 검색 버튼 클릭
time.sleep(1)

## 끝 페이지로 이동
end_page=driver.find_element(By.CLASS_NAME, 'btn-page-end')
end_page.click()

a_tags = driver.find_elements(By.CSS_SELECTOR, 'a.ellip')
href = a_tags[-1].get_attribute('href')
end_id = re.search(r'/consensus/view/(\d+)', href).group(1)
end_id = int(end_id)

 

 마지막으로 마지막 리포트의 id ~ 첫 리포트의 id를 순회하면서 pdf를 지정된 구글 드라이브 directory에 저장하며 끝낸다. 

import requests
from io import BytesIO
from PyPDF2 import PdfReader

base_url = "https://markets.hankyung.com/consensus/view/"
# Selenium 세션에서 쿠키 가져오기
cookies = driver.get_cookies()

# requests 세션 생성
session = requests.Session()
for cookie in cookies:
    session.cookies.set(cookie['name'], cookie['value'])

for i in range(end_id, start_id):
  driver.get(base_url + str(i))
  time.sleep(3)

  title=driver.find_element(By.CLASS_NAME, "report-tit")
  title=title.text.strip()
  title=re.sub(r'[\\/*?:"<>|]', "", title)

  file_url=driver.find_element(By.CSS_SELECTOR, 'a.btn.btn-default').get_attribute('href')
  driver.get(file_url)
  time.sleep(3)

  response = session.get(file_url)
  if response.status_code == 200:
      with open("#pdf 저장할 dir"+title+".pdf", "wb") as f:
          f.write(response.content)
      print("PDF 다운로드 완료.")
  else:
      print("PDF 파일 다운로드 실패.")

---

 

 크롤링 - 네이버 증권 리서치 

https://finance.naver.com/research/

리서치 : 네이버페이 증권

 

 네이버 증권 리서치의 장점은 로그인 없이 리포트 pdf 파일에 접근할 수 있다는 점이다. 한경 컨센서스처럼 별도로 원격 로그인 하는 코드를 쓸 필요가 없다. 다만, 종목/산업/시황/투자정보 리포트가 약간 다른 형태의 html로 주어져서 관련해서 살짝만 수정해 주면 된다. 또, 종목/산업 리포트끼리는 형식이 같고, 시황/투자정보 리포트끼리도 형식이 같다. 나는 종목/산업 분석 리포트를 중심으로 작성했다. 사실 시황/투자정보라고 하더라도 인덱스만 확인하고 살짝 수정해주면 된다.

# url과 driver set
url = "https://finance.naver.com/research/company_list.naver?&page=" #종목분석 리포트
#url = "https://finance.naver.com/research/industry_list.naver?page=" #산업분석 리포트

save_dir = "/" # pdf 저장할 directory
driver =  webdriver.Chrome(options=chrome_options)

 

 앞선 한경 컨센서스 크롤링에서는 검색탭에 주어지는 기간 설정을 이용했지만, 네이버 금융 리서치의 경우 그냥 직접 날짜를 확인하는 편이 더 간편해 보여서 그렇게 진행했다.

# 크롤링 기간 설정
from datetime import datetime, timedelta

end_date = datetime.now() # 오늘 날짜
start_date = end_date - timedelta(days=7) # n일 전 날짜

import requests
from io import BytesIO
from PyPDF2 import PdfReader

# Selenium 세션에서 쿠키 가져오기
cookies = driver.get_cookies()

# requests 세션 생성
session = requests.Session()
for cookie in cookies:
    session.cookies.set(cookie['name'], cookie['value'])

page = 1
while True:
  try:
    driver.get(url + str(page))
    time.sleep(3)
  except Exception:
    continue

  reports_tab = driver.find_element(By.XPATH, "/html/body/div[3]/div[2]/div[2]/div[1]/div[2]/table[1]/tbody")
  reports = reports_tab.find_elements(By.TAG_NAME, "tr") # 리포트 목록
  
  for r in reports:
    tds = r.find_elements(By.TAG_NAME, "td")
    if len(tds) < 4:
       continue

    company = tds[0].text.strip() # 회사 이름
    date = tds[4].text.strip() # 작성일

    ## 기간 확인
    try:
      report_date = datetime.strptime(date, "%y.%m.%d")
    except ValueError:
            continue
    if not (start_date <= report_date <= end_date):
        continue  # 원하는 날짜 범위가 아니면 스킵

    ## pdf 파일 url 추출
    file = tds[3].find_element(By.TAG_NAME, "a").get_attribute("href")
    driver.get(file)
    time.sleep(3)
    
    ## pdf 파일 저장
    response = session.get(file)
    time.sleep(3)
    if response.status_code == 200:
        with open(save_dir+company+"_"+date+".pdf", "wb") as f:
            f.write(response.content)
        print("PDF 다운로드 완료.")
    else:
        print("PDF 파일 다운로드 실패.")

  page += 1