categories: “爬虫日记”
环境
window10
python3
re
requests
etree
threading
threadpool
redis
文章目录
源码地址
前言
这次的升级版主要修复了上个程序的一些不足的地方,并且是为了用到之前总结过的redis只是而特意去应用,有一些应试教育的感觉。
上期博客
程序整体流程
流程图如下
程序开始进行一次初始化,主要是是将任务写进到redis
中,判断img
文件夹是否存在,不存在就原地新建一个。
之后开启一个线程,从redis
去获取每一个网页url
,判断一下历史记录中有没有处理过这个url
,没有就处理,处理完就写进历史。
然后用线程池去下载图片,同样在本地文件夹进行判断是否存在。
程序解说
环境
python环境
这里用到的包有:
import os.path
import random
import re
import threading
import time
import requests
import redis
from lxml import etree
都是常见的包,没有什么稀奇古怪的,使用pycharm
的话,直接把鼠标放在导包出现红色的地方,然后点安装软件包即可。
使用其他的可以使用pip
在命令行中安装,示例如下:
pip install lxml
编码申明
Python中默认的编码格式是 ASCII
格式,在没修改编码格式时无法正确打印汉字,所以在读取中文时会报错。
解决办法:在文件开头指定编码格式
# -*- coding: UTF-8 -*-
# 或者
# coding=utf-8
注:一定是第一行呀。
程序初始化
根据 http://www.netbian.com/index.htm 网站每一页的url
规则,将要采集的url
添加到redis
中。
这个程序下载图片的位置是本地同级目录img
,所以先建一个这样的文件夹。(当然要是本来就有那就不用程序去建)
# 初始化
with open('task.txt', 'w', encoding='utf-8') as f:
f.writelines('http://www.netbian.com/index.htm' + '\n')
# 当前网站最多页为,1223,即下方的1000最大可以替换成1224
for i in range(2, 1000):
f.writelines("http://www.netbian.com/index_{}.htm".format(i) + '\n')
if not os.path.exists('./img'):
os.mkdir('./img')
# 代理线程,定时更新代理
proxy_time_task(20)
time.sleep(3)
print("初始化结束")
访问列表页获取图片名
这边主要是从本地文件读取任务及历史,如果这一页有历史数据,那么就跳过,否则就采集对应的数据。
def get_source_code(url):
# 构造一个普通的请求头
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/94.0.4606.81 Safari/537.36',
}
# 发送请求,拿到数据
# 因为代理的问题,多次尝试
for i in range(20):
proxy = random.choice(proxy_list) if proxy_list else None
try:
if proxy:
if 'https' in proxy:
proxies = {'https': proxy}
else:
proxies = {'http': proxy}
print("尝试第{}次访问\t{}".format(i + 1, url))
response = requests.request("GET", url, headers=headers, timeout=3,
proxies=proxies)
if response.status_code == 200:
print("访问通过---")
break
else:
print("直接访问")
response = requests.request("GET", url, headers=headers)
except:
pass
if response:
# 返回网页源码
text = response.content.decode('gbk')
data_list = re.findall("<img src=\"http://img.netbian.com/file/\d{4}/\d{4}/small.*?\.jpg\" alt=\".*?\" />",
text)
for data in data_list:
try:
small_pic_url = re.findall("src=\"(.*?)\"", data)[0]
pic_id = re.search(r"/small(.*?)\.jpg", data).groups()[0][:-10]
pic_url = re.sub(re.findall("/(small.*?)\.jpg", small_pic_url)[0], pic_id, small_pic_url)
pic_name = re.findall("alt=\"(.*?)\"", data)[0]
threading.Thread(target=deonload_img, args=(pic_name, pic_url)).start()
except:
print("出现意外。。。")
if url + '\n' not in history:
with open('history.txt', 'a+', encoding='utf-8') as f:
f.writelines(url + '\n')
下载图片
和之前的方法一样,使用request
直接写入本地,但是这次加上了文件名。
有些东西是为了用到所学而故意为之,让整个程序变成了能用的屎山。。。
def deonload_img(img_name, img_url):
global proxy_list
if os.path.exists('img/' + str(img_name) + '.jpg'):
print("已存在,pass")
return
else:
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/94.0.4606.81 Safari/537.36',
}
# 发送请求,拿到数据
# 因为代理质量差,多次尝试
for i in range(20):
try:
proxy = random.choice(proxy_list) if proxy_list else None
if proxy:
# print("使用代理", proxy)
# if 'https' in proxy:
# proxies = {'https': proxy}
# else:
# proxies = {'http': proxy}
print("尝试第{}次下载图片\t{}".format(i + 1, img_name))
resp = requests.request("GET", img_url, headers=headers, timeout=3,
proxies={'http': proxy, 'https': proxy.replace("http", "https")})
if resp.status_code == 200:
open('img/' + str(img_name) + '.jpg', 'wb').write(resp.content) # 将内容写入图片
print("下载图片成功", img_name)
break
else:
print("直接访问")
resp = requests.get(img_url, headers=headers, stream=True)
if resp.status_code == 200:
open('img/' + str(img_name) + '.jpg', 'wb').write(resp.content) # 将内容写入图片
print("下载图片成功", img_name)
return
except:
pass
with open('img_download_fail.txt', 'a+', encoding='utf-8') as f:
f.writelines(img_name + ' ' + img_url + '\n')
代理相关
这里使用 小幻HTTP代理 ,主要是看它上面对代理的更新和检测的相对够快。
def get_proxy_list():
url = "https://ip.ihuan.me/address/5Lit5Zu9.html"
payload = {}
headers = {
'authority': 'ip.ihuan.me',
'sec-ch-ua': '"Chromium";v="94", "Google Chrome";v="94", ";Not A Brand";v="99"',
'sec-ch-ua-mobile': '?0',
'sec-ch-ua-platform': '"Windows"',
'upgrade-insecure-requests': '1',
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/94.0.4606.81 Safari/537.36',
'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9',
'sec-fetch-site': 'same-origin',
'sec-fetch-mode': 'navigate',
'sec-fetch-user': '?1',
'sec-fetch-dest': 'document',
'accept-language': 'zh-CN,zh;q=0.9'
}
response = requests.request("GET", url, headers=headers, data=payload)
res = []
_ = etree.HTML(response.text)
type_dct = {
"HTTP": "http://",
"HTTPS": "https://"
}
data_list = _.xpath("//tbody/tr")
for data in data_list:
ip = data.xpath("./td[1]/a/text()")[0]
port = data.xpath("./td[2]/text()")[0]
type = "HTTP"
res.append(type_dct[type] + ip + ':' + port)
return res
# 测试代理
def check(proxy):
href = 'https://www.baidu.com'
# href = 'http://img.netbian.com/file/2022/0103/224706qcowj.jpg'
if 'https' in proxy:
proxies = {'https': proxy}
else:
proxies = {'http': proxy}
headers = {
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 11_2_2) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/90.0.4396.0 Safari/537.36'
}
try:
r = requests.get(href, proxies=proxies, timeout=1, headers=headers)
if r.status_code == 200:
return True
except:
return False
def get_proxy():
global proxy_list
p_list = get_proxy_list()
pp_list = []
count = 0
for p in p_list:
if check(p):
pp_list.append(p)
count += 1
proxy_list = pp_list[:]
print("更新了{}个代理".format(count))
完整程序
# -*- coding:utf-8 -*-
import os.path
import random
import re
import threading
import time
import requests
import redis
from lxml import etree
proxy_list = []
def get_source_code(url):
# 构造一个普通的请求头
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/94.0.4606.81 Safari/537.36',
}
# 发送请求,拿到数据
# 因为代理的问题,多次尝试
for i in range(20):
proxy = random.choice(proxy_list) if proxy_list else None
try:
if proxy:
if 'https' in proxy:
proxies = {'https': proxy}
else:
proxies = {'http': proxy}
print("尝试第{}次访问\t{}".format(i + 1, url))
response = requests.request("GET", url, headers=headers, timeout=3,
proxies=proxies)
if response.status_code == 200:
print("访问通过---")
break
else:
print("直接访问")
response = requests.request("GET", url, headers=headers)
except:
pass
if response:
# 返回网页源码
text = response.content.decode('gbk')
data_list = re.findall("<img src=\"http://img.netbian.com/file/\d{4}/\d{4}/small.*?\.jpg\" alt=\".*?\" />",
text)
for data in data_list:
try:
small_pic_url = re.findall("src=\"(.*?)\"", data)[0]
pic_id = re.search(r"/small(.*?)\.jpg", data).groups()[0][:-10]
pic_url = re.sub(re.findall("/(small.*?)\.jpg", small_pic_url)[0], pic_id, small_pic_url)
pic_name = re.findall("alt=\"(.*?)\"", data)[0]
threading.Thread(target=deonload_img, args=(pic_name, pic_url)).start()
except:
print("出现意外。。。")
if url + '\n' not in history:
with open('history.txt', 'a+', encoding='utf-8') as f:
f.writelines(url + '\n')
def deonload_img(img_name, img_url):
global proxy_list
if os.path.exists('img/' + str(img_name) + '.jpg'):
print("已存在,pass")
return
else:
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/94.0.4606.81 Safari/537.36',
}
# 发送请求,拿到数据
# 因为代理质量差,多次尝试
for i in range(20):
try:
proxy = random.choice(proxy_list) if proxy_list else None
if proxy:
# print("使用代理", proxy)
# if 'https' in proxy:
# proxies = {'https': proxy}
# else:
# proxies = {'http': proxy}
print("尝试第{}次下载图片\t{}".format(i + 1, img_name))
resp = requests.request("GET", img_url, headers=headers, timeout=3,
proxies={'http': proxy, 'https': proxy.replace("http", "https")})
if resp.status_code == 200:
open('img/' + str(img_name) + '.jpg', 'wb').write(resp.content) # 将内容写入图片
print("下载图片成功", img_name)
break
else:
print("直接访问")
resp = requests.get(img_url, headers=headers, stream=True)
if resp.status_code == 200:
open('img/' + str(img_name) + '.jpg', 'wb').write(resp.content) # 将内容写入图片
print("下载图片成功", img_name)
return
except:
pass
with open('img_download_fail.txt', 'a+', encoding='utf-8') as f:
f.writelines(img_name + ' ' + img_url + '\n')
def get_proxy_list():
url = "https://ip.ihuan.me/address/5Lit5Zu9.html"
payload = {}
headers = {
'authority': 'ip.ihuan.me',
'sec-ch-ua': '"Chromium";v="94", "Google Chrome";v="94", ";Not A Brand";v="99"',
'sec-ch-ua-mobile': '?0',
'sec-ch-ua-platform': '"Windows"',
'upgrade-insecure-requests': '1',
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/94.0.4606.81 Safari/537.36',
'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9',
'sec-fetch-site': 'same-origin',
'sec-fetch-mode': 'navigate',
'sec-fetch-user': '?1',
'sec-fetch-dest': 'document',
'accept-language': 'zh-CN,zh;q=0.9'
}
response = requests.request("GET", url, headers=headers, data=payload)
res = []
_ = etree.HTML(response.text)
type_dct = {
"HTTP": "http://",
"HTTPS": "https://"
}
data_list = _.xpath("//tbody/tr")
for data in data_list:
ip = data.xpath("./td[1]/a/text()")[0]
port = data.xpath("./td[2]/text()")[0]
type = "HTTP"
res.append(type_dct[type] + ip + ':' + port)
return res
# 测试代理
def check(proxy):
href = 'https://www.baidu.com'
# href = 'http://img.netbian.com/file/2022/0103/224706qcowj.jpg'
if 'https' in proxy:
proxies = {'https': proxy}
else:
proxies = {'http': proxy}
headers = {
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 11_2_2) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/90.0.4396.0 Safari/537.36'
}
try:
r = requests.get(href, proxies=proxies, timeout=1, headers=headers)
if r.status_code == 200:
return True
except:
return False
def get_proxy():
global proxy_list
p_list = get_proxy_list()
pp_list = []
count = 0
for p in p_list:
if check(p):
pp_list.append(p)
count += 1
proxy_list = pp_list[:]
print("更新了{}个代理".format(count))
def proxy_time_task(inc):
try:
get_proxy()
except:
print("获取代理出现错误,稍后重试。")
t = threading.Timer(inc, proxy_time_task, (20,))
t.start()
if __name__ == '__main__':
# 初始化
with open('task.txt', 'w', encoding='utf-8') as f:
f.writelines('http://www.netbian.com/index.htm' + '\n')
# 当前网站最多页为,1223,即下方的1000最大可以替换成1224
for i in range(2, 1000):
f.writelines("http://www.netbian.com/index_{}.htm".format(i) + '\n')
if not os.path.exists('./img'):
os.mkdir('./img')
# 代理线程,定时更新代理
proxy_time_task(20)
time.sleep(3)
print("初始化结束")
with open('task.txt', 'r', encoding='utf-8') as f:
# 读取任务
task_list = f.readlines()
# 读取历史
if not os.path.exists('history.txt'):
history = []
else:
with open('history.txt', 'r', encoding='utf-8') as f:
history = f.readlines()
# 开始任务
for url in task_list:
url = url.replace('\r', '').replace('\n', '').strip()
if url + '\n' not in history:
get_source_code(url)
time.sleep(2)
print(url, "done!")
else:
# 如果是历史数据,那么就跳过
print("已存在,跳过")
效果
写在最后
这个爬虫有很多不完善的地方,比如:
代理质量差,毕竟是免费代理;
列表页顺序执行,下载图片多线程,如果直接关闭程序会丢失一部分图片数据,(把history删掉让程序重新开始可以解决,就是有点慢);
如果代理速度足够满,会导致线程越来越多;
页数比较后面的列表页,图片的连接格式好像有变化,但是这个不重要了,我又不是为了采集壁纸;
巴拉巴拉。。。
爬虫仅做学习交流使用。
总之这个爬虫还有很大的优化空间,下次抽点时间完善一下。
实力有限,才疏学浅,如有错误,欢迎指正。
- 我的个人博客 菜猫子小六 - 博客 (codesix.site)
- 我的简书 菜猫子小六 - 简书 (jianshu.com)
- 我的CSDN 菜猫子小六 - CSDN