feat(lps): 네이버 최저가를 모바일 크롤로 복구(오픈API 종료 대체)

shop.json 이 2026-07-31 종료(404 SE05)되고 NCP API HUB 에도 승계되지 않아
가격을 얻을 공식 경로가 사라졌다 → 쿠팡과 같은 스택(patchright+실제 Chrome)으로 크롤 전환.

경로: msearch.shopping.naver.com (PC 는 405/418 로 막힘). 7/9 스파이크 때 모바일은
로그인 리다이렉트였는데 그 사이 열렸다.

통과 조건 3개 — 하나라도 빠지면 WTM 캡차(실측):
- **한국 IP**: 해외 residential 은 즉시 하드차단(2.6KB) → kr.decodo.com 게이트웨이
  ([DecodoConfig].kr_host, DecodoProxy(host=...) 로 주입. 쿠팡은 기존 월드와이드 유지)
- **ko-KR 로케일/시간대**: KR IP + en-US 조합을 봇으로 본다
  (BrowserSearchAdapter.context_options 훅 추가)
- **리소스 차단 금지**: route 를 걸면 즉시 캡차. image/media/font 만 막아도 동일 →
  '무엇을 막느냐'가 아니라 요청 가로채기 자체가 탐지 신호. 대신 검색당 ~3MB(~$0.009)

파서는 '정확한 상품의 최저가'를 기준으로 취사선택한다:
- 광고/슈퍼적립/브랜드블록 카드 제외(멤버십·쿠폰 조건부 가격)
- 쿠폰할인가를 price 로 쓰지 않음(조건부라 실구매가보다 싸게 잡힘)
- 가격비교('최저 N원') 카드는 유지하고 mall_name="네이버"(옛 lprice 와 같은 의미)
- **배송비 확보** — 옛 오픈API 는 필드 자체가 없어 전 소스 None 이었다
- 가격 함정 3종 회귀 테스트: 단위가격(548원)·가격노드 안의 배송비(3,900원)·정상가/할인율

source 는 "naver" 유지 — price_history.naver_lowest·MALL_BY_SOURCE·프론트 그래프 계약이
구현(API→크롤) 교체와 무관하게 살아야 한다.

테스트 12건 추가(축약 픽스처 + 합성 함정) · 전체 182 passed.
This commit is contained in:
민헌 2026-08-05 09:05:38 +09:00
parent 55c18e6a83
commit 23c1c48634
12 changed files with 441 additions and 6 deletions

View File

@ -107,6 +107,7 @@ model = "gpt-4o-mini"
# DECODO residential 프록시 (쿠팡 전용, 포트기반 sticky). 값 다 채우면 활성(비면 프록시 미사용). # DECODO residential 프록시 (쿠팡 전용, 포트기반 sticky). 값 다 채우면 활성(비면 프록시 미사용).
[DecodoConfig] [DecodoConfig]
host = "" # 예: gate.decodo.com host = "" # 예: gate.decodo.com
kr_host = "kr.decodo.com" # 한국 타깃 게이트웨이(네이버용). 비우면 host 사용
username = "" # 대시보드 USERNAME (예: sppd6a3ze3) username = "" # 대시보드 USERNAME (예: sppd6a3ze3)
password = "" # 대시보드 PASSWORD password = "" # 대시보드 PASSWORD
port_start = 0 # 예: 10001 port_start = 0 # 예: 10001

View File

@ -96,6 +96,10 @@ class DecodoConfig(ConfigModel):
"""DECODO residential 프록시(쿠팡 전용, 포트기반 sticky). 값이 다 차야 활성.""" """DECODO residential 프록시(쿠팡 전용, 포트기반 sticky). 값이 다 차야 활성."""
host: str = "" host: str = ""
# 한국 residential 타깃 게이트웨이(kr.decodo.com). 네이버는 해외 IP 를 즉시 하드차단하므로
# 국내 사이트 전용으로 쓴다(실측 2026-08-04: 해외 IP=2.6KB 차단페이지, KR IP=정상).
# 비우면 host 를 그대로 쓴다(=국가 무지정 월드와이드).
kr_host: str = ""
username: str = "" username: str = ""
password: str = "" password: str = ""
port_start: int = 0 port_start: int = 0

View File

@ -71,6 +71,9 @@ class BrowserSearchAdapter(SearchAdapter):
blocked_resource_types: set = _BLOCKED_RESOURCES # 차단할 resource_type(사이트별 override — 오픈마켓은 CSS/JS 유지) blocked_resource_types: set = _BLOCKED_RESOURCES # 차단할 resource_type(사이트별 override — 오픈마켓은 CSS/JS 유지)
scroll_steps: int = 0 # >0 이면 렌더 대기 전 스크롤(지연 로딩 트리거, 예: 11번가) scroll_steps: int = 0 # >0 이면 렌더 대기 전 스크롤(지연 로딩 트리거, 예: 11번가)
max_proxy_retries: int = 2 # 프록시 전송오류(포트 사망) 시 IP 회전 재시도 횟수 max_proxy_retries: int = 2 # 프록시 전송오류(포트 사망) 시 IP 회전 재시도 횟수
# launch_persistent_context 에 얹을 사이트별 옵션. 네이버 WTM 은 **한국 IP + en-US 로케일** 조합을
# 봇으로 본다(실측: 같은 IP·같은 브라우저에서 locale 만 ko-KR 로 주면 캡차→정상). 기본은 비움.
context_options: dict = {}
def __init__(self, headless: bool = False, user_data_dir: str | None = None, rate_limiter: RateLimiter | None = None, 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, proxy=None, block_resources: bool | None = None, on_detect=None, max_block_retries: int = 1,
@ -159,6 +162,7 @@ class BrowserSearchAdapter(SearchAdapter):
if self._pw is None: if self._pw is None:
self._pw = await async_playwright().start() self._pw = await async_playwright().start()
kwargs = dict(user_data_dir=self._user_data_dir, headless=self._headless, no_viewport=True) kwargs = dict(user_data_dir=self._user_data_dir, headless=self._headless, no_viewport=True)
kwargs.update(self.context_options) # 사이트별 컨텍스트 옵션(예: 네이버 locale/timezone)
if _CHROME_EXECUTABLE: if _CHROME_EXECUTABLE:
kwargs["executable_path"] = _CHROME_EXECUTABLE # 컨테이너: 시스템 chromium kwargs["executable_path"] = _CHROME_EXECUTABLE # 컨테이너: 시스템 chromium
# 컨테이너(root)에선 sandbox 불가 → --no-sandbox 필수(없으면 런칭 행). /dev/shm 부족 크래시 방지. # 컨테이너(root)에선 sandbox 불가 → --no-sandbox 필수(없으면 런칭 행). /dev/shm 부족 크래시 방지.

View File

@ -0,0 +1,6 @@
"""네이버 쇼핑 크롤(모바일 msearch) — 종료된 쇼핑 오픈API 의 대체 경로."""
from services.search.naver_shop.adapter import NaverShopAdapter, detect_block
from services.search.naver_shop.parser import parse_search_html
__all__ = ["NaverShopAdapter", "detect_block", "parse_search_html"]

View File

@ -0,0 +1,67 @@
"""네이버 쇼핑 크롤 어댑터 (모바일 msearch).
배경: 2026-07-31 네이버가 쇼핑 검색 오픈API 를 종료(404 SE05)했고 NCP API HUB 에도
승계되지 않았다 → 가격을 얻을 공식 경로가 없어 크롤로 돌아왔다.
경로 선택(2026-08-04 실측):
PC search.shopping.naver.com/search/all 405 + WTM 캡차
PC /api/search/all 418(봇 차단)
모바일 msearch.shopping.naver.com **200 · 카드 40건** ← 이 경로
7/9 스파이크 때 모바일은 로그인 리다이렉트였는데 그 사이 열렸다. 다시 닫힐 수 있으므로
차단 마커를 넉넉히 잡고, 막히면 IP 회전(쿠팡과 동일한 BrowserSearchAdapter machinery)에 맡긴다.
프록시: **한국 IP 필수**(worker_main 이 kr.decodo.com 게이트웨이로 주입). 실측 차이가 크다 —
해외 residential IP 즉시 하드차단(2.6KB 'Access Denied' 계열)
한국 residential IP 정상. 단일 IP 로 연달아 두드리면 캡차로 넘어가므로 회전은 그대로 필요
"""
from urllib.parse import quote
from services.search.browser_base import BrowserSearchAdapter, detect_block as _detect_block
from services.search.contract import NormalizedProduct
from services.search.naver_shop.parser import parse_search_html
from services.search.naver_shop.selectors import SELECTORS
_SEARCH_URL = "https://msearch.shopping.naver.com/search/all?query={q}"
# 차단 flavor: WTM 캡차 페이지 / 접근제한 안내 / 418 teapot 본문.
_BLOCK_MARKERS = (
"wtm_captcha", "비정상적인 접근", "일시적으로 제한", "자동입력 방지",
"정상적인 서비스 이용", "nid.naver.com/nidlogin",
)
# 정상 결과 페이지는 1.3MB+ 다. 차단 페이지는 실측 48~65KB → 그 사이에 임계를 둔다.
# (0건일 때만 적용되므로 '검색결과 없음'이 커도 오탐하지 않는다)
_MIN_RESULT_HTML = 150_000
def detect_block(html: str, product_count: int) -> str | None:
"""0건 응답의 차단 여부 판정(순수 함수 — 브라우저 무관, 단위 테스트 가능)."""
return _detect_block(html, product_count, _BLOCK_MARKERS, _MIN_RESULT_HTML)
class NaverShopAdapter(BrowserSearchAdapter):
# source 는 구현(API/크롤)이 아니라 **도메인 정체성**이다. "naver" 를 유지해야
# price_history 의 naver_lowest/name/url, MALL_BY_SOURCE, 프론트 그래프 계약이 그대로 산다.
source = "naver"
block_markers = _BLOCK_MARKERS
min_result_html = _MIN_RESULT_HTML
ready_selector = SELECTORS.card
ready_timeout_ms = 20000
# 모바일은 무한스크롤 — 초기 20건, 스크롤하면 40건까지 늘고 그 이상은 안 나온다(실측).
scroll_steps = 3
# ⚠️ 리소스 차단을 **켜면 안 된다**. route 를 걸면 WTM 이 즉시 캡차로 넘긴다(실측 2026-08-04):
# 차단 없음 → 정상 14건 · 전송 3.07MB
# image/media/font 만 차단 → 차단 · 0.58MB ← CSS 를 살려도 안 통한다
# image/media/font/css 차단 → 차단
# 부분 차단조차 막히는 걸 보면 '무엇을 막느냐'가 아니라 **요청 가로채기(CDP Fetch) 자체**가
# 탐지 신호다. 그래서 대역폭을 포기하고 끈다 — 검색당 ~3MB(DECODO $3/GB 기준 ~$0.009).
block_resources_default = False
# 한국어 로케일/시간대 필수. 한국 IP 로 들어가면서 브라우저가 en-US 면 WTM 이 캡차를 띄운다
# (실측: 같은 KR 프록시·같은 브라우저에서 이 두 줄 유무로 캡차↔정상이 갈렸다).
context_options = {"locale": "ko-KR", "timezone_id": "Asia/Seoul"}
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_search_html(html, source=self.source)

View File

@ -0,0 +1,138 @@
"""네이버 모바일 쇼핑 카드 → NormalizedProduct (순수 함수, 저장 HTML 로 테스트).
최종 목적은 **정확한 상품의 최저가**다. 그래서 파싱은 '많이 긁기'가 아니라
'가격의 의미가 확실한 것만 남기기' 쪽으로 판단한다.
카드 종류에 따라 가격의 의미가 다르다 — 둘 다 담되 구분한다:
판매처 카드 "21,900원" + 몰명 → 그 판매처의 실제 판매가
가격비교 카드 "최저7,790원" + 판매처 N곳 → 카탈로그 최저가. 몰명이 없어 mall_name="네이버"
(옛 오픈API 의 lprice 와 같은 의미 — 그래서 버리지 않고 유지한다)
버리는 것과 이유:
광고/슈퍼적립/브랜드블록 카드 판매 오퍼가 아니라 노출 상품 — 가격이 멤버십·쿠폰 조건부
쿠폰할인가 조건부(1인 1회·선착순) — price 로 쓰면 실구매가보다 싸게 잡힘
가격 파싱 함정(전부 실측에서 실제로 밟은 것들):
"21,900원(100ml당 548원)" 단위가격을 먼저 떼지 않으면 548 을 집는다
"최저7,790원(1개입당 31원)배송비3,900원" 가격 노드가 배송비까지 감싼다 → 마지막 매치를 쓰면 3,900 이 가격이 된다
"정상가60,000원할인율38%36,840원" 정상가·할인율을 걷어내야 실판매가가 남는다
"최저7,790원" '최저'와 숫자 사이에 공백이 없다(\\s 를 요구하면 못 잡음)
"""
import re
from selectolax.parser import HTMLParser
from common.logger import LOG
from services.search.contract import NormalizedProduct
from services.search.naver_shop.selectors import SELECTORS as S
_NUM = re.compile(r"([\d,]{3,})\s*원")
_LOWEST = re.compile(r"최저\s*[\d]") # "최저7,790원" — 공백 없는 표기가 실제로 나온다
_CATALOG_MALL = "네이버" # 가격비교 카드의 판매처 표기(옛 오픈API 의 mallName 과 동일)
def parse_search_html(html: str, source: str = "naver") -> list[NormalizedProduct]:
"""검색 결과 HTML → 정규화 상품. 파싱 불가/의미 불명 카드는 조용히 버린다."""
tree = HTMLParser(html)
out: list[NormalizedProduct] = []
skipped = {"ad": 0, "no_price": 0, "no_title": 0}
catalog = 0
for card in tree.css(S.card):
cls = (card.attributes.get("class") or "").strip()
# adProduct_item_inner 도 'product_item_inner' 를 포함하므로 prefix 로 유기 카드만 남긴다
if not cls.startswith(S.card_class_prefix):
skipped["ad"] += 1
continue
if card.css_first(S.ad_marker) is not None:
skipped["ad"] += 1
continue
title = _text(card, S.title)
if not title:
skipped["no_title"] += 1
continue
price, price_raw, fee, ship_type = _prices(card)
if not price:
skipped["no_price"] += 1
continue
is_catalog = bool(_LOWEST.search(price_raw))
if is_catalog:
catalog += 1
out.append(NormalizedProduct(
source=source,
name=title,
price=price,
mall_name=_text(card, S.mall) or (_CATALOG_MALL if is_catalog else None),
detail_url=_href(card),
shipping_fee=fee,
shipping_type=ship_type,
))
if out or any(skipped.values()):
LOG.d(f"[naver_shop] 파싱 {len(out)}건(가격비교 {catalog}) · 제외 {skipped}")
return out
def _text(card, sel: str) -> str:
node = card.css_first(sel)
return re.sub(r"\s+", " ", node.text()).strip() if node else ""
def _prices(card) -> tuple[int | None, str, int | None, str | None]:
"""(상품가, 가격원문, 배송비, 배송유형)을 한 번에 뽑는다.
순서가 중요해서 한 함수에 가뒀다: 가격 노드에서 '가격이 아닌 금액'(단위가격·배송비·쿠폰가·
정상가)을 decompose 로 떼어내는데, decompose 는 트리에서 노드를 아예 지우므로
**배송비를 먼저 읽어야** 한다. 밖에서 호출 순서를 지키게 하면 언젠가 반드시 깨진다.
"""
fee, ship_type = _shipping(card)
node = card.css_first(S.price)
if node is None:
return None, "", fee, ship_type
for sel in (S.unit_price, S.delivery_fee, S.coupon_price, S.org_price):
for junk in node.css(sel):
junk.decompose()
raw = re.sub(r"\s+", " ", node.text()).strip()
return _to_won(raw), raw, fee, ship_type
def _to_won(text: str) -> int | None:
"""'할인율38% 36,840원' → 36840. '원' 앞 숫자에 앵커링하고 **첫** 매치를 쓴다.
첫 매치인 이유: 노드에 금액이 여러 개 남았다면 뒤쪽은 배송비·적립처럼 상품가가 아닌 것들이다
(할인율 '38%' 는 '원'이 안 붙어 애초에 매치되지 않는다).
"""
matches = _NUM.findall(text)
if not matches:
return None
try:
price = int(matches[0].replace(",", ""))
except ValueError:
return None
return price if price > 0 else None
def _shipping(card) -> tuple[int | None, str | None]:
"""'배송비무료' → (0, 'free') · '배송비3,000원' → (3000, 'paid') · 없으면 (None, None).
옛 오픈API 는 배송비 필드 자체가 없어 전 소스 None 이었다(lprice=배송비 제외 상품가).
크롤은 화면 그대로라 배송비를 얻는다 — 최저가 비교의 정확도가 올라가는 지점.
"""
text = _text(card, S.delivery_fee)
if not text:
return None, None
if "무료" in text:
return 0, "free"
won = _to_won(text)
return (won, "paid") if won else (None, None)
def _href(card) -> str | None:
node = card.css_first(S.link)
href = node.attributes.get("href") if node else None
return href or None

View File

@ -0,0 +1,37 @@
"""네이버 모바일 쇼핑(msearch) 카드 셀렉터 — 2026-08-04 실측 기준.
클래스는 webpack 해시(`product_price__O3ZGH`)라 **prefix 매칭**으로만 잡는다(쿠팡과 동일 전략).
해시가 바뀌어도 prefix 는 유지되므로 릴리스마다 깨지지 않는다.
카드 계열(같은 페이지에 5종이 섞여 있다):
product_item_inner 유기 검색결과 ← 우리가 쓰는 것
adProduct_item_inner 광고(‘광고’ 배지) ← 제외
superSavingProduct_item_ 슈퍼적립 프로모션 ← 제외(멤버십 조건부 가격)
productList_product_item 브랜드/기획 블록 ← 제외(브랜드가 고른 상품)
product_items 연관상품 추천 ← 가격 없음
"""
from dataclasses import dataclass
@dataclass(frozen=True)
class Selectors:
# 카드 — 유기 결과만. startswith 로 adProduct_item_inner 와 구분한다(둘 다 'product_item_inner' 를 포함).
card: str = "[class*=product_item_inner]"
card_class_prefix: str = "product_item_inner"
title: str = "[class*=product_info_tit]"
# 즉시 판매가. 내부에 단위가격(product_unit_price)이 끼어 있어 그대로 파싱하면 안 된다.
price: str = "[class*=product_immediate_price]"
unit_price: str = "[class*=product_unit_price]" # "(100ml당 548원)" — 가격 아님, 제거 대상
org_price: str = "[class*=product_org_price]" # "정상가 60,000원" — 가격 아님
coupon_price: str = "[class*=product_discount_price]" # 쿠폰할인가(조건부) — price 로 쓰지 않는다
delivery_fee: str = "[class*=product_delivery_fee]" # "배송비무료" | "배송비3,000원"
mall: str = "[class*=product_mall]"
link: str = "[class*=product_btn_link]"
# 광고 배지. adProduct 계열에만 존재하지만, 유기 카드에 광고가 섞이는 경우까지 방어한다.
ad_marker: str = "[class*=_link_ad]"
SELECTORS = Selectors()

View File

@ -19,9 +19,10 @@ from config.server_configs import decodo_config
class DecodoProxy: class DecodoProxy:
def __init__(self, cfg=None): def __init__(self, cfg=None, host: str | None = None):
cfg = cfg if cfg is not None else decodo_config cfg = cfg if cfg is not None else decodo_config
self.host = cfg.host # host 를 넘기면 그 게이트웨이를 쓴다 — 국가 타깃(kr.decodo.com)용. 포트/자격증명은 동일.
self.host = host or cfg.host
self.username = cfg.username self.username = cfg.username
self.password = cfg.password self.password = cfg.password
self.port_start = cfg.port_start self.port_start = cfg.port_start

22
lps/tests/fixtures/naver_msearch.html vendored Normal file

File diff suppressed because one or more lines are too long

View File

@ -126,10 +126,11 @@ def _live_adapters():
from services.search.coupang.adapter import CoupangAdapter from services.search.coupang.adapter import CoupangAdapter
from services.search.esm.adapter import EsmAdapter from services.search.esm.adapter import EsmAdapter
from services.search.st11.adapter import ElevenStAdapter from services.search.st11.adapter import ElevenStAdapter
from services.search.naver.adapter import NaverAdapter from services.search.naver_shop.adapter import NaverShopAdapter
p = DecodoProxy() p = DecodoProxy()
return [ return [
("naver", NaverAdapter()), # 네이버는 오픈API 종료(2026-07-31)로 크롤 전환. headless 는 WTM 에 걸리므로 headful.
("naver", NaverShopAdapter(headless=False)),
("coupang", CoupangAdapter(headless=True, proxy=p)), ("coupang", CoupangAdapter(headless=True, proxy=p)),
("gmarket", EsmAdapter("gmarket", headless=True, proxy=p)), ("gmarket", EsmAdapter("gmarket", headless=True, proxy=p)),
("auction", EsmAdapter("auction", headless=True, proxy=p)), ("auction", EsmAdapter("auction", headless=True, proxy=p)),

View File

@ -0,0 +1,145 @@
"""네이버 모바일 쇼핑 파서 테스트 (저장 HTML — 네트워크 불필요).
fixtures/naver_msearch.html 은 2026-08-04 실측 응답에서 카드만 추린 축약본이다
(판매처 6 · 가격비교 6 · 광고 3 · 슈퍼적립 1 · 브랜드블록 2).
가격 함정 회귀 테스트를 합성 HTML 로 따로 둔다 — 픽스처는 네이버가 마크업을 바꾸면
같이 늙지만, 함정 자체(단위가격·배송비·정상가)는 계속 지켜야 하는 계약이라서다.
"""
from pathlib import Path
from services.search.naver_shop.adapter import detect_block
from services.search.naver_shop.parser import parse_search_html
_FIXTURE = Path(__file__).parent / "fixtures" / "naver_msearch.html"
def _fixture() -> str:
return _FIXTURE.read_text()
def _card(inner: str, cls: str = "product_item_inner__9nwv0") -> str:
return f'<html><body><div class="{cls}">{inner}</div></body></html>'
# ── 픽스처 기반 ──────────────────────────────────────────────────────────
def test_parses_organic_cards_and_drops_ads():
ps = parse_search_html(_fixture())
assert len(ps) == 12 # 광고 3 + 슈퍼적립 1 은 제외됨
assert all(p.source == "naver" for p in ps)
assert all(p.price > 0 and p.name for p in ps)
def test_seller_and_catalog_cards_are_both_kept():
ps = parse_search_html(_fixture())
catalog = [p for p in ps if p.mall_name == "네이버"] # "최저 N원" = 가격비교
seller = [p for p in ps if p.mall_name and p.mall_name != "네이버"]
assert len(catalog) == 6 and len(seller) == 6
# 판매처 카드는 실제 몰명이 들어온다(빈 문자열/None 아님)
assert all(len(p.mall_name) > 1 for p in seller)
def test_shipping_is_captured():
"""옛 오픈API 는 배송비 필드가 없어 전 소스 None 이었다 — 크롤의 이점이라 회귀로 막는다."""
ps = parse_search_html(_fixture())
assert all(p.shipping_fee is not None for p in ps)
assert {p.shipping_type for p in ps} <= {"free", "paid"}
assert any(p.shipping_type == "free" and p.shipping_fee == 0 for p in ps)
assert any(p.shipping_type == "paid" and p.shipping_fee > 0 for p in ps)
def test_price_is_never_the_shipping_fee():
"""실측에서 실제로 밟은 버그: 가격 노드가 배송비까지 감싸 3,900원이 상품가로 잡혔다."""
ps = parse_search_html(_fixture())
assert not [p for p in ps if p.shipping_fee and p.price == p.shipping_fee]
def test_detail_url_present():
ps = parse_search_html(_fixture())
assert all(p.detail_url and p.detail_url.startswith("http") for p in ps)
# ── 가격 함정 회귀 (합성) ────────────────────────────────────────────────
def test_unit_price_is_not_mistaken_for_price():
"""'21,900원(100ml당 548원)' → 21900. 단위가격을 먼저 떼지 않으면 548 을 집는다."""
html = _card("""
<div class="product_info_tit__UOCqq">테스트 섬유유연제 4L</div>
<span class="product_price__O3ZGH">
<span class="product_immediate_price__m3YQU">21,900<span class="product_unit__seYZk">원</span>
<span class="product_unit_price__uLsbL">(100ml당 548원)</span>
</span>
</span>
<span class="product_mall__gUvbk">테스트몰</span>
<a class="product_btn_link__AhZaM" href="https://x.test/1">상품 바로가기</a>""")
(p,) = parse_search_html(html)
assert p.price == 21900
def test_delivery_fee_inside_price_node_is_excluded():
"""가격비교 카드 구조: 가격 노드가 배송비를 품는다. 상품가 7,790 / 배송비 3,900 로 갈려야 한다."""
html = _card("""
<div class="product_info_tit__UOCqq">커클랜드 드라이시트, 250매, 1개</div>
<span class="product_price__O3ZGH">
<span class="product_immediate_price__m3YQU">최저7,790<span class="product_unit__seYZk">원</span>
<span class="product_unit_price__uLsbL">(1개입당 31원)</span>
<span class="product_delivery_fee__agcRS">배송비3,900원</span>
</span>
</span>
<a class="product_btn_link__AhZaM" href="https://x.test/2">상품 바로가기</a>""")
(p,) = parse_search_html(html)
assert p.price == 7790
assert p.shipping_fee == 3900 and p.shipping_type == "paid"
assert p.mall_name == "네이버" # 판매처 표기가 없는 가격비교 카드
def test_list_price_and_discount_rate_are_ignored():
"""'정상가60,000원 할인율38% 36,840원' → 36840."""
html = _card("""
<div class="product_info_tit__UOCqq">다우니 미스티크 1L 6개</div>
<span class="product_price__O3ZGH">
<del class="product_org_price__Rc5Dx">정상가60,000원</del>
<span class="product_immediate_price__m3YQU">
<em class="product_rate__41yIl">38%</em>36,840<span class="product_unit__seYZk">원</span>
</span>
</span>
<span class="product_mall__gUvbk">공식몰</span>
<a class="product_btn_link__AhZaM" href="https://x.test/3">상품 바로가기</a>""")
(p,) = parse_search_html(html)
assert p.price == 36840
def test_coupon_price_is_not_used_as_price():
"""쿠폰할인가는 조건부(1인 1회·선착순)라 실구매가 하한으로 쓰면 최저가가 왜곡된다."""
html = _card("""
<div class="product_info_tit__UOCqq">테스트 상품</div>
<span class="product_price__O3ZGH">
<span class="product_immediate_price__m3YQU">21,900<span class="product_unit__seYZk">원</span></span>
<span class="product_discount_price__lL8iq">21,680원</span>
<span class="product_discount_title__arcCm">쿠폰할인가</span>
</span>
<span class="product_mall__gUvbk">테스트몰</span>
<a class="product_btn_link__AhZaM" href="https://x.test/4">상품 바로가기</a>""")
(p,) = parse_search_html(html)
assert p.price == 21900
def test_ad_card_family_is_dropped():
html = _card("""
<div class="adProduct_info_tit__x">광고 상품</div>
<span class="adProduct_price__DN4Al"><span class="adProduct_immediate_price__q">9,900원</span></span>
<a class="adProduct_link_ad__SqeqU">광고</a>""", cls="adProduct_item_inner__abc")
assert parse_search_html(html) == []
# ── 차단 판정 ────────────────────────────────────────────────────────────
def test_detect_block_flags_captcha_and_short_html():
assert detect_block("<html>wtm_captcha</html>", 0) == "wtm_captcha"
assert detect_block("<html>비정상적인 접근입니다</html>", 0) == "비정상적인 접근"
assert detect_block("<html>" + "x" * 5000 + "</html>", 0).startswith("short_html")
# 결과가 있으면 차단이 아니다 — 짧아도 통과
assert detect_block("<html>x</html>", 3) is None
def test_detect_block_passes_real_page():
assert detect_block(_fixture(), 12) is None

View File

@ -21,7 +21,7 @@ from crud.ip_session import IpSessionLog
from crud.price_history import PriceHistory from crud.price_history import PriceHistory
from services.search.proxy import DecodoProxy from services.search.proxy import DecodoProxy
from services.search.coupang.adapter import CoupangAdapter from services.search.coupang.adapter import CoupangAdapter
from services.search.naver.adapter import NaverAdapter from services.search.naver_shop.adapter import NaverShopAdapter
from services.search.esm.adapter import EsmAdapter from services.search.esm.adapter import EsmAdapter
from services.search.st11.adapter import ElevenStAdapter from services.search.st11.adapter import ElevenStAdapter
from services.ai.similarity import SimilarityJudge from services.ai.similarity import SimilarityJudge
@ -53,9 +53,14 @@ def _build_worker(i: int, concurrency: int, has_openai: bool, neg_cache, history
프로필 분리(user_data_dir_w{i}) + 워커별 다른 프록시 포트(=다른 IP)로 진짜 병렬을 보장한다.""" 프로필 분리(user_data_dir_w{i}) + 워커별 다른 프록시 포트(=다른 IP)로 진짜 병렬을 보장한다."""
# 워커별 프록시(다른 포트=다른 IP). 100포트를 워커 수로 균등 분할해 시작점을 벌린다. # 워커별 프록시(다른 포트=다른 IP). 100포트를 워커 수로 균등 분할해 시작점을 벌린다.
proxy = DecodoProxy() proxy = DecodoProxy()
# 네이버 전용 프록시 — 같은 계정·같은 포트지만 **한국 타깃 게이트웨이**를 쓴다.
# 해외 residential IP 로는 msearch 가 즉시 하드차단된다(2026-08-04 실측). kr_host 가 비면
# 기본 게이트웨이로 떨어지고, 그때는 네이버가 막힐 수 있다(로그의 BOT-DETECTED 로 드러남).
kr_proxy = DecodoProxy(host=decodo_config.kr_host or None)
if proxy.enabled and concurrency > 1: if proxy.enabled and concurrency > 1:
n = proxy.port_end - proxy.port_start + 1 n = proxy.port_end - proxy.port_start + 1
proxy.seed_offset(i * max(1, n // concurrency)) proxy.seed_offset(i * max(1, n // concurrency))
kr_proxy.seed_offset(i * max(1, n // concurrency))
bot_log = BotDetectionLog() bot_log = BotDetectionLog()
ip_log = IpSessionLog() ip_log = IpSessionLog()
suffix = f"_w{i}" if concurrency > 1 else "" suffix = f"_w{i}" if concurrency > 1 else ""
@ -67,7 +72,11 @@ def _build_worker(i: int, concurrency: int, has_openai: bool, neg_cache, history
adapters = { adapters = {
"coupang": CoupangAdapter(headless=False, user_data_dir=_pf("coupang"), proxy=proxy, "coupang": CoupangAdapter(headless=False, user_data_dir=_pf("coupang"), proxy=proxy,
on_detect=bot_log.record, on_session_end=ip_log.record), on_detect=bot_log.record, on_session_end=ip_log.record),
"naver": NaverAdapter(), # httpx 직접(프록시 미경유) — 워커별 인스턴스(last_bytes 경합 회피) # 네이버는 오픈API(shop.json)가 2026-07-31 종료돼 404 SE05 만 돌아온다 → 모바일 크롤로 전환.
# 쿠팡과 같은 프록시를 물린다: 단일 IP 로 연달아 두드리면 WTM 캡차로 넘어간다(실측 —
# 스파이크에서 십수 회 만에 홈 IP 가 탔다). 회전·예산·차단기록은 베이스가 처리한다.
"naver": NaverShopAdapter(headless=False, user_data_dir=_pf("naver"), proxy=kr_proxy,
on_detect=bot_log.record, on_session_end=ip_log.record),
} }
# 폴백은 기본 비활성(LPS_FALLBACKS 로 켬 — 상단 주석 참고). 켤 땐 데드라인이 상한이라 # 폴백은 기본 비활성(LPS_FALLBACKS 로 켬 — 상단 주석 참고). 켤 땐 데드라인이 상한이라
# 봇감지 재시도(챌린지 대기 2배)를 끈다(max_block_retries=0) — 빠르게 포기·스킵. # 봇감지 재시도(챌린지 대기 2배)를 끈다(max_block_retries=0) — 빠르게 포기·스킵.