Scrapy + Playwright 매뉴얼 - 기본부터 실전까지

Crawling

Scrapy + Playwright 매뉴얼 - 기본부터 실전까지

bluebamus 2025. 3. 2. 00:51

1. 설치 및 기본 설정

1.1. 의존성 설치

- 첫 번째 명령어는 scrapy-playwright 패키지를 설치하고, 두 번째 명령어는 필요한 브라우저(Chromium, Firefox, WebKit)를 설치한다.

pip install scrapy-playwright
playwright install

1.2. Scrapy 설정 파일 (settings.py) 수정

- Scrapy와 Playwright의 통합을 위해 settings.py에 다음 설정을 추가한다.

# settings.py

DOWNLOADER_MIDDLEWARES = {
    'scrapy_playwright.middleware.PlaywrightMiddleware': 543,
}

# Playwright 미들웨어 활성화
DOWNLOAD_HANDLERS = {
    "http": "scrapy_playwright.handler.PlaywrightDownloadHandler",
    "https": "scrapy_playwright.handler.PlaywrightDownloadHandler",
}

# 사용할 브라우저 종류 (chromium, firefox, webkit 중 선택)
PLAYWRIGHT_BROWSER_TYPE = "chromium"

# Playwright 실행 옵션 설정 (headless 모드, sandbox 옵션 등)
PLAYWRIGHT_LAUNCH_OPTIONS = {
    "headless": True,
    "args": ["--no-sandbox"],
}

# 기본 브라우저 타입 설정 (chromium, firefox, webkit 중 선택)
PLAYWRIGHT_BROWSER_TYPE = "chromium"

# 페이지 로딩이나 요소 대기 시간 등의 기본 타임아웃 설정 (필요 시 추가)
PLAYWRIGHT_DEFAULT_NAVIGATION_TIMEOUT = 30000  # 밀리초 단위
PLAYWRIGHT_DEFAULT_PAGE_WAIT_FOR_TIMEOUT = 10000

1.3. 고급 설정 옵션

- 브라우저 및 컨텍스트 설정
- settings.py에서 다양한 브라우저 옵션을 설정할 수 있다.

# 브라우저 시작 옵션
PLAYWRIGHT_LAUNCH_OPTIONS = {
    "headless": False,  # 헤드리스 모드 비활성화
    "slowMo": 50,  # 동작 속도 조절 (밀리초)
    "proxy": {  # 프록시 설정
        "server": "http://proxy.example.com:8080",
        "username": "username",
        "password": "password"
    },
    "args": ["--disable-gpu"],  # 추가 브라우저 인수
}

# 브라우저 컨텍스트 설정
PLAYWRIGHT_BROWSER_CONTEXT_ARGS = {
    "viewport": {"width": 1920, "height": 1080},  # 뷰포트 크기
    "deviceScaleFactor": 1.0,  # 장치 스케일 팩터
    "userAgent": "Custom User Agent",  # 사용자 에이전트 설정
    "locale": "ko-KR",  # 로케일 설정
    "geolocation": {"latitude": 37.5665, "longitude": 126.9780},  # 지오로케이션 설정
    "permissions": ["geolocation"],  # 권한 설정
}

# 기본 타임아웃 설정
PLAYWRIGHT_DEFAULT_NAVIGATION_TIMEOUT = 60000  # 60초

1.4. 로깅 설정

# 로깅 레벨 설정
LOG_LEVEL = "DEBUG"

# Playwright 관련 로그만 필터링
PLAYWRIGHT_LOGGER = "playwright"

1.5. 병렬 처리

# settings.py에서 동시 요청 제한 설정
CONCURRENT_REQUESTS = 4
PLAYWRIGHT_CONTEXTS_LIMIT = 4  # 동시에 유지할 컨텍스트 수 제한

2. 기본 사용법

- Scrapy Spider 내에서 Playwright를 사용하려면 요청 시 meta 옵션에 "playwright": True를 설정해야 한다.

- 필요한 경우 페이지 객체를 직접 조작할 수 있도록 "playwright_include_page": True도 함께 지정한다.

2.1. 기본 요청 예시

import scrapy

class ExampleSpider(scrapy.Spider):
    name = "example"
    start_urls = ["https://example.com"]

    def start_requests(self):
        for url in self.start_urls:
            yield scrapy.Request(
                url=url,
                meta={
                    "playwright": True,  # Playwright 사용 활성화
                    "playwright_include_page": True,  # 페이지 객체를 meta에 포함 (필요 시)
                },
                callback=self.parse
            )

    def parse(self, response):
        # Playwright를 통해 렌더링된 페이지에서 데이터 추출
        title = response.xpath('//title/text()').get()
        yield {"title": title}

        # 만약 playwright_include_page를 사용했다면 페이지 객체를 반드시 닫아줘야 함
        if "playwright_page" in response.meta:
            response.meta["playwright_page"].close()

2.2. Playwright 페이지 메서드 사용

- 추가적으로 페이지 상에서 특정 동작(예: 특정 요소 대기, 스크롤, 클릭 등)을 실행할 수 있다.

- 이를 위해 meta에 "playwright_page_methods" 옵션을 지정한다.

yield scrapy.Request(
    url="https://example.com",
    meta={
        "playwright": True,
        "playwright_page_methods": [
            {"method": "wait_for_selector", "selector": "div.content"}
        ],
    },
    callback=self.parse
)

- 위 예시는 요청한 페이지에서 div.content 요소가 나타날 때까지 대기하도록 한다.

3. 설정 옵션 상세 정리

3.1. PLAYWRIGHT_BROWSER_TYPE

- 설명:

- Playwright에서 사용할 브라우저 종류를 지정한다.
- 사용 가능 값:

- "chromium", "firefox", "webkit"
- 기본값:

- 일반적으로 "chromium"이 사용된다.

3.2. PLAYWRIGHT_LAUNCH_OPTIONS

- 설명:

- 브라우저 실행 시 전달할 옵션들을 지정한다.

- 예시 옵션:

- headless:

- 브라우저를 헤드리스 모드(화면 없이 백그라운드 실행)로 실행할지 여부
- args:

- 추가적인 커맨드라인 인자 (예: "--no-sandbox")

- 활용:

- 성능 최적화나 CI/CD 환경에서의 안정성 확보를 위해 활용된다.

3.3. PLAYWRIGHT_DEFAULT_NAVIGATION_TIMEOUT 및 PAGE_WAIT_FOR_TIMEOUT

- 설명:

- 페이지 로딩 및 요소 대기 시간에 대한 타임아웃을 설정한다.

- 단위:

- 밀리초(ms) 단위로 설정하며, 페이지 로딩이 오래 걸리는 사이트에 대해 유연하게 대처할 수 있다.

3.4. PLAYWRIGHT_CONTEXT_ARGS

- 설명:

- 브라우저 컨텍스트 생성 시 추가 옵션(예: 사용자 에이전트, 뷰포트 크기 등)을 지정할 수 있다.

- 예시:

PLAYWRIGHT_CONTEXT_ARGS = {
    "user_agent": "Mozilla/5.0 (compatible; CustomBot/1.0)",
    "viewport": {"width": 1280, "height": 800},
}

4. 일반적으로 사용되는 Playwright 활용 기법

4.1. 동적 페이지 처리

- 대기 메서드:

- wait_for_selector, wait_for_load_state 등을 활용해 동적 요소가 로드될 때까지 기다린다.
- 예시:

- AJAX로 로딩되는 콘텐츠를 대기 후 추출

4.2. 페이지 조작 및 상호작용

- 메서드 사용:

- page.click, page.fill, page.press 등을 사용하여 버튼 클릭, 입력 필드 채우기, 키 입력 등의 동작을 구현할 수 있다.
- 활용:

- 로그인 폼 제출, 필터 적용 등 사용자 인터랙션이 필요한 상황

4.3. 네트워크 요청 가로채기

- route 메서드:

- 특정 리소스(예: 광고, 추적 스크립트)를 차단하거나 수정하여 페이지 로딩 속도를 개선하거나 데이터 수집에 집중할 수 있다.

4.4. 스크린샷 및 PDF 저장

- 메서드:

- page.screenshot, page.pdf를 이용하여 페이지의 스냅샷이나 PDF 파일로 저장할 수 있다.

- 활용:

- 크롤링 결과의 시각적 증거 확보나 보고서 작성 시 유용

5. 일반적인 사용 패턴

스크린샷 캡쳐

def parse_with_page(self, response):
    page = response.meta["playwright_page"]
    
    # 스크린샷 저장
    screenshot_path = f"screenshots/{response.url.split('/')[-1]}.png"
    await page.screenshot(path=screenshot_path, full_page=True)
    
    await page.close()
    yield {"screenshot": screenshot_path}

6. 폼 입력 및 제출

from scrapy_playwright.page import PageMethod

def start_requests(self):
    yield scrapy.Request(
        "https://example.com/login",
        meta={
            "playwright": True,
            "playwright_include_page": True,
            "playwright_page_methods": [
                PageMethod("wait_for_selector", "#login-form"),
                # 폼 채우기
                PageMethod("fill", "#username", "my_username"),
                PageMethod("fill", "#password", "my_password"),
                # 폼 제출
                PageMethod("click", "#submit-button"),
                # 로그인 후 페이지 로드 기다리기
                PageMethod("wait_for_navigation"),
            ]
        },
        callback=self.parse_after_login
    )

7. 무한 스크롤 처리

def parse_infinite_scroll(self, response):
    page = response.meta["playwright_page"]
    
    # 초기 데이터 추출
    items = response.css("div.item::text").getall()
    
    # 더 많은 콘텐츠를 로드하기 위해 스크롤
    for _ in range(5):  # 5번 스크롤 다운
        await page.evaluate("window.scrollTo(0, document.body.scrollHeight)")
        # 새 콘텐츠 로드 대기
        await page.wait_for_timeout(1000)  # 1초 대기
    
    # 최종 데이터 추출 (스크롤 후)
    final_items = await page.query_selector_all("div.item")
    for item in final_items:
        text = await item.text_content()
        items.append(text)
    
    await page.close()
    yield {"items": items}

8. 동적 콘텐츠 기다리기 간단 예시

def start_requests(self):
    yield scrapy.Request(
        "https://example.com/dynamic-content",
        meta={
            "playwright": True,
            "playwright_include_page": True,
            "playwright_page_methods": [
                # 네트워크 요청이 완료될 때까지 대기
                PageMethod("wait_for_load_state", "networkidle"),
                # JavaScript 실행이 완료될 때까지 대기
                PageMethod("wait_for_load_state", "domcontentloaded"),
                # 특정 AJAX 요청이 완료될 때까지 대기
                PageMethod("wait_for_response", "https://api.example.com/data"),
            ]
        },
        callback=self.parse_dynamic_content
    )

9. 쿠키 관리

# settings.py에서 쿠키 관리 설정
COOKIES_ENABLED = True
COOKIES_DEBUG = True

# 스파이더에서 쿠키 설정
def start_requests(self):
    yield scrapy.Request(
        "https://example.com",
        meta={
            "playwright": True,
            "playwright_include_page": True,
            "playwright_page_methods": [
                # 쿠키 설정
                PageMethod("add_cookies", [
                    {"name": "session_id", "value": "abc123", "domain": "example.com"}
                ]),
            ]
        },
        callback=self.parse_with_cookies
    )

def parse_with_cookies(self, response):
    page = response.meta["playwright_page"]
    # 현재 쿠키 가져오기
    cookies = await page.context.cookies()
    yield {"cookies": cookies}
    await page.close()

10. 실무 기반 프로젝트 예제

10.1. 뉴스 사이트 동적 크롤링

- 뉴스 사이트는 동적으로 콘텐츠를 로드하는 경우가 많다. 아래 예제는 뉴스 기사 리스트를 추출하는 Spider이다.

import scrapy

class NewsSpider(scrapy.Spider):
    name = "news"
    start_urls = ["https://news.example.com"]

    def start_requests(self):
        for url in self.start_urls:
            yield scrapy.Request(
                url=url,
                meta={
                    "playwright": True,
                    # 뉴스 기사 컨테이너(div.news-container)가 로드될 때까지 대기
                    "playwright_page_methods": [
                        {"method": "wait_for_selector", "selector": "div.news-container"}
                    ],
                },
                callback=self.parse
            )

    def parse(self, response):
        # XPath를 통해 뉴스 기사들을 추출
        articles = response.xpath('//div[@class="article"]')
        for article in articles:
            title = article.xpath('.//h2/text()').get()
            link = article.xpath('.//a/@href').get()
            yield {
                "title": title,
                "link": response.urljoin(link)
            }

- 동작 설명:

- meta 옵션:

- "playwright":

- True로 Playwright를 활성화하며, "playwright_page_methods"를 통해 특정 요소(div.news-container)가 나타날 때까지 대기한다.

- 데이터 추출:

- XPath를 사용하여 뉴스 기사의 제목과 링크를 추출한다.

- 링크 처리:

- 상대 경로를 절대 경로로 변환하기 위해 response.urljoin을 사용한다.

10.2. 제품 리뷰 웹사이트 스크래핑 (무한 스크롤 처리)

- 제품 리뷰 사이트는 무한 스크롤 방식으로 데이터를 로드하는 경우가 많다. 아래 예제는 스크롤을 통해 리뷰 데이터를 추출하는 Spider이다.

import scrapy

class ReviewSpider(scrapy.Spider):
    name = "reviews"
    start_urls = ["https://reviews.example.com/products"]

    def start_requests(self):
        for url in self.start_urls:
            yield scrapy.Request(
                url=url,
                meta={
                    "playwright": True,
                    # playwright_page 객체를 meta에 포함하여 직접 제어 (예: 스크롤 후 추가 로딩)
                    "playwright_include_page": True,
                    "playwright_page_methods": [
                        # 페이지의 가장 아래로 스크롤하여 추가 콘텐츠 로드
                        {"method": "evaluate", "script": "window.scrollBy(0, document.body.scrollHeight)"},
                        # 스크롤 후 3초 대기 (네트워크 요청 및 콘텐츠 로딩 시간 확보)
                        {"method": "wait_for_timeout", "timeout": 3000}
                    ],
                },
                callback=self.parse
            )

    def parse(self, response):
        # playwright_include_page 옵션으로 페이지 객체를 meta에 포함함
        page = response.meta["playwright_page"]

        # XPath로 리뷰 섹션 추출
        reviews = response.xpath('//div[@class="review"]')
        for review in reviews:
            reviewer = review.xpath('.//span[@class="name"]/text()').get()
            content = review.xpath('.//p[@class="content"]/text()').get()
            yield {
                "reviewer": reviewer,
                "content": content
            }
        # 메모리 누수를 방지하기 위해 페이지 객체를 닫음
        page.close()

- 동작 및 설명:

- 무한 스크롤 구현:

- "playwright_page_methods"를 통해 페이지 내에서 JavaScript 코드를 실행하여 스크롤 동작을 수행한다.

- evaluate:

- JavaScript를 실행해 페이지를 아래로 스크롤한다.

- wait_for_timeout:

- 스크롤 후 콘텐츠 로딩을 위해 3초 대기한다.

- 데이터 추출:

- 리뷰어 이름과 리뷰 내용을 XPath로 추출한다.

- 리소스 정리:

- Playwright 페이지 객체를 사용한 후 반드시 page.close()를 호출해 브라우저 리소스를 해제한다.

10.3. 대기 조건 설정하기

- 페이지가 완전히 로드될 때까지 기다리기 위한 대기 조건을 설정할 수 있다.

yield scrapy.Request(
    "https://example.com",
    meta={
        "playwright": True,
        "playwright_include_page": True,  # page 객체 포함
        "playwright_page_methods": [
            PageMethod("wait_for_selector", "div.content"),  # 특정 선택자가 나타날 때까지 대기
        ]
    },
    callback=self.parse_with_page
)

def parse_with_page(self, response):
    page = response.meta["playwright_page"]
    # page 객체를 사용하여 추가 작업 수행
    title = page.title()
    
    # 작업 완료 후 페이지 닫기
    await page.close()
    
    yield {"title": title}

10.4. 이커머스 사이트 동적 상품 정보 크롤러

# ecommerce_crawler/spiders/product_spider.py

import scrapy
import json
from scrapy_playwright.page import PageMethod
import logging
from datetime import datetime

class ProductSpider(scrapy.Spider):
    name = "product_spider"
    
    # 크롤링할 카테고리 및 페이지 수 설정
    custom_settings = {
        'FEEDS': {
            f'data/products_{datetime.now().strftime("%Y%m%d_%H%M%S")}.json': {
                'format': 'json',
                'encoding': 'utf8',
                'indent': 4,
            },
        },
        'PLAYWRIGHT_DEFAULT_NAVIGATION_TIMEOUT': 60000,  # 60초로 타임아웃 설정
        'DOWNLOAD_DELAY': 2,  # 요청 간 딜레이 2초
    }
    
    def start_requests(self):
        # 여러 카테고리 페이지 크롤링
        categories = [
            {"id": "electronics", "pages": 3},
            {"id": "clothing", "pages": 2},
        ]
        
        for category in categories:
            for page in range(1, category["pages"] + 1):
                # 카테고리별 URL 생성
                url = f"https://example-ecommerce.com/category/{category['id']}?page={page}"
                
                self.logger.info(f"크롤링 시작: {url}")
                
                yield scrapy.Request(
                    url=url,
                    callback=self.parse_category_page,
                    meta={
                        "playwright": True,  # Playwright 활성화
                        "playwright_include_page": True,  # page 객체 포함
                        "playwright_page_methods": [
                            # 상품 목록이 로드될 때까지 대기
                            PageMethod("wait_for_selector", ".product-grid", timeout=10000),
                            # 페이지가 완전히 로드될 때까지 대기
                            PageMethod("wait_for_load_state", "networkidle"),
                            # 필터 메뉴가 펼쳐져 있다면 닫기 (선택적)
                            PageMethod("evaluate", """
                                () => {
                                    const filterBtn = document.querySelector('.filter-collapse-btn');
                                    if (filterBtn && filterBtn.getAttribute('aria-expanded') === 'true') {
                                        filterBtn.click();
                                    }
                                }
                            """),
                        ],
                        "category_id": category["id"],
                        "page_num": page,
                    }
                )
    
    async def parse_category_page(self, response):
        """
        카테고리 페이지에서 개별 상품 URL을 추출하고 각 상품 페이지로 요청을 생성합니다.
        """
        page = response.meta["playwright_page"]
        category_id = response.meta["category_id"]
        page_num = response.meta["page_num"]
        
        # 스크린샷 저장 (디버깅 용도)
        await page.screenshot(path=f"screenshots/{category_id}_page{page_num}.png", full_page=True)
        
        try:
            # 상품 목록에서 각 상품의 URL 추출
            product_links = response.css(".product-item a.product-link::attr(href)").getall()
            
            self.logger.info(f"카테고리 {category_id}, 페이지 {page_num}에서 {len(product_links)}개 상품 URL 추출됨")
            
            # 페이지 닫기 (메모리 관리를 위해 중요)
            await page.close()
            
            # 각 상품 페이지에 대한 요청 생성
            for product_url in product_links:
                # 상대 URL을 절대 URL로 변환
                if product_url.startswith('/'):
                    product_url = f"https://example-ecommerce.com{product_url}"
                
                yield scrapy.Request(
                    url=product_url,
                    callback=self.parse_product_page,
                    meta={
                        "playwright": True,
                        "playwright_include_page": True,
                        "playwright_page_methods": [
                            # 상품 상세 정보가 로드될 때까지 대기
                            PageMethod("wait_for_selector", ".product-details", timeout=10000),
                            # 동적으로 로드되는 가격 정보 대기
                            PageMethod("wait_for_selector", ".price-info", timeout=10000),
                            # AJAX 로드 완료 대기
                            PageMethod("wait_for_load_state", "networkidle"),
                        ],
                        "category_id": category_id,
                    }
                )
        except Exception as e:
            self.logger.error(f"카테고리 페이지 파싱 중 오류 발생: {str(e)}")
            await page.close()
    
    async def parse_product_page(self, response):
        """
        개별 상품 페이지에서 상세 정보를 추출합니다.
        동적으로 로드되는 재고, 리뷰, 가격 정보 등을 처리합니다.
        """
        page = response.meta["playwright_page"]
        category_id = response.meta["category_id"]
        
        try:
            # 기본 상품 정보 추출
            product_id = response.css("meta[property='product:id']::attr(content)").get()
            product_name = response.css("h1.product-title::text").get().strip()
            
            # 가격 정보 추출 (종종 동적으로 로드됨)
            price_info = await page.evaluate("""
                () => {
                    const priceElement = document.querySelector('.price-info .current-price');
                    return priceElement ? priceElement.innerText.trim() : null;
                }
            """)
            
            # 상품 이미지 URL 추출
            image_urls = response.css(".product-gallery img::attr(src)").getall()
            
            # 설명 정보 추출
            description = response.css(".product-description::text").get()
            if description:
                description = description.strip()
            
            # 재고 정보 (동적으로 로드되는 경우 JavaScript로 추출)
            stock_status = await page.evaluate("""
                () => {
                    const stockElement = document.querySelector('.stock-info');
                    return stockElement ? stockElement.innerText.trim() : "재고 정보 없음";
                }
            """)
            
            # 사양 정보 추출
            specifications = {}
            spec_rows = response.css(".product-specifications tr")
            for row in spec_rows:
                spec_name = row.css("th::text").get()
                spec_value = row.css("td::text").get()
                if spec_name and spec_value:
                    specifications[spec_name.strip()] = spec_value.strip()
            
            # 리뷰 정보 로드 (버튼 클릭을 통해 동적으로 로드되는 경우)
            review_count = 0
            average_rating = 0
            
            # 리뷰 탭 존재 여부 확인
            review_tab_exists = await page.query_selector(".reviews-tab")
            if review_tab_exists:
                # 리뷰 탭 클릭
                await page.click(".reviews-tab")
                await page.wait_for_selector(".review-list", timeout=5000)
                
                # 리뷰 수와 평균 평점 추출
                review_data = await page.evaluate("""
                    () => {
                        const reviewCountElement = document.querySelector('.review-count');
                        const ratingElement = document.querySelector('.average-rating');
                        return {
                            count: reviewCountElement ? parseInt(reviewCountElement.innerText.match(/\\d+/)[0]) : 0,
                            rating: ratingElement ? parseFloat(ratingElement.innerText) : 0
                        };
                    }
                """)
                
                review_count = review_data.get("count", 0)
                average_rating = review_data.get("rating", 0)
            
            # 최근 본 상품에 추가 (사용자 세션 추적에 유용)
            await page.evaluate(f"""
                () => {{
                    const recentlyViewed = JSON.parse(localStorage.getItem('recentlyViewed') || '[]');
                    if (!recentlyViewed.includes("{product_id}")) {{
                        recentlyViewed.unshift("{product_id}");
                        if (recentlyViewed.length > 10) recentlyViewed.pop();
                        localStorage.setItem('recentlyViewed', JSON.stringify(recentlyViewed));
                    }}
                }}
            """)
            
            # 스크린샷 저장
            await page.screenshot(path=f"screenshots/product_{product_id}.png")
            
            # 데이터 수집 완료, 페이지 닫기
            await page.close()
            
            # 결과 반환
            yield {
                "product_id": product_id,
                "product_name": product_name,
                "category_id": category_id,
                "price": price_info,
                "stock_status": stock_status,
                "description": description,
                "specifications": specifications,
                "image_urls": image_urls,
                "review_count": review_count,
                "average_rating": average_rating,
                "url": response.url,
                "crawled_at": datetime.now().isoformat(),
            }
            
        except Exception as e:
            self.logger.error(f"상품 페이지 파싱 중 오류 발생 ({response.url}): {str(e)}")
            # 오류 발생 시에도 페이지 닫기
            await page.close()


# ecommerce_crawler/settings.py

BOT_NAME = 'ecommerce_crawler'

SPIDER_MODULES = ['ecommerce_crawler.spiders']
NEWSPIDER_MODULE = 'ecommerce_crawler.spiders'

# Playwright 설정
DOWNLOAD_HANDLERS = {
    "http": "scrapy_playwright.handler.PlaywrightDownloadHandler",
    "https": "scrapy_playwright.handler.PlaywrightDownloadHandler",
}

DOWNLOADER_MIDDLEWARES = {
    'scrapy_playwright.middleware.PlaywrightMiddleware': 543,
}

# 브라우저 타입 설정
PLAYWRIGHT_BROWSER_TYPE = "chromium"

# 브라우저 시작 옵션
PLAYWRIGHT_LAUNCH_OPTIONS = {
    "headless": True,  # 헤드리스 모드 활성화
    "timeout": 30000,  # 브라우저 시작 타임아웃
    "args": [
        "--disable-gpu",
        "--no-sandbox",
        "--disable-dev-shm-usage",
        "--disable-setuid-sandbox",
    ],
}

# 브라우저 컨텍스트 설정
PLAYWRIGHT_BROWSER_CONTEXT_ARGS = {
    "viewport": {"width": 1366, "height": 768},
    "user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36",
    "locale": "ko-KR",
}

# 동시 요청 제한
CONCURRENT_REQUESTS = 4
PLAYWRIGHT_CONTEXTS_LIMIT = 4

# 재시도 설정
RETRY_ENABLED = True
RETRY_TIMES = 3
RETRY_HTTP_CODES = [500, 502, 503, 504, 522, 524, 408, 429]

# 저장 경로 설정
FEEDS_TEMPDIR = 'tmp/'

# 캐시 설정
HTTPCACHE_ENABLED = True
HTTPCACHE_EXPIRATION_SECS = 86400  # 24시간
HTTPCACHE_DIR = 'httpcache'

# 로깅 설정
LOG_LEVEL = 'INFO'
LOG_FILE = 'logs/ecommerce_crawler.log'

# 아이템 파이프라인 설정
ITEM_PIPELINES = {
    'ecommerce_crawler.pipelines.PriceCleaningPipeline': 300,
    'ecommerce_crawler.pipelines.ImageDownloadPipeline': 400,
}


# ecommerce_crawler/pipelines.py

import re
from scrapy.pipelines.images import ImagesPipeline
from scrapy import Request
import os

class PriceCleaningPipeline:
    """가격 데이터 정제 파이프라인"""
    
    def process_item(self, item, spider):
        # 가격 데이터 정제
        if 'price' in item and item['price']:
            # 쉼표 제거 및 숫자만 추출
            price_str = item['price']
            price_num = re.sub(r'[^\d.]', '', price_str)
            
            try:
                item['price_value'] = float(price_num)
                item['price_currency'] = self._extract_currency(price_str)
            except ValueError:
                spider.logger.warning(f"가격 변환 실패: {price_str}")
                item['price_value'] = None
        
        return item
    
    def _extract_currency(self, price_str):
        """가격 문자열에서 통화 기호 추출"""
        currency_map = {
            '₩': 'KRW',
            '원': 'KRW',
            '$': 'USD',
            '€': 'EUR',
            '£': 'GBP',
        }
        
        for symbol, code in currency_map.items():
            if symbol in price_str:
                return code
        
        return 'UNKNOWN'


class ImageDownloadPipeline(ImagesPipeline):
    """상품 이미지 다운로드 파이프라인"""
    
    def get_media_requests(self, item, info):
        if 'image_urls' in item and item['image_urls']:
            for img_url in item['image_urls']:
                yield Request(
                    url=img_url,
                    meta={'product_id': item.get('product_id', 'unknown')}
                )
    
    def file_path(self, request, response=None, info=None, *, item=None):
        product_id = request.meta.get('product_id', 'unknown')
        image_name = request.url.split('/')[-1]
        return f'product_images/{product_id}/{image_name}'
    
    def item_completed(self, results, item, info):
        if results:
            # 다운로드된 이미지 경로 저장
            image_paths = [x['path'] for ok, x in results if ok]
            if image_paths:
                item['image_paths'] = image_paths
        
        return item


# 실행 명령:
# scrapy crawl product_spider

10.5. 소셜 미디어 모니터링 크롤러

# social_media_monitor/spiders/twitter_spider.py

import scrapy
import json
import re
from scrapy_playwright.page import PageMethod
from datetime import datetime, timedelta
import logging
import csv
import os
from urllib.parse import quote

class TwitterSpider(scrapy.Spider):
    name = "twitter_monitor"
    
    # 크롤링 대상 키워드 및 설정
    custom_settings = {
        'DOWNLOAD_DELAY': 3,  # 요청 간 딜레이 3초
        'FEEDS': {
            f'data/twitter_posts_{datetime.now().strftime("%Y%m%d_%H%M%S")}.json': {
                'format': 'json',
                'encoding': 'utf8',
                'indent': 4,
            },
        },
        'PLAYWRIGHT_DEFAULT_NAVIGATION_TIMEOUT': 90000,  # 90초로 타임아웃 설정
    }
    
    def __init__(self, keywords=None, days_back=3, max_scrolls=10, *args, **kwargs):
        """
        초기화 함수로 크롤링 매개변수를 설정합니다.
        
        Args:
            keywords (str): 쉼표로 구분된 검색 키워드 목록
            days_back (int): 오늘부터 몇 일 전까지의 게시물을 크롤링할지 지정
            max_scrolls (int): 페이지당 최대 스크롤 횟수
        """
        super(TwitterSpider, self).__init__(*args, **kwargs)
        
        # 키워드 설정 (기본값 제공)
        self.keywords = keywords.split(',') if keywords else ["Python", "데이터분석", "인공지능"]
        
        # 날짜 범위 계산
        self.days_back = int(days_back)
        self.end_date = datetime.now()
        self.start_date = self.end_date - timedelta(days=self.days_back)
        
        # 스크롤 제한 설정
        self.max_scrolls = int(max_scrolls)
        
        # 결과 저장 디렉토리 생성
        os.makedirs('data', exist_ok=True)
        os.makedirs('screenshots', exist_ok=True)
        
        # CSV 요약 파일 초기화
        self.csv_path = f'data/twitter_summary_{datetime.now().strftime("%Y%m%d_%H%M%S")}.csv'
        with open(self.csv_path, 'w', newline='', encoding='utf-8') as f:
            writer = csv.writer(f)
            writer.writerow(['키워드', '총 게시물 수', '총 좋아요 수', '총 리트윗 수', '총 댓글 수', '평균 참여율'])
    
    def start_requests(self):
        """
        각 키워드에 대한 검색 URL을 생성하고 요청합니다.
        """
        # 날짜 형식 변환
        start_date_str = self.start_date.strftime("%Y-%m-%d")
        end_date_str = self.end_date.strftime("%Y-%m-%d")
        
        self.logger.info(f"크롤링 시작: 키워드 {self.keywords}, 기간 {start_date_str} ~ {end_date_str}")
        
        for keyword in self.keywords:
            # 검색 URL 인코딩 (고급 검색 쿼리 포함)
            encoded_query = quote(f"{keyword} since:{start_date_str} until:{end_date_str}")
            search_url = f"https://twitter.com/search?q={encoded_query}&src=typed_query&f=live"
            
            self.logger.info(f"키워드 '{keyword}' 검색 시작: {search_url}")
            
            yield scrapy.Request(
                url=search_url