feat(lps): 오픈마켓 폴백 오케스트레이션 + 몰 dedup

사용자 규칙 '네이버로 그 몰 값 확보 성공→그 값, 실패(몰 없음)→실사이트 크롤' 구현.

- handler: found 라운드 뒤 _enrich_with_fallback — 네이버 매칭이 커버 못 한 타깃 몰만
  (canonical_mall 로 판정) 크롤 어댑터로 검색→같은상품 판정→병합. 크롤 실패는 격리(무시).
  fallback_crawl STAGE 기록.
- summarize_by_mall: 키를 (source,mall)→몰명(canonical)으로 — 네이버노출 vs 직접크롤 같은 몰
  중복을 최저가 1건으로 병합(dedup).
- card_parser: MALL_BY_SOURCE + canonical_mall() 추가.
- worker_main: gmarket/auction/st11 폴백 어댑터 등록(lazy 기동, 리소스차단 OFF).
- 테스트: 커버몰 크롤생략·미커버 크롤·실패격리·몰 dedup 4종.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

lps/services/pipeline/core.py

@@ -52,13 +52,13 @@ def apply_filters(
 
 
 def summarize_by_mall(products: list[NormalizedProduct]) -> list[dict]:
-    """같은상품 매칭 후보를 (소스, 판매몰)별 최저가로 분해한다.
-    네이버 결과엔 G마켓·옥션·11번가 등이 mall_name 으로 이미 들어오므로,
-    추가 요청 없이 '오픈마켓 몰별 최저가'를 노출할 수 있다(가격 오름차순).
-    네이버 가격비교(mall_name='네이버')는 여러 판매자 중 최저가 롤업이다."""
-    best: dict[tuple[str, str], NormalizedProduct] = {}
+    """같은상품 매칭 후보를 판매몰별 최저가로 분해한다(가격 오름차순).
+    네이버 결과엔 G마켓·옥션·11번가 등이 mall_name 으로 이미 들어오고, 크롤 폴백도 같은 몰명을 쓴다.
+    **몰명(canonical)으로 dedup** — 같은 몰이 네이버 노출과 직접 크롤 양쪽에서 와도 최저가 1건으로 병합
+    (네이버 가격비교 mall_name='네이버'는 여러 판매자 최저가 롤업이라 별도 몰로 취급)."""
+    best: dict[str, NormalizedProduct] = {}
     for p in products:
-        key = (p.source, (p.mall_name or "기타"))
+        key = p.mall_name or p.source or "기타"  # 몰 정체성 = 몰명(소스 무관 병합)
         cur = best.get(key)
         if cur is None or p.price < cur.price:
             best[key] = p

lps/services/search/card_parser.py

@@ -12,6 +12,14 @@ from selectolax.parser import HTMLParser
 from services.search.contract import NormalizedProduct
 from services.search.util import extract_price, clean_name, shipping_from_text
 
+# 소스 코드 → 표시 몰명(canonical). 폴백 커버리지 판정·dedup 에 쓴다.
+MALL_BY_SOURCE = {"naver": "네이버", "coupang": "쿠팡", "gmarket": "G마켓", "auction": "옥션", "st11": "11번가"}
+
+
+def canonical_mall(product) -> str:
+    """상품의 판매몰 정체성(몰명 우선, 없으면 소스 매핑)."""
+    return product.mall_name or MALL_BY_SOURCE.get(product.source, product.source)
+
 
 @dataclass(frozen=True)
 class CardConfig:

lps/tests/test_search_handler.py

@@ -55,8 +55,8 @@ class FakeNegCache:
         self.puts.append(key)
 
 
-def _np(source, price):
-    return NormalizedProduct(source=source, name=f"{source}-{price}", price=price)
+def _np(source, price, mall=None):
+    return NormalizedProduct(source=source, name=f"{source}-{price}", price=price, mall_name=mall)
 
 
 def _job(**payload):
@@ -120,3 +120,44 @@ async def test_technical_failure_with_zero_match_raises():
     adapters = {"coupang": FakeAdapter("coupang", fail=True), "naver": FakeAdapter("naver", by_query={})}
     with pytest.raises(RuntimeError):
         await build_search_handler(adapters)(_job())  # 0매칭 + 차단 → 기술 재시도
+
+
+# ── 오픈마켓 폴백 크롤 (네이버 미커버 몰만) ──────────────────────────
+async def test_fallback_crawls_only_uncovered_malls():
+    # 네이버 매칭에 G마켓은 있고(→크롤 생략), 11번가는 없음(→크롤). 옥션도 없음(→크롤).
+    adapters = {"naver": FakeAdapter("naver", products=[_np("naver", 5000, mall="G마켓")])}
+    gmarket = FakeAdapter("gmarket", products=[_np("gmarket", 4000, mall="G마켓")])
+    auction = FakeAdapter("auction", products=[_np("auction", 4500, mall="옥션")])
+    st11 = FakeAdapter("st11", products=[_np("st11", 3000, mall="11번가")])
+    handler = build_search_handler(
+        adapters, judge=FakeJudge(lambda c: True),
+        fallback_adapters={"gmarket": gmarket, "auction": auction, "st11": st11},
+    )
+    r = await handler(_job())
+    assert gmarket.calls == []                    # 네이버가 G마켓 커버 → 크롤 생략
+    assert auction.calls and st11.calls           # 미커버 → 크롤함
+    assert r["lowest"]["price"] == 3000           # 11번가 크롤가가 전체 최저
+    malls = {m["mall_name"] for m in r["by_mall"]}
+    assert malls == {"G마켓", "옥션", "11번가"}     # 네이버 G마켓 + 크롤 옥션·11번가
+
+
+async def test_fallback_failure_is_isolated():
+    adapters = {"naver": FakeAdapter("naver", products=[_np("naver", 9000, mall="네이버")])}
+    st11 = FakeAdapter("st11", fail=True)          # 크롤 실패
+    r = await build_search_handler(
+        adapters, judge=FakeJudge(lambda c: True),
+        fallback_adapters={"st11": st11},
+    )(_job())
+    assert r["outcome"] == "found" and r["lowest"]["price"] == 9000   # 폴백 실패해도 정상 종료
+
+
+async def test_fallback_dedup_same_mall_keeps_lowest():
+    # 네이버 매칭에 G마켓 없음 → 크롤. 크롤 G마켓이 네이버 '네이버몰'보다 싸면 최저가 갱신.
+    adapters = {"naver": FakeAdapter("naver", products=[_np("naver", 8000, mall="네이버")])}
+    gmarket = FakeAdapter("gmarket", products=[_np("gmarket", 6000, mall="G마켓"), _np("gmarket", 7000, mall="G마켓")])
+    r = await build_search_handler(
+        adapters, judge=FakeJudge(lambda c: True),
+        fallback_adapters={"gmarket": gmarket},
+    )(_job())
+    gm = [m for m in r["by_mall"] if m["mall_name"] == "G마켓"]
+    assert len(gm) == 1 and gm[0]["price"] == 6000   # 몰별 1건(최저)로 dedup

lps/worker/handlers.py

@@ -18,6 +18,7 @@ import asyncio
 from common.enums import JobType
 from common.logger import LOG
 from services.search.contract import SearchAdapter, NormalizedProduct
+from services.search.card_parser import canonical_mall, MALL_BY_SOURCE
 from services.search.util import parse_price
 from services.pipeline.core import apply_filters, rank_result, summarize_by_mall
 
@@ -50,12 +51,15 @@ def build_search_handler(
     max_rounds: int = 3,
     neg_cache=None,
     history=None,
+    fallback_adapters: dict[str, SearchAdapter] | None = None,
 ):
     """검색 핸들러 생성.
     judge: SimilarityJudge(같은 상품 판정) / keyword_gen: KeywordGenerator(정밀·광역 재검색어) /
-    neg_cache: NegativeCache(TTL not_found 캐시) / history: PriceHistory(최저가 스냅샷).
+    neg_cache: NegativeCache(TTL not_found 캐시) / history: PriceHistory(최저가 스냅샷) /
+    fallback_adapters: 오픈마켓 크롤(gmarket/auction/st11) — 네이버가 그 몰을 커버 못 했을 때만 크롤(폴백).
     모두 선택 — 없으면 해당 단계 생략."""
     use = list(sources) if sources else list(adapters.keys())
+    fallbacks = fallback_adapters or {}
 
     async def _record_history(product_code: str, job_id, outcome: str, matched: list):
         if history is None:
@@ -80,6 +84,35 @@ def build_search_handler(
                 per_source[src] = {"count": len(res)}
         return products, per_source, tech_failed
 
+    async def _match(target: dict, products: list, base_price):
+        """필터 → (있으면) AI 같은상품 판정 → 매칭 후보. 폴백 크롤 결과 판정에도 재사용."""
+        candidates, _ = apply_filters(products, base_price=base_price)
+        if judge is not None and candidates:
+            verdicts = await judge.judge(target, candidates)
+            candidates = [c for c, v in zip(candidates, verdicts) if v.is_match]
+        return candidates
+
+    async def _enrich_with_fallback(target: dict, query: str, matched: list, base_price):
+        """네이버가 커버 못 한 오픈마켓만 직접 크롤(폴백) → 같은상품 판정 후 병합.
+        사용자 규칙: '네이버로 그 몰 값 확보 성공 → 그 값, 실패(몰 없음) → 실사이트 크롤'."""
+        if not fallbacks:
+            return matched
+        covered = {canonical_mall(p) for p in matched}
+        for src, adapter in fallbacks.items():
+            mall = MALL_BY_SOURCE.get(src, src)
+            if mall in covered:  # 네이버가 이미 그 몰 최저가 확보 → 크롤 생략
+                continue
+            try:
+                crawled = await adapter.search(query, limit=limit)
+            except Exception as ex:
+                LOG.w(f"[fallback:{src}] 크롤 실패(무시): {type(ex).__name__}: {ex}")
+                continue
+            hits = await _match(target, crawled, base_price)
+            if hits:
+                LOG.d(f"[fallback:{src}] 크롤 {len(crawled)}건 중 같은상품 {len(hits)}건 병합")
+                matched = matched + hits
+        return matched
+
     async def _round_queries(base_query: str, target: dict):
         """라운드 쿼리 지연 생성: 원본 → (0매칭 시에만 LLM 호출로) 정밀 → 광역."""
         yield ("original", base_query)
@@ -125,7 +158,11 @@ def build_search_handler(
                 candidates = matched
             last_stages, last_sources = stages, per_source
 
-            if candidates:  # 찾음 → 조기 종료
+            if candidates:  # 찾음 → 오픈마켓 폴백 보강 후 종료
+                before = len(candidates)
+                candidates = await _enrich_with_fallback(target, query, candidates, base_price)
+                if len(candidates) > before:
+                    stages.append({"stage": "fallback_crawl", "in": before, "out": len(candidates)})
                 result = rank_result(candidates, len(products), stages, top_n)
                 result.update(outcome="found", query=query, round=label, rounds_tried=rounds_done, sources=per_source)
                 await _record_history(cache_key, job.get("job_id"), "found", candidates)

lps/worker_main.py

@@ -17,6 +17,8 @@ from crud.price_history import PriceHistory
 from services.search.proxy import DecodoProxy
 from services.search.coupang.adapter import CoupangAdapter
 from services.search.naver.adapter import NaverAdapter
+from services.search.esm.adapter import EsmAdapter
+from services.search.st11.adapter import ElevenStAdapter
 from services.ai.similarity import SimilarityJudge
 from services.ai.keyword import KeywordGenerator
 from worker.handlers import build_search_handler
@@ -38,6 +40,14 @@ async def main(concurrency: int = 1):
         "coupang": CoupangAdapter(headless=False, proxy=proxy, on_detect=bot_log.record),
         "naver": NaverAdapter(),
     }
+    # 오픈마켓 폴백 크롤러: 네이버가 그 몰을 커버 못 했을 때만 lazy 하게 실사이트 크롤(브라우저는 첫 사용 시 기동).
+    #   G마켓·옥션(ESM '잠시만' 챌린지) + 11번가(PC). 리소스차단 OFF(렌더/챌린지 보호)는 어댑터 기본값.
+    fallback_adapters = {
+        "gmarket": EsmAdapter("gmarket", headless=False, proxy=proxy, on_detect=bot_log.record),
+        "auction": EsmAdapter("auction", headless=False, proxy=proxy, on_detect=bot_log.record),
+        "st11": ElevenStAdapter(headless=False, proxy=proxy, on_detect=bot_log.record),
+    }
+    LOG.i(f"오픈마켓 폴백 크롤: {', '.join(fallback_adapters)} (네이버 미커버 몰만)")
     # OpenAI 키 있으면 '같은 상품' AI 판정 + 재검색어 생성 활성화
     has_openai = bool(openai_config.api_key)
     judge = SimilarityJudge() if has_openai else None
@@ -46,6 +56,7 @@ async def main(concurrency: int = 1):
     handler = build_search_handler(
         adapters, judge=judge, keyword_gen=keyword_gen,
         neg_cache=NegativeCache(), history=PriceHistory(),
+        fallback_adapters=fallback_adapters,
     )
 
     stop = asyncio.Event()
@@ -68,7 +79,7 @@ async def main(concurrency: int = 1):
         stop.set()
         for listener in listeners:
             await listener.close()
-        for adapter in adapters.values():
+        for adapter in list(adapters.values()) + list(fallback_adapters.values()):
             await adapter.close()