CrawlSpider 클래스 개요
CrawlSpider는 Scrapy의 기본 Spider 클래스를 상속받아 확장된 클래스입니다. 기본 Spider는 start_urls에 정의된 초기 URL만 크롤링하는 데 적합하지만, CrawlSpider는 추출된 링크를 자동으로 따라가며 탐색하는 규칙(Rule) 기반 메커니즘을 제공합니다. 대규모 웹사이트나 페이지네이션이 있는 사이트를 크롤링할 때 매우 효율적입니다.
아래 명령어를 통해 CrawlSpider 템플릿 기반의 스파이더 파일을 생성할 수 있습니다:
scrapy genspider -t crawl tech_jobs techcorp.com
내부 동작 원리
CrawlSpider의 핵심 동작 방식을 이해하기 위해 내부 로직을 요약한 추상화 코드를 살펴보겠습니다. 원본 소스코드의 구조와 변수명을 변경하여 재구성한 코드는 다음과 같습니다:
class AutoCrawler(BaseSpider):
navigation_rules = ()
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._build_rules()
def start_parse(self, response):
# start_urls의 응답을 처리하며, 초기 콜백과 링크 추후(follow=True)를 설정
return self._analyze_response(response, self.initial_parse, follow=True)
def initial_parse(self, response):
return []
def _fetch_links(self, response):
if not isinstance(response, HtmlResponse):
return
visited = set()
# 정의된 규칙들을 순회하며 링크 추출
for idx, nav_rule in enumerate(self._active_rules):
found_links = [lnk for lnk in nav_rule.extractor.get_links(response) if lnk not in visited]
if found_links and nav_rule.link_filter:
found_links = nav_rule.link_filter(found_links)
for lnk in found_links:
visited.add(lnk)
req = Request(url=lnk.url, callback=self._download_handler)
req.meta.update(rule_idx=idx, link_txt=lnk.text)
yield nav_rule.req_filter(req)
def _download_handler(self, response):
active_rule = self._active_rules[response.meta['rule_idx']]
return self._analyze_response(response, active_rule.handler, active_rule.handler_args, active_rule.track)
def _analyze_response(self, response, handler, handler_args, track=True):
# 콜백 함수가 있으면 실행하여 Item 또는 Request 반환
if handler:
output = handler(response, **handler_args) or ()
for result in output:
yield result
# follow 설정이 True면 추가 링크 추출 및 요청 생성
if track and self._auto_follow:
for req in self._fetch_links(response):
yield req
LinkExtractor
LinkExtractor는 Response 객체에서 링크를 추출하는 역할을 수행합니다.
extract_links() 메서드를 통해 매칭된 링크들을 반환합니다.
class scrapy.linkextractors.LinkExtractor(
allow = (),
deny = (),
allow_domains = (),
deny_domains = (),
deny_extensions = None,
restrict_xpaths = (),
tags = ('a','area'),
attrs = ('href'),
canonicalize = True,
unique = True,
process_value = None
)
주요 파라미터:
allow: 지정된 정규표현식과 매칭되는 URL만 추출합니다. 비어있으면 모든 URL이 매칭됩니다.deny: 해당 정규표현식과 매칭되는 URL을 제외합니다.allow_domains: 링크 추출을 허용할 도메인 리스트입니다.deny_domains: 링크 추출을 차단할 도메인 리스트입니다.restrict_xpaths: XPath 표현식을 사용하여 특정 영역 내의 링크만 추출하도록 제한합니다. allow와 함께 사용할 수 있습니다.
Rule 객체
CrawlSpider의
rules 속성은 Rule 객체의 튜플을 포함합니다. 각 Rule은 링크 추출 및 후속 동작을 정의합니다. 여러 Rule이 동일한 링크와 매칭될 경우, 리스트에 정의된 순서상 첫 번째 Rule이 적용됩니다.
class scrapy.spiders.Rule(
link_extractor,
callback = None,
cb_kwargs = None,
follow = None,
process_links = None,
process_request = None
)
주요 파라미터:
link_extractor: LinkExtractor 인스턴스를 지정하여 추출 대상 링크를 정의합니다.callback: 추출된 링크의 Response를 처리할 콜백 함수입니다. 주의: CrawlSpider의 기본 동작을 오버라이드하는 parse 메서드를 콜백으로 사용하면 스파이더가 정상 작동하지 않으므로 반드시 다른 이름을 사용해야 합니다.follow: 추출된 링크에서 다시 링크를 추출하여 탐색할지 여부를 결정합니다. callback이 None이면 기본값은 True, 지정되어 있으면 False입니다.process_links: LinkExtractor가 추출한 링크 리스트를 필터링 또는 변형할 때 호출되는 스파이더 내부 메서드입니다.process_request: Rule에 의해 생성된 각 Request 객체를 필터링할 때 호출되는 메서드입니다.
실전 예제: 채용 정보 크롤링
# items.py
import scrapy
class JobPostItem(scrapy.Item):
job_title = scrapy.Field()
detail_url = scrapy.Field()
category = scrapy.Field()
vacancies = scrapy.Field()
workplace = scrapy.Field()
post_date = scrapy.Field()
# pipelines.py
import json
class JsonExportPipeline:
def open_spider(self, spider):
self.file_handler = open("job_results.json", "w")
def process_item(self, item, spider):
serialized = json.dumps(dict(item), ensure_ascii=False) + "\n"
self.file_handler.write(serialized)
return item
def close_spider(self, spider):
self.file_handler.close()
# tech_jobs_spider.py
import scrapy
from scrapy.spiders import CrawlSpider, Rule
from scrapy.linkextractors import LinkExtractor
from myproject.items import JobPostItem
class TechJobCrawler(CrawlSpider):
name = "tech_jobs"
allowed_domains = ["hr.techcorp.com"]
start_urls = ["http://hr.techcorp.com/positions.php?&start=0"]
page_extractor = LinkExtractor(allow=(r"start=\d+"))
rules = (
Rule(page_extractor, callback="extract_job_data", follow=True),
)
def extract_job_data(self, response):
for row in response.xpath("//tr[@class='even'] | //tr[@class='odd']"):
item = JobPostItem()
item['job_title'] = row.xpath("./td[1]/a/text()").get()
item['detail_url'] = row.xpath("./td[1]/a/@href").get()
item['category'] = row.xpath("./td[2]/text()").get()
item['vacancies'] = row.xpath("./td[3]/text()").get()
item['workplace'] = row.xpath("./td[4]/text()").get()
item['post_date'] = row.xpath("./td[5]/text()").get()
yield item
Scrapy 로깅 설정
settings.py 파일을 통해 크롤링 로그를 파일로 저장하고 레벨을 조정할 수 있습니다.
LOG_ENABLED = True
LOG_ENCODING = 'utf-8'
LOG_FILE = "scrapy_run.log"
LOG_LEVEL = "DEBUG"
LOG_STDOUT = False
로그 레벨은 다음 5단계로 구성됩니다:
- CRITICAL: 심각한 오류
- ERROR: 일반 오류
- WARNING: 경고 메시지
- INFO: 일반 정보 메시지
- DEBUG: 디버깅용 상세 메시지
실전 예제: 민원 게시판 크롤링 (CrawlSpider vs Spider)
민원 게시판(
http://wz.example.com/index.php/question/questionType?type=4&page=)의 모든 페이지에서 제목, 번호, 내용을 추출하는 두 가지 방식을 비교합니다.
데이터 모델 및 파이프라인
# items.py
class ComplaintItem(scrapy.Item):
subject = scrapy.Field()
ticket_id = scrapy.Field()
body_text = scrapy.Field()
source_link = scrapy.Field()
# pipelines.py
import codecs
import json
class UnicodeJsonPipeline:
def open_spider(self, spider):
self.file_stream = codecs.open("complaint_data.json", "w", encoding="utf-8")
def process_item(self, item, spider):
line = json.dumps(dict(item), ensure_ascii=False) + "\n"
self.file_stream.write(line)
return item
def close_spider(self, spider):
self.file_stream.close()
CrawlSpider 기반 구현
# civic_spider.py
import scrapy
from scrapy.linkextractors import LinkExtractor
from scrapy.spiders import CrawlSpider, Rule
from myproject.items import ComplaintItem
class CivicForumCrawler(CrawlSpider):
name = 'civic_crawl'
allowed_domains = ['wz.example.com']
start_urls = ['http://wz.example.com/index.php/question/type?type=4&page=']
list_page_extractor = LinkExtractor(allow=(r"type=4"))
detail_page_extractor = LinkExtractor(allow=(r"/html/question/\d+/\d+.shtml"))
rules = (
Rule(list_page_extractor, process_links="sanitize_links"),
Rule(detail_page_extractor, callback="parse_complaint")
)
def sanitize_links(self, extracted_links):
# 웹서버의 비정상 URL 파라미터 변형을 원래 형태로 치환
for link in extracted_links:
link.url = link.url.replace("?","&").replace("Type&","Type?")
return extracted_links
def parse_complaint(self, response):
item = ComplaintItem()
item['subject'] = response.xpath('//div[contains(@class, "pagecenter p3")]//strong/text()').get()
item['ticket_id'] = item['subject'].split(' ')[-1].split(":")[-1]
body = response.xpath('//div[@class="contentext"]/text()').getall()
if not body:
body = response.xpath('//div[@class="c1 text14_2"]/text()').getall()
item['body_text'] = "".join(body).strip()
item['source_link'] = response.url
yield item
Spider 기반 수동 탐색 구현
# manual_spider.py
import scrapy
from myproject.items import ComplaintItem
class ManualForumCrawler(scrapy.Spider):
name = 'manual_civic'
allowed_domains = ['wz.example.com']
base_url = 'http://wz.example.com/index.php/question/type?type=4&page='
current_page = 0
start_urls = [base_url + str(current_page)]
def parse(self, response):
post_links = response.xpath('//div[@class="greyframe"]/table//td/a[@class="news14"]/@href').getall()
for url in post_links:
yield scrapy.Request(url, callback=self.parse_complaint)
if self.current_page <= 71160:
self.current_page += 30
yield scrapy.Request(self.base_url + str(self.current_page), callback=self.parse)
def parse_complaint(self, response):
item = ComplaintItem()
item['subject'] = response.xpath('//div[contains(@class, "pagecenter p3")]//strong/text()').get()
item['ticket_id'] = item['subject'].split(' ')[-1].split(":")[-1]
body = response.xpath('//div[@class="contentext"]/text()').getall()
if not body:
body = response.xpath('//div[@class="c1 text14_2"]/text()').getall()
item['body_text'] = "".join(body).strip()
item['source_link'] = response.url
yield item