MyRequestsHelper

标签：get url self MyRequestsHelper requests proxies response

import requests
import time


class StateCodeError(Exception):
    """状态码异常"""
    pass


class ContentError(Exception):
    """内容异常"""
    pass


class RequestHelper:
    """未使用代理的requests"""

    # 判断内容是不是有问题
    # 判断状态码是不是200
    # 判断异常错误

    def request(self, method, url, **kwargs):
        response = None
        for retry_time in range(10):  # 处理问题
            try:
                response = requests.request(method, url, **kwargs)
                # if response.status_code != 200:
                #     raise StateCodeError("状态码异常")
                # 判断内容是否正常
                # if "<title>安全验证</title>" in response.content.decode():
                #     raise ContentError("内容异常")
                return response
            except requests.exceptions.ConnectTimeout:
                print("连接超时")
            except requests.exceptions.ConnectionError:
                print("连接错误")

    def get(self, url, params=None, headers=None):
        return self.request('get', url, params=params, headers=headers)

    def post(self, url, data=None, json=None, headers=None):
        return self.request('post', url, data=data, json=json, headers=headers)


class RequestsProxyHelper:
    """使用代理的requests"""

    def __init__(self):
        self.proxies = self.get_proxies_private()
        # 常见错误状态码
        self.error_code = [403, 418, 429, 454]

    def get_proxies_private(self):
        """私密代理"""
        time.sleep(5)
        url = '快代理api地址'
        # 这个也可能存在请求异常
        ip = requests.get(url).content.decode()
        proxies = {
            "http": f"http://xxx:xxx@{ip}",
            "https": f"http://xxx:xxx@{ip}"
        }
        print(f"获取新代理：{ip}")
        return proxies

    def request(self, method, url, **kwargs):
        response = None
        for retry_time in range(1, 11):
            try:
                response = requests.request(method, url, **kwargs, proxies=self.proxies)
                if response.status_code in self.error_code:
                    raise StateCodeError("异常状态码")
                # 判断内容是否正常
                # if "<title>安全验证</title>" in response.content.decode():
                #     raise ContentError("内容异常")
                return response
            except requests.exceptions.ProxyError:
                print("代理失效过期，重新获取")
                self.proxies = self.get_proxies_private()
            except requests.exceptions.ConnectTimeout:
                print("连接超时")
                if retry_time % 3 == 0:
                    self.proxies = self.get_proxies_private()
            except requests.exceptions.ConnectionError:
                print("连接错误")
                if retry_time % 3 == 0:
                    self.proxies = self.get_proxies_private()
            except StateCodeError as e:
                print(e)
                self.proxies = self.get_proxies_private()
            except ContentError as e:
                print(e)
                self.proxies = self.get_proxies_private()
        return response

    def get(self, url, params=None, headers=None):
        return self.request('get', url, params=params, headers=headers)

    def post(self, url, data=None, json=None, headers=None):
        return self.request('post', url, data=data, json=json, headers=headers)

# requests_helper = RequestsProxyHelper()
#
# url = 'https://www.baidu.com'
# response = requests_helper.post(url)
# if response:
#     print(response.status_code)
#     print(response.content.decode())

标签：get,url,self,MyRequestsHelper,requests,proxies,response
From： https://www.cnblogs.com/lsixu/p/18398855