From e27b73c018cc4e76b79eb06972a0861d97101985 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EB=AF=BC=ED=97=8C?= Date: Wed, 8 Jul 2026 15:59:48 +0900 Subject: [PATCH] =?UTF-8?q?feat(lps):=20=EC=BF=A0=ED=8C=A1=20=EA=B2=80?= =?UTF-8?q?=EC=83=89=20=EC=96=B4=EB=8C=91=ED=84=B0=20=E2=80=94=20Akamai=20?= =?UTF-8?q?=EC=9A=B0=ED=9A=8C(Patchright)=20+=20=EC=A0=95=EA=B7=9C?= =?UTF-8?q?=ED=99=94=20=ED=8C=8C=EC=84=9C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 소스 어댑터 패턴으로 크롤링을 시작한다. 쿠팡은 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) --- lps/requirements.txt | 7 ++ lps/services/search/contract.py | 60 +++++++++++++++ lps/services/search/coupang/adapter.py | 94 ++++++++++++++++++++++++ lps/services/search/coupang/parser.py | 72 ++++++++++++++++++ lps/services/search/coupang/selectors.py | 22 ++++++ lps/services/search/proxy.py | 25 +++++++ lps/services/search/rate_limiter.py | 26 +++++++ lps/services/search/util.py | 18 +++++ lps/tests/fixtures/coupang_search.html | 3 + lps/tests/test_coupang_parser.py | 27 +++++++ 10 files changed, 354 insertions(+) create mode 100644 lps/services/search/contract.py create mode 100644 lps/services/search/coupang/adapter.py create mode 100644 lps/services/search/coupang/parser.py create mode 100644 lps/services/search/coupang/selectors.py create mode 100644 lps/services/search/proxy.py create mode 100644 lps/services/search/rate_limiter.py create mode 100644 lps/services/search/util.py create mode 100644 lps/tests/fixtures/coupang_search.html create mode 100644 lps/tests/test_coupang_parser.py diff --git a/lps/requirements.txt b/lps/requirements.txt index 7da937d..1b3c169 100644 --- a/lps/requirements.txt +++ b/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 설치 필요 diff --git a/lps/services/search/contract.py b/lps/services/search/contract.py new file mode 100644 index 0000000..1fa7d64 --- /dev/null +++ b/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) diff --git a/lps/services/search/coupang/adapter.py b/lps/services/search/coupang/adapter.py new file mode 100644 index 0000000..1422038 --- /dev/null +++ b/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 diff --git a/lps/services/search/coupang/parser.py b/lps/services/search/coupang/parser.py new file mode 100644 index 0000000..4ada7ab --- /dev/null +++ b/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 diff --git a/lps/services/search/coupang/selectors.py b/lps/services/search/coupang/selectors.py new file mode 100644 index 0000000..3aa6cff --- /dev/null +++ b/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() diff --git a/lps/services/search/proxy.py b/lps/services/search/proxy.py new file mode 100644 index 0000000..ebf65cf --- /dev/null +++ b/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) diff --git a/lps/services/search/rate_limiter.py b/lps/services/search/rate_limiter.py new file mode 100644 index 0000000..66a5903 --- /dev/null +++ b/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() diff --git a/lps/services/search/util.py b/lps/services/search/util.py new file mode 100644 index 0000000..3df92d1 --- /dev/null +++ b/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 diff --git a/lps/tests/fixtures/coupang_search.html b/lps/tests/fixtures/coupang_search.html new file mode 100644 index 0000000..e4fb402 --- /dev/null +++ b/lps/tests/fixtures/coupang_search.html @@ -0,0 +1,3 @@ + \ No newline at end of file diff --git a/lps/tests/test_coupang_parser.py b/lps/tests/test_coupang_parser.py new file mode 100644 index 0000000..89189a2 --- /dev/null +++ b/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