헤드리스 Playwright 메모리 최적화 및 누수 방지

from playwright.sync_api import sync_playwright
 
with sync_playwright() as p:
    # 1. 브라우저 인스턴스는 1개만 실행
    browser = p.chromium.launch(
        headless=True,
        args=["--no-sandbox", "--disable-dev-shm-usage"]
    )
 
    while True:
        # 2. [핵심] 수집 작업마다 독립된 가상 브라우저 컨텍스트 생성
        context = browser.new_context()
        page = context.new_page()
        try:
            page.goto("https://target-site.com", timeout=30000)
            data = extract_data(page)
            save_to_database(data)
        except Exception as e:
            logger.error(f"수집 실패: {e}")
        finally:
            # 3. [핵심] 컨텍스트를 닫는 순간 해당 세션의 모든 DOM 힙과 캐시가 100% 즉시 증발
            context.close()

def block_unnecessary_resources(page):
    """이미지, 미디어, 웹폰트 다운로드를 네트워크 단에서 즉각 거절"""
    def route_filter(route):
        # 차단할 리소스 타입 정의
        if route.request.resource_type in ["image", "media", "font"]:
            route.abort()
        # 구글 애널리틱스 등 불필요한 외부 추적기 차단
        elif any(domain in route.request.url for domain in ["google-analytics.com", "doubleclick.net", "facebook.net"]):
            route.abort()
        else:
            route.continue_()
 
    page.route("**/*", route_filter)

import os
import time
import psutil
import logging
 
logger = logging.getLogger("RecycleWorker")
 
MAX_TASKS_PER_PROCESS = 500   # 500회 작업 후 자율 재시작
MAX_UPTIME_SECONDS = 12 * 3600 # 12시간 가동 후 자율 재시작
MAX_MEMORY_MB = 400            # 메모리가 400MB를 넘으면 자율 재시작
 
def get_current_memory_mb():
    """현재 프로세스의 실제 물리 메모리 RSS 점유량 측정"""
    process = psutil.Process(os.getpid())
    return process.memory_info().rss / (1024 * 1024)
 
def check_lifecycle_and_recycle(task_count, start_time):
    current_mem = get_current_memory_mb()
    uptime = time.time() - start_time
 
    if task_count >= MAX_TASKS_PER_PROCESS or uptime >= MAX_UPTIME_SECONDS or current_mem >= MAX_MEMORY_MB:
        logger.info(
            f"자율 재활용 임계치 도달 (작업: {task_count}회, 가동: {int(uptime)}초, 메모리: {current_mem:.1f}MB). "
            "프로세스를 정상 종료하여 메모리를 100% 초기화합니다."
        )
        # DashCloud 런타임이 프로세스 종료를 감지하면 깨끗한 컨테이너로 자동 재기동함
        os._exit(0)

import os
import time
import logging
from playwright.sync_api import sync_playwright
 
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("CrawlerEngine")
 
def run_crawler_pipeline():
    start_time = time.time()
    task_count = 0
 
    with sync_playwright() as p:
        logger.info("Chromium 엔진 기동...")
        browser = p.chromium.launch(
            headless=True,
            args=[
                "--no-sandbox",
                "--disable-dev-shm-usage",
                "--disable-gpu",
                "--disable-extensions",
            ]
        )
 
        while True:
            # 1. 독립 컨텍스트 생성
            context = browser.new_context(viewport={"width": 1280, "height": 720})
            page = context.new_page()
 
            # 2. 이미지/미디어 네트워크 차단
            page.route(
                "**/*", 
                lambda route: route.abort() 
                if route.request.resource_type in ["image", "media", "font"] 
                else route.continue_()
            )
 
            try:
                # 3. 데이터 수집 실행
                page.goto("https://news.ycombinator.com", timeout=20000, wait_until="domcontentloaded")
                title = page.title()
                task_count += 1
                logger.info(f"[{task_count}회차] 수집 성공: {title}")
 
            except Exception as e:
                logger.error(f"수집 예외 발생: {e}")
            finally:
                # 4. 컨텍스트 및 페이지 메모리 즉시 회수
                context.close()
 
            # 5. 자율 재시작 체크 (12시간 경과 또는 500회 수집)
            if task_count >= 500 or (time.time() - start_time) > 43200:
                logger.info("메모리 완전 초기화를 위해 프로세스를 정상 종료하고 재기동합니다.")
                browser.close()
                os._exit(0)
 
            # 다음 수집 주기까지 대기 (예: 30초)
            time.sleep(30)
 
if __name__ == "__main__":
    run_crawler_pipeline()