diff --git a/docker-compose.yml b/docker-compose.yml index d3d5cb4..ecb8d21 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -164,6 +164,8 @@ services: LPS_PROFILE_DIR: /profiles # Chrome 프로필을 영속 볼륨에 → 재시작해도 cf_clearance 유지(재웜업 회피) # LPS_FALLBACKS: "gmarket,auction,st11" # 오픈마켓 폴백(기본 OFF — 켜기 전 라이브 스모크로 셀렉터 점검) # LPS_JOB_DEADLINE_SEC: "300" # 잡 1건 처리 상한(행 방어) — 기본 300s + # LPS_IP_REQUEST_BUDGET: "3" # IP당 요청 예산 — 도달 시 차단 전 선제 회전(0=비활성). 기본 3 + # LPS_PORT_COOLDOWN_SEC: "1800" # 차단 감지된 프록시 포트 격리 시간 — 기본 max(sticky, 30분) # ── 시크릿 주입(이미지엔 없음 — 필수). 리포 루트 .env 에 값 채움(.env.example 참고) ── OPENAI_API_KEY: ${OPENAI_API_KEY:-} # 비면 AI 판정 OFF NAVER_KEYS: ${NAVER_KEYS:-} # "id1:secret1,id2:secret2" — 비면 네이버 검색 실패 diff --git a/lps/common/database/model/models.py b/lps/common/database/model/models.py index 78c0bce..2e96c5e 100644 --- a/lps/common/database/model/models.py +++ b/lps/common/database/model/models.py @@ -105,6 +105,33 @@ class price_history(MAIN_BASE): ) +class ip_session(MAIN_BASE): + """IP(프록시 포트) 세션 종료 이력 — '이 IP 로 몇 번 요청하고 어떻게 끝났나'를 매 세션 기록. + bot_detection 은 차단된 세션만 남지만 여기엔 무사 종료도 남아, 요청 예산(LPS_IP_REQUEST_BUDGET) + 상한 튜닝의 원천 데이터가 된다. (예: end_reason='block' 의 requests 분포 → 안전 상한 산출)""" + + @staticmethod + def DBType(): + return DBType.MAIN.value + + __tablename__ = "ip_session" + + id = Column(UUID(as_uuid=True), primary_key=True, server_default=text("gen_random_uuid()")) + source = Column(String(20), nullable=False) # coupang 등 + proxy_port = Column(Integer, nullable=True) # 사용 포트(=IP 세션), 프록시 미사용이면 NULL + requests = Column(Integer, nullable=False) # 이 IP 로 보낸 요청 수 + ok_count = Column(Integer, nullable=False, server_default=text("0")) # 성공 검색 수 + blocked_count = Column(Integer, nullable=False, server_default=text("0")) # 차단 감지 수 + elapsed_sec = Column(Integer, nullable=True) # 세션 지속 시간(초) + end_reason = Column(String(20), nullable=False) # budget/block/proxy_error/window/idle/shutdown/rotate + created_at = Column(DateTime(timezone=True), nullable=False, server_default=text("now()")) # 세션 종료 시각 + + __table_args__ = ( + # 상한 튜닝 쿼리(소스·기간별 종료 사유 분포) 최적화 + Index("ix_ip_session_source", "source", "created_at"), + ) + + class bot_detection(MAIN_BASE): """봇 감지 이력 — '이 IP로 몇 번째 요청에서, 어떤 방식으로 차단됐나'를 축적해 패턴 분석. (예: SELECT avg(ip_request_no) → IP당 평균 몇 요청 만에 감지되는지)""" diff --git a/lps/crud/ip_session.py b/lps/crud/ip_session.py new file mode 100644 index 0000000..ee5e709 --- /dev/null +++ b/lps/crud/ip_session.py @@ -0,0 +1,41 @@ +"""IP 세션 종료 이력 CRUD — 세션당 요청 수·종료 사유를 축적(요청 예산 상한 튜닝용).""" + +from sqlalchemy import text + +from common.database.db_session_manager import DB_SESSION_MNG +from common.enums import DBType, DBWRType + + +class IpSessionLog: + DB = DBType.MAIN.value + + async def record(self, event: dict): + """세션 종료 이벤트 1건 저장. 기록 실패가 검색을 막지 않도록 호출부에서 예외를 삼킨다.""" + sql = text(""" + INSERT INTO ip_session (source, proxy_port, requests, ok_count, blocked_count, elapsed_sec, end_reason) + VALUES (:source, :proxy_port, :requests, :ok_count, :blocked_count, :elapsed_sec, :end_reason) + """) + params = {k: event.get(k) for k in + ("source", "proxy_port", "requests", "ok_count", "blocked_count", "elapsed_sec", "end_reason")} + 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) + + async def recent_stats(self, minutes: int = 60) -> dict: + """최근 N분 세션 요약 — 종료 사유별 건수(모니터링·알림용). 예: {"budget": 12, "block": 1}""" + sql = text(""" + SELECT end_reason, count(*) FROM ip_session + WHERE created_at > now() - make_interval(mins => :m) GROUP BY end_reason + """) + s = await DB_SESSION_MNG.start_session(self.DB, DBWRType.DB_READ.value) + try: + rows = (await s.execute(sql, {"m": minutes})).all() + return {r[0]: int(r[1]) for r in rows} + finally: + await DB_SESSION_MNG.end_session(self.DB, DBWRType.DB_READ.value) diff --git a/lps/docs/architecture.md b/lps/docs/architecture.md index 534ff1e..52c131d 100644 --- a/lps/docs/architecture.md +++ b/lps/docs/architecture.md @@ -78,6 +78,7 @@ **핵심 메커니즘** - **IP 회전(DECODO)**: 같은 IP로 계속 두드리면 차단 → 시간창 기반 sticky + 봇감지/전송오류 시 즉시 회전. 감지 이력(`bot_detection`)을 기록해 패턴 분석. **프록시 전송오류(407/터널)** 도 사이트 차단과 구분해 회전. +- **선제 회전(요청 예산)**: IP당 요청 수가 예산(`LPS_IP_REQUEST_BUDGET`, 기본 3 — 실측상 5회 부근 차단)에 닿으면 **차단당하기 전에** 회전. 선제 교체된 포트는 평판이 깨끗해 로테이션 복귀 시 재사용됩니다. 반면 **차단 감지된 포트는 쿨다운**(`LPS_PORT_COOLDOWN_SEC`, 기본 max(sticky, 30분)) 동안 격리 — sticky 만료 후 복귀라 사실상 새 IP. 세션마다 `ip_session`(요청 수·종료 사유)을 남겨 예산 상한을 데이터로 튜닝합니다(쿼리는 database.md). - **시작 프리플라이트 + 웜업**: 기동 시 살아있는 프록시 포트를 선점(egress IP 로그)하고, 챌린지 소스를 미리 1회 풀어 **쿠키를 선점**(나쁜 IP는 회전 재시도) → 실 작업은 웜(빠름). - **동적 리소스 차단**: 이미지·폰트 등을 차단해 대역폭↓. 단 **Turnstile은 리소스 차단을 봇 신호로 감지**하므로, ESM은 챌린지 solving 중(콜드)엔 차단을 풀고 **cf_clearance 확보 후(웜)에만 차단**합니다. - **폴백 데드라인**: 오픈마켓 크롤은 '보강'이라 각 크롤에 시간 상한(기본 15초)을 둬, 한 몰이 안 풀려도 전체 지연이 늘지 않게 합니다. diff --git a/lps/docs/database.md b/lps/docs/database.md index 4a4b8c9..546f946 100644 --- a/lps/docs/database.md +++ b/lps/docs/database.md @@ -14,6 +14,7 @@ | `price_history` | 최저가 이력 — 그래프용 시계열 스냅샷 | | `search_negative` | 네거티브 캐시 — "없음"으로 확인된 상품을 일정 시간 기억 | | `bot_detection` | 봇 감지 이력 — 쿠팡이 차단한 패턴 기록 | +| `ip_session` | IP 세션 종료 이력 — 요청 예산(선제 회전) 상한 튜닝 데이터 | --- @@ -102,6 +103,33 @@ SELECT avg(ip_request_no), count(*) FROM bot_detection; --- +## 5. `ip_session` — IP(프록시 포트) 세션 종료 이력 + +브라우저(=IP 세션)가 끝날 때마다 기록. `bot_detection`은 **차단된** 세션만 남지만, +여기엔 **무사 종료**(예산 선제 회전·시간창 만료 등)도 남아 요청 예산(`LPS_IP_REQUEST_BUDGET`) +상한 튜닝의 원천 데이터가 됩니다. + +| 컬럼 | 뜻 | +|------|-----| +| `source` | 소스(coupang 등) | +| `proxy_port` | 사용 포트(=IP 세션). 프록시 미사용이면 NULL | +| `requests` | 이 IP로 보낸 요청 수 | +| `ok_count` / `blocked_count` | 성공 검색 수 / 차단 감지 수 | +| `elapsed_sec` | 세션 지속 시간(초) | +| `end_reason` | 종료 사유 — `budget`(예산 선제) / `block`(차단) / `proxy_error`(포트 사망) / `window`(시간창 만료) / `idle`(유휴 정리) / `shutdown`(종료) | +| `created_at` | 세션 종료 시각 | + +**예산 튜닝 쿼리** — 차단이 나기 시작하는 요청 수 분포를 보고 상한을 조정: +```sql +-- 종료 사유별 분포(최근 7일): budget 이 대다수 + block 0 이면 예산을 1씩 올려볼 수 있고, +-- block 이 보이면 그 세션들의 requests 최솟값보다 예산을 낮게 유지한다. +SELECT end_reason, count(*), avg(requests)::numeric(5,1) AS avg_req, min(requests), max(requests) + FROM ip_session WHERE created_at > now() - interval '7 days' + GROUP BY end_reason ORDER BY count(*) DESC; +``` + +--- + ## 스키마 생성/관리 - 개발·테스트: SQLAlchemy 모델에서 `create_all`로 자동 생성. diff --git a/lps/docs/operations.md b/lps/docs/operations.md index d97104e..ae2f1d1 100644 --- a/lps/docs/operations.md +++ b/lps/docs/operations.md @@ -191,6 +191,9 @@ docker ps # lps-worker "(healthy)" 확인 - **워커 헬스**: HEALTHCHECK(하트비트<120s)로 행 워커 감지. compose 의 `restart` 는 unhealthy 를 재시작하지 않으므로 **autoheal 컨테이너**(라벨 `autoheal=true` 감시)가 재시작 담당. k8s 는 liveness probe 로 대체. - **잡 데드라인**: 잡 1건 300s 상한(`LPS_JOB_DEADLINE_SEC`) — 크롤 행이 워커 슬롯을 영구 점유하지 못하게 함. +- **IP 선제 회전**: `LPS_IP_REQUEST_BUDGET`(기본 3) — IP당 요청 예산, 도달 시 차단 전에 회전(0=비활성). + `LPS_PORT_COOLDOWN_SEC`(기본 max(sticky, 1800)) — 차단 감지된 포트 격리 시간. 포트 수를 늘리면 + (DECODO_PORT_START/END) 자동 반영 — 코드에 포트 수 하드코딩 없음. 튜닝은 `ip_session` 분석 쿼리(database.md) 참고. **남은 배포 과제**: API 인증·레이트리밋(비용 남용 방지), 다중 레플리카 시 분산 레이트리밋/프록시 IP 조정. **비용**: 대역폭이 원가의 대부분(오픈마켓 크롤) — 같은 상품 재크롤을 줄이는 **TTL 캐시**가 다음 절감 후보. diff --git a/lps/migrations/2026-07-13-ip_session.sql b/lps/migrations/2026-07-13-ip_session.sql new file mode 100644 index 0000000..1ff4d5f --- /dev/null +++ b/lps/migrations/2026-07-13-ip_session.sql @@ -0,0 +1,17 @@ +-- IP(프록시 포트) 세션 종료 이력 — 요청 예산(LPS_IP_REQUEST_BUDGET) 상한 튜닝의 원천 데이터. +-- bot_detection 은 차단된 세션만 남지만, 여기엔 무사 종료(budget/window 등)도 남는다. +-- 적용: psql -h -U -d lps_db -f migrations/2026-07-13-ip_session.sql +-- (postgres-init/init-data/init.sql 의 lps_db 섹션에도 동일 DDL 반영됨 — 신규 설치는 그쪽이 소스) + +CREATE TABLE IF NOT EXISTS ip_session ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + source VARCHAR(20) NOT NULL, -- coupang 등 + proxy_port INTEGER NULL, -- 사용 포트(=IP 세션), 프록시 미사용이면 NULL + requests INTEGER NOT NULL, -- 이 IP 로 보낸 요청 수 + ok_count INTEGER NOT NULL DEFAULT 0, -- 성공 검색 수 + blocked_count INTEGER NOT NULL DEFAULT 0, -- 차단 감지 수 + elapsed_sec INTEGER NULL, -- 세션 지속 시간(초) + end_reason VARCHAR(20) NOT NULL, -- budget/block/proxy_error/window/idle/shutdown/rotate + created_at TIMESTAMPTZ NOT NULL DEFAULT now() -- 세션 종료 시각 +); +CREATE INDEX IF NOT EXISTS ix_ip_session_source ON ip_session (source, created_at); diff --git a/lps/services/search/browser_base.py b/lps/services/search/browser_base.py index f2842b5..486322c 100644 --- a/lps/services/search/browser_base.py +++ b/lps/services/search/browser_base.py @@ -33,6 +33,10 @@ _BLOCKED_RESOURCES = {"image", "media", "font", "stylesheet"} _CHROME_CHANNEL = os.environ.get("LPS_CHROME_CHANNEL", "chrome") _CHROME_EXECUTABLE = os.environ.get("LPS_CHROME_EXECUTABLE") or None +# IP(포트 세션)당 요청 예산 — 도달하면 차단당하기 **전에** 선제 회전해 IP 평판을 보존한다. +# 실측상 5회 부근에서 차단된 이력이 있어 보수적으로 3회. 0 이면 비활성(시간창 회전만). +_IP_REQUEST_BUDGET = int(os.environ.get("LPS_IP_REQUEST_BUDGET", "3")) + def detect_block(html: str, product_count: int, markers: tuple, min_len: int) -> str | None: """0건 응답의 차단 여부 판정(순수 함수). 반환: 차단 마커(차단) 또는 None(정상 빈결과). @@ -73,7 +77,8 @@ class BrowserSearchAdapter(SearchAdapter): max_proxy_retries: int = 2 # 프록시 전송오류(포트 사망) 시 IP 회전 재시도 횟수 def __init__(self, headless: bool = False, user_data_dir: str | None = None, rate_limiter: RateLimiter | None = None, - proxy=None, block_resources: bool | None = None, on_detect=None, max_block_retries: int = 1): + proxy=None, block_resources: bool | None = None, on_detect=None, max_block_retries: int = 1, + ip_request_budget: int | None = None, on_session_end=None): self._headless = headless self._user_data_dir = user_data_dir or f"/tmp/lps_{self.source}_profile" self._rl = rate_limiter or RateLimiter() @@ -82,6 +87,8 @@ class BrowserSearchAdapter(SearchAdapter): self._block_active = self._block_resources # 요청별 실제 차단 여부(_blocking_now 로 갱신) self._on_detect = on_detect # async def(event: dict) — 감지 영속화(선택) self._max_block_retries = max_block_retries + self._ip_budget = _IP_REQUEST_BUDGET if ip_request_budget is None else ip_request_budget + self._on_session_end = on_session_end # async def(event: dict) — IP 세션 종료 기록(선택, 상한 튜닝 데이터) self._pw = None self._ctx = None self._launched_at = 0.0 @@ -91,6 +98,9 @@ class BrowserSearchAdapter(SearchAdapter): self._lock = asyncio.Lock() self._ok = 0 self._blocked = 0 + self._sess_ok = 0 # 현재 IP 세션의 성공/차단(세션 종료 기록용, 재기동 시 리셋) + self._sess_blocked = 0 + self._end_reason = None # 이번 세션이 끝나는 이유(budget/block/proxy_error/window/idle/shutdown) self._last_used = 0.0 # 마지막 검색 시각(monotonic) — 유휴 브라우저 정리 판단용 self._cdp = None # CDP 세션(실제 네트워크 바이트 계측용). 미지원 시 None → DOM 크기 폴백 self._net_bytes = 0 # 현재 검색의 실제 전송 바이트(encodedDataLength 누적) @@ -143,6 +153,8 @@ class BrowserSearchAdapter(SearchAdapter): if self._ctx is not None: if self._recycle_due(): LOG.d(f"[{self.source}] 브라우저 재기동(IP 회전)") + if self._end_reason is None: # force 가 아닌 시간창 만료 재기동 + self._end_reason = "window" await self._close_ctx() else: return @@ -163,16 +175,35 @@ class BrowserSearchAdapter(SearchAdapter): await self._ctx.route("**/*", self._route) self._launched_at = time.monotonic() self._ip_requests = 0 + self._sess_ok = self._sess_blocked = 0 + self._end_reason = None self._force_recycle = False async def _close_ctx(self): self._cdp = None # 컨텍스트와 함께 CDP 세션도 죽음 → 다음 검색 때 재부착 if self._ctx is not None: + await self._record_session_end() try: await self._ctx.close() finally: self._ctx = None + async def _record_session_end(self): + """IP 세션 종료 1건 기록 — '이 IP 로 몇 번 요청하고 어떻게 끝났나'. 예산(상한) 튜닝의 원천 데이터. + 요청이 없던 세션(유휴 정리 등)은 노이즈라 기록하지 않는다. 기록 실패가 검색을 막지 않는다.""" + if self._on_session_end is None or self._ip_requests == 0: + self._end_reason = None + return + event = {"source": self.source, "proxy_port": self._current_port, + "requests": self._ip_requests, "ok_count": self._sess_ok, "blocked_count": self._sess_blocked, + "elapsed_sec": int(time.monotonic() - self._launched_at), + "end_reason": self._end_reason or "window"} + self._end_reason = None + try: + await self._on_session_end(event) + except Exception as ex: + LOG.e_no_callstack(f"[{self.source}] IP 세션 기록 실패(무시): {ex}") + def _add_net(self, event): """CDP Network.loadingFinished 콜백 — 실제 전송 바이트(encodedDataLength) 누적.""" try: @@ -196,12 +227,18 @@ class BrowserSearchAdapter(SearchAdapter): """이 어댑터가 프록시(DECODO)를 경유하는지 — 대역폭 비용 귀속용.""" return bool(self._proxy and self._proxy.enabled) - def _rotate_ip(self, reason: str): - """즉시 다음 IP(포트)로 회전 예약 + 다음 _ensure_browser 에서 브라우저 재기동.""" + def _budget_reached(self) -> bool: + """현재 IP 로 요청 예산을 소진했는지(선제 회전 트리거). 프록시 미사용·예산 0(비활성)이면 False.""" + return self.uses_proxy and self._ip_budget > 0 and self._ip_requests >= self._ip_budget + + def _rotate_ip(self, reason: str, kind: str = "rotate", warn: bool = True): + """즉시 다음 IP(포트)로 회전 예약 + 다음 _ensure_browser 에서 브라우저 재기동. + kind 는 세션 종료 사유로 기록된다(budget=선제/block=차단/proxy_error=포트사망).""" if self._proxy and self._proxy.enabled: self._proxy.rotate() self._force_recycle = True - LOG.w(f"[{self.source}] IP 회전 — {reason}") + self._end_reason = kind + (LOG.w if warn else LOG.i)(f"[{self.source}] IP 회전 — {reason}") # ---- 검색(프록시 전송오류·봇 감지 → IP 회전 인라인 재시도) -------- async def search(self, query: str, limit: int = 40) -> list[NormalizedProduct]: @@ -210,6 +247,10 @@ class BrowserSearchAdapter(SearchAdapter): proxy_retries, block_retries = self.max_proxy_retries, self._max_block_retries while True: await self._rl.wait() + # 예산 도달 → 차단당하기 전에 선제 회전. 이 포트는 불탄 게 아니라 쿨다운 없이 + # 로테이션 복귀 시 재사용된다(IP 평판 보존이 목적). + if self._budget_reached(): + self._rotate_ip(f"요청예산 {self._ip_budget}회 도달 — 선제 회전", kind="budget", warn=False) await self._ensure_browser() self._ip_requests += 1 page = self._ctx.pages[0] if self._ctx.pages else await self._ctx.new_page() @@ -225,7 +266,8 @@ class BrowserSearchAdapter(SearchAdapter): # 프록시 전송 실패(포트/IP 사망·407)면 사이트 문제가 아니므로 IP 회전 후 재시도 if self.uses_proxy and is_proxy_error(str(ex)) and proxy_retries > 0: proxy_retries -= 1 - self._rotate_ip(f"프록시 전송오류({type(ex).__name__}) 재시도 {self.max_proxy_retries - proxy_retries}/{self.max_proxy_retries}") + self._proxy.mark_burned(self._current_port) # 죽은 포트 — 쿨다운 뒤 복귀(sticky 만료로 새 IP) + self._rotate_ip(f"프록시 전송오류({type(ex).__name__}) 재시도 {self.max_proxy_retries - proxy_retries}/{self.max_proxy_retries}", kind="proxy_error") continue raise AdapterError(f"{self.source} 검색 실패: {ex}", source=self.source) from ex @@ -234,6 +276,7 @@ class BrowserSearchAdapter(SearchAdapter): products = self._parse(html) if products: self._ok += 1 + self._sess_ok += 1 LOG.d(f"[{self.source}] query={query!r} → {len(products)}건 (limit {limit}, ip_req#{self._ip_requests})") return products[:limit] @@ -241,13 +284,18 @@ class BrowserSearchAdapter(SearchAdapter): blocked = marker is not None self._blocked += 1 if blocked: + self._sess_blocked += 1 await self._report_detection(query, marker, len(html)) + if self.uses_proxy: + self._proxy.mark_burned(self._current_port) # 불탄 포트 — 쿨다운 격리(로테이션이 건너뜀) if blocked and self.uses_proxy and block_retries > 0: block_retries -= 1 - self._rotate_ip(f"봇 감지 재시도 {self._max_block_retries - block_retries}/{self._max_block_retries}") + self._rotate_ip(f"봇 감지 재시도 {self._max_block_retries - block_retries}/{self._max_block_retries}", kind="block") continue + if blocked: # 재시도 소진/비활성 — 불탄 포트로 다음 검색을 하지 않도록 회전만 예약하고 포기 + self._rotate_ip("봇 감지 — 다음 검색은 새 IP", kind="block") raise AdapterError(f"{self.source} 결과 없음/차단 (query={query!r}, blocked={blocked})", source=self.source, blocked=blocked) async def _report_detection(self, query: str, marker: str, html_len: int): @@ -279,12 +327,14 @@ class BrowserSearchAdapter(SearchAdapter): async with self._lock: if self._ctx is not None and time.monotonic() - self._last_used >= idle_sec: LOG.d(f"[{self.source}] 유휴 {idle_sec:.0f}s 초과 → 브라우저 정리(다음 검색 때 재기동)") + if self._end_reason is None: + self._end_reason = "idle" await self._close_ctx() async def close(self): - if self._ctx is not None: - await self._ctx.close() - self._ctx = None + if self._end_reason is None: + self._end_reason = "shutdown" + await self._close_ctx() if self._pw is not None: await self._pw.stop() self._pw = None diff --git a/lps/services/search/proxy.py b/lps/services/search/proxy.py index 977c1c4..26c9e5e 100644 --- a/lps/services/search/proxy.py +++ b/lps/services/search/proxy.py @@ -9,6 +9,7 @@ Decodo residential 은 **포트 기반 sticky** 모델이다: 자격증명/엔드포인트는 config.local.toml [DecodoConfig] 에서 로드(시크릿). """ +import os import time from urllib.parse import quote @@ -28,24 +29,52 @@ class DecodoProxy: self.port_end = cfg.port_end self.session_minutes = cfg.session_minutes or 10 self._rotate_offset = 0 # 봇 감지 등으로 '즉시 회전'이 필요할 때 증가 + # 불탄(차단 감지된) 포트 격리 시간. sticky 만료(session_minutes) 이상이어야 + # 쿨다운 복귀 시 같은 포트라도 사실상 새 IP 가 배정된다. 기본 max(sticky, 30분). + self.cooldown_sec = int(os.environ.get("LPS_PORT_COOLDOWN_SEC", "0")) \ + or max(self.session_minutes * 60, 1800) + self._burned: dict[int, float] = {} # port → 쿨다운 만료 시각(monotonic) @property def enabled(self) -> bool: return all([self.host, self.username, self.password, self.port_start, self.port_end]) def rotate(self): - """시간창과 무관하게 즉시 다음 포트(=새 IP)로 회전. 봇 감지 시 호출.""" + """시간창과 무관하게 즉시 다음 포트(=새 IP)로 회전. 봇 감지·예산 도달 시 호출.""" self._rotate_offset += 1 def seed_offset(self, k: int): """워커별 시작 포트 분산용 — 동시 워커가 같은 포트(=같은 IP)를 쓰지 않도록 시작점을 벌린다.""" self._rotate_offset = k + def mark_burned(self, port: int | None, cooldown_sec: float | None = None): + """차단 감지된 포트를 쿨다운 격리 — _port() 가 만료 전까지 건너뛴다. + 선제(예산) 회전된 포트는 부르지 않는다 — 불탄 게 아니므로 로테이션 복귀 시 재사용.""" + if port is None: + return + self._burned[port] = time.monotonic() + (cooldown_sec if cooldown_sec is not None else self.cooldown_sec) + LOG.i(f"[proxy] 포트 {port} 쿨다운 {int(cooldown_sec or self.cooldown_sec)}s — 활성 {self.available_ports()}/{self.port_end - self.port_start + 1}") + + def available_ports(self) -> int: + """쿨다운 중이 아닌 포트 수(관측·알림용).""" + self._prune_burned() + return (self.port_end - self.port_start + 1) - len(self._burned) + + def _prune_burned(self): + now = time.monotonic() + self._burned = {p: t for p, t in self._burned.items() if t > now} + def _port(self) -> int: - """시간창 + 수동 오프셋 기반 포트 선택. 창 안에선 동일 IP, rotate()나 창 변화 시 다음 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 + self._rotate_offset) % n) + self._prune_burned() + for k in range(n): + port = self.port_start + ((bucket + self._rotate_offset + k) % n) + if port not in self._burned: + return port + return min(self._burned, key=self._burned.get) @property def current_port(self): diff --git a/lps/tests/test_browser_base.py b/lps/tests/test_browser_base.py index f370c96..8a6beaa 100644 --- a/lps/tests/test_browser_base.py +++ b/lps/tests/test_browser_base.py @@ -49,8 +49,9 @@ class _MockCtx: class _MockProxy: enabled = True session_minutes = 10 - def __init__(self): self.rotations = 0 + def __init__(self): self.rotations, self.burned = 0, [] def rotate(self): self.rotations += 1 + def mark_burned(self, port, cooldown_sec=None): self.burned.append(port) def playwright_proxy(self): return None @property def current_port(self): return 10001 @@ -70,6 +71,7 @@ class _MockAdapter(BrowserSearchAdapter): self._ctx = _MockCtx(self._page) self._force_recycle = False self._ip_requests = 0 + self._current_port = self._proxy.current_port if self._proxy else None # 실제 _ensure_browser 와 동일 async def _ensure_net_meter(self, page): pass # CDP 없음 → last_bytes=DOM 크기 async def _wait_ready(self, page): pass @@ -110,11 +112,12 @@ async def test_non_proxy_goto_error_raises_no_rotation(): async def test_persistent_block_exhausts_and_raises_blocked(): - # 차단 2회 연속(max_block_retries=1) → 회전 1회 후 소진 → blocked=True 로 실패 + # 차단 2회 연속(max_block_retries=1) → 재시도 회전 1회 + 소진 후 '다음 검색용' 회전 1회 → blocked=True 로 실패 ad = _ad(["x", "x"]) with pytest.raises(AdapterError) as ei: await ad.search("q") - assert ei.value.blocked is True and ad._proxy.rotations == 1 + assert ei.value.blocked is True and ad._proxy.rotations == 2 + assert ad._proxy.burned == [10001, 10001] # 차단마다 해당 포트 쿨다운 격리 # ── 라이브 스모크(옵트인): 실제 사이트 셀렉터·안티봇 드리프트 감지 ────── diff --git a/lps/tests/test_ip_session.py b/lps/tests/test_ip_session.py new file mode 100644 index 0000000..8f34bf6 --- /dev/null +++ b/lps/tests/test_ip_session.py @@ -0,0 +1,136 @@ +"""IP 선제 로테이션 테스트 — 요청 예산 판정·세션 종료 기록(브라우저 불필요) + CRUD(실 lps_db).""" + +import pytest_asyncio +from sqlalchemy import text + +from config.config_models import DecodoConfig +from crud.ip_session import IpSessionLog +from services.search.coupang.adapter import CoupangAdapter +from services.search.proxy import DecodoProxy + + +def _proxy(): + return DecodoProxy(DecodoConfig(host="gate.decodo.com", username="u", password="p", + port_start=10001, port_end=10010, session_minutes=10)) + + +def _adapter(**kw): + # __init__ 은 브라우저를 띄우지 않는다 — 예산/세션 부기 로직만 검증 + return CoupangAdapter(headless=True, user_data_dir="/tmp/lps_test_profile", **kw) + + +# ---- 요청 예산(선제 회전) 판정 ------------------------------------------ + +def test_budget_reached_only_with_proxy(): + ad = _adapter(ip_request_budget=3) + ad._ip_requests = 3 + assert ad._budget_reached() is False # 프록시 미사용 — 예산 개념 없음 + ad._proxy = _proxy() + assert ad._budget_reached() is True + + +def test_budget_zero_disables_preemptive_rotation(): + ad = _adapter(ip_request_budget=0, proxy=_proxy()) + ad._ip_requests = 999 + assert ad._budget_reached() is False + + +def test_budget_not_reached_below_threshold(): + ad = _adapter(ip_request_budget=3, proxy=_proxy()) + ad._ip_requests = 2 + assert ad._budget_reached() is False + + +def test_default_budget_is_conservative(): + ad = _adapter(proxy=_proxy()) + assert ad._ip_budget == 3 # 실측상 5회 부근 차단 → 기본 3회 + + +def test_rotate_ip_records_end_reason_and_recycles(): + ad = _adapter(proxy=_proxy()) + before = ad._proxy.current_port + ad._rotate_ip("예산 도달", kind="budget", warn=False) + assert ad._end_reason == "budget" + assert ad._force_recycle is True + assert ad._proxy.current_port != before # 즉시 다음 포트 + + +# ---- 세션 종료 기록 ------------------------------------------------------- + +async def test_session_end_event_shape(): + events = [] + + async def cb(e): + events.append(e) + + ad = _adapter(proxy=_proxy(), on_session_end=cb) + ad._ip_requests, ad._sess_ok, ad._sess_blocked = 3, 2, 1 + ad._current_port, ad._end_reason = 10005, "budget" + await ad._record_session_end() + assert events == [{ + "source": "coupang", "proxy_port": 10005, "requests": 3, + "ok_count": 2, "blocked_count": 1, "elapsed_sec": events[0]["elapsed_sec"], + "end_reason": "budget", + }] + assert ad._end_reason is None # 기록 후 리셋 + + +async def test_session_end_skips_empty_session(): + events = [] + + async def cb(e): + events.append(e) + + ad = _adapter(on_session_end=cb) + ad._ip_requests = 0 # 요청 없던 세션(유휴 정리 등)은 노이즈 — 미기록 + await ad._record_session_end() + assert events == [] + + +async def test_session_end_defaults_to_window_reason(): + events = [] + + async def cb(e): + events.append(e) + + ad = _adapter(on_session_end=cb) + ad._ip_requests = 2 # end_reason 미지정 = 시간창 만료 재기동 + await ad._record_session_end() + assert events[0]["end_reason"] == "window" + + +async def test_session_end_callback_failure_is_swallowed(): + async def boom(e): + raise RuntimeError("db down") + + ad = _adapter(on_session_end=boom) + ad._ip_requests = 1 + await ad._record_session_end() # 예외가 검색을 막지 않는다 + + +# ---- CRUD (실 lps_db) ---------------------------------------------------- + +@pytest_asyncio.fixture +async def ip_log(db_engine): + async with db_engine.begin() as conn: + await conn.execute(text("TRUNCATE ip_session")) + return IpSessionLog() + + +async def test_record_persists_session(ip_log, db_engine): + await ip_log.record({"source": "coupang", "proxy_port": 10003, "requests": 3, + "ok_count": 3, "blocked_count": 0, "elapsed_sec": 120, "end_reason": "budget"}) + async with db_engine.begin() as conn: + row = (await conn.execute(text( + "SELECT source, proxy_port, requests, ok_count, end_reason FROM ip_session" + ))).first() + assert row.source == "coupang" and row.proxy_port == 10003 + assert row.requests == 3 and row.ok_count == 3 and row.end_reason == "budget" + + +async def test_recent_stats_groups_by_reason(ip_log, db_engine): + for reason in ("budget", "budget", "block"): + await ip_log.record({"source": "coupang", "requests": 3, "ok_count": 3, + "blocked_count": 0, "end_reason": reason}) + stats = await ip_log.recent_stats(60) + assert stats == {"budget": 2, "block": 1} diff --git a/lps/tests/test_proxy.py b/lps/tests/test_proxy.py index 49cf61e..86f348b 100644 --- a/lps/tests/test_proxy.py +++ b/lps/tests/test_proxy.py @@ -50,3 +50,38 @@ def test_rotate_advances_port_immediately(): assert 10001 <= after <= 10003 p.rotate(); p.rotate() # 3번 회전하면 한 바퀴 → 원위치 assert p._port() == before + + +def test_burned_port_is_skipped(): + p = _p(port_start=10001, port_end=10003) + burned = p._port() + p.mark_burned(burned) + assert p._port() != burned # 쿨다운 중인 포트는 건너뜀 + assert p.available_ports() == 2 + + +def test_burned_port_returns_after_cooldown(): + p = _p(port_start=10001, port_end=10003) + burned = p._port() + p.mark_burned(burned, cooldown_sec=0) # 즉시 만료 + assert p._port() == burned # 만료 후엔 다시 사용 가능 + assert p.available_ports() == 3 + + +def test_all_burned_falls_back_to_earliest_expiry(): + p = _p(port_start=10001, port_end=10002) # 포트 2개 + first, second = 10001, 10002 + p.mark_burned(first, cooldown_sec=10) # 먼저 만료 + p.mark_burned(second, cooldown_sec=999) + assert p._port() == first # 전 포트 쿨다운 — 만료 임박 포트 사용(가용성 우선) + + +def test_mark_burned_ignores_none_port(): + p = _p() + p.mark_burned(None) # 프록시 미사용 세션 — no-op + assert p.available_ports() == 10 + + +def test_cooldown_default_at_least_30min(): + assert _p(session_minutes=10).cooldown_sec == 1800 # max(sticky 600, 1800) + assert _p(session_minutes=60).cooldown_sec == 3600 # sticky 가 더 길면 sticky 만큼 diff --git a/lps/worker_main.py b/lps/worker_main.py index c0195fc..7ed1d3d 100644 --- a/lps/worker_main.py +++ b/lps/worker_main.py @@ -17,6 +17,7 @@ from config.server_configs import web_server_config, openai_config, decodo_confi from crud.job_crud import JobQueue from crud.negative_cache import NegativeCache from crud.bot_detection import BotDetectionLog +from crud.ip_session import IpSessionLog from crud.price_history import PriceHistory from services.search.proxy import DecodoProxy from services.search.coupang.adapter import CoupangAdapter @@ -56,6 +57,7 @@ def _build_worker(i: int, concurrency: int, has_openai: bool, neg_cache, history n = proxy.port_end - proxy.port_start + 1 proxy.seed_offset(i * max(1, n // concurrency)) bot_log = BotDetectionLog() + ip_log = IpSessionLog() suffix = f"_w{i}" if concurrency > 1 else "" def _pf(source): # 워커별 Chrome 프로필 경로(중복 실행 시 ProcessSingleton 충돌 방지) @@ -64,7 +66,8 @@ def _build_worker(i: int, concurrency: int, has_openai: bool, neg_cache, history return f"{base}/lps_{source}{suffix}" adapters = { - "coupang": CoupangAdapter(headless=False, user_data_dir=_pf("coupang"), proxy=proxy, on_detect=bot_log.record), + "coupang": CoupangAdapter(headless=False, user_data_dir=_pf("coupang"), proxy=proxy, + on_detect=bot_log.record, on_session_end=ip_log.record), "naver": NaverAdapter(), # httpx 직접(프록시 미경유) — 워커별 인스턴스(last_bytes 경합 회피) } # 폴백은 기본 비활성(LPS_FALLBACKS 로 켬 — 상단 주석 참고). 켤 땐 데드라인이 상한이라 @@ -72,9 +75,11 @@ def _build_worker(i: int, concurrency: int, has_openai: bool, neg_cache, history fallback_adapters = {} for name in _enabled_fallbacks(): if name == "st11": - fallback_adapters[name] = ElevenStAdapter(headless=False, user_data_dir=_pf("st11"), proxy=proxy, on_detect=bot_log.record, max_block_retries=0) + fallback_adapters[name] = ElevenStAdapter(headless=False, user_data_dir=_pf("st11"), proxy=proxy, + on_detect=bot_log.record, on_session_end=ip_log.record, max_block_retries=0) else: - fallback_adapters[name] = EsmAdapter(name, headless=False, user_data_dir=_pf(name), proxy=proxy, on_detect=bot_log.record, max_block_retries=0) + fallback_adapters[name] = EsmAdapter(name, headless=False, user_data_dir=_pf(name), proxy=proxy, + on_detect=bot_log.record, on_session_end=ip_log.record, max_block_retries=0) # AI 도 워커별 인스턴스 — 공유 상태(last_usage) 경합 원천 제거 judge = SimilarityJudge() if has_openai else None keyword_gen = KeywordGenerator() if has_openai else None diff --git a/postgres-init/init-data/init.sql b/postgres-init/init-data/init.sql index 09ada06..7995dee 100644 --- a/postgres-init/init-data/init.sql +++ b/postgres-init/init-data/init.sql @@ -744,6 +744,20 @@ CREATE TABLE IF NOT EXISTS price_history ( ); CREATE INDEX IF NOT EXISTS ix_price_history_product ON price_history (product_code, triggered_at); +-- IP(프록시 포트) 세션 종료 이력 — 요청 예산(LPS_IP_REQUEST_BUDGET) 상한 튜닝의 원천 데이터 +CREATE TABLE IF NOT EXISTS ip_session ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + source VARCHAR(20) NOT NULL, -- coupang 등 + proxy_port INTEGER NULL, -- 사용 포트(=IP 세션), 프록시 미사용이면 NULL + requests INTEGER NOT NULL, -- 이 IP 로 보낸 요청 수 + ok_count INTEGER NOT NULL DEFAULT 0, -- 성공 검색 수 + blocked_count INTEGER NOT NULL DEFAULT 0, -- 차단 감지 수 + elapsed_sec INTEGER NULL, -- 세션 지속 시간(초) + end_reason VARCHAR(20) NOT NULL, -- budget/block/proxy_error/window/idle/shutdown/rotate + created_at TIMESTAMPTZ NOT NULL DEFAULT now() -- 세션 종료 시각 +); +CREATE INDEX IF NOT EXISTS ix_ip_session_source ON ip_session (source, created_at); + -- 봇 감지 이력 — IP·요청순번·차단방식 축적(패턴 분석) CREATE TABLE IF NOT EXISTS bot_detection ( id uuid PRIMARY KEY DEFAULT gen_random_uuid(),