feat(lps): 쿠팡 검색 어댑터 — Akamai 우회(Patchright) + 정규화 파서

소스 어댑터 패턴으로 크롤링을 시작한다. 쿠팡은 Akamai Bot Manager 의
JS 행동 챌린지라 curl_cffi 단독 불가 → Patchright(스텔스 Playwright) +
실제 Chrome(channel=chrome)으로 챌린지를 통과하고 selectolax 로 파싱한다.

- contract: SearchAdapter(ABC)·NormalizedProduct·AdapterHealth·AdapterError
- coupang: adapter(브라우저 재사용, 챌린지 1회)·parser(순수)·selectors(외부화, webpack 해시 prefix 매칭)
- 공용: rate_limiter(2~8s 랜덤)·proxy(무프록시 off 인터페이스)·util.parse_price
- 가격 파싱: 단위가격 '(1개당 44,400원)'의 앞 '1' 오인 방지 — '원' 앞 숫자 앵커링
- deps: curl_cffi·selectolax·patchright(nodriver 는 py3.14 버그로 대체)
- tests: 파서 회귀(소형 fixture) — 라이브 2회 검색 + 파서 3/3 통과

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

lps/requirements.txt

@@ -6,3 +6,10 @@ asyncpg
 orjson
 pydantic>=2.0
 httpx  # 외부 사이트/오픈API(최저가 조회) 호출용 async HTTP 클라이언트
+
+# --- 크롤링(LPS) — 안티봇 대응은 어댑터 안에 격리. 트레드밀 대비 버전 pin. ---
+curl_cffi==0.15.0   # 쿠팡: TLS/HTTP2 지문 위장(impersonate) async HTTP 클라이언트
+selectolax==0.4.10  # 빠른 C 파서(쿠팡 HTML). bs4 대비 최대 30배
+patchright          # 스텔스 Playwright 포크. 쿠팡 Akamai JS 챌린지 통과(실제 Chrome, channel=chrome)
+                    #   ※ nodriver 는 Python 3.14 소스인코딩 버그로 미채택 → Patchright 로 대체
+                    #   ※ 실행엔 시스템 Google Chrome 필요(로컬) / 배포 이미지엔 chromium 설치 필요

lps/services/search/contract.py

@@ -0,0 +1,60 @@
+"""소스 어댑터 공통 계약.
+
+크롤링/검색 소스(네이버·쿠팡 …)는 각자 수집 방식과 안티봇 대응을 캡슐화하고,
+코어 파이프라인(필터·이상치·AI)은 정규화된 NormalizedProduct 만 본다.
+새 소스는 SearchAdapter 를 구현하기만 하면 코어 변경 없이 붙는다(트레드밀 격리).
+"""
+
+from abc import ABC, abstractmethod
+from typing import Optional
+
+from pydantic import BaseModel, Field
+
+
+class NormalizedProduct(BaseModel):
+    """소스 무관 정규화 상품 스키마. 어댑터의 유일한 출력 계약."""
+
+    source: str = Field(description="수집 소스 (naver|coupang)")
+    name: str = Field(description="상품명")
+    price: int = Field(description="판매가(원, 정수). 파싱 실패분은 어댑터에서 제외")
+    model: Optional[str] = Field(None, description="모델명(있으면)")
+    manufacturer: Optional[str] = Field(None, description="제조사(있으면)")
+    image_url: Optional[str] = Field(None, description="썸네일 URL")
+    detail_url: Optional[str] = Field(None, description="상품 상세 URL")
+    shipping_fee: Optional[int] = Field(None, description="배송비(원). 무료=0, 미확인=None")
+    mall_name: Optional[str] = Field(None, description="판매몰/스토어명")
+    external_id: Optional[str] = Field(None, description="소스 내 상품 식별자")
+
+
+class AdapterHealth(BaseModel):
+    """어댑터 건강도. 성공률 급락 = 레이아웃 변경/차단 신호 → 알림 훅."""
+
+    source: str
+    ok: bool = Field(description="현재 정상 동작 여부")
+    recent_success_rate: float = Field(0.0, description="최근 요청 성공률(0~1)")
+    blocked_rate: float = Field(0.0, description="최근 차단(봇탐지) 비율(0~1)")
+    note: str = ""
+
+
+class AdapterError(Exception):
+    """어댑터 수집 실패. blocked=True 면 안티봇 차단으로 판단(에스컬레이션/알림 트리거)."""
+
+    def __init__(self, message: str, *, source: str, blocked: bool = False):
+        super().__init__(message)
+        self.source = source
+        self.blocked = blocked
+
+
+class SearchAdapter(ABC):
+    """검색 소스 어댑터. 소스별 수집/에스컬레이션/안티봇을 내부에 캡슐화한다."""
+
+    source: str
+
+    @abstractmethod
+    async def search(self, query: str, limit: int = 40) -> list[NormalizedProduct]:
+        """query 로 검색해 정규화 상품 리스트를 반환. 차단 시 AdapterError(blocked=True)."""
+        raise NotImplementedError
+
+    async def health(self) -> AdapterHealth:
+        """기본 건강도. 어댑터가 관측 지표를 축적하면 override."""
+        return AdapterHealth(source=self.source, ok=True)

lps/services/search/coupang/adapter.py

@@ -0,0 +1,94 @@
+"""쿠팡 검색 어댑터.
+
+쿠팡은 Akamai Bot Manager 의 JS 행동 챌린지를 걸어 curl_cffi 단독으로는 통과 못 한다.
+→ Patchright(스텔스 Playwright) + 실제 Chrome 으로 챌린지를 통과한다.
+persistent context 로 브라우저를 재사용하므로 챌린지는 (쿠키 만료 전까지) 1회만 풀린다.
+
+향후 최적화(하이브리드): 검증된 Akamai 쿠키를 curl_cffi 로 넘겨 대량 후속 요청을
+브라우저 없이 처리 가능. 현재는 브라우저 재사용만으로도 후속 검색이 충분히 빠르다.
+"""
+
+import asyncio
+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/")
+
+
+class CoupangAdapter(SearchAdapter):
+    source = "coupang"
+
+    def __init__(self, headless: bool = False, user_data_dir: str = "/tmp/lps_coupang_profile", rate_limiter: RateLimiter | None = None):
+        self._headless = headless
+        self._user_data_dir = user_data_dir
+        self._rl = rate_limiter or RateLimiter()
+        self._pw = None
+        self._ctx = None
+        self._lock = asyncio.Lock()
+        self._ok = 0
+        self._blocked = 0
+
+    async def _ensure_browser(self):
+        if self._ctx is not None:
+            return
+        self._pw = await async_playwright().start()
+        self._ctx = await self._pw.chromium.launch_persistent_context(
+            user_data_dir=self._user_data_dir,
+            channel="chrome",
+            headless=self._headless,
+            no_viewport=True,
+        )
+
+    async def search(self, query: str, limit: int = 40) -> list[NormalizedProduct]:
+        async with self._lock:  # 인스턴스 내 검색은 직렬화(브라우저 컨텍스트 공유)
+            await self._rl.wait()
+            await self._ensure_browser()
+            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)
+                # Akamai 센서 실행 + 상품 렌더 대기
+                try:
+                    await page.wait_for_selector(SELECTORS.card, timeout=20000)
+                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 not products:
+                blocked = any(m in html for m in _BLOCK_MARKERS)
+                self._blocked += 1
+                LOG.w(f"[coupang] 결과 0건 (blocked={blocked}) query={query!r} len={len(html)}")
+                raise AdapterError(f"쿠팡 결과 없음/차단 (query={query!r})", source=self.source, blocked=blocked)
+
+            self._ok += 1
+            LOG.d(f"[coupang] query={query!r} → {len(products)}건 (limit {limit})")
+            return products[:limit]
+
+    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

lps/services/search/coupang/parser.py

@@ -0,0 +1,72 @@
+"""쿠팡 검색결과 HTML → NormalizedProduct[] (순수 함수, selectolax).
+
+브라우저/네트워크와 무관. 저장된 HTML 로 결정론적 단위 테스트가 가능하다.
+"""
+
+import re
+
+from selectolax.parser import HTMLParser
+
+from services.search.contract import NormalizedProduct
+from services.search.coupang.selectors import SELECTORS as S
+
+BASE = "https://www.coupang.com"
+
+# '원' 바로 앞의 숫자만 가격으로 인식('(1개당 44,400원)'의 앞 '1' 오인 방지).
+_WON = re.compile(r"([\d,]+)\s*원")
+
+
+def _sale_price(price_area) -> int | None:
+    """판매가 추출. 정가(del)·단위가격('~당', 괄호)·할인율(%)은 제외하고
+    본문 판매가 노드(문서 순서상 먼저 오는 '원' 값)를 취한다."""
+    if price_area is None:
+        return None
+    for node in price_area.css("span, div, strong"):
+        if node.tag == "del":
+            continue
+        text = node.text(strip=True) or ""
+        if "원" not in text or "당" in text or text.startswith("("):
+            continue  # 단위가격/부가문구 제외
+        m = _WON.search(text)
+        if m:
+            return int(m.group(1).replace(",", ""))
+    return None
+
+
+def parse_search_html(html: str, source: str = "coupang") -> list[NormalizedProduct]:
+    tree = HTMLParser(html)
+    products: list[NormalizedProduct] = []
+
+    for card in tree.css(S.card):
+        name_el = card.css_first(S.name)
+        name = name_el.text(strip=True) if name_el else None
+
+        img = card.css_first(S.image)
+        if not name and img:
+            name = img.attributes.get("alt")
+        image_url = img.attributes.get("src") if img else None
+
+        price_area = card.css_first(S.price_area)
+        price = _sale_price(price_area)
+
+        # 이름/가격이 없으면 유효 상품이 아니므로 스킵(광고 슬롯 등)
+        if not name or price is None:
+            continue
+
+        a = card.css_first(S.link)
+        href = a.attributes.get("href") if a else None
+        detail_url = (BASE + href) if href and href.startswith("/") else href
+
+        products.append(
+            NormalizedProduct(
+                source=source,
+                name=name,
+                price=price,
+                image_url=image_url,
+                detail_url=detail_url,
+                mall_name="쿠팡",
+                external_id=card.attributes.get(S.data_id_attr),
+            )
+        )
+
+    return products

lps/services/search/coupang/selectors.py

@@ -0,0 +1,22 @@
+"""쿠팡 검색결과 셀렉터(외부화).
+
+쿠팡 클래스는 webpack 해시 suffix(예: ProductUnit_productUnit__Qd6sv)를 달고 있어
+배포마다 suffix 가 바뀐다. 따라서 접두 부분 매칭([class*=...])으로 견고성을 확보한다.
+레이아웃이 바뀌면 이 파일만 고치면 되도록 파서에서 분리한다(new.md: 셀렉터 외부 설정).
+"""
+
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class CoupangSelectors:
+    card: str = "li[class*=ProductUnit_productUnit]"
+    name: str = "[class*=ProductUnit_productNameV2]"
+    price_area: str = "[class*=PriceArea_priceArea]"
+    orig_price: str = "del"            # price_area 내부: 정가(취소선)
+    link: str = "a[href]"
+    image: str = "figure img"
+    data_id_attr: str = "data-id"      # li 의 vendorItemId
+
+
+SELECTORS = CoupangSelectors()

lps/services/search/proxy.py

@@ -0,0 +1,25 @@
+"""프록시 풀 인터페이스.
+
+현재는 무프록시로 시작(residential 프록시 미보유). 인터페이스만 두고 off 상태로 동작한다.
+쿠팡 차단이 발생하면 residential/mobile 프록시 목록을 주입해 로테이션을 켠다.
+(datacenter IP 는 금방 차단되므로 residential/mobile 을 전제로 설계 — new.md)
+"""
+
+import itertools
+from typing import Optional
+
+
+class ProxyPool:
+    def __init__(self, proxies: Optional[list[str]] = None):
+        self._proxies = list(proxies or [])
+        self._cycle = itertools.cycle(self._proxies) if self._proxies else None
+
+    @property
+    def enabled(self) -> bool:
+        return bool(self._proxies)
+
+    def get(self) -> Optional[str]:
+        """다음 프록시 URL 반환. 비어있으면(off) None → 어댑터는 직접 연결."""
+        if self._cycle is None:
+            return None
+        return next(self._cycle)

lps/services/search/rate_limiter.py

@@ -0,0 +1,26 @@
+"""정중한 요청을 위한 랜덤 딜레이 레이트리미터.
+
+기계적 등간격 요청은 탐지 신호(new.md). 요청 사이에 2~8s 랜덤 간격을 둔다.
+소스(도메인)별로 인스턴스를 두고 마지막 요청 시각을 기준으로 최소 간격을 보장한다.
+"""
+
+import asyncio
+import random
+import time
+
+
+class RateLimiter:
+    def __init__(self, min_delay: float = 2.0, max_delay: float = 8.0):
+        self._min = min_delay
+        self._max = max_delay
+        self._last = 0.0
+        self._lock = asyncio.Lock()
+
+    async def wait(self) -> None:
+        """직전 요청으로부터 랜덤 간격이 지나도록 대기(동시 호출 직렬화)."""
+        async with self._lock:
+            target = random.uniform(self._min, self._max)
+            elapsed = time.monotonic() - self._last
+            if elapsed < target:
+                await asyncio.sleep(target - elapsed)
+            self._last = time.monotonic()

lps/services/search/util.py

@@ -0,0 +1,18 @@
+"""검색 어댑터 공용 순수 유틸(부작용 없음). 레퍼런스의 parse_price 를 정리 이식."""
+
+import re
+
+_NUM = re.compile(r"[\d,]+")
+
+
+def parse_price(value) -> int | None:
+    """'44,900원', 44900, '44.900' 등 다형 입력 → int. 실패 시 None."""
+    if value is None:
+        return None
+    if isinstance(value, (int, float)):
+        return int(value)
+    m = _NUM.search(str(value))
+    if not m:
+        return None
+    digits = m.group(0).replace(",", "")
+    return int(digits) if digits.isdigit() else None

lps/tests/test_coupang_parser.py

@@ -0,0 +1,27 @@
+# 쿠팡 파서 결정론적 단위 테스트(네트워크/브라우저 불필요).
+# fixture 는 실제 렌더된 검색결과에서 카드 4개를 추출한 것(단위가격 '1개당' 케이스 포함).
+from pathlib import Path
+
+from services.search.coupang.parser import parse_search_html
+
+FIXTURE = Path(__file__).parent / "fixtures" / "coupang_search.html"
+
+
+def test_parse_extracts_valid_products():
+    items = parse_search_html(FIXTURE.read_text())
+
+    assert len(items) >= 3
+    for p in items:
+        assert p.source == "coupang"
+        assert p.name and len(p.name) > 2
+        # 단위가격('1개당 44,400원')의 '1' 을 가격으로 오인하던 회귀 방지
+        assert p.price >= 100, f"이상 저가: {p.price} ({p.name})"
+        assert p.detail_url and p.detail_url.startswith("https://www.coupang.com/vp/products/")
+        assert p.external_id  # vendorItemId(data-id)
+
+
+def test_parse_price_is_sale_not_unit_price():
+    # fixture 첫 카드의 판매가는 44,400원(단위가격도 '1개당 44,400원'이라 값은 같지만
+    # 파서가 정가(del)나 '%'가 아닌 판매가를 정확히 집는지 확인)
+    items = parse_search_html(FIXTURE.read_text())
+    assert items[0].price == 44400