"""브라우저 기반 검색 어댑터 공통 베이스. 쿠팡·G마켓·옥션·11번가처럼 안티봇(Akamai/ESM 챌린지 등) 때문에 실제 Chrome(patchright)로 뚫어야 하는 소스의 공통 machinery 를 모은다: 브라우저 수명 관리 · 프록시 sticky 회전 · 리소스 차단(대역폭↓) · 차단 감지 + IP 회전 재시도 + 감지 기록. 사이트별 차이는 훅으로 분리한다: _search_url(query, limit) 검색 URL _parse(html) HTML → NormalizedProduct[] ready_selector 결과 렌더 완료 신호 셀렉터(챌린지 통과 대기용) block_markers/min_result_html 차단 판정(0건일 때) detect_block 은 순수 함수로 분리 — 브라우저 없이 단위 테스트 가능. """ import asyncio import time from abc import abstractmethod from patchright.async_api import async_playwright from common.logger import LOG from services.search.contract import SearchAdapter, NormalizedProduct, AdapterError, AdapterHealth from services.search.rate_limiter import RateLimiter # 대역폭 절감: 이미지/미디어/폰트/CSS 는 상품 데이터·안티봇(JS) 에 불필요 → 차단(프록시 per-GB 비용↓). _BLOCKED_RESOURCES = {"image", "media", "font", "stylesheet"} def detect_block(html: str, product_count: int, markers: tuple, min_len: int) -> str | None: """0건 응답의 차단 여부 판정(순수 함수). 반환: 차단 마커(차단) 또는 None(정상 빈결과). 상품이 있으면 항상 None. 알려진 마커 우선, 없으면 비정상적으로 짧은 HTML 을 미지의 차단으로 폴백.""" if product_count > 0: return None marker = next((m for m in markers if m in html), None) if marker is None and len(html) < min_len: marker = f"short_html({len(html)}B)" return marker class BrowserSearchAdapter(SearchAdapter): """patchright(스텔스 Chrome) 기반 검색 어댑터 베이스. persistent context 로 브라우저를 재사용한다.""" # 서브클래스 오버라이드 지점 block_markers: tuple = () min_result_html: int = 10000 ready_selector: str = "body" ready_timeout_ms: int = 20000 block_resources_default: bool = True # 쿠팡=True(대역폭↓). 오픈마켓은 리소스차단이 렌더/챌린지를 깨 False. scroll_steps: int = 0 # >0 이면 렌더 대기 전 스크롤(지연 로딩 트리거, 예: 11번가) def __init__(self, headless: bool = False, user_data_dir: str | None = None, rate_limiter: RateLimiter | None = None, proxy=None, block_resources: bool | None = None, on_detect=None, max_block_retries: int = 1): self._headless = headless self._user_data_dir = user_data_dir or f"/tmp/lps_{self.source}_profile" self._rl = rate_limiter or RateLimiter() self._proxy = proxy # DecodoProxy 등 (없으면 직접 연결) self._block_resources = self.block_resources_default if block_resources is None else block_resources self._on_detect = on_detect # async def(event: dict) — 감지 영속화(선택) self._max_block_retries = max_block_retries self._pw = None self._ctx = None self._launched_at = 0.0 self._ip_requests = 0 # 현재 브라우저(IP)로 보낸 요청 수(재기동 시 리셋) self._current_port = None self._force_recycle = False self._lock = asyncio.Lock() self._ok = 0 self._blocked = 0 # ---- 사이트별 훅 -------------------------------------------------- @abstractmethod def _search_url(self, query: str, limit: int) -> str: raise NotImplementedError @abstractmethod def _parse(self, html: str) -> list[NormalizedProduct]: raise NotImplementedError async def _wait_ready(self, page): """결과 렌더 대기. 지연 로딩(scroll_steps>0)이면 먼저 스크롤로 트리거하고, ready_selector 등장까지 대기. 챌린지형(ESM '잠시만')은 이 대기 시간 안에 자동 통과(IP 평판 좋을 때). 타임아웃은 예외로 두지 않는다 — 이후 parse 0건이면 차단 판정 로직(마커/짧은HTML)이 처리해 IP 회전을 유도.""" for _ in range(self.scroll_steps): try: await page.evaluate("window.scrollBy(0, 1500)") except Exception: break await page.wait_for_timeout(1000) try: await page.wait_for_selector(self.ready_selector, timeout=self.ready_timeout_ms) except Exception: pass # ---- 공통 브라우저 수명 ------------------------------------------- async def _route(self, route): if route.request.resource_type in _BLOCKED_RESOURCES: await route.abort() else: await route.continue_() def _recycle_due(self) -> bool: if self._force_recycle: return True if not (self._proxy and self._proxy.enabled): return False return (time.monotonic() - self._launched_at) > self._proxy.session_minutes * 60 async def _ensure_browser(self): if self._ctx is not None: if self._recycle_due(): LOG.d(f"[{self.source}] 브라우저 재기동(IP 회전)") await self._close_ctx() else: return if self._pw is None: self._pw = await async_playwright().start() kwargs = dict(user_data_dir=self._user_data_dir, channel="chrome", headless=self._headless, no_viewport=True) if self._proxy and self._proxy.enabled: kwargs["proxy"] = self._proxy.playwright_proxy() self._current_port = self._proxy.current_port self._ctx = await self._pw.chromium.launch_persistent_context(**kwargs) if self._block_resources: await self._ctx.route("**/*", self._route) self._launched_at = time.monotonic() self._ip_requests = 0 self._force_recycle = False async def _close_ctx(self): if self._ctx is not None: try: await self._ctx.close() finally: self._ctx = None # ---- 검색(차단 감지 + IP 회전 인라인 재시도) ---------------------- async def search(self, query: str, limit: int = 40) -> list[NormalizedProduct]: async with self._lock: # 인스턴스 내 검색 직렬화(브라우저 컨텍스트 공유) for attempt in range(self._max_block_retries + 1): await self._rl.wait() await self._ensure_browser() self._ip_requests += 1 page = self._ctx.pages[0] if self._ctx.pages else await self._ctx.new_page() url = self._search_url(query, limit) try: await page.goto(url, wait_until="domcontentloaded", timeout=40000) await self._wait_ready(page) html = await page.content() except Exception as ex: raise AdapterError(f"{self.source} 검색 실패: {ex}", source=self.source) from ex products = self._parse(html) if products: self._ok += 1 LOG.d(f"[{self.source}] query={query!r} → {len(products)}건 (limit {limit}, ip_req#{self._ip_requests})") return products[:limit] marker = detect_block(html, len(products), self.block_markers, self.min_result_html) blocked = marker is not None self._blocked += 1 if blocked: await self._report_detection(query, marker, len(html)) can_retry = blocked and self._proxy and self._proxy.enabled and attempt < self._max_block_retries if can_retry: self._proxy.rotate() self._force_recycle = True LOG.w(f"[{self.source}] 봇 감지 → IP 회전 후 재시도 ({attempt + 1}/{self._max_block_retries})") continue raise AdapterError(f"{self.source} 결과 없음/차단 (query={query!r}, blocked={blocked})", source=self.source, blocked=blocked) async def _report_detection(self, query: str, marker: str, html_len: int): elapsed = int(time.monotonic() - self._launched_at) LOG.w(f"[{self.source}][BOT-DETECTED] ip_req#{self._ip_requests} port={self._current_port} " f"elapsed={elapsed}s headless={self._headless} marker={marker!r} query={query!r} html_len={html_len}") if self._on_detect is not None: event = {"source": self.source, "query": query, "ip_request_no": self._ip_requests, "proxy_port": self._current_port, "elapsed_sec": elapsed, "marker": marker, "headless": self._headless, "html_len": html_len} try: await self._on_detect(event) except Exception as ex: LOG.e_no_callstack(f"[{self.source}] 감지 기록 실패(무시): {ex}") async def health(self) -> AdapterHealth: total = self._ok + self._blocked rate = (self._ok / total) if total else 0.0 return AdapterHealth(source=self.source, ok=(self._blocked == 0 or rate > 0.5), recent_success_rate=rate, blocked_rate=(self._blocked / total) if total else 0.0) async def close(self): if self._ctx is not None: await self._ctx.close() self._ctx = None if self._pw is not None: await self._pw.stop() self._pw = None