[Playwright/Python] 비동기 처리가 가능한 웹 스크래핑 라이브러리, Playwright

프로그래머스 데이터 엔지니어링 데브코스 프로젝트 중 동적 웹 페이지에서 브라우저 조작 및 스크래핑으로 데이터를 수집해야하는 작업이 있었다. 처음에는 기존에 사용하던 Selenium을 사용했지만, 수집할 데이터가 다수였던 만큼 시간이 너무 많이 소요된다는 단점이 있었다.

 

시간 비용 소모 문제를 해결하기 위해 Selenium을 사용하는 함수를 비동기 프로그래밍 처리하려고 했지만, Selenium은 기본적으로 비동기 처리를 지원하지 않기 때문에 다소 복잡한 과정을 거쳐야 한다.(driver를 가져오는 함수는 따로 작성하고 sync to async를 통해 처리해줘야 한다던지... 등등)

 

Selenium의 비동기 처리 문제를 해결하던 중, 우연히 검색을 통해 비동기 프로그래밍을 지원하는 브라우저 제어 라이브러리가 있다는 것을 알게되었다. 바로 Microsoft에서 개발한 Playwright다. Selenium과 거의 동일한 기능(더 좋은 기능도 많다.)을 가지고 있고, 여러 가지 브라우저(Chrome, firefox 등)를 지원한다. 실제로 웹 페이지 개발 뒤 QA 과정에서도 많이 사용되는 툴이라고 한다.

 

Playwright의 대표적인 기능

• 다중 브라우저 지원: Chromium, Firefox, WebKit 등 다양한 브라우저에서 테스트 가능
• 자동 대기: 요소가 준비될 때까지 자동으로 기다리는 기능을 제공
• 네트워크 인터셉션: 요청을 수정하거나 모의 응답 제공 가능
• 모바일 장치 에뮬레이션: 모바일 환경 시뮬레이션 가능
• 강력한 셀렉터 엔진: CSS, XPath 외에도 텍스트 내용, 레이블 등으로 요소 선택 가능
• 헤드리스 및 헤드풀 모드 지원: 다양한 환경에서 실행 가능(Linux 등)
• Codegen 기능: 사용자의 브라우저 조작을 기록하여 자동으로 테스트 코드 생성
• Tracing 및 Debugging: Tracing 기능을 제공하여 디버깅 시 유용한 정보를 제공 
  • Trace Viewer를 통해 기록된 트레이스를 시각적으로 분석 가능

 

 

Playwright 설치

Python에서는 pip 명령어를 통해 Playwright를 설치할 수 있다.

pip install playwright

 

설치 이후, 브라우저 바이너리를 추가로 설치해 주어야 라이브러리를 사용할 수 있다.

playwright install
# or
python -m playwright install

 

Playwright 기본예제

아래는 Playwright를 통해 웹 페이지의 title을 가져오는 예제이다. (동기 함수)

from playwright.sync_api import sync_playwright

with sync_playwright() as p:
    browser = p.chromium.launch(headless=False) # 크롬 브라우저 설치
    page = browser.new_page() # 새 페이지 생성
    page.goto('https://www.google.com') # 해당 링크로 접속
    print(page.title()) # title 출력
    browser.close()

 

Playwright로 비동기 처리하기

browser에서 .new_context() 메서드를 통해 독립적인 세션을 만들어준다. (크롬의 탭 같은 기능인 듯하다.) asyncio와 await 명령어를 통해 함수를 비동기 처리해주면 된다.

# 예제

import asyncio
from playwright.async_api import async_playwright

async def run(playwright):
    browser = await playwright.chromium.launch(headless=True)
    context = await browser.new_context() # 새로운 컨텍스트 생성(독립적인 세션 운영을 위함)
    page = await context.new_page()
    await page.goto('https://example.com')
    title = await page.title()
    print(title)
    await browser.close()

async def main():
    async with async_playwright() as playwright:
        await run(playwright)

asyncio.run(main())

 

이 함수를 병렬적으로 한번에 처리하려면 asyncio.gather() 메서드를 이용하면 된다. 프로젝트에서는 아래와 같이 구현했다. (접은 글 또는 Github 링크 참고 / https://github.com/yjbenkang/denpil (updater.py 파일))

async def get_purchase_data(url):
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=True)
        page = await browser.new_page()
        
        try:
            await page.goto(url, timeout=60000)  # 타임아웃을 60초로 증가
        except Exception as e:
            print(f"Error navigating to {url}: {e}")
            await browser.close()
            return {}

        SCROLL_PAUSE_TIME = 1
        scroll = 0
        purchase_data = {}

        while True:
            try:
                element = await page.query_selector('.Ere_prod_graphwrap_a')
                if element:
                    # print("Element found")  # 디버깅 출력
                    content = await element.inner_html()
                    soup = BeautifulSoup(content, 'html.parser')
                    data = soup.find_all('div', class_='per')
                    age = ['10대 여성', '10대 남성', '20대 여성', '20대 남성', '30대 여성', '30대 남성', '40대 여성', '40대 남성', '50대 여성', '50대 남성', '60대 이상 여성', '60대 이상 남성']

                    for i in range(0, len(age)):
                        purchase_data[age[i]] = float(data[i].text[:-1])
                    break
                else:
                    # print("Element not found")  # 디버깅 출력
                    pass
            except Exception as e:
                print(f"Error: {e}")

            scroll += 1500
            if scroll >= 13500:
                # print("Reached maximum scroll limit")  # 디버깅 출력
                break
            await page.evaluate(f"window.scrollTo(0, {scroll});")
            await asyncio.sleep(SCROLL_PAUSE_TIME)

        await browser.close()
        # print("Purchase data:", purchase_data)  # 디버깅 출력
        return purchase_data
        
        async def get_book_purchase_data_by_batch(batch_size=3):
    # start_time = time.time()
    links = await get_link()
    
    if not links:
        print("No links found")  # 디버깅 출력
        return

    tasks = [get_purchase_data(link) for link in links]

    # 작업을 배치로 나누기
    for i in range(0, len(tasks), batch_size):
        batch = tasks[i:i + batch_size]
        print(f'{i//batch_size + 1} 번째 배치 처리 중')
        
        # 배치를 비동기적으로 처리
        try:
            results = await asyncio.gather(*batch)
        except Exception as e:
            print(f"Error during batch processing: {e}")
            continue

        for link, purchase_data in zip(links[i:i + batch_size], results):
            if purchase_data:
                # print(f"Link: {link}, Purchase Data: {purchase_data}")
                await update_book_purchase_data(link, purchase_data)

 

Playwright 메서드 소개

페이지 조작

goto(url): 지정한 URL로 이동

page.goto("https://example.com")

 

click(selector): 선택한 요소 클릭

page.click("button#submit")

 

fill(selector, value): 입력 필드에 값을 입력

page.fill("input#username", "my_username")

 

screenshot(path="screenshot.png"): 현재 페이지의 스크린샷을 저장

page.screenshot(path="screenshot.png")

 

요소 선택 및 정보 가져오기

locator(selector): 특정 셀렉터에 해당하는 요소 찾기

locator = page.locator("div.content")

 

inner_text(): 요소의 내부 텍스트를 반환

text = locator.inner_text()

 

get_attribute(name): 요소의 특정 속성 값을 반환

attr_value = locator.get_attribute("href")

 

상태 확인

is_visible(): 요소가 보이는지 여부를 반환

visible = locator.is_visible()

 

is_enabled(): 요소가 활성화되어 있는지 여부를 반환

enabled = locator.is_enabled()

 

고급 조작

hover(): 요소 위에 마우스를 올리기

locator.hover()

 

drag_to(target_locator): 요소를 다른 위치로 드래그하여 놓기

source.drag_to(target)

 

evaluate(expression): 페이지 내에서 JavaScript 코드를 실행하고 결과를 반환

result = page.evaluate("document.title")