Django 크롤링 & 스크랩핑 하기
크롤링 & 스크랩핑 이란?
크롤링 (Web Crawling):
- 크롤링은 웹을 탐색하고 데이터를 수집하는 자동화된 프로세스다.
- 크롤러 또는 스파이더라고 불리는 프로그램이 웹 페이지를 순회하며 링크를 따라 이동하면서 정보를 수집한다.
주로 검색 엔진이 웹을 색인화하고 사용자에게 검색 결과를 제공하기 위해 크롤링을 사용한다 - 크롤러는 HTML을 분석하고 필요한 데이터를 추출하여 저장하거나 다른 시스템으로 전송한다.
스크래핑 (Web Scraping):
- 스크래핑은 웹 페이지에서 필요한 데이터를 추출하는 과정을 의미한다. 이는 주로 자동화된 방식으로 이루어진다.
- 웹 페이지의 HTML을 읽고 분석하여 특정 데이터를 추출하거나 원하는 형식으로 가공한다.
- 스크래핑은 크롤링 이후에 이루어지는 단계로, 크롤러가 수집한 데이터에서 필요한 정보를 추출하는 과정이다.
라이브러리 설명
파이썬에서 크롤링이라고 한다면 다들 떠올리는 것이 있다. 바로 `Selenium`이다. `Selenium`은 레거시이며 속도도 느리고 기능도 제한적이기 때문에 사용하기에 적합하지 않다. 하지만 오래된 라이브러리인 만큼 사용하는 곳에 따라 설루션이 많이 있을 수는 있다.
구세대가 아닌 차세대 크롤링 라이브러리인 `Puppeteer`, `Playwright`를 사용하는게 바람직하며 이는 크로미엄 기반 웹 브라우저에 대한 완전한 제어권과 원자성, 동시성 동기, 비동기를 전부 지원하고 공식 문서 또한 잘 정리되어 있어 손쉽게 사용할 수 있다. 즉 `race condition` 에 대해서 크게 고민할 필요 없이 사용해도 된다는 뜻이다.
기존 프로젝트에서는 `Next.js`로 테스트 작업을 했기에 `Node.js`라이브러리인 `Puppeteer`을 사용하였다.
이번 프로젝트에서는 `Django`로 테스트 작업을 하기 때문에 `Playwright`을 사용한다.
`Playwright` 는 전용 테스트 코드 작성 라이브러리 또한 제공되기 때문에 테스트 코드를 굉장히 쉽게 관리 작성할 수 있다!
https://playwright.dev/python/docs/intro
Installation | Playwright Python
Introduction
playwright.dev
크롤링 & 스크랩핑 구현
User-agent: *
Disallow: /admin
Disallow: /api
Disallow: /mypage
Disallow: /order
Allow: /
User-agent: bingbot
Crawl-delay: 10
User-agent: Googlebot
Allow: /
우선 작업을 하기 이전에 크롤링을 하고자하는 페이지에 /robots.txt을 `get`요청하여 정보를 받아오도록 한다.
무분별한 크롤링은 잘못하면 법적책임으로 까지 이어질 수 있기 때문에 조심해서 사용하도록한다.
이는 크롤링을 하기 위한 사이트에서 가져온 `robots.txt` 파일이다.
- `Disallow`에 관련된 페이지를 제외한 모든 페이지 경로에 대한 권한이 허용되어 크롤링이 가능하다.
- `Crawl-delay`은 각 크롤링 요청에 대한 지연시간이다 이는 초 단위로 구성된다. 현재는 `bingbot`에만 적용되어 있다.
이제 조건을 만족했으니 해당 사이트의 상품 목록을 크롤링해보도록 한다.
`models.py`
from os.path import splitext
from django.db import models
from django_lifecycle import LifecycleModelMixin, BEFORE_SAVE, hook
from theme.helper import make_thumb, uuid_name_upload_to
class Product(LifecycleModelMixin, models.Model):
brand = models.CharField(max_length=100)
thumb = models.ImageField(upload_to=uuid_name_upload_to)
name = models.CharField(max_length=100)
price = models.IntegerField()
sale_price = models.IntegerField()
created_at = models.DateTimeField(auto_now_add=True)
def __str__(self):
return f"{self.brand} - {self.thumb} - {self.name} - {self.price} - {self.sale_price} - {self.created_at}"
@hook(BEFORE_SAVE, when="thumb", has_changed=True)
def on_image_change(self):
if self.thumb:
image_width = self.thumb.width
image_extension = splitext(self.thumb.name)[-1].lower()
if image_width > 1024 or image_extension not in (".jpg", ".jpeg"):
thumb_file = make_thumb(self.thumb.file, 512, 512, 100)
self.thumb.save(thumb_file.name, thumb_file, save=False)
`Product` 모델을 사용하도록 한다. 현재는 테스트 작업을 위한 거 기 때문에 관계 테이블은 형성하지 않았다.
`product_crawling.py`
import itertools
import os
import urllib
from typing import Iterator
from urllib.parse import urlparse
from urllib.request import urlopen
import environ
from django.core.files.base import ContentFile
from django.core.management import BaseCommand
from playwright.sync_api import sync_playwright
from product.models import Product
class Command(BaseCommand):
help = "crawling test"
def handle(self, *args, **options):
env = environ.Env()
with sync_playwright() as p:
browser = p.chromium.launch()
page = browser.new_page()
page.goto(env.str("CRWAILNG_URL"))
product_list = page.query_selector_all(
"#best__product-list > li > article > div"
)
product_infos = (
Product(
thumb=self.convert_file(thumb_url),
brand=brand,
name=name,
sale_price=sale_price,
price=price,
)
for thumb_url, brand, name, sale_price, price in self.generate_product_details(
product_list
)
)
for chunks in self.get_chunks(product_infos, chunk_size=1000):
Product.objects.bulk_create(chunks, ignore_conflicts=True)
browser.close()
print("crawling finished")
@classmethod
def generate_product_details(cls, product_list):
for product in product_list:
thumb_url = product.query_selector(
"div.item__thumb > a > img"
).get_attribute("src")
if not thumb_url.startswith("https"):
thumb_url = "https:" + thumb_url
brand = product.query_selector("p.info__brand").inner_text()
name = product.query_selector("h3.info__title").inner_text()
sale_price = (
product.query_selector("p.info__price > span.ec-sale-rate")
.inner_text()
.rstrip("%")
if product.query_selector(
"p.info__price > span.ec-sale-rate"
).inner_text()
else "0"
)
price = (
product.query_selector("p.info__price > span:nth-of-type(2)")
.inner_text()
.replace(",", "")
.strip("₩")
)
yield thumb_url, brand, name, sale_price, price
@classmethod
def get_chunks(cls, iterable: Iterator, chunk_size: int = 100) -> Iterator:
iterator = iterable if hasattr(iterable, "__next__") else iter(iterable)
for first in iterator:
yield itertools.chain([first], itertools.islice(iterator, chunk_size - 1))
@classmethod
def convert_file(cls, url):
response = urllib.request.urlopen(url)
filename = os.path.basename(urlparse(url).path)
image_file = ContentFile(content=response.read(), name=filename)
return image_file
해당 포스팅에서는 제너레이터, Playwright에 대한 설명은 하지 않는다.
https://bingual76.tistory.com/299
에서 다루었던 `get_chunks` 메서드를 사용한다.
`generate_product_details` 메서드를 사용해서 스크랩 핑한 데이터를 `yield`한다.
Django ImageField는 `URL` 이미지 파일을 다운로드하는 기능을 기본적으로 지원하지 않는다.
`convert_file` 메서드를 이용해서 파일을 읽어 들이고 임시 파일을 만들어 저장한 뒤 `bulk_create` 메서드를 통해서 데이터 베이스에 일괄 적용한다.
DJANGO_ALLOW_ASYNC_UNSAFE = env.bool("DJANGO_ALLOW_ASYNC_UNSAFE", default=False)
장고에서는 기본적으로 ORM에 대한 비동기적인 동작을 지원하지 않기 때문에 `settings.py`에서 해당 작업을 허용해줘야 한다. 또는 스크랩핑 데이터를 CSV파일로 만들어 다운로드 한뒤 데이터 베이스에 삽입하는 방법이 있겠다.
해당 커스텀 커맨드를 실행하게 되면 아래와 같이 제대로 데이터를 데이터베이스에 삽입하는 것을 확인할 수 있다.
관련커밋
prodcut 모델 생성: 소스코드
crawling 커스텀 커맨드 생성: 소스코드
### ↑ 위의 코드들은 레거시 기준으로 작성됨 ###
Auto wait locator을 사용한 커밋: 소스코드