[Python 100일 챌린지] Day 58 - 웹 크롤러

게시 2025/04/27

Day 58 - 웹 크롤러

By YonYonWare

26 분읽는 시간

[Python 100일 챌린지] Day 58 - 웹 크롤러

visited = set() → to_visit = [url] → while to_visit: 링크 추출 → 1000개 페이지 자동 탐색! 😊

한 페이지에서 링크를 따라 자동으로 웹사이트 전체 크롤링! URL 큐 관리, 중복 방지, 도메인 체크까지 완벽한 크롤러 제작!

(40-50분 완독 ⭐⭐⭐)

🎯 오늘의 학습 목표

웹 크롤러 개념 이해하기
크롤링 전략 수립하기
URL 큐와 방문 기록 관리하기
효율적인 크롤러 구현하기

📚 사전 지식

Day 53: BeautifulSoup 기초 - HTML 파싱과 데이터 추출
Day 54: 웹 스크래핑 고급 - 에러 처리, 재시도 로직, robots.txt
Day 52: HTTP 메서드 기초 - HTTP 요청/응답

🎯 학습 목표 1: 웹 크롤러 개념 이해하기

웹 크롤러(Web Crawler)란?

웹 크롤러는 웹 페이지를 자동으로 탐색하며 링크를 따라 여러 페이지를 방문하는 프로그램입니다. 구글 검색 엔진, 가격 비교 사이트, 뉴스 수집 시스템 등에서 사용됩니다.

웹 스크래퍼 vs 웹 크롤러

구분	웹 스크래퍼 (Web Scraper)	웹 크롤러 (Web Crawler)
목적	특정 데이터 추출	링크 탐색 및 수집
범위	단일 또는 소수 페이지	여러 페이지 자동 탐색
링크 추적	하지 않음	링크를 따라 이동
예시	뉴스 헤드라인 추출	사이트맵 생성, 검색 엔진
복잡도	낮음	높음 (큐, 중복 방지 필요)

크롤러 작동 원리

[시작 URL] → 큐에 추가
[큐에서 URL 꺼내기] → 아직 방문 안 했으면 진행
[페이지 다운로드] → HTML 파싱
[데이터 추출] → 필요한 정보 저장
[링크 추출] → 모든 <a> 태그에서 href 추출
[새 링크 필터링] → 같은 도메인, 중복 제거
[큐에 추가] → 2번으로 돌아가기

핵심 개념:

큐(Queue): 방문할 URL 목록 (to_visit = [])
방문 기록(Visited Set): 이미 방문한 URL (visited = set())
도메인 필터링: 같은 도메인만 크롤링 (외부 링크 제외)
중복 방지: visited에 URL 저장하여 재방문 방지

크롤러 사용 사례

사용 사례	설명	예시
검색 엔진	웹 전체 색인 생성	Google, Bing
가격 비교	여러 쇼핑몰 가격 수집	다나와, 에누리
뉴스 집계	여러 뉴스 사이트 기사 수집	Google News
사이트맵 생성	웹사이트 구조 파악	SEO 도구
데이터 분석	경쟁사 제품 정보 수집	마케팅 리서치
아카이빙	웹사이트 백업 및 보관	Internet Archive

🎯 학습 목표 2: 크롤링 전략 수립하기

크롤링 전략 유형

1. BFS (Breadth-First Search) - 너비 우선 탐색

  
from collections import deque

class BFSCrawler:
    def __init__(self, start_url):
        self.to_visit = deque([start_url])  # 큐 사용
        self.visited = set()

    def crawl(self):
        while self.to_visit:
            url = self.to_visit.popleft()  # 먼저 들어온 것부터

            if url in self.visited:
                continue

            print(f"방문: {url}")
            self.visited.add(url)

            # 현재 페이지의 모든 링크를 큐 끝에 추가
            links = self.extract_links(url)
            self.to_visit.extend(links)

특징:

시작 URL에서 가까운 페이지부터 탐색
사이트 홈페이지 근처 페이지 우선 크롤링
넓게 퍼지며 탐색

2. DFS (Depth-First Search) - 깊이 우선 탐색

  
class DFSCrawler:
    def __init__(self, start_url):
        self.to_visit = [start_url]  # 리스트 사용
        self.visited = set()

    def crawl(self):
        while self.to_visit:
            url = self.to_visit.pop()  # 마지막에 추가된 것부터

            if url in self.visited:
                continue

            print(f"방문: {url}")
            self.visited.add(url)

            # 현재 페이지의 모든 링크를 스택 끝에 추가
            links = self.extract_links(url)
            self.to_visit.extend(links)

특징:

한 경로를 끝까지 탐색한 후 다른 경로 탐색
깊은 페이지를 빠르게 발견
메모리 효율적 (큐보다 작음)

3. 우선순위 기반 크롤링

  
import heapq

class PriorityCrawler:
    def __init__(self, start_url):
        # (우선순위, URL) 튜플 저장
        self.to_visit = [(0, start_url)]  # 우선순위 큐
        self.visited = set()

    def crawl(self):
        while self.to_visit:
            priority, url = heapq.heappop(self.to_visit)

            if url in self.visited:
                continue

            print(f"방문 (우선순위 {priority}): {url}")
            self.visited.add(url)

            links = self.extract_links(url)

            for link in links:
                # 우선순위 계산 (예: URL 깊이, 페이지 중요도)
                link_priority = self.calculate_priority(link)
                heapq.heappush(self.to_visit, (link_priority, link))

    def calculate_priority(self, url):
        """URL 우선순위 계산"""
        # 예: URL 경로 깊이가 낮을수록 우선순위 높음
        depth = url.count('/') - 2  # https://example.com/ = 깊이 0
        return depth

특징:

중요한 페이지를 먼저 크롤링
SEO 점수, 페이지 깊이, 업데이트 날짜 등 고려
효율적인 리소스 사용

크롤링 전략 비교

전략	장점	단점	사용 사례
BFS	홈 근처 페이지 우선, 균형적 탐색	메모리 많이 사용	사이트맵 생성, 얕은 크롤링
DFS	메모리 효율적, 깊은 페이지 발견	한쪽으로 치우칠 수 있음	깊은 구조 탐색
우선순위	중요한 페이지 우선, 효율적	복잡한 구현	검색 엔진, 제한된 시간

🎯 학습 목표 3: URL 큐와 방문 기록 관리하기

URL 정규화 (Normalization)

문제: 같은 페이지인데 URL이 다르면 중복 방문

  
# 모두 같은 페이지!
https://example.com/page
https://example.com/page/
https://example.com/page?
https://example.com/page#section

해결: URL 정규화

  
from urllib.parse import urlparse, urlunparse, urljoin

def normalize_url(url):
    """URL 정규화"""
    parsed = urlparse(url)

    # 1. 프래그먼트(#) 제거
    parsed = parsed._replace(fragment='')

    # 2. 쿼리 파라미터 정렬 (선택적)
    # query = '&'.join(sorted(parsed.query.split('&')))
    # parsed = parsed._replace(query=query)

    # 3. 경로 끝 슬래시 제거
    path = parsed.path.rstrip('/')
    if not path:
        path = '/'
    parsed = parsed._replace(path=path)

    # 4. 소문자 변환 (도메인만)
    netloc = parsed.netloc.lower()
    parsed = parsed._replace(netloc=netloc)

    return urlunparse(parsed)

# 테스트
urls = [
    'https://Example.com/Page',
    'https://example.com/page/',
    'https://example.com/page#section',
]

for url in urls:
    print(normalize_url(url))
# 모두 출력: https://example.com/page

도메인 필터링

  
from urllib.parse import urlparse

class DomainFilter:
    def __init__(self, allowed_domains):
        """허용된 도메인 목록"""
        self.allowed_domains = set(allowed_domains)

    def is_allowed(self, url):
        """URL이 허용된 도메인인지 확인"""
        domain = urlparse(url).netloc

        # 서브도메인 포함 (예: blog.example.com → example.com)
        for allowed in self.allowed_domains:
            if domain == allowed or domain.endswith('.' + allowed):
                return True

        return False

# 사용 예시
filter = DomainFilter(['example.com'])

print(filter.is_allowed('https://example.com/page'))  # True
print(filter.is_allowed('https://blog.example.com/post'))  # True
print(filter.is_allowed('https://other.com/page'))  # False

URL 패턴 필터링

  
import re

def should_crawl(url):
    """크롤링해야 할 URL인지 판단"""

    # 제외할 패턴
    exclude_patterns = [
        r'/login',
        r'/logout',
        r'/admin',
        r'\.pdf$',
        r'\.zip$',
        r'\?utm_',  # 추적 파라미터
    ]

    for pattern in exclude_patterns:
        if re.search(pattern, url, re.IGNORECASE):
            return False

    # 포함할 패턴 (선택적)
    include_patterns = [
        r'/blog/',
        r'/products/',
    ]

    if include_patterns:
        for pattern in include_patterns:
            if re.search(pattern, url, re.IGNORECASE):
                return True
        return False  # include_patterns가 있는데 매칭 안 되면 제외

    return True

# 테스트
urls = [
    'https://example.com/blog/post-1',  # ✅
    'https://example.com/products/item',  # ✅
    'https://example.com/login',  # ❌
    'https://example.com/file.pdf',  # ❌
]

for url in urls:
    print(f"{url}: {should_crawl(url)}")

중복 방지 전략

1. Set 사용 (기본)

  
visited = set()

def add_visited(url):
    visited.add(normalize_url(url))

def is_visited(url):
    return normalize_url(url) in visited

장점: 빠른 조회 (O(1)), 간단 단점: 메모리 많이 사용, 재시작 시 기록 손실

2. 파일 기반 저장 (대규모)

  
import json

class FileBasedVisited:
    def __init__(self, filepath='visited.json'):
        self.filepath = filepath
        self.visited = self.load()

    def load(self):
        """파일에서 방문 기록 로드"""
        try:
            with open(self.filepath, 'r') as f:
                return set(json.load(f))
        except FileNotFoundError:
            return set()

    def save(self):
        """파일에 방문 기록 저장"""
        with open(self.filepath, 'w') as f:
            json.dump(list(self.visited), f)

    def add(self, url):
        self.visited.add(normalize_url(url))
        self.save()  # 실시간 저장

    def contains(self, url):
        return normalize_url(url) in self.visited

장점: 영구 저장, 재시작 가능 단점: 디스크 I/O 느림

3. Bloom Filter (초대규모)

  
# pybloom-live 라이브러리 사용
from pybloom_live import BloomFilter

visited = BloomFilter(capacity=1000000, error_rate=0.001)

visited.add('https://example.com/page1')
print('https://example.com/page1' in visited)  # True

장점: 메모리 매우 효율적 (수백만 URL 저장 가능) 단점: False Positive 가능 (아주 낮은 확률로 “방문했다”고 잘못 판단)

🎯 학습 목표 4: 효율적인 크롤러 구현하기

완전한 웹 크롤러 클래스

  
import requests
from bs4 import BeautifulSoup
from urllib.parse import urljoin, urlparse, urlunparse
from collections import deque
import time
import logging

# 로깅 설정
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

class WebCrawler:
    def __init__(self, start_url, max_pages=50, delay=1):
        """
        웹 크롤러 초기화

        Args:
            start_url: 시작 URL
            max_pages: 최대 크롤링 페이지 수
            delay: 요청 간 지연 시간 (초)
        """
        self.start_url = start_url
        self.max_pages = max_pages
        self.delay = delay

        self.visited = set()  # 방문한 URL
        self.to_visit = deque([start_url])  # 방문할 URL 큐
        self.data = []  # 수집한 데이터

        # 도메인 추출
        self.allowed_domain = urlparse(start_url).netloc

        logging.info(f"크롤러 시작: {start_url} (도메인: {self.allowed_domain})")

    def normalize_url(self, url):
        """URL 정규화"""
        parsed = urlparse(url)

        # 프래그먼트 제거
        parsed = parsed._replace(fragment='')

        # 경로 정리
        path = parsed.path.rstrip('/')
        if not path:
            path = '/'
        parsed = parsed._replace(path=path)

        # 도메인 소문자
        netloc = parsed.netloc.lower()
        parsed = parsed._replace(netloc=netloc)

        return urlunparse(parsed)

    def is_valid_url(self, url):
        """URL이 크롤링 가능한지 확인"""
        parsed = urlparse(url)

        # 같은 도메인인지 확인
        if parsed.netloc != self.allowed_domain:
            return False

        # HTTP/HTTPS만 허용
        if parsed.scheme not in ['http', 'https']:
            return False

        # 제외할 확장자
        exclude_ext = ['.pdf', '.zip', '.jpg', '.png', '.gif', '.mp4']
        if any(url.lower().endswith(ext) for ext in exclude_ext):
            return False

        return True

    def extract_links(self, url, soup):
        """페이지에서 링크 추출"""
        links = []

        for link_tag in soup.find_all('a', href=True):
            href = link_tag['href']

            # 절대 URL로 변환
            full_url = urljoin(url, href)

            # 정규화
            normalized_url = self.normalize_url(full_url)

            # 유효성 검사
            if self.is_valid_url(normalized_url):
                links.append(normalized_url)

        return links

    def extract_data(self, url, soup):
        """페이지에서 데이터 추출 (커스터마이징 가능)"""
        # 예시: 제목과 첫 번째 단락 추출
        title = soup.find('h1')
        title_text = title.get_text(strip=True) if title else 'No Title'

        first_p = soup.find('p')
        content = first_p.get_text(strip=True) if first_p else ''

        return {
            'url': url,
            'title': title_text,
            'content': content[:200]  # 처음 200자
        }

    def crawl_page(self, url):
        """단일 페이지 크롤링"""
        try:
            logging.info(f"크롤링 중: {url}")

            response = requests.get(url, timeout=10, headers={
                'User-Agent': 'Mozilla/5.0 (compatible; MyCrawler/1.0)'
            })

            response.raise_for_status()

            soup = BeautifulSoup(response.text, 'html.parser')

            # 데이터 추출
            page_data = self.extract_data(url, soup)
            self.data.append(page_data)

            # 링크 추출
            links = self.extract_links(url, soup)

            # 새 링크를 큐에 추가
            for link in links:
                if link not in self.visited and link not in self.to_visit:
                    self.to_visit.append(link)

            return True

        except requests.RequestException as e:
            logging.error(f"요청 실패: {url} - {e}")
            return False

        except Exception as e:
            logging.error(f"파싱 오류: {url} - {e}")
            return False

    def crawl(self):
        """크롤링 시작"""
        while self.to_visit and len(self.visited) < self.max_pages:
            url = self.to_visit.popleft()

            # 이미 방문했으면 건너뛰기
            if url in self.visited:
                continue

            # 방문 기록 추가
            self.visited.add(url)

            # 페이지 크롤링
            success = self.crawl_page(url)

            if success:
                # 예의 있는 크롤링 (딜레이)
                time.sleep(self.delay)

        logging.info(f"크롤링 완료: {len(self.visited)}개 페이지 방문")
        logging.info(f"수집된 데이터: {len(self.data)}개")

    def save_data(self, filepath='crawler_data.json'):
        """수집한 데이터 저장"""
        import json

        with open(filepath, 'w', encoding='utf-8') as f:
            json.dump(self.data, f, ensure_ascii=False, indent=2)

        logging.info(f"데이터 저장 완료: {filepath}")

    def get_summary(self):
        """크롤링 요약 정보"""
        return {
            'total_visited': len(self.visited),
            'total_queued': len(self.to_visit),
            'data_collected': len(self.data)
        }

사용 예시:

  
# 크롤러 생성
crawler = WebCrawler(
    start_url='https://example.com',
    max_pages=20,
    delay=2
)

# 크롤링 실행
crawler.crawl()

# 데이터 저장
crawler.save_data('output.json')

# 요약 출력
summary = crawler.get_summary()
print(f"방문한 페이지: {summary['total_visited']}개")
print(f"대기 중인 URL: {summary['total_queued']}개")
print(f"수집한 데이터: {summary['data_collected']}개")

병렬 크롤링 (Multi-threading)

  
import concurrent.futures
import threading

class ParallelCrawler(WebCrawler):
    def __init__(self, start_url, max_pages=50, delay=1, max_workers=5):
        super().__init__(start_url, max_pages, delay)
        self.max_workers = max_workers
        self.lock = threading.Lock()  # Thread-safe를 위한 Lock

    def crawl_parallel(self):
        """병렬 크롤링"""
        with concurrent.futures.ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            futures = []

            while len(self.visited) < self.max_pages:
                # 큐에서 URL 가져오기 (Thread-safe)
                with self.lock:
                    if not self.to_visit:
                        break

                    url = self.to_visit.popleft()

                    if url in self.visited:
                        continue

                    self.visited.add(url)

                # 비동기로 크롤링
                future = executor.submit(self.crawl_page, url)
                futures.append(future)

                time.sleep(self.delay / self.max_workers)  # 조정된 딜레이

            # 모든 작업 완료 대기
            concurrent.futures.wait(futures)

        logging.info(f"병렬 크롤링 완료: {len(self.visited)}개 페이지")

주의사항:

너무 많은 스레드는 서버에 부담 (max_workers=3~5 권장)
Thread-safe한 자료구조 사용 (threading.Lock)
robots.txt와 사이트 정책 준수

robots.txt 준수

  
from urllib.robotparser import RobotFileParser

class RobotAwareCrawler(WebCrawler):
    def __init__(self, start_url, max_pages=50, delay=1):
        super().__init__(start_url, max_pages, delay)

        # robots.txt 파서 초기화
        self.robot_parser = RobotFileParser()
        robots_url = urljoin(start_url, '/robots.txt')
        self.robot_parser.set_url(robots_url)

        try:
            self.robot_parser.read()
            logging.info(f"robots.txt 읽기 완료: {robots_url}")
        except:
            logging.warning(f"robots.txt 읽기 실패: {robots_url}")

    def is_valid_url(self, url):
        """URL 유효성 + robots.txt 확인"""
        # 기본 검증
        if not super().is_valid_url(url):
            return False

        # robots.txt 확인
        if not self.robot_parser.can_fetch('*', url):
            logging.info(f"robots.txt에서 차단됨: {url}")
            return False

        return True

💡 실전 팁 & 주의사항

✅ DO (이렇게 하세요)

robots.txt 준수: 크롤링 전에 /robots.txt 확인하여 허용된 경로만 크롤링
User-Agent 설정: 요청 헤더에 크롤러 정보 명시 (User-Agent: MyBot/1.0)
딜레이 추가: 서버 부담 방지를 위해 요청 간 1~2초 대기
URL 정규화: 중복 방문 방지를 위해 URL 정규화
예외 처리: 네트워크 오류, 파싱 오류에 대한 예외 처리
데이터 저장: 크롤링 중간에 주기적으로 데이터 저장 (재시작 대비)

❌ DON’T (하지 마세요)

무한정 크롤링: 반드시 max_pages 제한 설정
딜레이 없이 요청: 초당 수십 건 요청은 서버 부담 및 IP 차단 위험
모든 링크 따라가기: 외부 도메인까지 크롤링하면 무한 루프
로그인 필요 페이지: 인증 없이 접근 시도하지 말 것
개인정보 수집: 법적 문제 발생 가능 (GDPR, 개인정보보호법)

크롤링 윤리

  
# ✅ 좋은 크롤러
crawler = WebCrawler(
    start_url='https://example.com',
    max_pages=100,
    delay=2  # 2초 대기
)

# 요청 헤더에 연락처 포함
headers = {
    'User-Agent': 'MyBot/1.0 (contact@example.com)'
}

# ❌ 나쁜 크롤러
bad_crawler = WebCrawler(
    start_url='https://example.com',
    max_pages=10000,  # 너무 많음
    delay=0.1  # 너무 빠름
)

🧪 연습 문제

문제 1: 뉴스 사이트 헤드라인 크롤러

요구사항:

뉴스 사이트의 메인 페이지에서 시작
기사 페이지 링크만 추출 (예: /news/, /article/ 경로)
각 기사에서 제목, 날짜, 본문 첫 단락 추출
최대 30개 기사 수집
JSON 파일로 저장

💡 힌트

is_valid_url을 오버라이드하여 기사 URL 패턴만 허용
extract_data에서 제목(h1), 날짜(time 또는 .date), 본문(article p) 추출
save_data로 JSON 저장

✅ 정답 예시

  
import re

class NewsCrawler(WebCrawler):
    def is_valid_url(self, url):
        """기사 URL만 허용"""
        if not super().is_valid_url(url):
            return False

        # /news/ 또는 /article/ 경로만
        if re.search(r'/(news|article)/', url):
            return True

        return False

    def extract_data(self, url, soup):
        """기사 데이터 추출"""
        # 제목
        title_tag = soup.find('h1') or soup.find('h2', class_='title')
        title = title_tag.get_text(strip=True) if title_tag else 'No Title'

        # 날짜
        date_tag = soup.find('time') or soup.find(class_=re.compile('date'))
        date = date_tag.get_text(strip=True) if date_tag else 'Unknown'

        # 본문 첫 단락
        article_tag = soup.find('article') or soup.find('div', class_='content')
        if article_tag:
            first_p = article_tag.find('p')
            content = first_p.get_text(strip=True) if first_p else ''
        else:
            content = ''

        return {
            'url': url,
            'title': title,
            'date': date,
            'content': content[:300]
        }

# 사용
crawler = NewsCrawler(
    start_url='https://news-site.com',
    max_pages=30,
    delay=2
)

crawler.crawl()
crawler.save_data('news_articles.json')

print(f"수집한 기사: {len(crawler.data)}개")

문제 2: 사이트맵 생성기