Python

1. requests

#! -*- encoding:utf-8 -*-

import requests
import random

# 要访问的目标页面
targetUrl = "http://httpbin.org/ip"

# 要访问的目标HTTPS页面
# targetUrl = "https://httpbin.org/ip"

# 代理服务器(产品官网 www.duoip.cn)
proxyHost = "tunnel.hahado.cn"
proxyPort = "31111"

# 代理验证信息
proxyUser = "username"
proxyPass = "password"

proxyMeta = "http://%(user)s:%(pass)s@%(host)s:%(port)s" % {
    "host" : proxyHost,
    "port" : proxyPort,
    "user" : proxyUser,
    "pass" : proxyPass,
}

# 设置 http和https访问都是用HTTP代理
proxies = {
    "http"  : proxyMeta,
    "https" : proxyMeta,
}


#  设置IP切换头
tunnel = random.randint(1,10000)
headers = {"Proxy-Tunnel": str(tunnel)}



resp = requests.get(targetUrl, proxies=proxies, headers=headers)

print (resp.status_code)
print (resp.text)

#! -*- encoding:utf-8 -*-
import requests
import random
import requests.adapters

# 要访问的目标页面
targetUrlList = [
    "https://httpbin.org/ip",
    "https://httpbin.org/headers",
    "https://httpbin.org/user-agent",
]

# 代理服务器(产品官网 www.duoip.cn)
proxyHost = "tunnel.hahado.cn"
proxyPort = "31111"

# 代理验证信息
proxyUser = "username"
proxyPass = "password"

proxyMeta = "http://%(user)s:%(pass)s@%(host)s:%(port)s" % {
    "host": proxyHost,
    "port": proxyPort,
    "user": proxyUser,
    "pass": proxyPass,
}

# 设置 http和https访问都是用HTTP代理
proxies = {
    "http": proxyMeta,
    "https": proxyMeta,
}

#  设置IP切换头
tunnel = random.randint(1, 10000)
headers = {"Proxy-Tunnel": str(tunnel)}

class HTTPAdapter(requests.adapters.HTTPAdapter):
    def proxy_headers(self, proxy):
        headers = super(HTTPAdapter, self).proxy_headers(proxy)
        if hasattr(self, 'tunnel'):
            headers['Proxy-Tunnel'] = self.tunnel
        return headers


# 访问三次网站，使用相同的tunnel标志，均能够保持相同的外网IP
for i in range(3):
    s = requests.session()

    a = HTTPAdapter()

    #  设置IP切换头
    a.tunnel = tunnel
    s.mount('https://', a)

    for url in targetUrlList:
        r = s.get(url, proxies=proxies)
        print r.text

#! -*- encoding:utf-8 -*-
import requests
import random
import requests.adapters

# 要访问的目标页面
targetUrlList = [
    "https://httpbin.org/ip",
    "https://httpbin.org/headers",
    "https://httpbin.org/user-agent",
]

# 代理服务器(产品官网 www.duoip.cn)
proxyHost = "tunnel.hahado.cn"  
proxyPort = "31111"  

# 代理验证信息
proxyUser = "username"  
proxyPass = "password" 

proxyMeta = "http://%(user)s:%(pass)s@%(host)s:%(port)s" % {
    "host": proxyHost,
    "port": proxyPort,
    "user": proxyUser,
    "pass": proxyPass,
}

# 设置 http和https访问都是用HTTP代理
proxies = {
    "http": proxyMeta,
    "https": proxyMeta,
}

# 访问三次网站，使用相同的Session(keep-alive)，均能够保持相同的外网IP
s = requests.session()

# 设置cookie
# cookie_dict = {"JSESSION":"123456789"}
# cookies = requests.utils.cookiejar_from_dict(cookie_dict, cookiejar=None, overwrite=True)
# s.cookies = cookies

for i in range(3):
    for url in targetUrlList:
        r = s.get(url, proxies=proxies)
        print r.text

### 2. urllib2

:::warning[urllib2无法使用Keep-alive]

urllib2对于HTTP/1.1默认会关闭连接请通过设置相同Proxy-Tunnel来保持相同的外网IP.

:::

#! -*- encoding:utf-8 -*-

from urllib import request

# 要访问的目标页面
targetUrl = "http://httpbin.org/ip"

# 代理服务器(产品官网 www.duoip.cn)
proxyHost = "tunnel.hahado.cn"
proxyPort = "31111"

# 代理验证信息
proxyUser = "username"
proxyPass = "password"


proxyMeta = "http://%(user)s:%(pass)s@%(host)s:%(port)s" % {
    "host" : proxyHost,
    "port" : proxyPort,
    "user" : proxyUser,
    "pass" : proxyPass,
}

proxy_handler = request.ProxyHandler({
    "http"  : proxyMeta,
    "https" : proxyMeta,
})        

opener = request.build_opener(proxy_handler)

request.install_opener(opener)
resp = request.urlopen(targetUrl).read()

print (resp)

#! -*- encoding:utf-8 -*-
import urllib2
import random
import httplib


class HTTPSConnection(httplib.HTTPSConnection):

    def set_tunnel(self, host, port=None, headers=None):
        httplib.HTTPSConnection.set_tunnel(self, host, port, headers)
        if hasattr(self, 'proxy_tunnel'):
            self._tunnel_headers['Proxy-Tunnel'] = self.proxy_tunnel


class HTTPSHandler(urllib2.HTTPSHandler):
    def https_open(self, req):
        return urllib2.HTTPSHandler.do_open(self, HTTPSConnection, req, context=self._context)


# 要访问的目标页面
targetUrlList = [
    "https://httpbin.org/ip",
    "https://httpbin.org/headers",
    "https://httpbin.org/user-agent",
]

# 代理服务器(产品官网 www.duoip.cn)
proxyHost = "tunnel.hahado.cn"
proxyPort = "31111"

# 代理验证信息
proxyUser = "username"
proxyPass = "password"

proxyMeta = "http://%(user)s:%(pass)s@%(host)s:%(port)s" % {
    "host": proxyHost,
    "port": proxyPort,
    "user": proxyUser,
    "pass": proxyPass,
}

# 设置 http和https访问都是用HTTP代理
proxies = {
    "http": proxyMeta,
    "https": proxyMeta,
}

#  设置IP切换头
tunnel = random.randint(1, 10000)
headers = {"Proxy-Tunnel": str(tunnel)}
HTTPSConnection.proxy_tunnel = tunnel


proxy = urllib2.ProxyHandler(proxies)
opener = urllib2.build_opener(proxy, HTTPSHandler)
urllib2.install_opener(opener)

# 访问三次网站，使用相同的tunnel标志，均能够保持相同的外网IP
for i in range(3):
    for url in targetUrlList:
        r = urllib2.Request(url)
        print(urllib2.urlopen(r).read())

### 3. scrapy ☝️ 在项目中新建middlewares.py文件(./项目名/middlewares.py) 修改项目配置文件 (./项目名/settings.py)

    DOWNLOADER_MIDDLEWARES = {
        '项目名.middlewares.ProxyMiddleware': 100,
    }

#! -*- encoding:utf-8 -*-
import base64            
import sys
import random

PY3 = sys.version_info[0] >= 3

def base64ify(bytes_or_str):
    if PY3 and isinstance(bytes_or_str, str):
        input_bytes = bytes_or_str.encode('utf8')
    else:
        input_bytes = bytes_or_str

    output_bytes = base64.urlsafe_b64encode(input_bytes)
    if PY3:
        return output_bytes.decode('ascii')
    else:
        return output_bytes

class ProxyMiddleware(object):                
    def process_request(self, request, spider):
        # 代理服务器(产品官网 www.duoip.cn)
        proxyHost = "tunnel.hahado.cn"
        proxyPort = "31111"

        # 代理验证信息
        proxyUser = "username"
        proxyPass = "password"

        # [版本>=2.6.2](https://docs.scrapy.org/en/latest/news.html?highlight=2.6.2#scrapy-2-6-2-2022-07-25)无需添加验证头,会自动在请求头中设置Proxy-Authorization     
        request.meta['proxy'] = "http://{0}:{1}@{2}:{3}".format(proxyUser,proxyPass，proxyHost，proxyPort)

✌️ Here's content that's only inside the second Tab. 💪 Here's content that's only inside the third Tab. 💪 Here's content that's only inside the third Tab. 💪 Here's content that's only inside the third Tab. ### 4. aiohttp

:::warning[注意]
aiohttp库实现了TCP链接池功能，如果没设置随机Proxy-Tunnel或断开TCP链接，会导致多个请求始终没有切换IP。如需切换IP，需每个请求新建一个session；同时设置connector_owner参数让session关闭后链接也关闭。如需切换IP，也可以设置Proxy-Tunnel为随机数，使用随机IP。
:::

import aiohttp, asyncio
import random
def main():

    targetUrl = "https://httpbin.org/headers"

    # 代理服务器(产品官网 www.duoip.cn)
    proxyHost = "tunnel.hahado.cn"
    proxyPort = "31111"

    # 代理验证信息
    proxyUser = "username"
    proxyPass = "password"

    proxyServer = "http://%(user)s:%(pass)s@%(host)s:%(port)s" % {
        "host" : proxyHost,
        "port" : proxyPort,
        "user" : proxyUser,
        "pass" : proxyPass,
    }

    userAgent = "Chrome/83.0.4103.61"
    # 所有请求使用同一个Proxy-Tunnel, 使用固定IP                            
    # proxy_tunnel = "{}".format(random.randint(1,10000))

    async def entry():            
        async with aiohttp.ClientSession(headers={"User-Agent": userAgent}) as session:
            while True:
                # 随机设Proxy-Tunnel,使用随机IP                          
                proxy_tunnel = "{}".format(random.randint(1,10000))
                async with session.get(targetUrl, proxy=proxyServer, proxy_headers={"Proxy-Tunnel":proxy_tunnel}) as resp:
                    body = await resp.read()            
                    print(resp.status)
                    print(body)

    loop = asyncio.get_event_loop()
    loop.run_until_complete(entry())
    loop.run_forever()


if __name__ == '__main__':
    main()

#! -*- encoding:utf-8 -*-

import aiohttp, asyncio


targetUrl = "http://httpbin.org/ip"

# 代理服务器(产品官网 www.duoip.cn)
proxyHost = "tunnel.hahado.cn"
proxyPort = "31111"

# 代理验证信息
proxyUser = "username"
proxyPass = "password"

proxyServer = "http://%(user)s:%(pass)s@%(host)s:%(port)s" % {
    "host" : proxyHost,
    "port" : proxyPort,
    "user" : proxyUser,
    "pass" : proxyPass,
}

userAgent = "Chrome/83.0.4103.61"

async def entry():
    while True: 
        conn = aiohttp.TCPConnector(verify_ssl=False)                       
        async with aiohttp.ClientSession(headers={"User-Agent": userAgent}, connector=conn) as session:
            async with session.get(targetUrl, proxy=proxyServer) as resp:
                body = await resp.read()

                print(resp.status)
                print(body)

loop = asyncio.get_event_loop()
loop.run_until_complete(entry())
loop.run_forever()

### 5. httpx ```python import asyncio import httpx # 代理服务器(产品官网 www.duoip.cn) proxyHost = "tunnel.hahado.cn" proxyPort = "31111"

代理验证信息

proxyUser = "username"
proxyPass = "password"

proxyMeta = "http://%(user)s:%(pass)s@%(host)s:%(port)s" % {
"host": proxyHost,
"port": proxyPort,
"user": proxyUser,
"pass": proxyPass,
}

设置 http和https访问都是用HTTP代理

proxies = {
"http://": proxyMeta,
"https://": proxyMeta,
}

client = httpx.AsyncClient(proxies=proxies)

开启http2.0支持，请使用 pip install httpx[http2]

client = httpx.AsyncClient(http2=True,proxies=proxies)

async def test():
resp = await client.get("https://httpbin.org/ip")
print(resp.http_version)
print(resp.text)

asyncio.run(test())

  </Tab>
  <Tab title="Proxy-Tunnel保持IP不变">

```python
import httpx
import random
# 代理服务器(产品官网 www.duoip.cn)
proxy_host = "tunnel.hahado.cn"
proxy_port = "31111"

# 代理验证信息
proxy_user = "username"
proxy_pwd = "password"

proxy_url = f"http://{proxy_user}:{proxy_pwd}@{proxy_host}:{proxy_port}"

proxy = httpx.Proxy(
    url=proxy_url,
    #  设置IP切换头，数不变，保持IP不变
    headers={"Proxy-Tunnel": f"{random.randint(1, 10000)}"}
)
print(proxy_url)
proxies = {
    "http://": proxy,
    "https://": proxy,
}
target_url = "https://httpbin.org/ip"


async def test_async():
    # 三次请求保持在同一个IP上
    # 开启http2.0支持，请使用 pip install httpx[http2]
    # client = httpx.AsyncClient(http2=True,proxies=proxies)
    for _ in range(3):
        async with httpx.AsyncClient(
                proxies=proxies,
        ) as client:
            response = await client.get(target_url)
            print("test_async:", response.text)


def test():
    # 三次请求保持在同一个IP上
    for _ in range(3):
        # 开启http2.0支持，请使用 pip install httpx[http2]
        # client = httpx.Client(http2=True,proxies=proxies)
        client = httpx.Client(
            proxies=proxies,
        )

        response = client.get(target_url)
        print("test:", response.text)

        client.close()


if __name__ == '__main__':                
    import asyncio

    test()
    asyncio.run(test_async())

最后修改时间： 2 个月前