로그인 세션 및 쿠키 영구화

[ 1회성 초기 인증 (내 로컬 PC) ]
개발자가 브라우저를 열고 로그인 (SMS/OTP 직접 입력)
  │
  ▼
context.storage_state(path="session.json") 호출
  │ (쿠키 + localStorage 토큰이 JSON 파일로 추출됨)
  ▼
대시보드 또는 CLI로 DashCloud 영구 볼륨(/app/data/session.json)에 업로드
 
─────────────────────────────────────────────────────────────
 
[ 24/7 무중단 수집 (DashCloud 런타임) ]
browser.new_context(storage_state="/app/data/session.json")
  │ (로그인 페이지를 거치지 않고 인증 완료 상태로 즉시 브라우저 로드)
  ▼
회원 전용 마이페이지 또는 특가 페이지로 다이렉트 이동

from pathlib import Path
from playwright.sync_api import sync_playwright
 
SESSION_FILE = Path("session.json")
 
def create_initial_session():
    with sync_playwright() as p:
        # 사람이 직접 로그인할 수 있도록 화면에 브라우저 표시
        browser = p.chromium.launch(headless=False)
        context = browser.new_context()
        page = context.new_page()
 
        print("로그인 페이지로 이동합니다. 브라우저에서 직접 로그인을 완료하세요...")
        page.goto("https://target-service.com/login")
 
        # 로그인 완료 후 이동하는 메인 페이지 URL 또는 특정 요소가 뜰 때까지 대기
        # 예: 마이페이지 링크나 로그아웃 버튼이 보일 때까지 대기 (최대 120초)
        page.wait_for_selector("a[href='/logout']", timeout=120000)
        print("로그인이 감지되었습니다! 세션 상태를 저장합니다.")
 
        # [핵심] 쿠키와 로컬스토리지 전체를 session.json으로 덤프
        context.storage_state(path=str(SESSION_FILE))
        print(f"세션 파일 저장 완료: {SESSION_FILE.resolve()}")
 
        context.close()
        browser.close()
 
if __name__ == "__main__":
    create_initial_session()

import os
import logging
from pathlib import Path
from playwright.sync_api import sync_playwright
 
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("AuthCrawler")
 
SESSION_PATH = Path("/app/data/session.json")
 
def run_authenticated_crawler():
    if not SESSION_PATH.exists():
        logger.error(f"세션 파일({SESSION_PATH})이 존재하지 않습니다! 먼저 세션을 업로드하세요.")
        return
 
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
 
        # [핵심] 저장된 세션 상태를 주입하여 컨텍스트 생성
        context = browser.new_context(
            storage_state=str(SESSION_PATH),
            viewport={"width": 1920, "height": 1080}
        )
 
        page = context.new_page()
        try:
            logger.info("회원 전용 페이지로 직접 이동합니다...")
            page.goto("https://target-service.com/wholesale/products", timeout=30000)
 
            # 1. 세션 만료 여부 검증 (로그아웃 버튼 존재 확인)
            if not page.is_visible("a[href='/logout']"):
                logger.warning("⚠️ 세션이 만료되어 비로그인 상태로 리다이렉트되었습니다!")
                handle_session_expired()
                return
 
            # 2. 회원 전용 데이터 수집 수행
            items = page.query_selector_all(".wholesale-item")
            logger.info(f"성공: {len(items)}개 도매 상품 정보를 수집했습니다.")
 
        except Exception as e:
            logger.error(f"크롤링 도중 예외 발생: {e}")
        finally:
            context.close()
            browser.close()
 
def handle_session_expired():
    # 텔레그램이나 디스코드로 관리자에게 세션 만료 알림 전송
    logger.critical("관리자 텔레그램으로 재로그인 요청 알림을 발송합니다.")
 
if __name__ == "__main__":
    run_authenticated_crawler()

def refresh_and_save_session(page, context, session_path: Path):
    """주기적으로 페이지를 방문하여 갱신된 세션 토큰을 영구 디스크에 저장"""
    try:
        page.goto("https://target-service.com/my-account", timeout=20000)
        # 최신 세션 상태를 다시 디스크에 덮어쓰기
        context.storage_state(path=str(session_path))
        logger.info("세션 상태가 성공적으로 갱신되었습니다.")
    except Exception as e:
        logger.warning(f"세션 갱신 실패: {e}")