o2o-negosium-original/lps/services/search/proxy.py

"""DECODO(구 Smartproxy) residential 프록시 제공자 — 쿠팡(Akamai) 전용.

Decodo residential 은 **포트 기반 sticky** 모델이다:
  gate.decodo.com : 10001..N  (각 포트 = 별도 sticky 세션, 대시보드에서 지속시간 지정: 예 10분)
  username/password 는 고정. → **IP 회전 = 포트를 바꾸는 것.**

매 요청 IP 변경은 Akamai 가 '쿠키-IP 불일치'로 재챌린지하므로 금물.
→ 시간창(now // window)으로 포트를 고른다: 창 안에선 같은 포트=같은 IP, 창이 지나면 다음 포트=새 IP.
자격증명/엔드포인트는 config.local.toml [DecodoConfig] 에서 로드(시크릿).
"""

import time
from urllib.parse import quote

import httpx

from common.logger import LOG
from config.server_configs import decodo_config


class DecodoProxy:
    def __init__(self, cfg=None, host: str | None = None, registry=None, owner: str = ""):
        cfg = cfg if cfg is not None else decodo_config
        # host 를 넘기면 그 게이트웨이를 쓴다 — 국가 타깃(kr.decodo.com)용. 포트/자격증명은 동일.
        self.host = host or cfg.host
        # registry 를 주면 포트를 **공용 중재자**에게서 배타 임대한다(소스·워커 간 IP 충돌·재사용 방지).
        # 없으면 기존 동작(시간창+오프셋 계산) 그대로 — 단독 사용·테스트 경로를 깨지 않는다.
        self._registry = registry
        self._owner = owner or "proxy"
        self._leased: int | None = None
        self.username = cfg.username
        self.password = cfg.password
        self.port_start = cfg.port_start
        self.port_end = cfg.port_end
        self.session_minutes = cfg.session_minutes or 10
        self._rotate_offset = 0  # 봇 감지 등으로 '즉시 회전'이 필요할 때 증가
        # 불탄(차단 감지된) 포트 격리 시간([DecodoConfig].port_cooldown_sec). sticky 만료(session_minutes)
        # 이상이어야 쿨다운 복귀 시 같은 포트라도 사실상 새 IP 가 배정된다. 0=자동 max(sticky, 30분).
        self.cooldown_sec = getattr(cfg, "port_cooldown_sec", 0) or max(self.session_minutes * 60, 1800)
        # 선제 회전으로 놓은 포트의 휴식 시간. 0=자동(sticky 수명) — 그 IP 의 세션이 만료돼
        # 다음에 잡힐 땐 사실상 새 IP 가 배정된다. 차단 쿨다운(cooldown_sec)보다 훨씬 짧다.
        self.rest_sec = getattr(cfg, "port_rest_sec", 0) or self.session_minutes * 60
        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, kind: str = "rotate"):
        """시간창과 무관하게 즉시 다음 포트(=새 IP)로 회전. 봇 감지·예산 도달 시 호출.

        kind="budget"(차단 전 선제 회전)이면 놓는 포트에 **휴식**을 준다 — 방금 쓴 IP 를
        다른 워커가 곧바로 집으면 요청률이 도로 올라가 예산의 의미가 사라진다.
        차단으로 태우는 건 mark_burned 가 따로 처리한다(휴식보다 훨씬 긴 쿨다운).
        """
        self._rotate_offset += 1
        if self._registry is not None and self._leased is not None:
            rest = self.rest_sec if kind == "budget" else 0
            self._registry.release(self.host, self._leased, self._owner, rest_sec=rest)
            self._leased = None

    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
        cd = cooldown_sec if cooldown_sec is not None else self.cooldown_sec
        if self._registry is not None:
            # 전역 격리 — 태운 포트를 다른 소스/워커가 곧바로 집는 걸 막는다.
            self._registry.burn(self.host, port, cd, owner=self._owner, reason="block")
            if self._leased == port:
                self._leased = None
            return
        self._burned[port] = time.monotonic() + cd
        LOG.i(f"[proxy] 포트 {port} 쿨다운 {int(cd)}s — 활성 {self.available_ports()}/{self.port_end - self.port_start + 1}")

    def available_ports(self) -> int:
        """쿨다운 중이 아닌 포트 수(관측·알림용)."""
        if self._registry is not None:
            return self._registry.available(self.host)
        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.
        쿨다운 중인 포트는 건너뛰고, 전 포트가 쿨다운이면 만료가 가장 임박한 포트를 쓴다(가용성 우선)."""
        if self._registry is not None:
            lease_sec = self.session_minutes * 60
            if self._leased is not None and self._registry.lease_expired(self.host, self._leased, self._owner, lease_sec):
                self._registry.release(self.host, self._leased, self._owner)   # sticky 수명 만료 → 새 IP
                self._leased = None
            if self._leased is None:
                # 시작 오프셋은 기존과 동일한 의미(워커별 분산). 실제 중복 방지는 registry 가 보장한다.
                bucket = int(time.time() // (self.session_minutes * 60))
                start = (bucket + self._rotate_offset) % (self.port_end - self.port_start + 1)
                self._leased = self._registry.acquire(self.host, self._owner, start_at=start, lease_sec=lease_sec)
            return self._leased

        n = self.port_end - self.port_start + 1
        bucket = int(time.time() // (self.session_minutes * 60))
        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):
        return self._port() if self.enabled else None

    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,
        }

    def _proxy_url(self, port: int) -> str:
        return f"http://{quote(self.username)}:{quote(self.password)}@{self.host}:{port}"

    async def healthcheck(self, timeout: float = 6.0) -> tuple[str | None, int | None]:
        """시작 프리플라이트: 현재 포트로 egress IP 확인, 실패하면 회전하며 살아있는 포트를 찾는다.
        반환: (egress_ip, port) 성공 / (None, None) 전 포트 실패. residential IP 는 실행 중에도
        죽으므로 이건 '빠른 실패+가시성'용이고, 실제 회복은 런타임 IP 회전이 담당한다."""
        if not self.enabled:
            return None, None
        n = self.port_end - self.port_start + 1
        for _ in range(n):
            port = self._port()
            try:
                async with httpx.AsyncClient(proxy=self._proxy_url(port), timeout=timeout) as c:
                    r = await c.get("https://ip.decodo.com/ip")
                    if r.status_code == 200:
                        return r.text.strip(), port
            except Exception as ex:
                LOG.d(f"[proxy] 포트 {port} 헬스체크 실패: {type(ex).__name__} → 회전")
            self.rotate()
        return None, None