From 9557c1b1af72785c135ea77cf13818b4cadc7aa4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EB=AF=BC=ED=97=8C?= Date: Thu, 9 Jul 2026 08:38:15 +0900 Subject: [PATCH] =?UTF-8?q?feat(lps):=20=EB=84=A4=EC=9D=B4=EB=B2=84=20?= =?UTF-8?q?=EC=96=B4=EB=8C=91=ED=84=B0=20+=20=EB=8B=A4=EC=A4=91=20?= =?UTF-8?q?=EC=86=8C=EC=8A=A4=20=EB=B3=91=ED=95=A9=20=EA=B2=80=EC=83=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 네이버 쇼핑 오픈API 어댑터(크롤링 불필요) + 핸들러를 다중 소스로 확장. 쿠팡(브라우저)+네이버(API)를 동시 검색·병합해 교차 최저가를 뽑는다. - search/naver/adapter: httpx + 오픈API + 키 로테이션(429/403 순환), .env 키 로드 - search/naver/transform: 순수 변환(태그/엔티티 정리, lprice). 가격비교(catalog) lprice 는 '여러 판매자 중 최저가'라 최저가 솔루션엔 핵심 → 유지 - handler: asyncio.gather 동시 검색 + 소스별 실패 격리(일부 죽어도 결과) + 전체 실패 시 잡 실패 - worker_main: adapters={coupang, naver} - tests: 네이버 변환 + 병합/실패격리/전체실패 4건 → 전체 31/31 - 라이브: 쿠팡30+네이버30 병합 top-N 최저가(소스 라벨 포함) Co-Authored-By: Claude Opus 4.8 (1M context) --- lps/services/search/naver/adapter.py | 104 +++++++++++++++++++++++++ lps/services/search/naver/transform.py | 49 ++++++++++++ lps/tests/test_naver_transform.py | 19 +++++ lps/tests/test_search_handler.py | 58 ++++++++++++++ lps/worker/handlers.py | 40 +++++++--- lps/worker_main.py | 4 +- 6 files changed, 262 insertions(+), 12 deletions(-) create mode 100644 lps/services/search/naver/adapter.py create mode 100644 lps/services/search/naver/transform.py create mode 100644 lps/tests/test_naver_transform.py create mode 100644 lps/tests/test_search_handler.py diff --git a/lps/services/search/naver/adapter.py b/lps/services/search/naver/adapter.py new file mode 100644 index 0000000..2c61ddf --- /dev/null +++ b/lps/services/search/naver/adapter.py @@ -0,0 +1,104 @@ +"""네이버 쇼핑 검색 어댑터. + +공식 오픈 API(https://openapi.naver.com/v1/search/shop.json)라 크롤링/브라우저 불필요. +레퍼런스의 핵심 자산인 **키 로테이션**을 이식: 429/403(쿼터/차단) 시 다음 키로 순환 재시도. +키는 .env(NAVER_CLIENT_ID/SECRET, +_2.._10)에서 로드 → server_configs 가 load_dotenv 로 주입. +""" + +import os + +import httpx + +from common.logger import LOG +from services.search.contract import SearchAdapter, NormalizedProduct, AdapterError, AdapterHealth +from services.search.rate_limiter import RateLimiter +from services.search.naver.transform import transform_items + +_API = "https://openapi.naver.com/v1/search/shop.json" +_MAX_START = 1000 # 네이버 start 상한 +_MAX_DISPLAY = 100 + + +def load_naver_keys() -> list[tuple[str, str]]: + """(client_id, client_secret) 쌍 목록. 기본(무접미) + _2.._10 로테이션 키.""" + keys: list[tuple[str, str]] = [] + cid, csec = os.environ.get("NAVER_CLIENT_ID"), os.environ.get("NAVER_CLIENT_SECRET") + if cid and csec: + keys.append((cid, csec)) + for i in range(2, 11): + cid, csec = os.environ.get(f"NAVER_CLIENT_ID_{i}"), os.environ.get(f"NAVER_CLIENT_SECRET_{i}") + if cid and csec: + keys.append((cid, csec)) + return keys + + +class NaverAdapter(SearchAdapter): + source = "naver" + + def __init__(self, keys: list[tuple[str, str]] | None = None, rate_limiter: RateLimiter | None = None, timeout: float = 10.0): + self._keys = keys if keys is not None else load_naver_keys() + self._idx = 0 + self._rl = rate_limiter or RateLimiter(0.1, 0.3) # 공식 API — 짧은 간격 + self._timeout = timeout + self._ok = 0 + self._blocked = 0 + + def _headers(self) -> dict: + cid, csec = self._keys[self._idx] + return {"X-Naver-Client-Id": cid, "X-Naver-Client-Secret": csec} + + def _rotate(self): + self._idx = (self._idx + 1) % len(self._keys) + + async def search(self, query: str, limit: int = 40) -> list[NormalizedProduct]: + if not self._keys: + raise AdapterError("네이버 API 키 없음(.env NAVER_CLIENT_ID/SECRET)", source=self.source) + + collected: list[dict] = [] + async with httpx.AsyncClient(timeout=self._timeout) as client: + start = 1 + while len(collected) < limit and start <= _MAX_START: + display = min(_MAX_DISPLAY, limit - len(collected)) + data = await self._request(client, { + "query": query, "display": display, "start": start, + "sort": "sim", "exclude": "used:rental:cbshop", + }) + items = data.get("items", []) + if not items: + break + collected.extend(items) + start += display + if len(items) < display: + break + + products = transform_items(collected, self.source) + self._ok += 1 + LOG.d(f"[naver] query={query!r} → {len(products)}건 (limit {limit})") + return products[:limit] + + async def _request(self, client: httpx.AsyncClient, params: dict) -> dict: + """키 개수만큼 재시도. 429/403 이면 다음 키로 로테이션.""" + last_status = None + for _ in range(max(1, len(self._keys))): + await self._rl.wait() + r = await client.get(_API, params=params, headers=self._headers()) + if r.status_code == 200: + return r.json() + if r.status_code in (429, 403): + self._blocked += 1 + last_status = r.status_code + LOG.w(f"[naver] {r.status_code} → 키 로테이션(idx {self._idx})") + self._rotate() + continue + raise AdapterError(f"네이버 API 오류 {r.status_code}: {r.text[:200]}", source=self.source) + raise AdapterError(f"네이버 API 쿼터/차단(모든 키 소진, last={last_status})", source=self.source, blocked=True) + + 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): + """httpx 클라이언트를 search 마다 생성/정리하므로 별도 정리 불필요(수명주기 통일용 no-op).""" + return diff --git a/lps/services/search/naver/transform.py b/lps/services/search/naver/transform.py new file mode 100644 index 0000000..e54b325 --- /dev/null +++ b/lps/services/search/naver/transform.py @@ -0,0 +1,49 @@ +"""네이버 쇼핑 API 응답 items → NormalizedProduct (순수 함수). + +네트워크 무관 — 저장된/모의 JSON 으로 결정론적 테스트 가능. +title 의 강조 태그·HTML 엔티티를 제거하고, 가격비교(catalog) 페이지는 제외한다. +""" + +import html +import re + +from services.search.contract import NormalizedProduct + +_TAG = re.compile(r"<[^>]+>") + + +def _clean(text: str) -> str: + return html.unescape(_TAG.sub("", text or "")).strip() + + +def transform_items(items: list[dict], source: str = "naver") -> list[NormalizedProduct]: + products: list[NormalizedProduct] = [] + for it in items: + link = it.get("link", "") or "" + # 가격비교(catalog, productType=1) 페이지의 lprice 는 '여러 판매자 중 최저가'라 + # 최저가 솔루션에는 오히려 핵심 신호 → 제외하지 않고 그대로 취한다. + + try: + price = int(it.get("lprice")) # lprice = 최저가 + except (TypeError, ValueError): + continue + if price <= 0: + continue + + name = _clean(it.get("title")) + if not name: + continue + + products.append( + NormalizedProduct( + source=source, + name=name, + price=price, + image_url=it.get("image") or None, + detail_url=link or None, + mall_name=it.get("mallName") or None, + manufacturer=(it.get("maker") or it.get("brand")) or None, + external_id=str(it["productId"]) if it.get("productId") else None, + ) + ) + return products diff --git a/lps/tests/test_naver_transform.py b/lps/tests/test_naver_transform.py new file mode 100644 index 0000000..5d49322 --- /dev/null +++ b/lps/tests/test_naver_transform.py @@ -0,0 +1,19 @@ +"""네이버 응답 변환 테스트 (순수, 네트워크 불필요).""" + +from services.search.naver.transform import transform_items + + +def test_transform_strips_tags_and_keeps_catalog(): + items = [ + {"title": "로지텍 무선 마우스 & 키보드", "link": "https://search.shopping.naver.com/catalog/1", + "image": "img1", "lprice": "13500", "mallName": "네이버", "maker": "로지텍", "productId": "1"}, + {"title": "0원상품", "link": "https://x", "lprice": "0"}, # 가격 0 → 제외 + {"title": "가격없음", "link": "https://y", "lprice": "abc"}, # 파싱 실패 → 제외 + ] + out = transform_items(items) + assert len(out) == 1 + p = out[0] + assert p.name == "로지텍 무선 마우스 & 키보드" # 제거 + 엔티티 복원 + assert p.price == 13500 and p.source == "naver" + assert p.mall_name == "네이버" and p.manufacturer == "로지텍" and p.external_id == "1" + assert "catalog/1" in p.detail_url # 가격비교(catalog) 최저가 유지 diff --git a/lps/tests/test_search_handler.py b/lps/tests/test_search_handler.py new file mode 100644 index 0000000..a538550 --- /dev/null +++ b/lps/tests/test_search_handler.py @@ -0,0 +1,58 @@ +"""다중 소스 검색 핸들러 테스트 — 병합·소스별 실패 격리·전체 실패 시 잡 실패 (fake 어댑터).""" + +import pytest + +from common.enums import JobType +from services.search.contract import NormalizedProduct, AdapterError +from worker.handlers import build_search_handler + + +class FakeAdapter: + def __init__(self, source, products=None, fail=False): + self.source = source + self._products = products or [] + self._fail = fail + + async def search(self, query, limit=40): + if self._fail: + raise AdapterError("boom", source=self.source, blocked=True) + return self._products + + +def _np(source, price): + return NormalizedProduct(source=source, name=f"{source}-{price}", price=price) + + +def _job(): + return {"job_type": JobType.SEARCH.value, "attempts": 1, "payload": {"product_name": "x"}} + + +async def test_merges_and_ranks_across_sources(): + adapters = { + "coupang": FakeAdapter("coupang", [_np("coupang", 3000), _np("coupang", 1000)]), + "naver": FakeAdapter("naver", [_np("naver", 2000), _np("naver", 500)]), + } + r = await build_search_handler(adapters, top_n=3)(_job()) + assert r["lowest"]["price"] == 500 and r["lowest"]["source"] == "naver" + assert [p["price"] for p in r["top"]] == [500, 1000, 2000] + assert r["sources"]["coupang"]["count"] == 2 and r["sources"]["naver"]["count"] == 2 + + +async def test_isolates_single_source_failure(): + adapters = { + "coupang": FakeAdapter("coupang", fail=True), + "naver": FakeAdapter("naver", [_np("naver", 900)]), + } + r = await build_search_handler(adapters)(_job()) + assert "error" in r["sources"]["coupang"] # 실패 격리 + assert r["sources"]["naver"]["count"] == 1 + assert r["lowest"]["price"] == 900 # 성공 소스로 결과 산출 + + +async def test_all_sources_fail_raises(): + adapters = { + "coupang": FakeAdapter("coupang", fail=True), + "naver": FakeAdapter("naver", fail=True), + } + with pytest.raises(RuntimeError): + await build_search_handler(adapters)(_job()) diff --git a/lps/worker/handlers.py b/lps/worker/handlers.py index b80275f..94bdda0 100644 --- a/lps/worker/handlers.py +++ b/lps/worker/handlers.py @@ -1,17 +1,22 @@ """잡 핸들러 — job_type 별 처리. 현재는 SEARCH(검색)만. -검색 핸들러: 소스 어댑터로 검색 → 코어 파이프라인(필터·이상치·top-N 최저가) → 결과. +검색 핸들러: 여러 소스 어댑터를 동시 검색 → 병합 → 코어 파이프라인(필터·이상치·top-N 최저가). +소스별 실패는 격리한다(한 소스가 죽어도 나머지로 결과 산출). 모든 소스 실패 시에만 잡 실패(재시도). AI 유사도 판정은 파이프라인 슬롯에 키 준비 시 결합한다. """ +import asyncio + from common.enums import JobType +from common.logger import LOG from services.search.contract import SearchAdapter from services.search.util import parse_price from services.pipeline.core import run_price_pipeline -def build_search_handler(adapters: dict[str, SearchAdapter], default_source: str = "coupang", limit: int = 40, top_n: int = 5): - """검색 핸들러 생성. adapters = {source: SearchAdapter}.""" +def build_search_handler(adapters: dict[str, SearchAdapter], sources: list[str] | None = None, limit: int = 40, top_n: int = 5): + """검색 핸들러 생성. adapters = {source: SearchAdapter}. sources 미지정 시 전체 사용.""" + use = list(sources) if sources else list(adapters.keys()) async def handler(job: dict) -> dict: if job["job_type"] != JobType.SEARCH.value: @@ -21,17 +26,30 @@ def build_search_handler(adapters: dict[str, SearchAdapter], default_source: str query = (payload.get("product_name") or "").strip() if not query: raise ValueError("empty product_name") - - adapter = adapters.get(default_source) - if adapter is None: - raise ValueError(f"no adapter for source: {default_source}") - - products = await adapter.search(query, limit=limit) - # 요청 현재가(있으면)를 가격 밴드 기준으로 사용 → 엉뚱한 저가/고가 targeted 컷 base_price = parse_price(payload.get("price")) + + # 소스 동시 검색 (실패는 예외로 수거해 격리) + results = await asyncio.gather( + *[adapters[s].search(query, limit=limit) for s in use], + return_exceptions=True, + ) + + products = [] + per_source: dict[str, dict] = {} + for src, res in zip(use, results): + if isinstance(res, Exception): + LOG.w(f"[{src}] 검색 실패: {type(res).__name__}: {res}") + per_source[src] = {"error": f"{type(res).__name__}: {res}"} + else: + products.extend(res) + per_source[src] = {"count": len(res)} + + if not products and all("error" in v for v in per_source.values()): + raise RuntimeError(f"모든 소스 검색 실패: {per_source}") # 잡 실패 → 재시도 + result = run_price_pipeline(products, base_price=base_price, top_n=top_n) - result["source"] = adapter.source result["query"] = query + result["sources"] = per_source # 소스별 건수/에러 (관측) return result return handler diff --git a/lps/worker_main.py b/lps/worker_main.py index e35900f..dc6cf6d 100644 --- a/lps/worker_main.py +++ b/lps/worker_main.py @@ -12,6 +12,7 @@ from common.logger import LOG from config.server_configs import web_server_config from crud.job_crud import JobQueue from services.search.coupang.adapter import CoupangAdapter +from services.search.naver.adapter import NaverAdapter from worker.handlers import build_search_handler from worker.notify import JobListener from worker.runner import Worker, run_reaper @@ -21,7 +22,8 @@ LOG.SetPrefix(f"{web_server_config.server_name}-worker") async def main(concurrency: int = 1): queue = JobQueue() - adapters = {"coupang": CoupangAdapter(headless=False)} # 워커들이 공유(내부 직렬화) + # 쿠팡(브라우저, 무거움) + 네이버(오픈API, 가벼움) 동시 검색 → 병합 최저가 + adapters = {"coupang": CoupangAdapter(headless=False), "naver": NaverAdapter()} handler = build_search_handler(adapters) stop = asyncio.Event()