feat(lps): 프록시 전송오류 IP회전 + 시작 프리플라이트 + DECODO/컴포넌트별 비용
프록시(DECODO) 포트/IP 사망(407/ERR_TUNNEL/ERR_HTTP_RESPONSE_CODE_FAILURE)이
봇차단과 구분 없이 예외로 튕겨 같은 죽은 포트로 재시도만 하다 DEAD 되던 문제를 고친다.
- browser_base: is_proxy_error(순수함수) + search 루프에서 프록시 전송오류 시 IP 회전 재시도
(max_proxy_retries=2). 봇감지 회전과 통합. uses_proxy 프로퍼티.
- 쿠팡 어댑터를 BrowserSearchAdapter 로 통합 — 중복 machinery 제거, 회전 로직 한 곳에서 공유
(detect_block 순수함수는 유지, 테스트 호환).
- proxy.healthcheck(): 시작 프리플라이트 — 살아있는 포트 선점 + egress IP 로그(빠른 실패·가시성).
worker_main 기동 시 호출.
- 비용: DecodoConfig.cost_per_gb 추가. metrics 에 proxy_bytes(네이버 직접 제외) + 컴포넌트별
cost{ai_usd, proxy_usd, total_usd}. FE 원가 타일에 AI/DECODO 분해·프록시 바이트.
- 테스트: is_proxy_error 8종 + 비용 분해 1종.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
민헌
parent
9c478e80fd
commit
c955a8f367
@ -71,3 +71,4 @@ class DecodoConfig(ConfigModel):
|
||||
port_start: int = 0
|
||||
port_end: int = 0
|
||||
session_minutes: int = 10
|
||||
cost_per_gb: float = 0.0 # DECODO residential 요금($/GB) — 검색 원가의 대역폭 비용 산정용(플랜에 맞게 설정)
|
||||
|
||||
@ -30,14 +30,16 @@ def estimate_cost(model: str, prompt_tokens: int, completion_tokens: int) -> flo
|
||||
class SearchMetrics:
|
||||
"""검색 1건의 누적 계측기. 스레드/코루틴 공유 X — 잡마다 새로 만든다."""
|
||||
|
||||
def __init__(self, model: str = ""):
|
||||
def __init__(self, model: str = "", proxy_cost_per_gb: float = 0.0):
|
||||
self._start = time.monotonic()
|
||||
self._model = model
|
||||
self._proxy_rate = proxy_cost_per_gb # DECODO $/GB
|
||||
self.ai_calls = 0
|
||||
self.ai_prompt = 0
|
||||
self.ai_completion = 0
|
||||
self.fetches = 0
|
||||
self.html_bytes = 0
|
||||
self.proxy_bytes = 0 # 프록시(DECODO) 경유 바이트만 — 네이버(직접) 제외
|
||||
self.source_ms: dict[str, int] = {}
|
||||
self.crawled_malls: list[str] = []
|
||||
|
||||
@ -49,27 +51,39 @@ class SearchMetrics:
|
||||
self.ai_prompt += getattr(usage, "prompt_tokens", 0) or 0
|
||||
self.ai_completion += getattr(usage, "completion_tokens", 0) or 0
|
||||
|
||||
def add_fetch(self, source: str, html_bytes: int, ms: int, crawl: bool = False):
|
||||
"""외부 fetch 1건(소스·바이트·소요) 누적. crawl=True 면 오픈마켓 폴백 크롤."""
|
||||
def add_fetch(self, source: str, html_bytes: int, ms: int, crawl: bool = False, via_proxy: bool = False):
|
||||
"""외부 fetch 1건(소스·바이트·소요) 누적. crawl=폴백 크롤, via_proxy=DECODO 경유(비용 귀속)."""
|
||||
self.fetches += 1
|
||||
self.html_bytes += html_bytes or 0
|
||||
html_bytes = html_bytes or 0
|
||||
self.html_bytes += html_bytes
|
||||
if via_proxy:
|
||||
self.proxy_bytes += html_bytes
|
||||
self.source_ms[source] = self.source_ms.get(source, 0) + ms
|
||||
if crawl and source not in self.crawled_malls:
|
||||
self.crawled_malls.append(source)
|
||||
|
||||
def snapshot(self) -> dict:
|
||||
ai_usd = estimate_cost(self._model, self.ai_prompt, self.ai_completion)
|
||||
proxy_usd = round(self.proxy_bytes / (1024 ** 3) * self._proxy_rate, 6)
|
||||
return {
|
||||
"duration_ms": int((time.monotonic() - self._start) * 1000),
|
||||
"ai": {
|
||||
"calls": self.ai_calls,
|
||||
"prompt_tokens": self.ai_prompt,
|
||||
"completion_tokens": self.ai_completion,
|
||||
"est_cost_usd": estimate_cost(self._model, self.ai_prompt, self.ai_completion),
|
||||
"est_cost_usd": ai_usd,
|
||||
},
|
||||
"crawl": {
|
||||
"fetches": self.fetches,
|
||||
"html_bytes": self.html_bytes,
|
||||
"proxy_bytes": self.proxy_bytes,
|
||||
"malls_crawled": self.crawled_malls,
|
||||
},
|
||||
# 컴포넌트별 비용($) + 총합. proxy=DECODO 대역폭(근사=처리 바이트, 하한).
|
||||
"cost": {
|
||||
"ai_usd": ai_usd,
|
||||
"proxy_usd": proxy_usd,
|
||||
"total_usd": round(ai_usd + proxy_usd, 6),
|
||||
},
|
||||
"source_ms": self.source_ms,
|
||||
}
|
||||
|
||||
@ -38,6 +38,20 @@ def detect_block(html: str, product_count: int, markers: tuple, min_len: int) ->
|
||||
return marker
|
||||
|
||||
|
||||
# 프록시 전송 실패 마커 — 사이트 차단이 아니라 DECODO 포트/IP 사망·세션만료. 봇 감지와 별개로 IP 회전 트리거.
|
||||
_PROXY_ERR_MARKERS = (
|
||||
"ERR_TUNNEL_CONNECTION_FAILED", "ERR_PROXY_CONNECTION_FAILED", "ERR_HTTP_RESPONSE_CODE_FAILURE",
|
||||
"ERR_NO_SUPPORTED_PROXIES", "ERR_SOCKS_CONNECTION_FAILED", "ERR_CONNECTION_CLOSED",
|
||||
"Proxy Authentication", "status code 407", "407 ",
|
||||
)
|
||||
|
||||
|
||||
def is_proxy_error(msg: str) -> bool:
|
||||
"""예외 메시지가 프록시 전송 실패(포트/IP 사망·407)인지(순수 함수, 단위 테스트 가능).
|
||||
True 면 사이트 차단이 아니라 프록시 문제 → 다른 IP 로 회전하면 회복 가능."""
|
||||
return any(m in (msg or "") for m in _PROXY_ERR_MARKERS)
|
||||
|
||||
|
||||
class BrowserSearchAdapter(SearchAdapter):
|
||||
"""patchright(스텔스 Chrome) 기반 검색 어댑터 베이스. persistent context 로 브라우저를 재사용한다."""
|
||||
|
||||
@ -48,6 +62,7 @@ class BrowserSearchAdapter(SearchAdapter):
|
||||
ready_timeout_ms: int = 20000
|
||||
block_resources_default: bool = True # 쿠팡=True(대역폭↓). 오픈마켓은 리소스차단이 렌더/챌린지를 깨 False.
|
||||
scroll_steps: int = 0 # >0 이면 렌더 대기 전 스크롤(지연 로딩 트리거, 예: 11번가)
|
||||
max_proxy_retries: int = 2 # 프록시 전송오류(포트 사망) 시 IP 회전 재시도 횟수
|
||||
|
||||
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):
|
||||
@ -134,10 +149,23 @@ class BrowserSearchAdapter(SearchAdapter):
|
||||
finally:
|
||||
self._ctx = None
|
||||
|
||||
# ---- 검색(차단 감지 + IP 회전 인라인 재시도) ----------------------
|
||||
@property
|
||||
def uses_proxy(self) -> bool:
|
||||
"""이 어댑터가 프록시(DECODO)를 경유하는지 — 대역폭 비용 귀속용."""
|
||||
return bool(self._proxy and self._proxy.enabled)
|
||||
|
||||
def _rotate_ip(self, reason: str):
|
||||
"""즉시 다음 IP(포트)로 회전 예약 + 다음 _ensure_browser 에서 브라우저 재기동."""
|
||||
if self._proxy and self._proxy.enabled:
|
||||
self._proxy.rotate()
|
||||
self._force_recycle = True
|
||||
LOG.w(f"[{self.source}] IP 회전 — {reason}")
|
||||
|
||||
# ---- 검색(프록시 전송오류·봇 감지 → 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):
|
||||
proxy_retries, block_retries = self.max_proxy_retries, self._max_block_retries
|
||||
while True:
|
||||
await self._rl.wait()
|
||||
await self._ensure_browser()
|
||||
self._ip_requests += 1
|
||||
@ -148,6 +176,11 @@ class BrowserSearchAdapter(SearchAdapter):
|
||||
await self._wait_ready(page)
|
||||
html = await page.content()
|
||||
except Exception as ex:
|
||||
# 프록시 전송 실패(포트/IP 사망·407)면 사이트 문제가 아니므로 IP 회전 후 재시도
|
||||
if self.uses_proxy and is_proxy_error(str(ex)) and proxy_retries > 0:
|
||||
proxy_retries -= 1
|
||||
self._rotate_ip(f"프록시 전송오류({type(ex).__name__}) 재시도 {self.max_proxy_retries - proxy_retries}/{self.max_proxy_retries}")
|
||||
continue
|
||||
raise AdapterError(f"{self.source} 검색 실패: {ex}", source=self.source) from ex
|
||||
|
||||
self.last_bytes = len(html.encode("utf-8"))
|
||||
@ -163,11 +196,9 @@ class BrowserSearchAdapter(SearchAdapter):
|
||||
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})")
|
||||
if blocked and self.uses_proxy and block_retries > 0:
|
||||
block_retries -= 1
|
||||
self._rotate_ip(f"봇 감지 재시도 {self._max_block_retries - block_retries}/{self._max_block_retries}")
|
||||
continue
|
||||
|
||||
raise AdapterError(f"{self.source} 결과 없음/차단 (query={query!r}, blocked={blocked})", source=self.source, blocked=blocked)
|
||||
|
||||
@ -1,187 +1,46 @@
|
||||
"""쿠팡 검색 어댑터.
|
||||
|
||||
쿠팡은 Akamai Bot Manager 의 JS 행동 챌린지를 걸어 curl_cffi 단독으로는 통과 못 한다.
|
||||
→ Patchright(스텔스 Playwright) + 실제 Chrome 으로 챌린지를 통과한다.
|
||||
→ Patchright(스텔스 Playwright) + 실제 Chrome 으로 챌린지를 통과한다(BrowserSearchAdapter 공유).
|
||||
persistent context 로 브라우저를 재사용하므로 챌린지는 (쿠키 만료 전까지) 1회만 풀린다.
|
||||
|
||||
향후 최적화(하이브리드): 검증된 Akamai 쿠키를 curl_cffi 로 넘겨 대량 후속 요청을
|
||||
브라우저 없이 처리 가능. 현재는 브라우저 재사용만으로도 후속 검색이 충분히 빠르다.
|
||||
쿠팡 차단은 여러 flavor 다 — Akamai JS 챌린지 / Edge Access Denied / 권한제한 페이지.
|
||||
detect_block 은 순수 함수로 분리해 단위 테스트한다(browser_base.detect_block 에 쿠팡 마커 주입).
|
||||
"""
|
||||
|
||||
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.browser_base import BrowserSearchAdapter, detect_block as _detect_block
|
||||
from services.search.contract import NormalizedProduct
|
||||
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}"
|
||||
# 쿠팡 차단은 여러 flavor 다 — 하나만 잡으면 나머지는 blocked=False 로 오판(→회전·기록 누락).
|
||||
# 1) Akamai JS 챌린지: sec-if-cpt-container / /akam/ / Powered and protected
|
||||
# 2) Edge Access Denied(수백 B): errors.edgesuite.net / You don't have permission to access
|
||||
# 3) 권한 제한 페이지: '사용권한이 제한된' / '쿠팡을 찾아주신 고객님'
|
||||
# 차단 마커 3계열: Akamai JS 챌린지 / Edge Access Denied(수백 B) / 권한제한 페이지.
|
||||
_BLOCK_MARKERS = (
|
||||
"sec-if-cpt-container", "Powered and protected", "/akam/",
|
||||
"errors.edgesuite.net", "You don't have permission to access",
|
||||
"사용권한이 제한된", "쿠팡을 찾아주신 고객님",
|
||||
)
|
||||
# 정상 검색결과·'검색결과 없음' 페이지는 전체 chrome 포함이라 수십 KB+ 다.
|
||||
# 이보다 짧은데 0건이면 미지의 차단/에러 페이지로 간주(마커 없어도 회전 트리거).
|
||||
# 정상 검색결과·'검색결과 없음'은 수십 KB+. 이보다 짧은데 0건이면 미지의 차단으로 간주(회전 트리거).
|
||||
_MIN_RESULT_HTML = 10000
|
||||
# 대역폭 절감: 이미지/미디어/폰트/CSS 는 상품 데이터·Akamai(JS) 에 불필요 → 차단(프록시 per-GB 비용↓).
|
||||
_BLOCKED_RESOURCES = {"image", "media", "font", "stylesheet"}
|
||||
|
||||
|
||||
def detect_block(html: str, product_count: int) -> str | None:
|
||||
"""0건 응답의 차단 여부 판정(순수 함수 — 브라우저 무관, 단위 테스트 가능).
|
||||
반환: 차단 마커 문자열(차단) 또는 None(정상 빈결과). 상품이 있으면 항상 None.
|
||||
알려진 마커 우선, 없으면 비정상적으로 짧은 HTML 을 미지의 차단으로 폴백 처리."""
|
||||
if product_count > 0:
|
||||
return None
|
||||
marker = next((m for m in _BLOCK_MARKERS if m in html), None)
|
||||
if marker is None and len(html) < _MIN_RESULT_HTML:
|
||||
marker = f"short_html({len(html)}B)"
|
||||
return marker
|
||||
"""쿠팡 0건 응답의 차단 여부 판정(순수 함수 — 브라우저 무관, 단위 테스트 가능)."""
|
||||
return _detect_block(html, product_count, _BLOCK_MARKERS, _MIN_RESULT_HTML)
|
||||
|
||||
|
||||
class CoupangAdapter(SearchAdapter):
|
||||
class CoupangAdapter(BrowserSearchAdapter):
|
||||
source = "coupang"
|
||||
block_markers = _BLOCK_MARKERS
|
||||
min_result_html = _MIN_RESULT_HTML
|
||||
ready_selector = SELECTORS.card
|
||||
ready_timeout_ms = 20000
|
||||
block_resources_default = True # 이미지/미디어/폰트/CSS 차단 → 대역폭↓(Akamai·상품데이터엔 불필요)
|
||||
|
||||
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
|
||||
self.last_bytes = 0 # 직전 search 의 처리 HTML 바이트(계측용)
|
||||
def _search_url(self, query: str, limit: int) -> str:
|
||||
return _SEARCH_URL.format(q=quote(query), n=limit)
|
||||
|
||||
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
|
||||
|
||||
self.last_bytes = len(html.encode("utf-8"))
|
||||
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건 — 차단 여부 판정(알려진 마커 + 짧은 HTML 폴백)
|
||||
marker = detect_block(html, len(products))
|
||||
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
|
||||
def _parse(self, html: str) -> list[NormalizedProduct]:
|
||||
return parse_search_html(html, source=self.source)
|
||||
|
||||
@ -35,6 +35,8 @@ class NaverAdapter(SearchAdapter):
|
||||
self._blocked = 0
|
||||
self.last_bytes = 0 # 직전 search 의 응답 바이트(계측용)
|
||||
|
||||
uses_proxy = False # 네이버는 오픈API 직접 호출(프록시 미경유) — DECODO 대역폭 비용 없음
|
||||
|
||||
def _headers(self) -> dict:
|
||||
cid, csec = self._keys[self._idx]
|
||||
return {"X-Naver-Client-Id": cid, "X-Naver-Client-Secret": csec}
|
||||
|
||||
@ -10,7 +10,11 @@ Decodo residential 은 **포트 기반 sticky** 모델이다:
|
||||
"""
|
||||
|
||||
import time
|
||||
from urllib.parse import quote
|
||||
|
||||
import httpx
|
||||
|
||||
from common.logger import LOG
|
||||
from config.server_configs import decodo_config
|
||||
|
||||
|
||||
@ -52,3 +56,25 @@ class DecodoProxy:
|
||||
"username": self.username,
|
||||
"password": self.password,
|
||||
}
|
||||
|
||||
def _proxy_url(self, port: int) -> str:
|
||||
return f"http://{quote(self.username)}:{quote(self.password)}@{self.host}:{port}"
|
||||
|
||||
async def healthcheck(self, timeout: float = 6.0) -> tuple[str | None, int | None]:
|
||||
"""시작 프리플라이트: 현재 포트로 egress IP 확인, 실패하면 회전하며 살아있는 포트를 찾는다.
|
||||
반환: (egress_ip, port) 성공 / (None, None) 전 포트 실패. residential IP 는 실행 중에도
|
||||
죽으므로 이건 '빠른 실패+가시성'용이고, 실제 회복은 런타임 IP 회전이 담당한다."""
|
||||
if not self.enabled:
|
||||
return None, None
|
||||
n = self.port_end - self.port_start + 1
|
||||
for _ in range(n):
|
||||
port = self._port()
|
||||
try:
|
||||
async with httpx.AsyncClient(proxy=self._proxy_url(port), timeout=timeout) as c:
|
||||
r = await c.get("https://ip.decodo.com/ip")
|
||||
if r.status_code == 200:
|
||||
return r.text.strip(), port
|
||||
except Exception as ex:
|
||||
LOG.d(f"[proxy] 포트 {port} 헬스체크 실패: {type(ex).__name__} → 회전")
|
||||
self.rotate()
|
||||
return None, None
|
||||
|
||||
@ -5,6 +5,7 @@ from pathlib import Path
|
||||
import pytest
|
||||
|
||||
from services.search.card_parser import parse_cards
|
||||
from services.search.browser_base import is_proxy_error
|
||||
from services.search.esm.selectors import GMARKET, AUCTION
|
||||
from services.search.st11.selectors import CARDS as ST11
|
||||
|
||||
@ -40,6 +41,27 @@ def test_shipping_type_valid(fixture, cfg, source, mall):
|
||||
assert p.shipping_fee == 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize("msg", [
|
||||
"Page.goto: net::ERR_TUNNEL_CONNECTION_FAILED at https://...",
|
||||
"net::ERR_HTTP_RESPONSE_CODE_FAILURE at https://www.coupang.com/...",
|
||||
"HTTP ERROR 407 Proxy Authentication Required",
|
||||
"net::ERR_PROXY_CONNECTION_FAILED",
|
||||
])
|
||||
def test_is_proxy_error_true(msg):
|
||||
# 프록시 전송 실패(포트/IP 사망·407) → IP 회전 대상
|
||||
assert is_proxy_error(msg) is True
|
||||
|
||||
|
||||
@pytest.mark.parametrize("msg", [
|
||||
"쿠팡 결과 없음/차단 (blocked=True)",
|
||||
"net::ERR_NAME_NOT_RESOLVED", # DNS — 프록시 문제 아님
|
||||
"Timeout 40000ms exceeded", # 단순 타임아웃(사이트 지연)
|
||||
"",
|
||||
])
|
||||
def test_is_proxy_error_false(msg):
|
||||
assert is_proxy_error(msg) is False
|
||||
|
||||
|
||||
def test_dedup_by_link():
|
||||
# 동일 링크 카드가 반복돼도 1건으로 축약
|
||||
card = ('<div class="box__item-container"><a href="/x"><span class="text__item-title">상품명 물병</span></a>'
|
||||
|
||||
@ -8,11 +8,13 @@ from worker.handlers import build_search_handler
|
||||
|
||||
|
||||
class FakeAdapter:
|
||||
def __init__(self, source, by_query=None, products=None, fail=False):
|
||||
def __init__(self, source, by_query=None, products=None, fail=False, uses_proxy=False, last_bytes=0):
|
||||
self.source = source
|
||||
self._by_query = by_query # {query: [products]}
|
||||
self._products = products or []
|
||||
self._fail = fail
|
||||
self.uses_proxy = uses_proxy # DECODO 경유 여부(비용 귀속)
|
||||
self.last_bytes = last_bytes
|
||||
self.calls = []
|
||||
|
||||
async def search(self, query, limit=40):
|
||||
@ -171,6 +173,23 @@ async def test_metrics_recorded_in_result():
|
||||
assert "naver" in m["source_ms"] and "duration_ms" in m
|
||||
|
||||
|
||||
async def test_metrics_cost_split_ai_and_proxy():
|
||||
# 네이버(직접, 프록시X) + 프록시 경유 크롤 폴백 → proxy_usd 는 프록시 바이트만, ai_usd 는 토큰만
|
||||
adapters = {"naver": FakeAdapter("naver", products=[_np("naver", 5000, mall="네이버")], last_bytes=2000)}
|
||||
st11 = FakeAdapter("st11", products=[_np("st11", 3000, mall="11번가")], uses_proxy=True, last_bytes=1024**3) # 1GB
|
||||
judge = FakeJudge(lambda c: True, usage=_Usage(1_000_000, 0)) # 1M prompt 토큰
|
||||
handler = build_search_handler(
|
||||
adapters, judge=judge, ai_model="gpt-4o-mini",
|
||||
fallback_adapters={"st11": st11}, proxy_cost_per_gb=3.0,
|
||||
)
|
||||
m = (await handler(_job()))["metrics"]
|
||||
# 네이버 2000B 는 프록시 경유 아님 → proxy_bytes = 1GB(st11)만
|
||||
assert m["crawl"]["proxy_bytes"] == 1024**3
|
||||
assert m["cost"]["proxy_usd"] == 3.0 # 1GB × $3
|
||||
assert m["cost"]["ai_usd"] == round(m["ai"]["prompt_tokens"]/1e6*0.15, 6) # gpt-4o-mini input 단가
|
||||
assert m["cost"]["total_usd"] == round(m["cost"]["ai_usd"] + 3.0, 6)
|
||||
|
||||
|
||||
async def test_metrics_counts_fallback_crawl():
|
||||
adapters = {"naver": FakeAdapter("naver", products=[_np("naver", 5000, mall="네이버")])}
|
||||
st11 = FakeAdapter("st11", products=[_np("st11", 3000, mall="11번가")])
|
||||
|
||||
@ -55,6 +55,7 @@ def build_search_handler(
|
||||
history=None,
|
||||
fallback_adapters: dict[str, SearchAdapter] | None = None,
|
||||
ai_model: str = "",
|
||||
proxy_cost_per_gb: float = 0.0,
|
||||
):
|
||||
"""검색 핸들러 생성.
|
||||
judge: SimilarityJudge(같은 상품 판정) / keyword_gen: KeywordGenerator(정밀·광역 재검색어) /
|
||||
@ -75,13 +76,14 @@ def build_search_handler(
|
||||
|
||||
async def _timed_search(adapter, query: str, source: str, metrics: SearchMetrics, crawl: bool):
|
||||
"""어댑터 검색 1건을 타이밍+바이트 계측하며 실행. 예외는 그대로 전파(호출부가 처리)."""
|
||||
via_proxy = bool(getattr(adapter, "uses_proxy", False))
|
||||
t0 = time.monotonic()
|
||||
try:
|
||||
res = await adapter.search(query, limit=limit)
|
||||
metrics.add_fetch(source, getattr(adapter, "last_bytes", 0), int((time.monotonic() - t0) * 1000), crawl=crawl)
|
||||
metrics.add_fetch(source, getattr(adapter, "last_bytes", 0), int((time.monotonic() - t0) * 1000), crawl=crawl, via_proxy=via_proxy)
|
||||
return res
|
||||
except Exception:
|
||||
metrics.add_fetch(source, 0, int((time.monotonic() - t0) * 1000), crawl=crawl)
|
||||
metrics.add_fetch(source, 0, int((time.monotonic() - t0) * 1000), crawl=crawl, via_proxy=via_proxy)
|
||||
raise
|
||||
|
||||
async def _search_round(query: str, metrics: SearchMetrics):
|
||||
@ -153,7 +155,7 @@ def build_search_handler(
|
||||
target = {k: payload.get(k, "") for k in ("product_name", "model", "specification", "company")}
|
||||
base_price = parse_price(payload.get("price"))
|
||||
cache_key = payload.get("product_code") or base_query
|
||||
metrics = SearchMetrics(ai_model) # 검색 1건의 리소스/비용/시간 계측
|
||||
metrics = SearchMetrics(ai_model, proxy_cost_per_gb) # 검색 1건의 리소스/비용/시간 계측
|
||||
|
||||
# 0) 네거티브 캐시 — 최근 not_found면 재검색 생략
|
||||
if neg_cache is not None and await neg_cache.is_negative(cache_key):
|
||||
|
||||
@ -9,7 +9,7 @@ import asyncio
|
||||
import os
|
||||
|
||||
from common.logger import LOG
|
||||
from config.server_configs import web_server_config, openai_config
|
||||
from config.server_configs import web_server_config, openai_config, decodo_config
|
||||
from crud.job_crud import JobQueue
|
||||
from crud.negative_cache import NegativeCache
|
||||
from crud.bot_detection import BotDetectionLog
|
||||
@ -34,6 +34,14 @@ async def main(concurrency: int = 1):
|
||||
# DECODO 프록시: .env 에 값 있으면 쿠팡만 sticky+주기적 회전으로 경유(없으면 직접 연결)
|
||||
proxy = DecodoProxy()
|
||||
LOG.i(f"DECODO 프록시: {'ON(sticky ' + str(proxy.session_minutes) + '분 회전)' if proxy.enabled else 'OFF(미설정)'}")
|
||||
# 시작 프리플라이트: 살아있는 프록시 포트를 선점하고 egress IP 를 로그로 남긴다(빠른 실패·가시성).
|
||||
# residential IP 는 실행 중에도 죽으므로 실제 회복은 런타임 IP 회전(전송오류·봇감지)이 담당.
|
||||
if proxy.enabled:
|
||||
egress_ip, egress_port = await proxy.healthcheck()
|
||||
if egress_ip:
|
||||
LOG.i(f"DECODO 프리플라이트 OK — egress IP {egress_ip} (port {egress_port})")
|
||||
else:
|
||||
LOG.w("DECODO 프리플라이트 실패 — 살아있는 포트를 못 찾음(런타임 회전으로 재시도)")
|
||||
# 봇 감지 시: 감지 기록(DB) + 새 IP 로 회전 후 재시도
|
||||
bot_log = BotDetectionLog()
|
||||
adapters = {
|
||||
@ -58,6 +66,7 @@ async def main(concurrency: int = 1):
|
||||
neg_cache=NegativeCache(), history=PriceHistory(),
|
||||
fallback_adapters=fallback_adapters,
|
||||
ai_model=openai_config.model,
|
||||
proxy_cost_per_gb=decodo_config.cost_per_gb,
|
||||
)
|
||||
|
||||
stop = asyncio.Event()
|
||||
|
||||
Loading…
Reference in New Issue
Block a user