diff --git a/lps/config/config_models.py b/lps/config/config_models.py index e7cf59a..a6b279f 100644 --- a/lps/config/config_models.py +++ b/lps/config/config_models.py @@ -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) — 검색 원가의 대역폭 비용 산정용(플랜에 맞게 설정) diff --git a/lps/services/metrics.py b/lps/services/metrics.py index 09dfb4e..f977a9f 100644 --- a/lps/services/metrics.py +++ b/lps/services/metrics.py @@ -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, } diff --git a/lps/services/search/browser_base.py b/lps/services/search/browser_base.py index 17a468f..7a196bd 100644 --- a/lps/services/search/browser_base.py +++ b/lps/services/search/browser_base.py @@ -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) diff --git a/lps/services/search/coupang/adapter.py b/lps/services/search/coupang/adapter.py index 3053334..9afbe67 100644 --- a/lps/services/search/coupang/adapter.py +++ b/lps/services/search/coupang/adapter.py @@ -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) diff --git a/lps/services/search/naver/adapter.py b/lps/services/search/naver/adapter.py index abee4a3..2995c24 100644 --- a/lps/services/search/naver/adapter.py +++ b/lps/services/search/naver/adapter.py @@ -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} diff --git a/lps/services/search/proxy.py b/lps/services/search/proxy.py index dbd7c9f..bbad8f3 100644 --- a/lps/services/search/proxy.py +++ b/lps/services/search/proxy.py @@ -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 diff --git a/lps/tests/test_openmarket_parser.py b/lps/tests/test_openmarket_parser.py index e2adcc4..c8c8e2a 100644 --- a/lps/tests/test_openmarket_parser.py +++ b/lps/tests/test_openmarket_parser.py @@ -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 = ('
상품명 물병' diff --git a/lps/tests/test_search_handler.py b/lps/tests/test_search_handler.py index 2ff668a..dfec092 100644 --- a/lps/tests/test_search_handler.py +++ b/lps/tests/test_search_handler.py @@ -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번가")]) diff --git a/lps/worker/handlers.py b/lps/worker/handlers.py index 13dccee..5973547 100644 --- a/lps/worker/handlers.py +++ b/lps/worker/handlers.py @@ -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): diff --git a/lps/worker_main.py b/lps/worker_main.py index d3e7749..1632aa0 100644 --- a/lps/worker_main.py +++ b/lps/worker_main.py @@ -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()