✕
代理数据保存清洗
运行效果:
然后我的IP就给封了
代理测试
代码没问题。。。
不过短短几分钟抓了6000条代理,也算是不错了
需要下载的模块
pip install tinydb
# 主要用到的包
import requests
from lxml import etree
import pprint
import time
from tinydb import TinyDB, Query # 以前发布过使用
更新了下写入文件的初始化操作
class daili():
def __init__(self):
''' 初始化操作'''
# 起始url
self.url = "https://www.xicidaili.com/nn"
# 使用的是手机浏览器
self.headers = {
'User-Agent':'Mozilla/5.0 (Linux; Android 6.0; Nexus 5 Build/MRA58N) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/75.0.3770.142 Mobile Safari/537.36'
}
# 代理字典
#self.dl_list = list()
self.db = TinyDB("db.json")
self.db2 = TinyDB("db2.json")
self.Fruit = Query()
更新查找下页代码
# 查找下一页url
next_page = html.xpath('//*[@id="body"]/div[@class="pagination"]/a[@class="next_page"]/@href')
添加代码测试以及写入文件函数
def dlcs(self, proxies):
try:
# 查询是否存在 存在则跳过
if not self.db2.search(self.Fruit.HTTP == proxies['HTTP']):
# 测试代理可用性, 超时时间5秒
requests.get('http://www.baidu.com', headers=self.headers, proxies=proxies, timeout=5)
print("{} 可用".format(proxies))
self.db2.insert(proxies)
else:
print("{} 已存在".format(proxies))
except:
print("{} 不可用".format(proxies))
def dlqx(self):
''' 代理测试'''
proxies = [] # 代理列表
print(len(self.db))
for i in self.db:
proxies.append({i['type'] : i['type'] + "://" + i['IP'] + ':' + i['port']})
self.dlcs(proxies[-1])
import requests
from lxml import etree
import pprint
import time
from tinydb import TinyDB, Query
class daili():
def __init__(self):
''' 初始化操作'''
# 起始url
self.url = "https://www.xicidaili.com/nn"
# 使用的是手机浏览器
self.headers = {
'User-Agent':'Mozilla/5.0 (Linux; Android 6.0; Nexus 5 Build/MRA58N) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/75.0.3770.142 Mobile Safari/537.36'
}
# 爬虫数据存储数据库
self.db = TinyDB("db.json")
# 整理后数据库
self.db2 = TinyDB("db2.json")
# 数据查询是否相同
self.Fruit = Query()
def dl_pool(self, url):
# 请求url
txt = requests.get(url, headers=self.headers)
# 获取url内容用于整理
html = etree.HTML(txt.text)
# 第一条是表头 一页100行数据
for i in range(2, 102):
item = {}
# 遍历出一行数据
ip_text = html.xpath('//*[@id="ip_list"]/tr[{}]/td/text()'.format(i))
# 添加字典的对应键值对
item["IP"] = ip_text[0]
item["port"] = ip_text[1]
item["anonymity"] = ip_text[4]
item["type"] = ip_text[5]
item["survival"] = ip_text[-2]
item["proof"] = ip_text[-1]
# 打印每页数据
pprint.pprint(item)
# 存进数据库
self.db.insert(item)
# 查找下一页url
next_page = html.xpath('//*[@id="body"]/div[@class="pagination"]/a[@class="next_page"]/@href')
if next_page:
# 下一页url拼接
next_url = next_page[0].split('/')
next_url = self.url + '/' + next_url[-1]
print('下页地址',next_url)
# 暂停 看下效果
time.sleep(1)
# 继续下一页
return self.dl_pool(next_url)
else:
return
def dlcs(self, proxies):
try:
# 查询是否存在 存在则跳过
if not self.db2.search(self.Fruit.HTTP == proxies['HTTP']):
# 测试代理可用性, 超时时间5秒
requests.get('http://www.baidu.com', headers=self.headers, proxies=proxies, timeout=5)
print("{} 可用".format(proxies))
self.db2.insert(proxies)
else:
print("{} 已存在".format(proxies))
except:
print("{} 不可用".format(proxies))
def dlqx(self):
''' 代理测试'''
proxies = [] # 代理列表
print(len(self.db))
for i in self.db:
proxies.append({i['type'] : i['type'] + "://" + i['IP'] + ':' + i['port']})
self.dlcs(proxies[-1])
def __call__(self, *args, **kwargs):
''' 流程控制 '''
try:
self.dl_pool(self.url)
except AttributeError:
print("貌似IP被封???")
# 测试代理
self.dlce()
def main():
pp = daili()
# pp()
pp.dlqx()
if __name__ == '__main__':
main()