"""잡 핸들러 — job_type 별 처리. 현재는 SEARCH(검색)만. 검색 핸들러 = 한정된 재정제 루프 + 명시적 outcome: 0. 네거티브 캐시 확인(최근 not_found면 즉시 반환) 각 라운드(원본 → 정밀(LLM) → 광역, 최대 max_rounds): 소스 동시 검색 → 필터 → 이상치 → AI 같은상품 판정 ├ 매칭 있음 → DONE(outcome=found), 조기 종료 ├ 0매칭 + 기술적 실패(차단/예외) 있음 → raise → 큐가 잡 전체 백오프 재시도(→소진 시 DEAD) └ 0매칭 + 소스 정상 → 다음 라운드 라운드 소진 → DONE(outcome=not_found) + 네거티브 캐시 기록 두 재시도 축을 분리한다: 기술적(큐 attempts/백오프) ≠ 검색어(refine 라운드, 유한). '못 찾음'은 정상 종료(DONE)지 dead-letter 가 아니다. """ import asyncio import time from common.enums import JobType, SourceState from common.logger import LOG from services.metrics import SearchMetrics 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 # 데드라인 초과로 버린 폴백 태스크의 강한 참조(asyncio 는 태스크를 약참조만 유지 — 없으면 GC 로 중도 파괴될 수 있음). # 완료 시 콜백에서 스스로 제거된다. 테스트는 이 집합을 gather 해 잔여 태스크를 배수(drain)할 수 있다. _abandoned_fallbacks: set[asyncio.Task] = set() def _reap_abandoned(task: asyncio.Task): """버려진 폴백 태스크 종료 시 예외를 회수 — 'Future exception was never retrieved' 노이즈 방지.""" _abandoned_fallbacks.discard(task) if task.cancelled(): return ex = task.exception() if ex is not None: LOG.d(f"[fallback] 데드라인 초과 태스크 종료(무시): {type(ex).__name__}") def _finalize_states(per_source: dict, matched: list[NormalizedProduct]) -> dict: """수집 성공 소스의 상태를 **매칭 결과로 확정**한다(제자리 수정 후 반환). 수집 단계에선 '가져왔다'까지만 알 수 있다. '같은 상품이었나'는 AI 판정을 거쳐야 알므로 여기서 MATCHED / NO_MATCH 를 가른다. 실패 상태(BLOCKED·EMPTY 등)는 이미 확정이라 건드리지 않는다. """ hit = {p.source for p in matched} collected = SourceState.MATCHED.name.lower() for src, info in per_source.items(): if info.get("state") == collected and src not in hit: info["state"] = SourceState.NO_MATCH.name.lower() return per_source def _price_snapshot(matched: list[NormalizedProduct]) -> dict: """매칭 목록에서 소스별 최저가 + 전체 최저가 스냅샷을 만든다(price_history 기록용).""" def lowest(src): items = [p for p in matched if p.source == src] return min(items, key=lambda p: p.price) if items else None n, c = lowest("naver"), lowest("coupang") # 최종 최저가는 소스 무관 전체 매칭 중 최저(G마켓·옥션·11번가 등 폴백 포함). f = min(matched, key=lambda p: p.price) if matched else None return { "matched_count": len(matched), "naver_lowest": n.price if n else None, "naver_name": n.name if n else None, "naver_url": n.detail_url if n else None, "coupang_lowest": c.price if c else None, "coupang_name": c.name if c else None, "coupang_url": c.detail_url if c else None, "final_lowest": f.price if f else None, "final_source": f.source if f else None, # 최저가 오퍼의 신뢰 신호 — 리뷰·평점이 없으면 '살 수 없는 가격'일 수 있다(유령상품). "final_rating": f.rating if f else None, "final_review_count": f.review_count if f else None, # 배송은 순위에 쓰지 않지만(주체가 다르면 비교 무의미) 기록은 남긴다 — 사후 분석용. "final_shipping_fee": f.shipping_fee if f else None, "final_shipping_type": f.shipping_type if f else None, "final_shipping_label": f.shipping_label if f else None, "by_mall": summarize_by_mall(matched), # 몰별 최저가 스냅샷(열린 스키마) } def build_search_handler( adapters: dict[str, SearchAdapter], sources: list[str] | None = None, limit: int = 40, top_n: int = 5, judge=None, keyword_gen=None, max_rounds: int = 3, neg_cache=None, history=None, fallback_adapters: dict[str, SearchAdapter] | None = None, ai_model: str = "", proxy_cost_per_gb: float = 0.0, fallback_deadline_sec: float = 15.0, ): """검색 핸들러 생성. judge: SimilarityJudge(같은 상품 판정) / keyword_gen: KeywordGenerator(정밀·광역 재검색어) / 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: return event = {"product_code": product_code, "job_id": job_id, "outcome": outcome, **_price_snapshot(matched)} try: await history.record(event) except Exception as ex: LOG.e_no_callstack(f"[history] 스냅샷 기록 실패(무시): {ex}") 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, via_proxy=via_proxy) return res except BaseException: # CancelledError(데드라인 취소) 포함 — 소요/바이트는 계측하고 재전파 metrics.add_fetch(source, getattr(adapter, "last_bytes", 0), int((time.monotonic() - t0) * 1000), crawl=crawl, via_proxy=via_proxy) raise async def _search_round(query: str, metrics: SearchMetrics): """한 라운드: 모든 소스 동시 검색 → (products, per_source, ok_sources). 소스별 시간/바이트 계측. ok_sources 는 **정상 응답한 소스 목록**이다. 예전엔 '하나라도 실패했나'(bool)만 봤는데, 그러면 한 소스가 막혔을 때 다른 소스가 멀쩡히 가져온 결과까지 버리게 된다(아래 호출부 참고). """ results = await asyncio.gather(*[_timed_search(adapters[s], query, s, metrics, False) for s in use], return_exceptions=True) products, per_source, ok_sources = [], {}, [] for src, res in zip(use, results): if isinstance(res, Exception): # ⚠️ 예외를 문자열로만 남기면 '그 몰에 없었다'와 '그 몰을 못 봤다'가 같아진다. # AdapterError 는 이미 state 로 그걸 알고 있으므로 **구조화해서 보존**한다. # (AdapterError 가 아닌 예외 = 우리 코드 버그 → 못 본 것으로 본다) state = getattr(res, "state", None) or SourceState.UNAVAILABLE per_source[src] = {"state": state.name.lower(), "error": f"{type(res).__name__}: {res}"} LOG.w(f"[{src}] 검색 실패({state.name}): {type(res).__name__}: {res}") if state.confirmed: ok_sources.append(src) # EMPTY = 봤는데 없던 것 → '정상 응답'으로 센다 else: products.extend(res) per_source[src] = {"state": SourceState.MATCHED.name.lower(), "count": len(res)} ok_sources.append(src) return products, per_source, ok_sources async def _match(target: dict, products: list, base_price, metrics: SearchMetrics): """필터 → (있으면) AI 같은상품 판정 → 매칭 후보. AI 토큰은 metrics 에 누적. ⚠️ 동시성 불변식: judge.judge 가 last_usage 를 세팅한 뒤 여기서 읽기까지 await 이 없어야 한다(asyncio 협조 스케줄링상 그 사이 다른 코루틴이 last_usage 를 덮어쓸 수 없음). 병렬 폴백 안전.""" candidates, _ = apply_filters(products, base_price=base_price) if judge is not None and candidates: verdicts = await judge.judge(target, candidates) metrics.add_ai(judge.last_usage) # ← await 직후 즉시 읽음(사이에 await 금지) 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, metrics: SearchMetrics): """네이버가 커버 못 한 오픈마켓만 직접 크롤(폴백) → 같은상품 판정 후 병합. 사용자 규칙: '네이버로 그 몰 값 확보 성공 → 그 값, 실패(몰 없음) → 실사이트 크롤'. 미커버 몰들을 **동시 크롤**한다(각 어댑터=별 브라우저 인스턴스라 병렬 안전, 소요=합→최댓값).""" if not fallbacks: return matched covered = {canonical_mall(p) for p in matched} todo = [(src, ad) for src, ad in fallbacks.items() if MALL_BY_SOURCE.get(src, src) not in covered] if not todo: return matched async def _crawl_match(src, adapter): # 폴백은 '있으면 좋은' 보강이라 데드라인을 건다 — 초과 시 그 몰만 스킵(전체 지연에 상한). # cancel 하지 않고 버린다(asyncio.wait): in-flight page.goto 를 취소하면 patchright 내부 # future 가 미회수 예외 노이즈를 남기고 페이지가 어중간한 상태로 남는다. 버려진 크롤은 # 백그라운드에서 자체 타임아웃(goto 40s 등)으로 끝나고 _reap_abandoned 가 예외를 회수한다. task = asyncio.ensure_future(_timed_search(adapter, query, src, metrics, crawl=True)) done, _ = await asyncio.wait({task}, timeout=fallback_deadline_sec) if not done: _abandoned_fallbacks.add(task) task.add_done_callback(_reap_abandoned) LOG.w(f"[fallback:{src}] 데드라인 {fallback_deadline_sec:.0f}s 초과 → 스킵(크롤은 백그라운드 종료)") return [] try: crawled = task.result() except Exception as ex: LOG.w(f"[fallback:{src}] 크롤 실패(무시): {type(ex).__name__}: {ex}") return [] hits = await _match(target, crawled, base_price, metrics) if hits: LOG.d(f"[fallback:{src}] 크롤 {len(crawled)}건 중 같은상품 {len(hits)}건 병합") return hits results = await asyncio.gather(*[_crawl_match(s, a) for s, a in todo]) for hits in results: matched = matched + hits return matched async def _round_queries(base_query: str, target: dict, metrics: SearchMetrics): """라운드 쿼리 지연 생성: 원본 → (0매칭 시에만 LLM 호출로) 정밀 → 광역.""" yield ("original", base_query) if keyword_gen is not None: kw = await keyword_gen.generate(target) # 원본이 실패해 여기까지 온 경우에만 호출됨 metrics.add_ai(keyword_gen.last_usage) seen = {base_query} for label, q in (("precise", kw.precise), ("broad", kw.broad)): q = (q or "").strip() if q and q not in seen: seen.add(q) yield (label, q) async def handler(job: dict) -> dict: if job["job_type"] != JobType.SEARCH.value: raise ValueError(f"unsupported job_type: {job['job_type']}") payload = job.get("payload") or {} base_query = (payload.get("product_name") or "").strip() if not base_query: raise ValueError("empty product_name") 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, proxy_cost_per_gb) # 검색 1건의 리소스/비용/시간 계측 # 0) 네거티브 캐시 — 최근 not_found면 재검색 생략. # force=True(사용자가 '다시 검색'을 명시적으로 누름)면 기록을 지우고 실제 검색을 돌린다. # 캐시 키가 product_code 라, 상품명·모델을 고쳐 재시도하는 경우 이 우회가 없으면 영원히 막힌다. if neg_cache is not None and payload.get("force"): await neg_cache.drop(cache_key) if neg_cache is not None and not payload.get("force") and await neg_cache.is_negative(cache_key): # 캐시 히트도 '이 잡의 결과'이므로 이력을 남긴다. 남기지 않으면 잡은 완료인데 # price_history 에 새 행이 없어, 이를 폴링하는 소비자(negodata 최저가 모달)가 # 결과를 영영 못 받고 로딩만 돈다(실측 버그). await _record_history(cache_key, job.get("job_id"), "not_found", []) return {"outcome": "not_found", "cached": True, "query": base_query, "rounds_tried": 0, "lowest": None, "top": [], "stages": [], "sources": {}, "metrics": metrics.snapshot()} rounds_done = 0 last_stages, last_sources = [], {} last_ok, last_failed = [], [] async for label, query in _round_queries(base_query, target, metrics): if rounds_done >= max_rounds: break rounds_done += 1 products, per_source, ok_sources = await _search_round(query, metrics) failed_sources = [s for s in use if s not in ok_sources] candidates, stages = apply_filters(products, base_price=base_price) if judge is not None and candidates: verdicts = await judge.judge(target, candidates) metrics.add_ai(judge.last_usage) matched = [c for c, v in zip(candidates, verdicts) if v.is_match] stages.append({"stage": "ai_match", "in": len(candidates), "out": len(matched)}) candidates = matched _finalize_states(per_source, candidates) # 수집만 된 소스를 matched/no_match 로 확정 last_stages, last_sources = stages, per_source last_ok, last_failed = ok_sources, failed_sources if candidates: # 찾음 → 오픈마켓 폴백 보강 후 종료 before = len(candidates) candidates = await _enrich_with_fallback(target, query, candidates, base_price, metrics) 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, sources_ok=ok_sources, sources_failed=failed_sources, partial=bool(failed_sources), metrics=metrics.snapshot()) if failed_sources: LOG.w(f"[partial] {failed_sources} 없이 결과를 냈습니다 — 그 몰의 더 싼 값은 못 봤을 수 있습니다") await _record_history(cache_key, job.get("job_id"), "found", candidates) return result # 0매칭 + 일부 소스 실패. **살아있는 소스가 하나라도 있으면 그 결과로 진행한다** — # 한 소스가 막혔다고 다른 소스가 멀쩡히 가져온 결과까지 버리면 사용자는 아무것도 못 받는다. # (실측 2026-08-06: 쿠팡이 막힌 동안 네이버가 40건씩 가져왔는데도 잡이 전부 DEAD 로 갔고, # DEAD 는 price_history 에 행을 안 남겨 프론트가 '탐색 중'에서 멈췄다) # 전부 죽었을 때만 '없음'이라 단정할 수 없으므로 재시도한다. if not ok_sources: # 마지막 시도면 재시도해도 갈 곳이 없다(다음은 DEAD). DEAD 는 price_history 에 행을 # 남기지 않아, 이를 폴링하는 소비자(negodata 최저가 모달)가 결과를 영영 못 받고 # '탐색 중'에서 멈춘다(2026-08-06 실측). 잡을 죽이는 대신 **실패를 결과로 기록**해 # 화면이 '검색 실패'를 보여줄 수 있게 한다. if job.get("attempts", 0) >= job.get("max_attempts", 3): LOG.w(f"[{cache_key}] 모든 소스 실패 + 재시도 소진 → error 로 종료(이력은 남긴다)") result = rank_result([], 0, stages, top_n) result.update(outcome="error", query=query, rounds_tried=rounds_done, sources=per_source, sources_ok=[], sources_failed=failed_sources, partial=True, metrics=metrics.snapshot()) await _record_history(cache_key, job.get("job_id"), "error", []) return result raise RuntimeError(f"모든 소스 실패로 0매칭(round={label}) — 잡 재시도: {per_source}") # 모든 라운드 0매칭 → not_found 로 정상 종료 # 단, 일부 소스가 막혀 있었다면 **네거티브 캐시에 넣지 않는다** — 그 소스엔 있었을 수 있는데 # '없음'으로 굳혀버리면 TTL 동안 재검색이 막힌다(사용자가 '다시 검색'을 눌러도 캐시 히트). partial = bool(last_failed) if neg_cache is not None and not partial: await neg_cache.put(cache_key, reason=f"not_found after {rounds_done} rounds") result = rank_result([], 0, last_stages, top_n) result.update(outcome="not_found", query=base_query, rounds_tried=rounds_done, sources=last_sources, sources_ok=last_ok, sources_failed=last_failed, partial=partial, metrics=metrics.snapshot()) if partial: LOG.w(f"[partial] {last_failed} 없이 not_found — 네거티브 캐시는 건너뜁니다(그 몰엔 있었을 수 있음)") await _record_history(cache_key, job.get("job_id"), "not_found", []) return result return handler