0
点赞
收藏
分享

微信扫一扫

爬虫日记-一键下载****张壁纸-升级版版 彼岸 桌面壁纸 附源码 单文件一键可执行

小龟老师 2022-01-05 阅读 57
爬虫

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
举报

相关推荐

神器 ｜ 一键下载海量高清无码壁纸

python结果集网络请求

安卓一键换壁纸

android

如何用python一键保存绝美***壁纸？

pythonjsonpycharm

【京东一键获取CK电脑版】

开发语言

python爬取桌面壁纸（初学者），一看就会

python

一键脚本快速安装容器版GrayLog

dockerelasticsearchmongodb办公效率

使用Linux桌面壁纸应用variety发现的一些问题

Linux杂谈desktoplinux系统安装系统/运维

想要好看的壁纸图片，用这个网站一键解决，不用爬虫也能实现爬虫效果，一键爬取图片网站所有的图片

爬虫python数据挖掘Spark大数据

一键部署你的私人 ChatGPT 网页版

产品运营

docker一键部署GPU版ChatGLM3

djangohttp前端
0 条评论