多个站点同时抓取！怎么做到的？ | 珊瑚贝

class Hello(object):
def hello(self, name='world'):
print('Hello, %s.' % name)

class ProxyMetaclass(type):
"""
定义 ProxyMetaclass ，按照默认习惯，metaclass 的类名总是以 Metaclass 结尾，以便清楚地表示这是一个metaclass ：
元类，在 FreeProxyGetter 类中加入
CrawlFunc_list 和 CrawlFuncCount
两个参数，分别表示爬虫函数，和爬虫函数的数量。
"""
def __new__(cls, name, bases, attrs):
count = 0
attrs['CrawlFunc_List'] = [] # 添加：CrawlFunc_List 列表方法
for k, v in attrs.items():
if 'crawl_' in k:
# 判断这个函数里面是否携带 crawl_ 也就是利用 value in xxx
attrs['CrawlFunc_List'].append(k)
count += 1
attrs['CrawlFuncCount'] = count # 检测添加的函数的总数量
return type.__new__(cls, name, bases, attrs) # 返回所修改的

"""
project = 'Code', file_name = 'utils.py', author = 'AI悦创'
time = '2020/6/1 12:34', product_name = PyCharm, 公众号：AI悦创
code is far away from bugs with the god animal protecting
I love animals. They taste delicious.
"""
import requests
import asyncio
import aiohttp
from requests.exceptions import ConnectionError
from fake_useragent import UserAgent,FakeUserAgentError
import random

def get_page(url, options={}):
"""
构造随机请求头，如果不理解的可以阅读此文章：
两行代码设置 Scrapy UserAgent：https://www.aiyc.top/archives/533.html
"""
try:
ua = UserAgent()
except FakeUserAgentError:
pass
# 生成随机的请求头，加 try...except... 使代码更加健壮
base_headers = {
'User-Agent': ua.random,
'Accept-Encoding': 'gzip, deflate, sdch',
'Accept-Language': 'zh-CN,zh;q=0.8'
}
# 如果使用者有传入请求头，则将此请求头和随机生成的合成在一起
headers = dict(base_headers, **options)
# 当前请求的 url
print('Getting', url)
try:
r = requests.get(url, headers=headers)
print('Getting result', url, r.status_code)
if r.status_code == 200:
return r.text
return None
except ConnectionError:
print('Crawling Failed', url)
return None

class Downloader(object):
"""
一个异步下载器，可以对代理源异步抓取，但是容易被 BAN。
self._htmls: 把请求的 html 放入列表当中
升级版的请求类
"""

def __init__(self, urls):
self.urls = urls
self._htmls = []

async def download_single_page(self, url):
"""
下载单页
"""
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
self._htmls.append(await response.text())

def download(self):
loop = asyncio.get_event_loop()
tasks = [self.download_single_page(url) for url in self.urls]
loop.run_until_complete(asyncio.wait(tasks))

@property
def htmls(self):
self.download()
return self._htmls

def crawl_kuaidaili(self):
"""
快代理
:return:
"""
for page in range(1, 4):
# 国内高匿代理
start_url = 'https://www.kuaidaili.com/free/inha/{}/'.format(page)
html = get_page(start_url)
# print(html)
pattern = re.compile(
'<td data-title="IP">(.*)</td>s*<td data-title="PORT">(w+)</td>'
)
# s * 匹配空格，起到换行作用
# ip_addres = re.findall(pattern, html) # 写法一
ip_addres = pattern.findall(str(html))
for adress, port in ip_addres:
# print(adress, port)
result = f"{adress}:{port}".strip()
yield result

def crawl_xicidaili(self):
"""
西刺代理
:return:
"""
for page in range(1, 4):
start_url = 'https://www.xicidaili.com/wt/{}'.format(page)
html = get_page(start_url)
# print(html)
ip_adress = re.compile(
'<td class="country"><img src="//fs.xicidaili.com/images/flag/cn.png" alt="Cn" /></td>s*<td>(.*?)</td>s*<td>(.*?)</td>'
)
# s* 匹配空格，起到换行作用
re_ip_adress = ip_adress.findall(str(html))
for adress, port in re_ip_adress:
result = f"{adress}:{port}".strip()
yield result

def crawl_data5u(self):
"""
无忧代理
:return:
"""
start_url = 'http://www.data5u.com'
html = get_page(start_url)
# print(html)
ip_adress = re.compile(
'<ul class="l2">s*<span><li>(.*?)</li></span>s*<span style="width: 100px;"><li class=".*">(.*?)</li></span>'
)
# s * 匹配空格，起到换行作用
re_ip_adress = ip_adress.findall(str(html))
for adress, port in re_ip_adress:
result = f"{adress}:{port}"
yield result.strip()

def crawl_kxdaili(self):
"""
开心代理-高匿
:return:
"""
for i in range(1, 4):
start_url = 'http://www.kxdaili.com/dailiip/1/{}.html'.format(i)
try:
html = requests.get(start_url)
if html.status_code == 200:
html.encoding = 'utf-8'
# print(html.text)
ip_adress = re.compile('<tr.*?>s*<td>(.*?)</td>s*<td>(.*?)</td>')
# s* 匹配空格，起到换行作用
re_ip_adress = ip_adress.findall(str(html.text))
for adress, port in re_ip_adress:
result = f"{adress}:{port}"
yield result.strip()
return None
except:
pass

def IP3366Crawler(self):
"""
云代理
parse html file to get proxies
:return:
"""
start_url = 'http://www.ip3366.net/free/?stype=1&page={page}'
urls = [start_url.format(page=i) for i in range(1, 8)]
# s * 匹配空格，起到换行作用
ip_address = re.compile('<tr>s*<td>(.*?)</td>s*<td>(.*?)</td>')
for url in urls:
html = get_page(url)
re_ip_address = ip_address.findall(str(html))
for adress, port in re_ip_address:
result = f"{adress}:{port}"
yield result.strip()

# files:schedule.py
import asyncio
import time
from multiprocessing.context import Process

import aiohttp
from db import RedisClient
from Spider import FreeProxyGetter
from error import ResourceDepletionError
from aiohttp import ServerDisconnectedError, ClientResponseError, ClientConnectorError
from socks import ProxyConnectionError

get_proxy_timeout = 9

class ValidityTester(object):
"""
代理的有效性测试
"""
def __init__(self):
self.raw_proxies = None
self.usable_proxies = [] # 可用代理
self.test_api = 'http://www.baidu.com'

def set_raw_proxies(self, proxies):
self.raw_proxies = proxies # 临时存储一些代理

async def test_single_proxy(self, proxy):
"""
python3.5 之后出现的新特性
text one proxy, if valid, put them to usable_proxies.
"""
try:
async with aiohttp.ClientSession() as session:
try:
if isinstance(proxy, bytes):
proxy = proxy.decode('utf-8')
real_proxy = 'http://' + proxy
print('Testing', proxy)
async with session.get(self.test_api, proxy=real_proxy, timeout=get_proxy_timeout) as response:
if response.status == 200:
print('Valid proxy', proxy)
except (ProxyConnectionError, TimeoutError, ValueError):
print('Invalid proxy', proxy)
except (ServerDisconnectedError, ClientResponseError, ClientConnectorError) as s:
print(s)
pass

def test(self):
"""
aio test all proxies.
"""
print('ValidityTester is working')
try:
loop = asyncio.get_event_loop()
tasks = [self.test_single_proxy(proxy) for proxy in self.raw_proxies]
loop.run_until_complete(asyncio.wait(tasks))
except ValueError:
print('Async Error')

class Schedule(object):
@staticmethod
def valid_proxy():
"""
Get half of proxies which in Spider
"""
tester = ValidityTester()
free_proxy_getter = FreeProxyGetter()
# 获取 Spider 数据库数据
# 调用测试类
tester.set_raw_proxies(free_proxy_getter.run())
# 开始测试
tester.test()

def run(self):
print('Ip processing running')
valid_process = Process(target=Schedule.valid_proxy) # 获取代理并筛选
valid_process.start()
if __name__ == '__main__':
schedule = Schedule().run()