f9714c3f06
사용자 규칙 '네이버로 그 몰 값 확보 성공→그 값, 실패(몰 없음)→실사이트 크롤' 구현. - 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>
104 lines
4.1 KiB
Python
104 lines
4.1 KiB
Python
"""최저가 코어 파이프라인.
|
|
|
|
검색 결과(NormalizedProduct[]) → 필터 → 이상치 제거 → 정렬 → top-N 최저가.
|
|
각 STAGE 의 in/out 건수를 기록해 관측성을 확보한다(레퍼런스 pipeline_log 패턴 계승).
|
|
|
|
AI 유사도 판정(같은 상품인지)은 키 준비 시 이 사이(이상치 제거 뒤, top-N 앞)에 끼운다 —
|
|
지금은 슬롯만 비워두고 규칙 매칭을 하지 않는다(고정 파서 취약성 회피).
|
|
"""
|
|
|
|
from services.search.contract import NormalizedProduct
|
|
from services.pipeline.filters import keep_only_mall, filter_out_malls, filter_by_price_band
|
|
from services.pipeline.outliers import remove_price_outliers
|
|
|
|
|
|
def apply_filters(
|
|
products: list[NormalizedProduct],
|
|
*,
|
|
base_price: int | None = None,
|
|
keep_mall: str | None = None,
|
|
banned_malls=(),
|
|
band_tolerance: float = 0.7,
|
|
remove_outliers: bool = True,
|
|
) -> tuple[list[NormalizedProduct], list[dict]]:
|
|
"""mall → 가격밴드 → 이상치(IQR) 순으로 후보를 좁힌다. (후보, STAGE로그) 반환.
|
|
AI 유사도 판정은 이 뒤(top-N 앞)에 결합한다."""
|
|
stages: list[dict] = []
|
|
cur = list(products)
|
|
|
|
def stage(name: str, before: list, after: list):
|
|
stages.append({"stage": name, "in": len(before), "out": len(after)})
|
|
|
|
if keep_mall:
|
|
before = cur
|
|
cur = keep_only_mall(cur, keep_mall)
|
|
stage("keep_mall", before, cur)
|
|
elif banned_malls:
|
|
before = cur
|
|
cur = filter_out_malls(cur, banned_malls)
|
|
stage("filter_out_malls", before, cur)
|
|
|
|
if base_price:
|
|
before = cur
|
|
cur = filter_by_price_band(cur, base_price, band_tolerance)
|
|
stage("price_band", before, cur)
|
|
|
|
if remove_outliers:
|
|
before = cur
|
|
cur, _ = remove_price_outliers(cur)
|
|
stage("outlier", before, cur)
|
|
|
|
return cur, stages
|
|
|
|
|
|
def summarize_by_mall(products: list[NormalizedProduct]) -> list[dict]:
|
|
"""같은상품 매칭 후보를 판매몰별 최저가로 분해한다(가격 오름차순).
|
|
네이버 결과엔 G마켓·옥션·11번가 등이 mall_name 으로 이미 들어오고, 크롤 폴백도 같은 몰명을 쓴다.
|
|
**몰명(canonical)으로 dedup** — 같은 몰이 네이버 노출과 직접 크롤 양쪽에서 와도 최저가 1건으로 병합
|
|
(네이버 가격비교 mall_name='네이버'는 여러 판매자 최저가 롤업이라 별도 몰로 취급)."""
|
|
best: dict[str, NormalizedProduct] = {}
|
|
for p in products:
|
|
key = p.mall_name or p.source or "기타" # 몰 정체성 = 몰명(소스 무관 병합)
|
|
cur = best.get(key)
|
|
if cur is None or p.price < cur.price:
|
|
best[key] = p
|
|
rows = sorted(best.values(), key=lambda p: p.price)
|
|
return [{
|
|
"source": p.source, "mall_name": p.mall_name, "price": p.price,
|
|
"shipping_fee": p.shipping_fee, "shipping_type": p.shipping_type,
|
|
"name": p.name, "detail_url": p.detail_url,
|
|
} for p in rows]
|
|
|
|
|
|
def rank_result(products: list[NormalizedProduct], total_found: int, stages: list[dict], top_n: int = 5) -> dict:
|
|
"""최저가순 정렬 + top-N + 결과 봉투 조립(top_n STAGE 포함)."""
|
|
ranked = sorted(products, key=lambda p: p.price)
|
|
top = ranked[:top_n]
|
|
stages = stages + [{"stage": "top_n", "in": len(ranked), "out": len(top)}]
|
|
return {
|
|
"total_found": total_found,
|
|
"kept": len(ranked),
|
|
"lowest": top[0].model_dump() if top else None,
|
|
"top": [p.model_dump() for p in top],
|
|
"by_mall": summarize_by_mall(ranked),
|
|
"stages": stages,
|
|
}
|
|
|
|
|
|
def run_price_pipeline(
|
|
products: list[NormalizedProduct],
|
|
*,
|
|
base_price: int | None = None,
|
|
keep_mall: str | None = None,
|
|
banned_malls=(),
|
|
band_tolerance: float = 0.7,
|
|
remove_outliers: bool = True,
|
|
top_n: int = 5,
|
|
) -> dict:
|
|
"""AI 없는 동기 파이프라인(테스트/기본 경로). apply_filters + rank_result 조합."""
|
|
cur, stages = apply_filters(
|
|
products, base_price=base_price, keep_mall=keep_mall,
|
|
banned_malls=banned_malls, band_tolerance=band_tolerance, remove_outliers=remove_outliers,
|
|
)
|
|
return rank_result(cur, len(products), stages, top_n)
|