데이터 엔지니어링/Database

6. 파이썬 기반 크롤러

427paul 2026. 6. 5. 21:54

1. 크롤링(Crawling) vs 스크래핑(Scraping) 개념 및 기술 스택

  • 웹 크롤링 (Web Crawling): 조직적·자동화된 방법으로 웹을 순회하며 하이퍼링크를 탐색·수집하고 인덱싱(색인)하는 '무한 웹 순회 및 링크 수집 기술' (ex: 구글 검색 엔진 봇)
  • 웹 스크래핑 (Web Scraping): 정해진 특정 웹 페이지에 타겟팅하여 우리가 원하는 특정 알갱이 데이터(가격, 제목, 리뷰 등)만 도려내어 구조화된 파일로 보관하는 '데이터 추출 기술'

🛠️ 수집 대상에 따른 핵심 기술 스택

  • 정적 페이지 데이터 추출 (HTTP 통신):
    • requests, urllib: 서버에 HTTP 요청을 보내고 순수한 HTML 소스 코드를 응답받는 라이브러리
    • BeautifulSoup: 받아온 HTML 문자열을 메모리에 DOM 트리 구조로 변환(파싱)하여 태그 및 데이터 추출을 쉽게 돕는 파서 라이브러리
  • 동적 페이지 및 SPA(Single Page Application) 대응:
    • Selenium, Playwright: 실제 브라우저를 원격 제어하여 자바스크립트(JavaScript) 동적 렌더링, 클릭, 스크롤, 대기(Wait) 등을 사람처럼 시뮬레이션하는 웹 드라이버 도구

2. 크롤러의 동작 아키텍처 및 순환 메커니즘

전체적인 크롤러 시스템은 대기열과 요청, 파싱, 저장이 유기적으로 맞물려 돌아가는 순환 루프 구조를 가집니다.

  1. 시작점 및 대기열 (Seed URLs & Frontier Queue): 최초의 기초 주소 리스트(Seed)를 주입받아 스케줄러 대기열(Queue)에 쌓아두고 하나씩 꺼내어 탐색
  2. 요청 및 응답 (Request & Response): 대기열의 URL을 기반으로 타겟 웹 서버에 HTTP GET 요청을 보내고, 뼈대가 되는 HTML 코드를 반환
  3. 파싱 및 제어 (Page Parser): BeautifulSoup 등을 통해 HTML 원본을 DOM 구조로 해석하고, 다중 페이지 수집을 위한 페이징 파라미터를 계산
  4. 데이터 추출 및 검증 (Extract Data & Error Handling): 목표로 하는 정보(제목, 가격 등)를 추출한 뒤, 레이아웃 변경이나 차단 여부를 검증 (요소가 없다면 Break 및 예외 처리)
  5. 링크 발견 및 대기열 피딩 (Link Queue): 페이지 내 또 다른 하이퍼링크(<a href="...">)를 모조리 긁어모아 다시 스케줄러 대기열(Frontier Queue) 끝에 추가하여 스스로 탐색 범위를 무한히 확장
  6. 데이터 영속화 (Store Results): 검증된 최종 데이터를 JSON, CSV 파일이나 데이터베이스(DB)에 영구 저장
  7. Politeness (신사 협정): 대기열에서 다음 요청을 보낼 때, 서버 과부하를 막기 위해 의도적인 지연 시간(time.sleep)을 부여하고 robots.txt 규약을 검증하는 통제 장치가 함께 가동

3. 라이선스, 법적 주의점 및 방어 트렌드 (Compliance)

무분별한 데이터 수집은 강력한 민·형사상 처벌 대상이 되므로 반드시 아래 컴플라이언스를 준수해야 합니다.

  • 로봇 배제 표준 (robots.txt):
    • 웹사이트 루트 경로(ex: 사이트주소/robots.txt)에 명시된 수집 거부 규칙을 무조건 확인하고 준수.
    • 최근에는 AI 모델의 무단 무손실 학습을 거부하는 ai.txt, llms.txt 가 새로운 규격으로 자리 잡고 있음
  • 서비스 이용약관(ToS) 및 개인정보: 로그인이 필요한 서비스(Gated Data)의 경우 "자동화 수단 이용 금지" 약관 위반 시 민사 손해배상 대상이 되며, 이름·전화번호 등 개인식별정보(PII)를 대량 수집하는 행위는 개인정보보호법(국내) 및 GDPR(유럽)에 정면 위반
  • 업계의 Anti-Scraping(방어) 및 크롤러의 진화:
    • 방어측: Cloudflare, Akamai 등 보안 솔루션을 통해 단순 IP 차단을 넘어 브라우저 핑거프린팅과 행동 패턴을 분석해 봇을 실시간 차단하며, AI 기반 고도화 캡차(CAPTCHA)를 요구
    • 크롤러측: 인간의 마우스 움직임과 스크롤 속도를 모방하는 헤드리스 브라우저 고도화와 정규식/CSS 선택자 없이 HTML 스냅샷을 통째로 LLM 프롬프트에 넣어 JSON 구조로만 정제해 오는 '에이전트형 스크래핑(Agentic Scraping)'으로 패러다임이 이동 중

4. 실습 예제 기반 구현 전략 요약

🔹사전 준비

  # 가상환경 생성 및 활성화
  python -m venv crawler
  cd crawler
  source ./bin/activate

  # 필요 라이브러리 설치
  pip install requests beautifulsoup4

🔹 ① 네이버 뉴스 검색 결과 수집 (정적 requests vs 동적 Selenium 구조 격파)

  • Version 1 (requests + BeautifulSoup):
  • 네이버의 검색 결과 페이징 공식(start = page * 10 + 1)을 이용해 정적으로 접근
  • 차단 방지를 위해 User-Agent 헤더 설정이 필수적이나, 네이버의 HTML 구조 패치나 봇 차단 솔루션 작동 시 요소 탐색 실패 위험 존재
  import sys
  import time
  import requests
  from bs4 import BeautifulSoup

  def scrape_naver_news(search_keyword, max_pages=2):
      """네이버 뉴스에서 키워드를 검색하여 제목과 링크를 수집하는 함수

      :param search_keyword: 검색할 키워드 (str)
      :param max_pages: 수집할 페이지 수 (int)
      """
      print(f"[{search_keyword}] 관련 뉴스 수집을 시작합니다. (최대 {max_pages}페이지)")
      print("-" * 60)

      # 1. 크롤링 시 차단을 방지하기 위한 필수 헤더(User-Agent) 설정
      headers = {
          "User-Agent": (
              "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
              "AppleWebKit/537.36 (KHTML, like Gecko) "
              "Chrome/120.0.0.0 Safari/537.36"
          )
      }

      results_count = 0

      for page in range(max_pages):
          # 네이버 뉴스 검색 페이징 공식: 1페이지(start=1), 2페이지(start=11), 3페이지(start=21) ...
          start_val = page * 10 + 1
          url = f"<https://search.naver.com/search.naver?where=news&query={search_keyword}&start={start_val}>"

          try:
              # 2. HTTP GET 요청 보냄
              response = requests.get(url, headers=headers, timeout=5)

              # 응답 상태 코드가 200(정상)이 아니면 예외 발생
              response.raise_for_status()

          except requests.exceptions.RequestException as e:
              print(f"[에러] 페이지 요청 중 문제가 발생했습니다: {e}")
              break

          # 3. HTML 파싱
          soup = BeautifulSoup(response.text, "html.parser")

          # 4. 뉴스 기사 블록 요소 선택 (네이버 검색 개편 후 기준 최신 Selector 기재)
          # 각 뉴스 기사 카드는 'bx' 클래스를 가진 'li' 태그 내에 존재함
          news_items = soup.select("ul.list_news > li.bx")

          if not news_items:
              print(f"{page+1}페이지에서 검색 결과 요소를 찾을 수 없습니다. 구조 변경을 확인하세요.")
              break

          print(f"[진행] {page+1} 페이지 수집 중...")

          for item in news_items:
              # 뉴스 제목과 링크가 있는  태그 선택 (news_tit 클래스)
              title_element = item.select_one("a.news_tit")

              if title_element:
                  title = title_element.get_text(strip=True)
                  link = title_element["href"]

                  results_count += 1
                  print(f"{results_count}. {title}")
                  print(f"   링크: {link}")

          # 5. 과도한 트래픽 유발 방지를 위한 정중한 매너 (Politeness Delay)
          # 서버에 부담을 주지 않기 위해 페이지 전환 시 1.5초간 대기
          time.sleep(1.5)

      print("-" * 60)
      print(f"수집 완료: 총 {results_count}개의 뉴스 데이터를 안전하게 수집했습니다.")

  if __name__ == "__main__":
      # 인코딩 문제 방지 (윈도우 환경 대응)
      if sys.platform == "win32":
          import io

          sys.stdout = io.TextIOWrapper(sys.stdout.detach(), encoding="utf-8")

      # 실습 키워드 설정 (예: AI 기술)
      keyword = "인공지능 트렌드"
      scrape_naver_news(keyword, max_pages=2)

  • Version 2 (Selenium 가동 우회 버전) ⭐:
    • 봇 감지 바를 숨기고(excludeSwitches), 웹 드라이버 감지 변수(navigator.webdriver)를 undefined로 강제 변환 조작하여 보안 솔루션을 우회
    • 클래스명 의존성을 100% 제거하기 위해 특정 클래스(news_tit) 대신, 모든 <a> 태그 중 URL 패턴(article, news.naver.com 등)과 글자 수 제한(10자 이상)을 결합하여 구조 개편에도 절대 깨지지 않는 무적의 추출 패턴을 구현
  import sys
  import time
  from bs4 import BeautifulSoup
  from selenium import webdriver
  from selenium.webdriver.chrome.options import Options

  def scrape_naver_news_absolute(search_keyword):
      print(f"🚀 [구조 격파 가동] '{search_keyword}' 뉴스 수집을 시작합니다.")
      print("-" * 60)

      chrome_options = Options()
      # Headless 모드 해제 (실제 브라우저 구동으로 보안 통과)
      chrome_options.add_argument("--no-sandbox")
      chrome_options.add_argument("--disable-dev-shm-usage")

      # 자동화 봇 감지 바 숨기기
      chrome_options.add_experimental_option(
          "excludeSwitches", ["enable-automation"]
      )
      chrome_options.add_experimental_option("useAutomationExtension", False)

      driver = webdriver.Chrome(options=chrome_options)

      # navigator.webdriver 변수 조작 (보안 솔루션 우회)
      driver.execute_cdp_cmd(
          "Page.addScriptToEvaluateOnNewDocument",
          {
              "source": "Object.defineProperty(navigator, 'webdriver', {get: () => undefined})"
          },
      )

      try:
          # 뉴스 탭 주소로 진입
          url = f"<https://search.naver.com/search.naver?where=news&query={search_keyword}>"
          driver.get(url)

          # 화면에 뉴스 데이터가 바인딩될 때까지 3.5초간 충분히 대기
          time.sleep(3.5)

          # 브라우저가 완성한 최종 가상 DOM 소스 확보
          final_html = driver.page_source
          soup = BeautifulSoup(final_html, "html.parser")

          # [⭐ 핵심 수정: 클래스명 의존성 100% 제거]
          # 클래스명(news_tit 등)이 네이버 패치로 바뀌어도, '뉴스 기사 링크'라는 본질은 변하지 않습니다.
          # href 주소에 일반 언론사나 네이버 뉴스 포맷이 들어간 모든  태그를 포괄적으로 수집합니다.
          all_links = soup.find_all("a")

          news_results = []
          for link_element in all_links:
              href = link_element.get("href", "")
              title = link_element.get_text(strip=True)

              # 네이버 뉴스 검색 결과 링크들의 공통적인 URL 패턴 및 글자 수 필터링
              # 보통 제목 링크는 언론사 도메인이거나 네이버 뉴스 링크이며, 제목 텍스트가 10자 이상으로 깁니다.
              if (
                  "sp_nws" in href
                  or "news.naver.com" in href
                  or "sid=" in href
                  or "article" in href
              ) and len(title) > 10:

                  # 중복 수집 방지 및 무효 데이터 제거
                  if (
                      "네이버뉴스" not in title
                      and {"title": title, "link": href} not in news_results
                  ):
                      news_results.append({"title": title, "link": href})

          if not news_results:
              print(
                  "⚠️ 브라우저 화면은 떴으나 데이터 추출 패턴이 일치하지 않습니다."
              )
              return

          print(f"[성공] 화면 렌더링 검증 완료. 데이터를 출력합니다.\\n")

          for index, res in enumerate(news_results, start=1):
              print(f"{index}. {res['title']}")
              print(f"   🔗 링크: {res['link']}")

      except Exception as e:
          print(f"❌ 에러 발생: {e}")

      finally:
          # 강사님이 결과를 눈으로 편하게 확인하실 수 있도록 5초간 창을 유지한 뒤 닫습니다.
          time.sleep(5)
          driver.quit()

  if __name__ == "__main__":
      if sys.platform == "win32":
          import io

          sys.stdout = io.TextIOWrapper(sys.stdout.detach(), encoding="utf-8")

      keyword = "인공지능 트렌드"
      scrape_naver_news_absolute(keyword)

🔹 ② 네이버 페이 증권 기반 주가 정보 획득 (데이터 구조화)

  • 특정 기업의 6자리 종목 코드(ex: 삼성전자 005930)를 받아 실시간 주가 데이터를 딕셔너리로 맵핑하는 정석 스크래퍼
  • .no_today 영역의 blind 클래스를 통해 현재가를 도려내고, .no_exday 내의 아이콘 텍스트(ico)를 분석해 상승(▲)과 하락(▼) 기호를 가독성 있게 가공
  • 전일종가, 시가, 고가, 거래량 등의 다각화 데이터를 순서대로 매핑하여 API 서비스 부재 시 대안으로 쓸 수 있는 템플릿 구조
  import sys
  import time
  import requests
  from bs4 import BeautifulSoup

  def get_naver_stock_info(company_code):
      """네이버 금융에서 특정 종목의 현재 주가 정보를 가져오는 함수

      :param company_code: 6자리 종목 코드 (str)
      """
      # 1. 대상 URL 설정 (네이버 금융 개별 종목 홈)
      url = f"<https://finance.naver.com/item/main.naver?code={company_code}>"

      # 2. 크롤링 차단 방지를 위한 브라우저 헤더 설정
      headers = {
          "User-Agent": (
              "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
              "AppleWebKit/537.36 (KHTML, like Gecko) "
              "Chrome/120.0.0.0 Safari/537.36"
          )
      }

      try:
          # HTTP GET 요청 및 예외 처리
          response = requests.get(url, headers=headers, timeout=5)
          response.raise_for_status()
      except requests.exceptions.RequestException as e:
          print(f"📢 [네트워크 에러] 데이터 요청 실패: {e}")
          return None

      # 3. HTML 파싱 (BeautifulSoup 객체 생성)
      soup = BeautifulSoup(response.text, "html.parser")

      # 4. 주가 정보 영역 데이터 추출 (wrap_company 클래스 및 today 클래스 타겟)
      # 종목명 추출
      wrap_company = soup.select_one(".wrap_company")
      if not wrap_company:
          print(
              f"⚠️ 종목 코드를 확인해 주세요. ({company_code} 데이터를 찾을 수 없습니다.)"
          )
          return None

      company_name = wrap_company.select_one("h2 > a").get_text(strip=True)

      # 실시간 주가 정보 테이블 탐색 (.no_today 영역)
      no_today = soup.select_one(".no_today")
      if not no_today:
          print("⚠️ 주가 데이터 레이아웃을 분석할 수 없습니다.")
          return None

      # 현재가 추출 (blind 클래스 내 텍스트 활용)
      current_price = no_today.select_one(".blind").get_text(strip=True)

      # 전일대비 및 등락률 영역 탐색 (.no_exday 영역)
      no_exday = soup.select_one(".no_exday")
      # 상승, 하락 여부를 아이콘 텍스트로 판별
      ico_direction = no_exday.select_one(".ico").get_text(strip=True)
      # 변동 금액과 등락률 (blind 클래스 안의 텍스트 추출)
      blinds = no_exday.select(".blind")

      change_amount = blinds[0].get_text(strip=True) if len(blinds) > 0 else "0"
      change_rate = blinds[1].get_text(strip=True) if len(blinds) > 1 else "0%"

      # 기호 가독성 처리
      if "상승" in ico_direction or "상한" in ico_direction:
          direction_sign = "▲"
      elif "하락" in ico_direction or "하한" in ico_direction:
          direction_sign = "▼"
      else:
          direction_sign = ""

      # 5. 거래량 및 시가/고가/저가 추가 정보 추출 (실무 데이터 다각화용)
      # .no_info 테이블 내부의 blind 값을 순서대로 매핑
      no_info = soup.select_one(".no_info")
      info_blinds = no_info.select(".blind") if no_info else []

      prev_close = info_blinds[0].get_text(strip=True) if len(info_blinds) > 0 else "-"  # 전일
      market_open = info_blinds[1].get_text(strip=True) if len(info_blinds) > 1 else "-"  # 시가
      high_price = info_blinds[5].get_text(strip=True) if len(info_blinds) > 5 else "-"  # 고가
      volume = info_blinds[3].get_text(strip=True) if len(info_blinds) > 3 else "-"  # 거래량

      # 결과 데이터 구조화 (딕셔너리 반환)
      stock_data = {
          "종목명": company_name,
          "종목코드": company_code,
          "현재가": current_price,
          "전일대비": f"{direction_sign} {change_amount}",
          "등락률": f"{direction_sign}{change_rate}",
          "거래량": volume,
          "시가": market_open,
          "고가": high_price,
          "전일종가": prev_close,
      }

      return stock_data

  if __name__ == "__main__":
      # 윈도우 환경 콘솔 출력 인코딩 방어 코드
      if sys.platform == "win32":
          import io

          sys.stdout = io.TextIOWrapper(sys.stdout.detach(), encoding="utf-8")

      # 실습용 타겟 종목 정의 (삼성전자: 005930, SK하이닉스: 000660)
      target_stocks = ["005930", "000660"]

      print("📊 [네이버 금융] 주요 종목 실시간 주가 정보 모니터링")
      print("=" * 60)

      for code in target_stocks:
          data = get_naver_stock_info(code)

          if data:
              print(f"■ {data['종목명']} ({data['종목코드']})")
              print(f"  - 현재가: {data['현재가']} 원")
              print(f"  - 등락률: {data['등락률']} ({data['전일대비']} 원)")
              print(f"  - 거래량: {data['거래량']} 주")
              print(
                  f"  - 시가/고가: {data['시가']} 원 / {data['고가']} 원"
              )
              print("-" * 60)

          # 연속 요청 시 매너 필터 (서버 과부하 방지)
          time.sleep(1.0)

'데이터 엔지니어링 > Database' 카테고리의 다른 글

5. ERD 정규화 및 비정규화  (0) 2026.06.05
4. ERD 설계  (0) 2026.06.04
3. ERD 개요  (0) 2026.06.04
2. SQL 기초  (0) 2026.05.29
1. Database 개요  (0) 2026.05.29