"""쿠팡 검색 어댑터. 쿠팡은 Akamai Bot Manager 의 JS 행동 챌린지를 걸어 curl_cffi 단독으로는 통과 못 한다. → Patchright(스텔스 Playwright) + 실제 Chrome 으로 챌린지를 통과한다. persistent context 로 브라우저를 재사용하므로 챌린지는 (쿠키 만료 전까지) 1회만 풀린다. 향후 최적화(하이브리드): 검증된 Akamai 쿠키를 curl_cffi 로 넘겨 대량 후속 요청을 브라우저 없이 처리 가능. 현재는 브라우저 재사용만으로도 후속 검색이 충분히 빠르다. """ import asyncio import time from urllib.parse import quote 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 from services.search.coupang.parser import parse_search_html from services.search.coupang.selectors import SELECTORS _SEARCH_URL = "https://www.coupang.com/np/search?q={q}&channel=user&listSize={n}" _BLOCK_MARKERS = ("sec-if-cpt-container", "Powered and protected", "/akam/") # 대역폭 절감: 이미지/미디어/폰트/CSS 는 상품 데이터·Akamai(JS) 에 불필요 → 차단(프록시 per-GB 비용↓). _BLOCKED_RESOURCES = {"image", "media", "font", "stylesheet"} class CoupangAdapter(SearchAdapter): source = "coupang" def __init__(self, headless: bool = False, user_data_dir: str = "/tmp/lps_coupang_profile", rate_limiter: RateLimiter | None = None, proxy=None, block_resources: bool = True, on_detect=None, max_block_retries: int = 1): self._headless = headless self._user_data_dir = user_data_dir self._rl = rate_limiter or RateLimiter() self._proxy = proxy # DecodoProxy 등 (없으면 직접 연결) self._block_resources = block_resources self._on_detect = on_detect # async def(event: dict) — 감지 이벤트 영속화(선택) self._max_block_retries = max_block_retries # 감지 시 새 IP 로 인라인 재시도 횟수 self._pw = None self._ctx = None self._launched_at = 0.0 self._ip_requests = 0 # 현재 브라우저(IP)로 보낸 요청 수(재기동 시 리셋) self._current_port = None # 현재 사용 중인 프록시 포트(감지 로그용) self._force_recycle = False # 감지 등으로 다음 _ensure_browser 에서 강제 재기동 self._lock = asyncio.Lock() self._ok = 0 self._blocked = 0 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: """강제 재기동 플래그(봇 감지)거나, 프록시 sticky 세션창이 지났으면 재기동해 새 IP 를 받는다.""" 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("[coupang] 브라우저 재기동(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) # 이미지/미디어/폰트/CSS 차단 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 async def search(self, query: str, limit: int = 40) -> list[NormalizedProduct]: async with self._lock: # 인스턴스 내 검색은 직렬화(브라우저 컨텍스트 공유) # 봇 감지 시: 새 IP 로 회전 후 인라인 재시도(최대 max_block_retries 회) 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 = _SEARCH_URL.format(q=quote(query), n=limit) try: await page.goto(url, wait_until="domcontentloaded", timeout=40000) try: await page.wait_for_selector(SELECTORS.card, timeout=20000) # Akamai 센서+렌더 대기 except Exception: pass html = await page.content() except Exception as ex: raise AdapterError(f"쿠팡 검색 실패: {ex}", source=self.source) from ex products = parse_search_html(html, source=self.source) if products: self._ok += 1 LOG.d(f"[coupang] query={query!r} → {len(products)}건 (limit {limit}, ip_req#{self._ip_requests})") return products[:limit] # 0건 — 봇 감지 여부 판정 marker = next((m for m in _BLOCK_MARKERS if m in html), None) 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() # 다음 포트 = 새 IP self._force_recycle = True # 다음 _ensure_browser 에서 재기동 LOG.w(f"[coupang] 봇 감지 → IP 회전 후 재시도 ({attempt + 1}/{self._max_block_retries})") continue raise AdapterError(f"쿠팡 결과 없음/차단 (query={query!r}, blocked={blocked})", source=self.source, blocked=blocked) async def _report_detection(self, query: str, marker: str, html_len: int): """봇 감지 기록 — 로그 + (있으면) DB 영속화. IP당 몇 번째 요청에서 감지됐는지 축적.""" elapsed = int(time.monotonic() - self._launched_at) LOG.w(f"[coupang][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"[coupang] 감지 기록 실패(무시): {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