feat(lps): DECODO residential 프록시 — 쿠팡 sticky 세션 + 주기적 IP 회전

datacenter IP + Akamai 대응. 매 요청 IP 변경은 쿠키-IP 불일치로 재챌린지를
유발하므로, sticky 세션(일정 시간 같은 IP) + 주기적 회전 방식 사용.

- search/proxy.DecodoProxy: username 에 -session-<시간창id>-sessionduration-<분> 부착
  → 창 안에선 같은 IP, 창이 지나면 새 IP. 자격증명은 .env(DECODO_*), 4개 다 있어야 활성
- coupang/adapter: proxy 주입 + 세션창 경과 시 브라우저 재기동(IP 회전). 무프록시면 직접 연결
- worker_main: 쿠팡에만 DecodoProxy 경유(네이버는 공식 API라 미적용)
- .env.example: DECODO_HOST/PORT/USERNAME/PASSWORD/SESSION_MINUTES 슬롯
- tests: 활성조건·sticky username·server형식·세션창 회전 4건 → 전체 43/43

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

lps/.env.example

@@ -13,3 +13,12 @@ NAVER_CLIENT_SECRET=""
 
 # ── AI (상품 유사도 판정 / 검색어 생성) — OpenAI ──────────────────────
 OPENAI_API_KEY=""
+
+# ── DECODO residential 프록시 (쿠팡 전용, sticky 세션+주기적 회전) ─────
+# DECODO 대시보드의 residential 게이트웨이/자격증명. 4개 다 채워야 활성화(비면 프록시 미사용).
+# username 은 base 값만(코드가 -session-<id>-sessionduration-<분> 을 자동으로 붙임).
+DECODO_HOST=""              # 예: gate.decodo.com
+DECODO_PORT=""              # 예: 7000
+DECODO_USERNAME=""
+DECODO_PASSWORD=""
+DECODO_SESSION_MINUTES="10" # 같은 IP 유지 시간(분) → 이 주기로 회전

lps/services/search/coupang/adapter.py

@@ -9,6 +9,7 @@ persistent context 로 브라우저를 재사용하므로 챌린지는 (쿠키
 """
 
 import asyncio
+import time
 from urllib.parse import quote
 
 from patchright.async_api import async_playwright
@@ -26,26 +27,45 @@ _BLOCK_MARKERS = ("sec-if-cpt-container", "Powered and protected", "/akam/")
 class CoupangAdapter(SearchAdapter):
     source = "coupang"
 
-    def __init__(self, headless: bool = False, user_data_dir: str = "/tmp/lps_coupang_profile", rate_limiter: RateLimiter | None = None):
+    def __init__(self, headless: bool = False, user_data_dir: str = "/tmp/lps_coupang_profile", rate_limiter: RateLimiter | None = None, proxy=None):
         self._headless = headless
         self._user_data_dir = user_data_dir
         self._rl = rate_limiter or RateLimiter()
+        self._proxy = proxy  # DecodoProxy 등 (없으면 직접 연결)
         self._pw = None
         self._ctx = None
+        self._launched_at = 0.0
         self._lock = asyncio.Lock()
         self._ok = 0
         self._blocked = 0
 
+    def _recycle_due(self) -> bool:
+        """프록시 sticky 세션창이 지났으면 브라우저를 재기동해 새 IP 를 받는다."""
+        if not (self._proxy and self._proxy.enabled):
+            return False
+        return (time.monotonic() - self._launched_at) > self._proxy.session_minutes * 60
+
     async def _ensure_browser(self):
         if self._ctx is not None:
-            return
+            if self._recycle_due():
-        self._pw = await async_playwright().start()
+                LOG.d("[coupang] 프록시 세션창 경과 → 브라우저 재기동(IP 회전)")
-        self._ctx = await self._pw.chromium.launch_persistent_context(
+                await self._close_ctx()
-            user_data_dir=self._user_data_dir,
+            else:
-            channel="chrome",
+                return
-            headless=self._headless,
+        if self._pw is None:
-            no_viewport=True,
+            self._pw = await async_playwright().start()
-        )
+        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._ctx = await self._pw.chromium.launch_persistent_context(**kwargs)
+        self._launched_at = time.monotonic()
+
+    async def _close_ctx(self):
+        if self._ctx is not None:
+            try:
+                await self._ctx.close()
+            finally:
+                self._ctx = None
 
     async def search(self, query: str, limit: int = 40) -> list[NormalizedProduct]:
         async with self._lock:  # 인스턴스 내 검색은 직렬화(브라우저 컨텍스트 공유)

lps/services/search/proxy.py

@@ -1,25 +1,47 @@
-"""프록시 풀 인터페이스.
+"""DECODO(구 Smartproxy) residential 프록시 제공자.
 
-현재는 무프록시로 시작(residential 프록시 미보유). 인터페이스만 두고 off 상태로 동작한다.
+쿠팡(Akamai) 전용. 핵심: **sticky 세션 + 주기적 회전**.
-쿠팡 차단이 발생하면 residential/mobile 프록시 목록을 주입해 로테이션을 켠다.
+매 요청 IP 변경은 Akamai 가 '쿠키-IP 불일치'로 재챌린지/차단하므로 금물.
-(datacenter IP 는 금방 차단되므로 residential/mobile 을 전제로 설계 — new.md)
+→ 일정 시간(session_minutes) 같은 IP 유지 후 회전한다.
+
+DECODO 는 username 에 세션 파라미터를 붙여 sticky 를 구현한다:
+  <username>-session-<id>-sessionduration-<minutes>
+세션 id 를 '시간창(now // window)'으로 만들면 창 안에선 같은 IP, 창이 바뀌면 새 IP 가 된다.
+자격증명은 .env(DECODO_*) 에서 로드(시크릿).
 """
 
-import itertools
+import os
-from typing import Optional
+import time
 
 
-class ProxyPool:
+class DecodoProxy:
-    def __init__(self, proxies: Optional[list[str]] = None):
+    def __init__(self, host=None, port=None, username=None, password=None, session_minutes=None, sticky=True):
-        self._proxies = list(proxies or [])
+        self.host = host or os.environ.get("DECODO_HOST")
-        self._cycle = itertools.cycle(self._proxies) if self._proxies else None
+        self.port = port or os.environ.get("DECODO_PORT")
+        self.username = username or os.environ.get("DECODO_USERNAME")
+        self.password = password or os.environ.get("DECODO_PASSWORD")
+        self.session_minutes = int(session_minutes or os.environ.get("DECODO_SESSION_MINUTES") or 10)
+        self.sticky = sticky
 
     @property
     def enabled(self) -> bool:
-        return bool(self._proxies)
+        return all([self.host, self.port, self.username, self.password])
 
-    def get(self) -> Optional[str]:
+    def _session_id(self) -> int:
-        """다음 프록시 URL 반환. 비어있으면(off) None → 어댑터는 직접 연결."""
+        """시간창 기반 세션 id — 같은 창에선 동일, 창이 지나면 증가(→새 IP)."""
-        if self._cycle is None:
+        return int(time.time() // (self.session_minutes * 60))
+
+    def _session_username(self) -> str:
+        if not self.sticky:
+            return self.username
+        return f"{self.username}-session-{self._session_id()}-sessionduration-{self.session_minutes}"
+
+    def playwright_proxy(self) -> dict | None:
+        """Playwright launch(proxy=...) 용 설정. 비활성 시 None(프록시 미사용)."""
+        if not self.enabled:
             return None
-        return next(self._cycle)
+        return {
+            "server": f"http://{self.host}:{self.port}",
+            "username": self._session_username(),
+            "password": self.password,
+        }

lps/tests/test_proxy.py

@@ -0,0 +1,40 @@
+"""DECODO 프록시 제공자 테스트 (순수, 네트워크 불필요)."""
+
+from services.search.proxy import DecodoProxy
+
+
+def _p(**kw):
+    base = dict(host="gate.decodo.com", port="7000", username="user1", password="pw", session_minutes=10)
+    base.update(kw)
+    return DecodoProxy(**base)
+
+
+def test_disabled_when_credentials_missing():
+    assert DecodoProxy(host="", port="", username="", password="").enabled is False
+    assert DecodoProxy(host="h", port="7000", username="", password="pw").enabled is False
+    assert _p().playwright_proxy() is not None  # 다 채우면 활성
+    assert DecodoProxy(host="", port="", username="", password="").playwright_proxy() is None
+
+
+def test_playwright_proxy_shape_and_sticky_username():
+    cfg = _p().playwright_proxy()
+    assert cfg["server"] == "http://gate.decodo.com:7000"
+    assert cfg["password"] == "pw"
+    # sticky: base username + 세션/유지시간 파라미터
+    assert cfg["username"].startswith("user1-session-")
+    assert cfg["username"].endswith("-sessionduration-10")
+
+
+def test_non_sticky_uses_plain_username():
+    cfg = _p(sticky=False).playwright_proxy()
+    assert cfg["username"] == "user1"
+
+
+def test_session_id_rotates_by_time_window():
+    p = _p(session_minutes=10)
+    sid = p._session_id()
+    assert isinstance(sid, int)
+    # 같은 창 안에서는 동일해야 함(연속 호출)
+    assert p._session_id() == sid
+    # 창 크기가 다르면(1분) 버킷 값도 달라짐(더 큰 수) — 시간창 분해 검증
+    assert _p(session_minutes=1)._session_id() > sid

lps/worker_main.py

@@ -12,6 +12,7 @@ from common.logger import LOG
 from config.server_configs import web_server_config
 from crud.job_crud import JobQueue
 from crud.negative_cache import NegativeCache
+from services.search.proxy import DecodoProxy
 from services.search.coupang.adapter import CoupangAdapter
 from services.search.naver.adapter import NaverAdapter
 from services.ai.similarity import SimilarityJudge
@@ -26,7 +27,10 @@ LOG.SetPrefix(f"{web_server_config.server_name}-worker")
 async def main(concurrency: int = 1):
     queue = JobQueue()
     # 쿠팡(브라우저, 무거움) + 네이버(오픈API, 가벼움) 동시 검색 → 병합 최저가
-    adapters = {"coupang": CoupangAdapter(headless=False), "naver": NaverAdapter()}
+    # 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()}
     # OPENAI_API_KEY 있으면 '같은 상품' AI 판정 + 재검색어 생성 활성화
     has_openai = bool(os.environ.get("OPENAI_API_KEY"))
     judge = SimilarityJudge() if has_openai else None