e27b73c018
소스 어댑터 패턴으로 크롤링을 시작한다. 쿠팡은 Akamai Bot Manager 의 JS 행동 챌린지라 curl_cffi 단독 불가 → Patchright(스텔스 Playwright) + 실제 Chrome(channel=chrome)으로 챌린지를 통과하고 selectolax 로 파싱한다. - contract: SearchAdapter(ABC)·NormalizedProduct·AdapterHealth·AdapterError - coupang: adapter(브라우저 재사용, 챌린지 1회)·parser(순수)·selectors(외부화, webpack 해시 prefix 매칭) - 공용: rate_limiter(2~8s 랜덤)·proxy(무프록시 off 인터페이스)·util.parse_price - 가격 파싱: 단위가격 '(1개당 44,400원)'의 앞 '1' 오인 방지 — '원' 앞 숫자 앵커링 - deps: curl_cffi·selectolax·patchright(nodriver 는 py3.14 버그로 대체) - tests: 파서 회귀(소형 fixture) — 라이브 2회 검색 + 파서 3/3 통과 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
61 lines
2.6 KiB
Python
61 lines
2.6 KiB
Python
"""소스 어댑터 공통 계약.
|
|
|
|
크롤링/검색 소스(네이버·쿠팡 …)는 각자 수집 방식과 안티봇 대응을 캡슐화하고,
|
|
코어 파이프라인(필터·이상치·AI)은 정규화된 NormalizedProduct 만 본다.
|
|
새 소스는 SearchAdapter 를 구현하기만 하면 코어 변경 없이 붙는다(트레드밀 격리).
|
|
"""
|
|
|
|
from abc import ABC, abstractmethod
|
|
from typing import Optional
|
|
|
|
from pydantic import BaseModel, Field
|
|
|
|
|
|
class NormalizedProduct(BaseModel):
|
|
"""소스 무관 정규화 상품 스키마. 어댑터의 유일한 출력 계약."""
|
|
|
|
source: str = Field(description="수집 소스 (naver|coupang)")
|
|
name: str = Field(description="상품명")
|
|
price: int = Field(description="판매가(원, 정수). 파싱 실패분은 어댑터에서 제외")
|
|
model: Optional[str] = Field(None, description="모델명(있으면)")
|
|
manufacturer: Optional[str] = Field(None, description="제조사(있으면)")
|
|
image_url: Optional[str] = Field(None, description="썸네일 URL")
|
|
detail_url: Optional[str] = Field(None, description="상품 상세 URL")
|
|
shipping_fee: Optional[int] = Field(None, description="배송비(원). 무료=0, 미확인=None")
|
|
mall_name: Optional[str] = Field(None, description="판매몰/스토어명")
|
|
external_id: Optional[str] = Field(None, description="소스 내 상품 식별자")
|
|
|
|
|
|
class AdapterHealth(BaseModel):
|
|
"""어댑터 건강도. 성공률 급락 = 레이아웃 변경/차단 신호 → 알림 훅."""
|
|
|
|
source: str
|
|
ok: bool = Field(description="현재 정상 동작 여부")
|
|
recent_success_rate: float = Field(0.0, description="최근 요청 성공률(0~1)")
|
|
blocked_rate: float = Field(0.0, description="최근 차단(봇탐지) 비율(0~1)")
|
|
note: str = ""
|
|
|
|
|
|
class AdapterError(Exception):
|
|
"""어댑터 수집 실패. blocked=True 면 안티봇 차단으로 판단(에스컬레이션/알림 트리거)."""
|
|
|
|
def __init__(self, message: str, *, source: str, blocked: bool = False):
|
|
super().__init__(message)
|
|
self.source = source
|
|
self.blocked = blocked
|
|
|
|
|
|
class SearchAdapter(ABC):
|
|
"""검색 소스 어댑터. 소스별 수집/에스컬레이션/안티봇을 내부에 캡슐화한다."""
|
|
|
|
source: str
|
|
|
|
@abstractmethod
|
|
async def search(self, query: str, limit: int = 40) -> list[NormalizedProduct]:
|
|
"""query 로 검색해 정규화 상품 리스트를 반환. 차단 시 AdapterError(blocked=True)."""
|
|
raise NotImplementedError
|
|
|
|
async def health(self) -> AdapterHealth:
|
|
"""기본 건강도. 어댑터가 관측 지표를 축적하면 override."""
|
|
return AdapterHealth(source=self.source, ok=True)
|