关于pandalive的API封IP的问题和建议

[xueli12]

作者大大和各位大佬：

现在pandalive的风控很严格。如果频繁请求IP，很快就会被封了，而且一封就是一整天

解决方法 1.找很多个IP轮流切换，但是这种方法会不会封账号就不清楚了。

2.在使用Python进行网络爬虫或API调用时,我们经常会使用requests库进行HTTP请求。每次发送请求时，都会创建一个新的连接, 完成请求后再关闭连接。然而，对于pandalive这种需要频繁请求的场景，这种方式效率较低，短时间内频繁请求可能会遭到封禁ip。为了提高性能和减少IP被封的可能性，建议使用长会话机制，通过复用己建立的连接来减少连接的创建和关闭。建议使用requests.session()保持长会话，在代码中封装一个req_session函数实现这个功能。目前我使用的另一个项目使用了异步的client长会话请求方式，封禁ip的情况极其少,下面是引用大佬的异步client部分的代码：

    async def request(self, method, url, **kwargs):
        try:
            response = await self.client.request(method, url, **kwargs)
            return response
        except httpx.ProtocolError as error:
            raise ConnectionError(f'{self.flag}直播检测请求协议错误\n{error}')
        except httpx.HTTPError as error:
            raise ConnectionError(f'{self.flag}直播检测请求错误\n{repr(error)}')
        except anyio.EndOfStream as error:
            raise ConnectionError(f'{self.flag}直播检测代理错误\n{error}')

    def get_client(self):
        client_kwargs = {
            'http2': True,
            'timeout': self.interval,
            'limits': httpx.Limits(max_keepalive_connections=100, keepalive_expiry=self.interval * 2),
            'headers': self.headers,
            'cookies': self.cookies
        }
        # 检查是否有设置代理
        if self.proxy:
            if 'socks' in self.proxy:
                client_kwargs['transport'] = AsyncProxyTransport.from_url(self.proxy)
            else:
                client_kwargs['proxies'] = self.proxy
        return httpx.AsyncClient(**client_kwargs)

    def get_cookies(self):
        if self.cookies:
            cookies = SimpleCookie()
            cookies.load(self.cookies)
            self.cookies = {k: v.value for k, v in cookies.items()}

但是request库貌似没有长会话的参数设置，后面询问gpt找了一个代码实现类似client的长会话功能供参考。

参考代码示例，具体实现还是要看作者大大和各位大佬：

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry


class YourClass:
    # 初始化方法, 根据你的需要设置属性
    def __init__(self, interval, headers, cookies, max_keepalive_connections=1):
        self.interval = interval
        self.headers = headers
        self.cookies = cookies
        self.max_keepalive_connections = max_keepalive_connections

    def get_session(self):
        session = requests.Session()
        session.headers.update(self.headers)
        session.cookies.update(self.cookies)

        # 设置连接适配器，自定义最大连接数和连接池过期时间
        adapter = HTTPAdapter(max_retries=Retry(total=3, backoff_factor=0.1),
                              pool_connections=self.max_keepalive_connections,
                              pool_maxsize=self.max_keepalive_connections,
                              pool_block=True)
        session.mount('http://', adapter)
        session.mount('https://', adapter)

        # 设置请求超时
        session.request = lambda method, url, **kwargs: requests.Session.request(
            session, method, url, timeout=self.interval, **kwargs)

        return session


# 使用示例
if __name__ == "__main__":
    your_class_instance = YourClass(
        interval=5,
        headers={"User-Agent": "Awesome-Octopus"},
        cookies={"session": "ABC123"},
        max_keepalive_connections=100
    )

    session = your_class_instance.get_session()
    response = session.get('http://example.com')
    print(response.text)