python3和scrapy使用亿牛云隧道代理问题以及代码

近期，我参与了一个需要爬取国家食品药品监督局数据的项目，但该网站存在IP屏蔽机制。因此，我需要在Scrapy框架中实现自动IP切换，才能完成任务。然而，尽管我使用了第三方库scrapy-proxys和代理API接口，但测试并不成功。爬取药监局数据是一项不容易完成的任务。这是因为该网站采用了多种反爬虫机制，如IP屏蔽、频率限制等，以避免窃取机密数据信息。

因此，在实施这项任务时，我们需要使用各种技术工具和方法来克服这些障碍。对于大多数企业，使用爬虫程序和库工具是一项不错的选择，其中最常用的是Scrapy和Python3。这些工具具有强大的功能，可以轻松地爬取网站上的数据。但要想成功抓取药监局的数据，我们还需要实现IP自动切换的功能，以确保IP被屏蔽后，程序可以顺利地继续运行下去。

亿牛云官方给出了python3和scrapy的参考示例python3示例以上是python的原本使用方式，下面提供scrapy的中间件示例在项目中新建middlewares.py文件(./项目名/middlewares.py)

import base64

import sys

PY3 = sys.version_info[0] >= 3

def base64ify(bytes_or_str):

input_bytes = bytes_or_str if PY3 else bytes_or_str.encode('utf8')

output_bytes = base64.urlsafe_b64encode(input_bytes)

return output_bytes.decode('ascii') if PY3 else output_bytes

class ProxyMiddleware2(object):

def process_request(self, request, spider):

# 选择一个代理服务器

proxyHost = "u6791.5.tn.16yun"

proxyPort = "31111"

# 设置IP地址和端口号

request.meta['proxy'] = "http://{0}:{1}".format(proxyHost, proxyPort)

# 设置代理用户名和密码(根据需要开启/关闭)

# proxyUser = "16EDRSSX"

# proxyPass = "214587"

# request.headers['Proxy-Authorization'] = 'Basic ' + base64ify(proxyUser + ":" + proxyPass)

# 设置隧道(根据需要开启/关闭)

# tunnel = random.randint(1,10000)

# request.headers['Proxy-Tunnel'] = str(tunnel)

# 修改连接方式为Close，每次都切换到新的IP

request.headers['Connection'] = "Close"

三、正式集成

在项目的middlewares.py中新增类import logging

import base64

import requests

import threading

proxyServer = "u6791.5.tn.16yun0"

proxyUser = "16EDRSSXxx"

proxyPass = "214587"

proxyAuth = "Basic " + base64.urlsafe_b64encode(bytes((proxyUser + ":" + proxyPass), "ascii")).decode("utf8")

proxies = {

"http": "http://{0}".format(proxyServer),

"https": "https://{0}".format(proxyServer),

}

headers = {

"Proxy-Authorization": proxyAuth,

}

LOCK = threading.Lock()

logging.basicConfig(filename='requests.log', level=logging.DEBUG)

def fetch_url(url):

try:

response = requests.get(url, proxies=proxies, headers=headers)

LOCK.acquire()

logging.debug(f'{url} Status Code: {response.status_code}')

print(threading.current_thread().name, response.status_code, url)

except Exception as e:

logging.error(f'{url} Error: {e}')

finally:

LOCK.release()

if __name__ == '__main__':

urls = ["http://www.example.com", "http://www.google.com", "http://www.github.com"]

threads = []

for url in urls:

t = threading.Thread(target=fetch_url, args=(url,))

threads.append(t)

t.start()

for t in threads:

t.join()

logging.debug('All URL requests finished')

在实现IP自动切换的功能之前，我们需要优先考虑的是网站可能存在的安全协议，如验证码。这些协议旨在防止机器人的恶意行为，例如恶意爬取数据，从而保护用户的隐私。虽然这些协议可确保网站安全，但却增加了我们爬取数据的难度。

同时，针对药监局数据的质量问题也需要考虑，如缺失数据、错误数据、重复数据等，这就需要对数据进行清洗和处理，以确保最终的数据质量。在处理数据时，我们必须小心谨慎，以避免产生不准确或不全面的数据，从而带来不必要的麻烦。

若有收获，就点个赞吧

import base64

import sys

PY3 = sys.version_info[0] >= 3

def base64ify(bytes_or_str):

input_bytes = bytes_or_str if PY3 else bytes_or_str.encode('utf8')

output_bytes = base64.urlsafe_b64encode(input_bytes)

return output_bytes.decode('ascii') if PY3 else output_bytes

class ProxyMiddleware2(object):

def process_request(self, request, spider):

# 选择一个代理服务器

proxyHost = "u6791.5.tn.16yun"

proxyPort = "31111"

# 设置IP地址和端口号

request.meta['proxy'] = "http://{0}:{1}".format(proxyHost, proxyPort)

# 设置代理用户名和密码(根据需要开启/关闭)

# proxyUser = "16EDRSSX"

# proxyPass = "214587"

# request.headers['Proxy-Authorization'] = 'Basic ' + base64ify(proxyUser + ":" + proxyPass)

# 设置隧道(根据需要开启/关闭)

# tunnel = random.randint(1,10000)

# request.headers['Proxy-Tunnel'] = str(tunnel)

# 修改连接方式为Close，每次都切换到新的IP

request.headers['Connection'] = "Close"

三、正式集成

在项目的middlewares.py中新增类import logging

import base64

import requests

import threading

proxyServer = "u6791.5.tn.16yun0"

proxyUser = "16EDRSSXxx"

proxyPass = "214587"

proxyAuth = "Basic " + base64.urlsafe_b64encode(bytes((proxyUser + ":" + proxyPass), "ascii")).decode("utf8")

proxies = {

"http": "http://{0}".format(proxyServer),

"https": "https://{0}".format(proxyServer),

}

headers = {

"Proxy-Authorization": proxyAuth,

}

LOCK = threading.Lock()

logging.basicConfig(filename='requests.log', level=logging.DEBUG)

def fetch_url(url):

try:

response = requests.get(url, proxies=proxies, headers=headers)

LOCK.acquire()

logging.debug(f'{url} Status Code: {response.status_code}')

print(threading.current_thread().name, response.status_code, url)

except Exception as e:

logging.error(f'{url} Error: {e}')

finally:

LOCK.release()

if __name__ == '__main__':

urls = ["http://www.example.com", "http://www.google.com", "http://www.github.com"]

threads = []

for url in urls:

t = threading.Thread(target=fetch_url, args=(url,))

threads.append(t)

t.start()

for t in threads:

t.join()

logging.debug('All URL requests finished')

若有收获，就点个赞吧