o2o-negosium-original/lps/services/search/proxy.py
민헌 2481c79e3f feat(lps): DECODO 포트기반 sticky + 대역폭 최적화(리소스 차단)
대시보드 실측 반영 — Decodo residential 은 포트 기반 sticky(gate:10001..N,
고정 user/pass, 각 포트=sticky 세션). IP 회전 = 포트 순환.

- proxy.DecodoProxy: 시간창 기반 포트 선택(창 안 동일 IP, 창 지나면 다음 포트=새 IP).
  _UNSET sentinel 로 '미지정(env)' vs '명시적 빈값' 구분(테스트 결정성)
- coupang/adapter: 이미지/미디어/폰트/CSS 차단(block_resources) — per-GB 대역폭 대폭 절감,
  Akamai(JS)·상품파싱엔 무영향. 라이브 검증: 차단 ON 에도 Akamai 통과+파싱 정상
- .env.example: DECODO_HOST/USERNAME/PASSWORD/PORT_START/PORT_END/SESSION_MINUTES
- tests: 포트범위/시간창 안정성/활성조건 5건 → 전체 44/44

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-09 09:48:54 +09:00

54 lines
2.4 KiB
Python

"""DECODO(구 Smartproxy) residential 프록시 제공자 — 쿠팡(Akamai) 전용.
Decodo residential 은 **포트 기반 sticky** 모델이다:
gate.decodo.com : 10001..N (각 포트 = 별도 sticky 세션, 대시보드에서 지속시간 지정: 예 10분)
username/password 는 고정. → **IP 회전 = 포트를 바꾸는 것.**
매 요청 IP 변경은 Akamai 가 '쿠키-IP 불일치'로 재챌린지하므로 금물.
→ 시간창(now // window)으로 포트를 고른다: 창 안에선 같은 포트=같은 IP, 창이 지나면 다음 포트=새 IP.
자격증명/엔드포인트는 .env(DECODO_*) 에서 로드(시크릿).
"""
import os
import time
_UNSET = object() # 미지정(→env 사용) vs 명시적 빈값("" 등, 그대로 사용) 구분
def _int(v, default=None):
try:
return int(v)
except (TypeError, ValueError):
return default
class DecodoProxy:
def __init__(self, host=_UNSET, username=_UNSET, password=_UNSET, port_start=_UNSET, port_end=_UNSET, session_minutes=_UNSET):
env = os.environ.get
self.host = env("DECODO_HOST") if host is _UNSET else host
self.username = env("DECODO_USERNAME") if username is _UNSET else username
self.password = env("DECODO_PASSWORD") if password is _UNSET else password
self.port_start = _int(env("DECODO_PORT_START") if port_start is _UNSET else port_start)
self.port_end = _int(env("DECODO_PORT_END") if port_end is _UNSET else port_end)
self.session_minutes = _int(env("DECODO_SESSION_MINUTES") if session_minutes is _UNSET else session_minutes, 10)
@property
def enabled(self) -> bool:
return all([self.host, self.username, self.password, self.port_start, self.port_end])
def _port(self) -> int:
"""시간창 기반 포트 선택 — 창 안에선 동일 포트(동일 sticky IP), 창이 지나면 다음 포트."""
n = self.port_end - self.port_start + 1
bucket = int(time.time() // (self.session_minutes * 60))
return self.port_start + (bucket % n)
def playwright_proxy(self) -> dict | None:
"""Playwright launch(proxy=...) 용 설정. 비활성 시 None(프록시 미사용)."""
if not self.enabled:
return None
return {
"server": f"http://{self.host}:{self._port()}",
"username": self.username,
"password": self.password,
}