Class StaticProxyProvider

proxy/proxy_ip_pool.py:155–188

class StaticProxyProvider(ProxyProvider):
    async def get_proxy(self, num: int) -> List[IpInfoModel]:
        proxy_url = getattr(config, "STATIC_PROXY_URL", "")
        if not proxy_url:
            utils.logger.warning("[StaticProxyProvider] STATIC_PROXY_URL is not configured!")
            return []

        try:
            parsed = urlparse(proxy_url)
            scheme = parsed.scheme or "http"
            if scheme not in {"http", "https"}:
                utils.logger.error(f"[StaticProxyProvider] Unsupported proxy scheme: {scheme}")
                return []

            ip = parsed.hostname or ""
            port = parsed.port or (443 if scheme == "https" else 80)
            if not ip:
                utils.logger.error("[StaticProxyProvider] STATIC_PROXY_URL host is empty!")
                return []

            return [
                IpInfoModel(
                    ip=ip,
                    port=port,
                    user=unquote(parsed.username or ""),
                    password=unquote(parsed.password or ""),
                    protocol=f"{scheme}://",
                    # Static proxy doesn't expire.
                    expired_time_ts=int(time.time()) + 99999999,
                )
            ]
        except Exception as e:
            utils.logger.error(f"[StaticProxyProvider] Parse static proxy url error: {e}")
            return []


IpProxyProvider: Dict[str, ProxyProvider] = {