From bca8032b79aee5fc4a83fd20ac84b864e65f1d5d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EB=AF=BC=ED=97=8C?= Date: Thu, 9 Jul 2026 11:04:07 +0900 Subject: [PATCH] =?UTF-8?q?feat(lps):=20=EB=B4=87=20=EA=B0=90=EC=A7=80=20?= =?UTF-8?q?=EC=8B=9C=20IP=20=ED=9A=8C=EC=A0=84+=EC=9E=AC=EC=8B=9C=EB=8F=84?= =?UTF-8?q?=20+=20=EA=B0=90=EC=A7=80=20=EC=9D=B4=EB=A0=A5=20=EA=B8=B0?= =?UTF-8?q?=EB=A1=9D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 쿠팡이 봇으로 감지(Akamai 차단 페이지)하면, 같은 IP로 재시도하던 것을 '브라우저 끄고 새 IP로 켜서 재시도'로 전환. 감지 패턴도 축적한다. - DecodoProxy.rotate(): 시간창 무관 즉시 다음 포트(=새 IP). current_port 노출 - coupang/adapter: 0건+차단마커 감지 시 → 감지기록 → proxy.rotate()+강제재기동 → 인라인 재시도(max_block_retries=1). IP당 요청수(ip_request_no) 추적, 재기동 시 리셋 - bot_detection 테이블 + crud: source/query/ip_request_no/proxy_port/elapsed_sec/marker/ headless/html_len 기록 → 'IP당 몇 요청 만에 감지되나' 분석 가능 - worker_main: on_detect=BotDetectionLog.record 주입 - 로그: [coupang][BOT-DETECTED] ip_req#N port=... elapsed=...s marker=... - tests: proxy rotate 즉시회전 + 감지기록 CRUD → 전체 47/47 Co-Authored-By: Claude Opus 4.8 (1M context) --- lps/common/database/model/models.py | 24 +++++++- lps/crud/bot_detection.py | 28 +++++++++ lps/services/search/coupang/adapter.py | 85 ++++++++++++++++++-------- lps/services/search/proxy.py | 13 +++- lps/tests/test_bot_detection.py | 34 +++++++++++ lps/tests/test_proxy.py | 11 ++++ lps/worker_main.py | 10 ++- 7 files changed, 176 insertions(+), 29 deletions(-) create mode 100644 lps/crud/bot_detection.py create mode 100644 lps/tests/test_bot_detection.py diff --git a/lps/common/database/model/models.py b/lps/common/database/model/models.py index 2be9127..b2bcbcc 100644 --- a/lps/common/database/model/models.py +++ b/lps/common/database/model/models.py @@ -1,4 +1,4 @@ -from sqlalchemy import Column, Index, SmallInteger, String, Text, DateTime +from sqlalchemy import Boolean, Column, Index, Integer, SmallInteger, String, Text, DateTime from sqlalchemy.dialects.postgresql import UUID, JSONB from sqlalchemy.orm import declarative_base from sqlalchemy.sql import text @@ -66,3 +66,25 @@ class search_negative(MAIN_BASE): until = Column(DateTime(timezone=True), nullable=False) # 이 시각까지 not_found 로 간주 reason = Column(String(200), nullable=True) # 종료 사유 메모(관측) created_at = Column(DateTime(timezone=True), nullable=False, server_default=text("now()")) + + +class bot_detection(MAIN_BASE): + """봇 감지 이력 — '이 IP로 몇 번째 요청에서, 어떤 방식으로 차단됐나'를 축적해 패턴 분석. + (예: SELECT avg(ip_request_no) → IP당 평균 몇 요청 만에 감지되는지)""" + + @staticmethod + def DBType(): + return DBType.MAIN.value + + __tablename__ = "bot_detection" + + id = Column(UUID(as_uuid=True), primary_key=True, server_default=text("gen_random_uuid()")) + source = Column(String(20), nullable=False) # coupang 등 + query = Column(String(300), nullable=True) # 감지 당시 검색어 + ip_request_no = Column(Integer, nullable=True) # 현재 IP(브라우저)로 몇 번째 요청이었나 + proxy_port = Column(Integer, nullable=True) # 사용 중이던 프록시 포트(=IP 세션) + elapsed_sec = Column(Integer, nullable=True) # 브라우저 실행 후 경과(초) + marker = Column(String(120), nullable=True) # 감지 근거(차단 페이지 마커) + headless = Column(Boolean, nullable=True) + html_len = Column(Integer, nullable=True) # 응답 길이(차단 페이지는 작음) + created_at = Column(DateTime(timezone=True), nullable=False, server_default=text("now()")) diff --git a/lps/crud/bot_detection.py b/lps/crud/bot_detection.py new file mode 100644 index 0000000..8d0618f --- /dev/null +++ b/lps/crud/bot_detection.py @@ -0,0 +1,28 @@ +"""봇 감지 이력 기록 CRUD — '몇 번째 요청/어떤 포트에서 감지됐나'를 축적(패턴 분석용).""" + +from sqlalchemy import text + +from common.database.db_session_manager import DB_SESSION_MNG +from common.enums import DBType, DBWRType + + +class BotDetectionLog: + DB = DBType.MAIN.value + + async def record(self, event: dict): + """감지 이벤트 1건 저장. 로깅 실패가 검색을 막지 않도록 호출부에서 예외를 삼킨다.""" + sql = text(""" + INSERT INTO bot_detection (source, query, ip_request_no, proxy_port, elapsed_sec, marker, headless, html_len) + VALUES (:source, :query, :ip_request_no, :proxy_port, :elapsed_sec, :marker, :headless, :html_len) + """) + params = {k: event.get(k) for k in + ("source", "query", "ip_request_no", "proxy_port", "elapsed_sec", "marker", "headless", "html_len")} + s = await DB_SESSION_MNG.start_session(self.DB, DBWRType.DB_WRITE.value) + try: + await s.execute(sql, params) + await s.commit() + except Exception: + await s.rollback() + raise + finally: + await DB_SESSION_MNG.end_session(self.DB, DBWRType.DB_WRITE.value) diff --git a/lps/services/search/coupang/adapter.py b/lps/services/search/coupang/adapter.py index bc8eb36..030fba4 100644 --- a/lps/services/search/coupang/adapter.py +++ b/lps/services/search/coupang/adapter.py @@ -29,15 +29,20 @@ _BLOCKED_RESOURCES = {"image", "media", "font", "stylesheet"} class CoupangAdapter(SearchAdapter): source = "coupang" - def __init__(self, headless: bool = False, user_data_dir: str = "/tmp/lps_coupang_profile", rate_limiter: RateLimiter | None = None, proxy=None, block_resources: bool = True): + def __init__(self, headless: bool = False, user_data_dir: str = "/tmp/lps_coupang_profile", rate_limiter: RateLimiter | None = None, proxy=None, block_resources: bool = True, on_detect=None, max_block_retries: int = 1): self._headless = headless self._user_data_dir = user_data_dir self._rl = rate_limiter or RateLimiter() self._proxy = proxy # DecodoProxy 등 (없으면 직접 연결) self._block_resources = block_resources + self._on_detect = on_detect # async def(event: dict) — 감지 이벤트 영속화(선택) + self._max_block_retries = max_block_retries # 감지 시 새 IP 로 인라인 재시도 횟수 self._pw = None self._ctx = None self._launched_at = 0.0 + self._ip_requests = 0 # 현재 브라우저(IP)로 보낸 요청 수(재기동 시 리셋) + self._current_port = None # 현재 사용 중인 프록시 포트(감지 로그용) + self._force_recycle = False # 감지 등으로 다음 _ensure_browser 에서 강제 재기동 self._lock = asyncio.Lock() self._ok = 0 self._blocked = 0 @@ -49,7 +54,9 @@ class CoupangAdapter(SearchAdapter): await route.continue_() def _recycle_due(self) -> bool: - """프록시 sticky 세션창이 지났으면 브라우저를 재기동해 새 IP 를 받는다.""" + """강제 재기동 플래그(봇 감지)거나, 프록시 sticky 세션창이 지났으면 재기동해 새 IP 를 받는다.""" + if self._force_recycle: + return True if not (self._proxy and self._proxy.enabled): return False return (time.monotonic() - self._launched_at) > self._proxy.session_minutes * 60 @@ -57,7 +64,7 @@ class CoupangAdapter(SearchAdapter): async def _ensure_browser(self): if self._ctx is not None: if self._recycle_due(): - LOG.d("[coupang] 프록시 세션창 경과 → 브라우저 재기동(IP 회전)") + LOG.d("[coupang] 브라우저 재기동(IP 회전)") await self._close_ctx() else: return @@ -66,10 +73,13 @@ class CoupangAdapter(SearchAdapter): kwargs = dict(user_data_dir=self._user_data_dir, channel="chrome", headless=self._headless, no_viewport=True) if self._proxy and self._proxy.enabled: kwargs["proxy"] = self._proxy.playwright_proxy() + self._current_port = self._proxy.current_port self._ctx = await self._pw.chromium.launch_persistent_context(**kwargs) if self._block_resources: await self._ctx.route("**/*", self._route) # 이미지/미디어/폰트/CSS 차단 self._launched_at = time.monotonic() + self._ip_requests = 0 + self._force_recycle = False async def _close_ctx(self): if self._ctx is not None: @@ -80,31 +90,58 @@ class CoupangAdapter(SearchAdapter): async def search(self, query: str, limit: int = 40) -> list[NormalizedProduct]: async with self._lock: # 인스턴스 내 검색은 직렬화(브라우저 컨텍스트 공유) - await self._rl.wait() - await self._ensure_browser() - page = self._ctx.pages[0] if self._ctx.pages else await self._ctx.new_page() - url = _SEARCH_URL.format(q=quote(query), n=limit) - try: - await page.goto(url, wait_until="domcontentloaded", timeout=40000) - # Akamai 센서 실행 + 상품 렌더 대기 + # 봇 감지 시: 새 IP 로 회전 후 인라인 재시도(최대 max_block_retries 회) + for attempt in range(self._max_block_retries + 1): + await self._rl.wait() + await self._ensure_browser() + self._ip_requests += 1 + page = self._ctx.pages[0] if self._ctx.pages else await self._ctx.new_page() + url = _SEARCH_URL.format(q=quote(query), n=limit) try: - await page.wait_for_selector(SELECTORS.card, timeout=20000) - except Exception: - pass - html = await page.content() - except Exception as ex: - raise AdapterError(f"쿠팡 검색 실패: {ex}", source=self.source) from ex + await page.goto(url, wait_until="domcontentloaded", timeout=40000) + try: + await page.wait_for_selector(SELECTORS.card, timeout=20000) # Akamai 센서+렌더 대기 + except Exception: + pass + html = await page.content() + except Exception as ex: + raise AdapterError(f"쿠팡 검색 실패: {ex}", source=self.source) from ex - products = parse_search_html(html, source=self.source) - if not products: - blocked = any(m in html for m in _BLOCK_MARKERS) + products = parse_search_html(html, source=self.source) + if products: + self._ok += 1 + LOG.d(f"[coupang] query={query!r} → {len(products)}건 (limit {limit}, ip_req#{self._ip_requests})") + return products[:limit] + + # 0건 — 봇 감지 여부 판정 + marker = next((m for m in _BLOCK_MARKERS if m in html), None) + blocked = marker is not None self._blocked += 1 - LOG.w(f"[coupang] 결과 0건 (blocked={blocked}) query={query!r} len={len(html)}") - raise AdapterError(f"쿠팡 결과 없음/차단 (query={query!r})", source=self.source, blocked=blocked) + if blocked: + await self._report_detection(query, marker, len(html)) - self._ok += 1 - LOG.d(f"[coupang] query={query!r} → {len(products)}건 (limit {limit})") - return products[:limit] + can_retry = blocked and self._proxy and self._proxy.enabled and attempt < self._max_block_retries + if can_retry: + self._proxy.rotate() # 다음 포트 = 새 IP + self._force_recycle = True # 다음 _ensure_browser 에서 재기동 + LOG.w(f"[coupang] 봇 감지 → IP 회전 후 재시도 ({attempt + 1}/{self._max_block_retries})") + continue + + raise AdapterError(f"쿠팡 결과 없음/차단 (query={query!r}, blocked={blocked})", source=self.source, blocked=blocked) + + async def _report_detection(self, query: str, marker: str, html_len: int): + """봇 감지 기록 — 로그 + (있으면) DB 영속화. IP당 몇 번째 요청에서 감지됐는지 축적.""" + elapsed = int(time.monotonic() - self._launched_at) + LOG.w(f"[coupang][BOT-DETECTED] ip_req#{self._ip_requests} port={self._current_port} " + f"elapsed={elapsed}s headless={self._headless} marker={marker!r} query={query!r} html_len={html_len}") + if self._on_detect is not None: + event = {"source": self.source, "query": query, "ip_request_no": self._ip_requests, + "proxy_port": self._current_port, "elapsed_sec": elapsed, "marker": marker, + "headless": self._headless, "html_len": html_len} + try: + await self._on_detect(event) + except Exception as ex: + LOG.e_no_callstack(f"[coupang] 감지 기록 실패(무시): {ex}") async def health(self) -> AdapterHealth: total = self._ok + self._blocked diff --git a/lps/services/search/proxy.py b/lps/services/search/proxy.py index dad3f83..dbd7c9f 100644 --- a/lps/services/search/proxy.py +++ b/lps/services/search/proxy.py @@ -23,16 +23,25 @@ class DecodoProxy: self.port_start = cfg.port_start self.port_end = cfg.port_end self.session_minutes = cfg.session_minutes or 10 + self._rotate_offset = 0 # 봇 감지 등으로 '즉시 회전'이 필요할 때 증가 @property def enabled(self) -> bool: return all([self.host, self.username, self.password, self.port_start, self.port_end]) + def rotate(self): + """시간창과 무관하게 즉시 다음 포트(=새 IP)로 회전. 봇 감지 시 호출.""" + self._rotate_offset += 1 + def _port(self) -> int: - """시간창 기반 포트 선택 — 창 안에선 동일 포트(동일 sticky IP), 창이 지나면 다음 포트.""" + """시간창 + 수동 오프셋 기반 포트 선택. 창 안에선 동일 IP, rotate()나 창 변화 시 다음 IP.""" n = self.port_end - self.port_start + 1 bucket = int(time.time() // (self.session_minutes * 60)) - return self.port_start + (bucket % n) + return self.port_start + ((bucket + self._rotate_offset) % n) + + @property + def current_port(self): + return self._port() if self.enabled else None def playwright_proxy(self) -> dict | None: """Playwright launch(proxy=...) 용 설정. 비활성 시 None(프록시 미사용).""" diff --git a/lps/tests/test_bot_detection.py b/lps/tests/test_bot_detection.py new file mode 100644 index 0000000..78a0d43 --- /dev/null +++ b/lps/tests/test_bot_detection.py @@ -0,0 +1,34 @@ +"""봇 감지 이력 기록 CRUD 테스트 (실 lps_db).""" + +import pytest_asyncio +from sqlalchemy import text + +from crud.bot_detection import BotDetectionLog + + +@pytest_asyncio.fixture +async def bd(db_engine): + async with db_engine.begin() as conn: + await conn.execute(text("TRUNCATE bot_detection")) + return BotDetectionLog() + + +async def test_record_persists_event(bd, db_engine): + await bd.record({ + "source": "coupang", "query": "맥심 커피", "ip_request_no": 7, + "proxy_port": 10003, "elapsed_sec": 42, "marker": "/akam/", + "headless": False, "html_len": 2604, + }) + async with db_engine.begin() as conn: + row = (await conn.execute(text( + "SELECT source, query, ip_request_no, proxy_port, marker FROM bot_detection" + ))).first() + assert row.source == "coupang" and row.query == "맥심 커피" + assert row.ip_request_no == 7 and row.proxy_port == 10003 and row.marker == "/akam/" + + +async def test_record_tolerates_missing_fields(bd, db_engine): + await bd.record({"source": "coupang"}) # 나머지는 None 허용 + async with db_engine.begin() as conn: + cnt = (await conn.execute(text("SELECT count(*) FROM bot_detection"))).scalar() + assert cnt == 1 diff --git a/lps/tests/test_proxy.py b/lps/tests/test_proxy.py index c8c8b64..49cf61e 100644 --- a/lps/tests/test_proxy.py +++ b/lps/tests/test_proxy.py @@ -39,3 +39,14 @@ def test_port_selected_within_range_and_stable_in_window(): def test_single_port_range(): p = _p(port_start=10001, port_end=10001) assert p._port() == 10001 # 포트 1개면 항상 그 포트 + + +def test_rotate_advances_port_immediately(): + p = _p(port_start=10001, port_end=10003) # 포트 3개 + before = p._port() + p.rotate() + after = p._port() + assert after != before # 즉시 다음 포트(새 IP) + assert 10001 <= after <= 10003 + p.rotate(); p.rotate() # 3번 회전하면 한 바퀴 → 원위치 + assert p._port() == before diff --git a/lps/worker_main.py b/lps/worker_main.py index 0033da5..816a15c 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, openai_config from crud.job_crud import JobQueue from crud.negative_cache import NegativeCache +from crud.bot_detection import BotDetectionLog from services.search.proxy import DecodoProxy from services.search.coupang.adapter import CoupangAdapter from services.search.naver.adapter import NaverAdapter @@ -29,8 +30,13 @@ async def main(concurrency: int = 1): # 쿠팡(브라우저, 무거움) + 네이버(오픈API, 가벼움) 동시 검색 → 병합 최저가 # DECODO 프록시: .env 에 값 있으면 쿠팡만 sticky+주기적 회전으로 경유(없으면 직접 연결) proxy = DecodoProxy() - LOG.i(f"DECODO 프록시: {'ON(sticky ' + str(proxy.session_minutes) + '분 회전)' if proxy.enabled else 'OFF(.env 미설정)'}") - adapters = {"coupang": CoupangAdapter(headless=False, proxy=proxy), "naver": NaverAdapter()} + LOG.i(f"DECODO 프록시: {'ON(sticky ' + str(proxy.session_minutes) + '분 회전)' if proxy.enabled else 'OFF(미설정)'}") + # 봇 감지 시: 감지 기록(DB) + 새 IP 로 회전 후 재시도 + bot_log = BotDetectionLog() + adapters = { + "coupang": CoupangAdapter(headless=False, proxy=proxy, on_detect=bot_log.record), + "naver": NaverAdapter(), + } # OpenAI 키 있으면 '같은 상품' AI 판정 + 재검색어 생성 활성화 has_openai = bool(openai_config.api_key) judge = SimilarityJudge() if has_openai else None