diff --git a/lps/services/search/browser_base.py b/lps/services/search/browser_base.py new file mode 100644 index 0000000..db8ddda --- /dev/null +++ b/lps/services/search/browser_base.py @@ -0,0 +1,198 @@ +"""브라우저 기반 검색 어댑터 공통 베이스. + +쿠팡·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 diff --git a/lps/services/search/card_parser.py b/lps/services/search/card_parser.py new file mode 100644 index 0000000..2877b50 --- /dev/null +++ b/lps/services/search/card_parser.py @@ -0,0 +1,66 @@ +"""오픈마켓 검색결과 공용 카드 파서(순수 함수, selectolax). + +G마켓·옥션·11번가처럼 '카드 목록 → 카드별 이름/가격/링크/배송' 구조가 같은 소스를 하나로 파싱한다. +사이트별 셀렉터·소스명·베이스URL 만 주입한다. 브라우저/네트워크 무관 — 저장 HTML 로 단위 테스트 가능. +""" + +from dataclasses import dataclass +from urllib.parse import urljoin + +from selectolax.parser import HTMLParser + +from services.search.contract import NormalizedProduct +from services.search.util import extract_price, clean_name, shipping_from_text + + +@dataclass(frozen=True) +class CardConfig: + source: str # naver 외 크롤 소스 코드 (gmarket|auction|st11) + mall_name: str # 표시 몰명 (G마켓|옥션|11번가) + base_url: str # 상대경로 링크 절대화용 + card: str + name: str + price: str + link: str = "a[href]" + + +def parse_cards(html: str, cfg: CardConfig) -> list[NormalizedProduct]: + tree = HTMLParser(html) + products: list[NormalizedProduct] = [] + seen: set[str] = set() + + for card in tree.css(cfg.card): + name_el = card.css_first(cfg.name) + name = clean_name(name_el.text(strip=True)) if name_el else None + if not name: + continue + + price_el = card.css_first(cfg.price) + price = extract_price(price_el.text(strip=True)) if price_el else None + if price is None or price <= 0: + continue # 광고/헤더/추천 슬롯 등 유효 상품 아님 + + a = card.css_first(cfg.link) + href = a.attributes.get("href") if a else None + detail_url = urljoin(cfg.base_url, href) if href else None + + key = detail_url or name # 동일 링크/광고 반복 제거 + if key in seen: + continue + seen.add(key) + + shipping_fee, shipping_type = shipping_from_text(card.text(separator=" "), name) + + products.append( + NormalizedProduct( + source=cfg.source, + name=name, + price=price, + detail_url=detail_url, + mall_name=cfg.mall_name, + shipping_fee=shipping_fee, + shipping_type=shipping_type, + ) + ) + + return products diff --git a/lps/services/search/esm/__init__.py b/lps/services/search/esm/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/lps/services/search/esm/adapter.py b/lps/services/search/esm/adapter.py new file mode 100644 index 0000000..b07068f --- /dev/null +++ b/lps/services/search/esm/adapter.py @@ -0,0 +1,38 @@ +"""ESM(G마켓·옥션) 검색 어댑터 — 네이버 폴백용(네이버에 해당 몰이 없을 때만 크롤). + +두 사이트는 eBay Korea 백엔드 공유 + '잠시만 기다리십시오…' JS 챌린지(수 초 후 자동 통과)가 +동일해, BrowserSearchAdapter 위에 site 파라미터 하나로 커버한다(카드/셀렉터만 분기). +챌린지는 ready_selector(카드) 가 렌더될 때까지 대기하면 자연히 통과한다. +""" + +from urllib.parse import quote + +from services.search.browser_base import BrowserSearchAdapter +from services.search.contract import NormalizedProduct +from services.search.card_parser import parse_cards +from services.search.esm.selectors import SITES + +# ESM 차단 마커. '잠시만 기다리십시오' 챌린지는 렌더 완료 후에도(parse=0) 남아 있으면 = 이 IP 가 +# 챌린지를 못 푼 것 → 차단으로 간주해 IP 회전 재시도를 유도한다(정상 시엔 카드가 떠 parse>0 이라 여기 안 옴). +_BLOCK_MARKERS = ("errors.edgesuite.net", "You don't have permission to access", "비정상적인 접근", "잠시만 기다리") + + +class EsmAdapter(BrowserSearchAdapter): + block_markers = _BLOCK_MARKERS + min_result_html = 8000 # 챌린지 페이지(~21KB)는 크므로 short 폴백은 명백한 에러만 + ready_timeout_ms = 22000 # '잠시만' 챌린지 통과(~9초) 여유 + block_resources_default = False # 리소스 차단은 ESM 챌린지/렌더를 깨뜨림 + + def __init__(self, site: str, **kwargs): + if site not in SITES: + raise ValueError(f"unknown ESM site: {site}") + self._site = SITES[site] + self.source = site # base __init__ 의 user_data_dir 기본값에 쓰임 + self.ready_selector = self._site.ready_selector + super().__init__(**kwargs) + + def _search_url(self, query: str, limit: int) -> str: + return self._site.search_url.format(q=quote(query)) + + def _parse(self, html: str) -> list[NormalizedProduct]: + return parse_cards(html, self._site.cards) diff --git a/lps/services/search/esm/selectors.py b/lps/services/search/esm/selectors.py new file mode 100644 index 0000000..718af80 --- /dev/null +++ b/lps/services/search/esm/selectors.py @@ -0,0 +1,38 @@ +"""ESM(G마켓·옥션, eBay Korea 공유 백엔드) 검색결과 셀렉터/설정. + +두 사이트는 백엔드를 공유하지만 검색결과 HTML 템플릿은 서로 달라, 카드/이름/가격 셀렉터가 +다르다(일부 가격 클래스만 공유). 사이트별 설정을 외부화한다 — 레이아웃 변경 시 여기만. +""" + +from dataclasses import dataclass + +from services.search.card_parser import CardConfig + + +@dataclass(frozen=True) +class EsmSite: + search_url: str # {q} 자리표시자 + ready_selector: str # 챌린지 통과·렌더 완료 신호 + cards: CardConfig + + +GMARKET = EsmSite( + search_url="https://browse.gmarket.co.kr/search?keyword={q}", + ready_selector="div.box__item-container", + cards=CardConfig( + source="gmarket", mall_name="G마켓", base_url="https://browse.gmarket.co.kr/", + card="div.box__item-container", name="[class*=text__item-title]", price="[class*=box__price-seller]", + ), +) + +AUCTION = EsmSite( + search_url="https://browse.auction.co.kr/search?keyword={q}", + ready_selector="div.itemcard", + cards=CardConfig( + source="auction", mall_name="옥션", base_url="https://browse.auction.co.kr/", + card="div.itemcard", name="span.text--title", price="strong.text__price-seller", + link="a.link--itemcard", + ), +) + +SITES = {"gmarket": GMARKET, "auction": AUCTION} diff --git a/lps/services/search/st11/__init__.py b/lps/services/search/st11/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/lps/services/search/st11/adapter.py b/lps/services/search/st11/adapter.py new file mode 100644 index 0000000..461bd8c --- /dev/null +++ b/lps/services/search/st11/adapter.py @@ -0,0 +1,30 @@ +"""11번가 검색 어댑터 — 네이버 폴백용(네이버에 11번가가 없을 때만 크롤). + +공식 오픈API 는 판매자 계정 필요라 불가 → PC 웹(search.11st.co.kr) 크롤. +챌린지 없이 렌더되므로 ready_selector(카드) 대기만으로 충분하다. +""" + +from urllib.parse import quote + +from services.search.browser_base import BrowserSearchAdapter +from services.search.contract import NormalizedProduct +from services.search.card_parser import parse_cards +from services.search.st11.selectors import SEARCH_URL, READY_SELECTOR, CARDS + +_BLOCK_MARKERS = ("captcha", "robot", "비정상적인 접근", "errors.edgesuite.net") + + +class ElevenStAdapter(BrowserSearchAdapter): + source = "st11" + block_markers = _BLOCK_MARKERS + min_result_html = 20000 # PC 정상 결과는 수백 KB — 차단/미렌더 페이지는 작다(→회전/재시도) + ready_selector = READY_SELECTOR + ready_timeout_ms = 20000 + block_resources_default = False # 리소스 차단 시 goto 가 hang / 렌더 실패 + scroll_steps = 3 # 결과가 지연 로딩 → 스크롤로 트리거 + + def _search_url(self, query: str, limit: int) -> str: + return SEARCH_URL.format(q=quote(query)) + + def _parse(self, html: str) -> list[NormalizedProduct]: + return parse_cards(html, CARDS) diff --git a/lps/services/search/st11/selectors.py b/lps/services/search/st11/selectors.py new file mode 100644 index 0000000..ef92f70 --- /dev/null +++ b/lps/services/search/st11/selectors.py @@ -0,0 +1,20 @@ +"""11번가 PC 검색결과 셀렉터/설정. + +11번가는 판매자 계정이 없어 공식 오픈API 불가 → PC 웹 크롤(모바일 m.11st 는 'robot' 차단). +PC total-search 는 챌린지 없이 렌더된다. +""" + +from services.search.card_parser import CardConfig + +SEARCH_URL = "https://search.11st.co.kr/pc/total-search?kwd={q}" +READY_SELECTOR = "[class*=c-card-item]" + +CARDS = CardConfig( + source="st11", mall_name="11번가", base_url="https://search.11st.co.kr/", + # 카드 토큰 '.c-card-item' (렌더마다 '--list' 접미사 유무가 달라 접미사 매칭은 취약). + # 토큰 매칭이라 서브요소(c-card-item__name 등)는 안 잡힌다. + card=".c-card-item", + name="[class*=c-card-item__name]", + price="[class*=c-card-item__price]", + link="a[href]", +) diff --git a/lps/services/search/util.py b/lps/services/search/util.py index 3df92d1..2d7889f 100644 --- a/lps/services/search/util.py +++ b/lps/services/search/util.py @@ -3,6 +3,10 @@ import re _NUM = re.compile(r"[\d,]+") +_WON = re.compile(r"([\d,]+)\s*원") # '원' 바로 앞 숫자(단위가격/퍼센트 오인 방지) +_NUM3 = re.compile(r"([\d,]{3,})") # 3자리+ 숫자 토큰(가격 후보) +_SHIP_FEE = re.compile(r"배송비\s*([\d,]+)\s*원") +_LABELS = ("상품명", "브랜드명", "판매가", "할인가") # 오픈마켓 카드의 a11y 라벨 프리픽스 def parse_price(value) -> int | None: @@ -16,3 +20,38 @@ def parse_price(value) -> int | None: return None digits = m.group(0).replace(",", "") return int(digits) if digits.isdigit() else None + + +def extract_price(text: str) -> int | None: + """가격 노드 텍스트 → int. '원' 앵커 우선, 없으면 첫 3자리+ 숫자. + ('1%할인가44,540원' → 44540, '44,540' → 44540). 실패 시 None.""" + if not text: + return None + m = _WON.search(text) + if m: + return int(m.group(1).replace(",", "")) + m = _NUM3.search(text) + return int(m.group(1).replace(",", "")) if m else None + + +def clean_name(text: str) -> str: + """오픈마켓 카드 이름에서 a11y 라벨('상품명'/'브랜드명' 등) 프리픽스를 제거.""" + s = (text or "").strip() + for lab in _LABELS: + s = s.replace(lab, " ") + return re.sub(r"\s+", " ", s).strip() + + +def shipping_from_text(text: str, name: str | None = None) -> tuple[int | None, str | None]: + """카드 텍스트에서 (배송비, 배송유형) 추출 — 오픈마켓 공용(로켓 없음). + '무료배송'→(0,'free'), '배송비 X원'→(X,'paid'), 그 외 (None, None). + 상품명에 '무료배송'이 들어간 오탐 방지를 위해 이름 제거 후 매칭.""" + t = text or "" + if name: + t = t.replace(name, " ") + if "무료배송" in t: + return 0, "free" + m = _SHIP_FEE.search(t) + if m: + return int(m.group(1).replace(",", "")), "paid" + return None, None diff --git a/lps/tests/fixtures/auction_search.html b/lps/tests/fixtures/auction_search.html new file mode 100644 index 0000000..6877fe3 --- /dev/null +++ b/lps/tests/fixtures/auction_search.html @@ -0,0 +1,3 @@ + \ No newline at end of file diff --git a/lps/tests/fixtures/gmarket_search.html b/lps/tests/fixtures/gmarket_search.html new file mode 100644 index 0000000..fbb6538 --- /dev/null +++ b/lps/tests/fixtures/gmarket_search.html @@ -0,0 +1,3 @@ + \ No newline at end of file diff --git a/lps/tests/fixtures/st11_search.html b/lps/tests/fixtures/st11_search.html new file mode 100644 index 0000000..d97e47a --- /dev/null +++ b/lps/tests/fixtures/st11_search.html @@ -0,0 +1,3 @@ + \ No newline at end of file diff --git a/lps/tests/test_openmarket_parser.py b/lps/tests/test_openmarket_parser.py new file mode 100644 index 0000000..e2adcc4 --- /dev/null +++ b/lps/tests/test_openmarket_parser.py @@ -0,0 +1,49 @@ +# 오픈마켓(G마켓·옥션·11번가) 크롤 파서 결정론적 단위 테스트(네트워크/브라우저 불필요). +# fixture 는 실제 렌더된 검색결과에서 카드 3개씩 추출한 것. +from pathlib import Path + +import pytest + +from services.search.card_parser import parse_cards +from services.search.esm.selectors import GMARKET, AUCTION +from services.search.st11.selectors import CARDS as ST11 + +FIX = Path(__file__).parent / "fixtures" + +CASES = [ + ("gmarket_search.html", GMARKET.cards, "gmarket", "G마켓"), + ("auction_search.html", AUCTION.cards, "auction", "옥션"), + ("st11_search.html", ST11, "st11", "11번가"), +] + + +@pytest.mark.parametrize("fixture,cfg,source,mall", CASES) +def test_parse_extracts_valid_products(fixture, cfg, source, mall): + items = parse_cards((FIX / fixture).read_text(), cfg) + assert len(items) >= 2, f"{source}: 카드 파싱 실패" + for p in items: + assert p.source == source + assert p.mall_name == mall + assert p.name and len(p.name) > 2 + assert "상품명" not in p.name and "브랜드명" not in p.name # a11y 라벨 제거 확인 + assert p.price >= 100, f"이상 저가: {p.price} ({p.name})" + + +@pytest.mark.parametrize("fixture,cfg,source,mall", CASES) +def test_shipping_type_valid(fixture, cfg, source, mall): + items = parse_cards((FIX / fixture).read_text(), cfg) + for p in items: + assert p.shipping_type in (None, "free", "paid") + if p.shipping_type == "paid": + assert p.shipping_fee and p.shipping_fee > 0 + if p.shipping_type == "free": + assert p.shipping_fee == 0 + + +def test_dedup_by_link(): + # 동일 링크 카드가 반복돼도 1건으로 축약 + card = ('
상품명 물병' + '
판매가5,000원
') + html = f"" + items = parse_cards(html, GMARKET.cards) + assert len(items) == 1 and items[0].price == 5000