爬D站的一次记录

资源分享 / 作者：

前言

这天逛博客，发现一个网站，里面有很多好看的图片，博客封面图片正好不多了，所以我心血来潮想要将将我喜欢的作者的图片全部爬下来，这里做一个记录。

正文

前面很多脚本由于一次电脑故障消失了（心痛），正好觉得之前写的东西都太杂了，并且为了安全，防止文件再次丢失，借助git进行了版本控制。

这篇文章主要是挑几个我写了很久的地方

selenium控制浏览器

由于D站是要你将页面往下滑动才能加载出更多的图片，由于没有学框架之类的，所以我只能用selenium来控制浏览器，让他尽量模拟人浏览页面，但是D站如果一直往下滚动，会直接离开正确的链接，所以如何判断页面滚动到页尾这是一个难点。

我也尝试了很多方式，但是最后我才用的是利用异常来判断。

1
2
3
4
5
6
 
def check_last(self, Browser):
    try:
        last_tag = Browser.find_element_by_xpath(self.get_xpath('tail'))
        return False
    except selenium.common.exceptions.NoSuchElementException:
        return True

当我们往下滚动的时候，首先是发现不了这个last_tag = Browser.find_element_by_xpath(self.get_xpath('tail'))last_tag的，所以当发现不了的时候就让他返回报错，然后用异常接住这个报错，让其返回True即可。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
 
while self.check_last(Browser):
    # 如果没有发现loggin是消失的话，就往下移动，否则就开始统计a的值
    all_hrefs.update([a_tag.get_attribute('href') for a_tag in
                      Browser.find_elements_by_xpath(self.get_xpath('a_hrefs_xpath'))])
    unseen.update(all_hrefs - seen)
    # 创造协程任务，使所有页面开始爬取图片的信息
    if len(unseen) != 0:
        tasks = [loop.create_task(self.get_img_info(session, target)) for target in unseen]
        Done, Pendding = await asyncio.wait(tasks)
        for item in Done:
            if item.result():
                # 添加图片信息
                img_info.append(item.result())
                print("添加的图片信息数量为{}".format(len(img_info)))

    # 等到加载图标消失后再往下移动
    try:
        WebDriverWait(Browser, 10).until(
            EC.invisibility_of_element_located((By.XPATH, self.get_xpath('logging_xpath')))
        )
    #找不到的话就往上滑动两下（有点不美观）。怕代理断连的时候没有加载loggin
    except selenium.common.exceptions.TimeoutException:
        Browser.find_element_by_xpath(self.get_xpath('a_hrefs_xpath')).send_keys(Keys.PAGE_UP)
        Browser.find_element_by_xpath(self.get_xpath('a_hrefs_xpath')).send_keys(Keys.PAGE_UP)
    # 为了切换协程
    await asyncio.sleep(0.1)
    seen.update(unseen)
    unseen.clear()
    Browser.find_element_by_xpath(self.get_xpath('a_hrefs_xpath')).send_keys(Keys.PAGE_DOWN)

这里还有一个点值得讲，就是selenium自带的WebDriverWait，相关文章

https://selenium-python.readthedocs.io/waits.html

等待页面加载完成

1
2
3
 
#它是可以直接这么使用的，until里面的条件成立的时候，他会返回true，如果不成的话，他会抛出一个异常。
WebDriverWait(Browser, 10).until(
                        EC.invisibility_of_element_located((By.XPATH, self.get_xpath('logging_xpath')))

如何给请求套上代理

由于D站是国外的一个站，所以如何让我们的所有请求走代理又成了一个重要的问题，当然如果把脚本放到vps上去的这个就完全没有问题了。但是发现服务器上不太好执行这个脚本，然后又不想写个服务器端，所以就只能找方法，为所有请求套上代理了。

为selenium加代理

这个代理是可以selenium自带的，格式如下

1
2
3
4
 
from selenium.webdriver.chrome.webdriver import Options
chrome_options = Options()
#设置socks5代理
chrome_options.add_argument('--proxy-server=socks5://127.0.0.1:1081')

为aiohttp加代理

由于aiohttp不自带的代理的，所以我在找了很多资料，其中最简单方法就是利用github的aiosocksy

相关网站和格式如下

romis2012/aiosocksy

Installation

Demo

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
 
import asyncio
import aiohttp
from aiosocksy import Socks5Auth
from aiosocksy.connector import ProxyConnector, ProxyClientRequest


async def fetch(url):
    auth = Socks5Auth(login='...', password='...')
    connector = ProxyConnector()
    socks = 'socks5://127.0.0.1:1080'
    async with aiohttp.ClientSession(connector=connector, request_class=ProxyClientRequest) as session:
        async with session.get(url, proxy=socks, proxy_auth=auth) as response:
            print(await response.text())


loop = asyncio.get_event_loop()
loop.run_until_complete(fetch('https://www.google.com/'))

为requests加代理

这个是这里面我觉得最简单的了，就直接用requests自带的可以了，但是我觉的下载东西光加代理不好玩，所以我还加了个随机UserAgent，我是直接用的fake-useragent这个库操作也很简单,代码片段如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
 
def download_img(self, img):
    res = requests.session()
    #设置代理，这里直接用的是Http的
    proxies = {
        'http': '127.0.0.1:1081',
        'https': '127.0.0.1:1081',
    }
    #随机UserAgent
    headers = {'UserAgent': UserAgent().random}
    if len(img) != 0:
        if self.check_img_exists(img['img_name']):
            filename = self.check_img_exists(img['img_name'])
            r = res.get(img['img_href'], headers=headers, proxies=proxies)
            if r.status_code == 200:
                with open(filename, 'wb') as f:
                    f.write(r.content)
                    print('下载成功')
                return filename
            else:
                print("文件下载失败,返回错误为{}".format(r.status_code))
                return None

如何复用链接

由于开代理，所以很容易代理一崩，整个程序就断了，特别是一个作者，图片有1.4k张，光是利用selenium将所有的图片的详细链接爬下来，代理都吃不住。

也查了很多资料，知乎上面有些人就是说，建立什么链接池，当链接失败的时候丢进去重新跑。其实并没有这么负载，直接使用python3.7之后自带的retrying就可以了。如果没有的话可以自己安装，我发现这个东西是真的简单，直接加两句话就行了

1
2
3
4
5
6
 
1.导入retry
from retrying import retry
2.在你的需要的函数面前加上@retry就可以了，当这个函数报错的时候，会自动重试这个函数
@retry
def funciton_you_need():
	....

相关的链接

Tenacity——Exception Retry 从此无比简单

Python之retrying

最后的脚本

里面有一些跟代理相关的，根据自己实际情况改。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
 
import asyncio
import time
import multiprocessing as mp
import os
import re
import aiohttp
import aiohttp.client_exceptions
import requests
import selenium
from aiosocksy.connector import ProxyConnector, ProxyClientRequest
from fake_useragent import UserAgent
from lxml import etree
from retrying import retry
from selenium import webdriver
from selenium.common.exceptions import NoSuchAttributeException
from selenium.webdriver.chrome.webdriver import Options
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.support.ui import WebDriverWait


class Config():
    def __init__(self):
        self.SCR_NUM_MAX = 3


class D_crawler():
    def __init__(self, author):
        self.author = author
        # 初始化url
        self.base_url = "https://www.deviantart.com/{}/gallery/all".format(self.author)
        # 图片信息
        self.img_info = []
        self.config = Config()
        # 当实例化这个类的时候就会自动创建一个文件夹
        self.mkdir()

    def mkdir(self):
        filelist = os.listdir('.')
        if self.author not in filelist:
            os.mkdir(self.author)
            r = requests.get(self.base_url)
            if r.status_code == 200:
                print("{}作者的文件夹已经建立".format(self.author))
            else:
                exit("输入的作者不存在")
        else:
            if len(os.listdir(self.author)):
                exit('文件夹已经存在，并且里面有东西，程序暂停')
            else:
                print("文件夹已经存在，但是里面没有东西，程序继续运行")

    def init_browser(self):
        print("浏览器初始化中...")
        chrome_options = Options()
        # 使用无头浏览器
        # chrome_options.add_argument('--headless')
        # 为浏览器设置代理
        chrome_options.add_argument('--proxy-server=socks5://127.0.0.1:1081')
        # 开启无图模式
        prefs = {"profile.managed_default_content_settings.images": 2}
        chrome_options.add_experimental_option("prefs", prefs)
        # 初始化浏览器
        Browser = webdriver.Chrome(options=chrome_options)
        Browser.get(self.base_url)
        # 当获取到body标签的时候才进行下面的任务
        WebDriverWait(Browser, 10).until(
            EC.presence_of_element_located((By.CSS_SELECTOR, 'body'))
        )
        # 点击浏览器一次
        Browser.find_element_by_css_selector('body').click()
        print("浏览器初始化完成")
        return Browser

    # 获取指定的xpath
    def get_xpath(self, xpath_name):
        xpath = {
            'a_hrefs_xpath': '//*[@class="_2vta_"]',
            'img_src_xpath': '//*[@id="root"]/main/div/div[1]/div[1]/div/div[2]/div[1]/img/@src',
            'logging_xpath': '//*[@class="_3UKUX"]',
            'all_num': '//*[@id="sub-folder-gallery"]/div[1]/div/div/div/div[1]/div/div/div/div/span',
            'tail': '//*[@class="_1BvgX"]'
        }
        return xpath[xpath_name]

    def check_last(self, Browser):
        try:
            last_tag = Browser.find_element_by_xpath(self.get_xpath('tail'))
            return False
        except selenium.common.exceptions.NoSuchElementException:
            return True

    # 让浏览器进行移动,并且动态获取a链接
    async def main_process(self):
        Browser = self.init_browser()
        await asyncio.sleep(0.1)
        all_num = str(Browser.find_element_by_xpath(self.get_xpath('all_num')).text)
        print("{}有{}张图片".format(self.author, all_num))
        scr_num = 0
        img_info = []
        connector = ProxyConnector()
        all_hrefs = set()
        unseen = set()
        seen = set()
        async with aiohttp.ClientSession(connector=connector, request_class=ProxyClientRequest) as session:
            while self.check_last(Browser):
                # 如果没有发现loggin是消失的话，就往下移动，否则就开始统计a的值
                all_hrefs.update([a_tag.get_attribute('href') for a_tag in
                                  Browser.find_elements_by_xpath(self.get_xpath('a_hrefs_xpath'))])
                unseen.update(all_hrefs - seen)
                # 创造协程任务，使所有页面开始爬取图片的信息
                if len(unseen) != 0:
                    tasks = [loop.create_task(self.get_img_info(session, target)) for target in unseen]
                    Done, Pendding = await asyncio.wait(tasks)
                    for item in Done:
                        if item.result():
                            # 添加图片信息
                            img_info.append(item.result())
                            print("添加的图片信息数量为{}".format(len(img_info)))

                # 等到加载图标消失后再往下移动
                try:
                    WebDriverWait(Browser, 10).until(
                        EC.invisibility_of_element_located((By.XPATH, self.get_xpath('logging_xpath')))
                    )
                # 为了切换协程
                except selenium.common.exceptions.TimeoutException:
                    Browser.find_element_by_xpath(self.get_xpath('a_hrefs_xpath')).send_keys(Keys.PAGE_UP)
                    Browser.find_element_by_xpath(self.get_xpath('a_hrefs_xpath')).send_keys(Keys.PAGE_UP)
                await asyncio.sleep(0.1)
                seen.update(unseen)
                unseen.clear()
                Browser.find_element_by_xpath(self.get_xpath('a_hrefs_xpath')).send_keys(Keys.PAGE_DOWN)
        Browser.quit()
        return img_info

    # 爬取图片的信息，返回的是图片的名字，下载地址，和找到图片的这个地址
    @retry
    async def get_img_info(self, session, url):
        img_name = "".join(re.findall(r'(?<=\/)[^\/]*(?=\-)', url)) + ".jpg"
        socks = 'socks5://127.0.0.1:1081'
        r = await session.get(url, proxy=socks)
        html = await r.text()
        await asyncio.sleep(0.1)  # slightly delay for downloading
        selector = etree.HTML(html)
        parse_hrefs = "".join(selector.xpath(self.get_xpath('img_src_xpath')))
        if parse_hrefs != '':
            img_info = {
                'img_name': img_name,
                'img_href': parse_hrefs,
                'a_href': url
            }
            return img_info
        else:
            print(img_name + '地址找不到')
            return False

    # 下载图片,img_info是一个所有的图片链接和图片的名字
    @retry
    def download_img(self, img):
        res = requests.session()
        proxies = {
            'http': '127.0.0.1:1081',
            'https': '127.0.0.1:1081',
        }
        headers = {'UserAgent': UserAgent().random}
        if len(img) != 0:
            if self.check_img_exists(img['img_name']):
                filename = self.check_img_exists(img['img_name'])
                r = res.get(img['img_href'], headers=headers, proxies=proxies)
                if r.status_code == 200:
                    with open(filename, 'wb') as f:
                        f.write(r.content)
                        print('下载成功')
                    return filename
                else:
                    print("文件下载失败,返回错误为{}".format(r.status_code))
                    return None

    # 检查文件是否已经被下载了
    def check_img_exists(self, filename):
        filelist = os.listdir(self.author + '/')
        if filename not in filelist:
            print('{}图片可以下载'.format(filename))
            return self.author + '/' + filename
        else:
            print('{}图片已经存在,已经跳过下载'.format(filename))
            return False


async def main(loop, authors):
    craws = [D_crawler(author) for author in authors]
    tasks = [loop.create_task(craw.main_process()) for craw in craws]
    Done, Pendding = await asyncio.wait(tasks)
    img_infos = [item.result() for item in Done]
    i = 0
    while i < len(craws):
        craw = craws[i]
        pool = mp.Pool(8)
        img_info = img_infos[i]
        download_jobs = [pool.apply_async(craw.download_img, args=(img,)) for img in img_info]
        print([job.get() for job in download_jobs])
        i += 1
        print("完毕")


if __name__ == '__main__':
    start_time = time.time()
    authors_num = int(input("请输入你想要几位作者\n"))
    authors = []
    for i in range(authors_num):
        authors.append(input("请输入第%d位作者的名字\n" % int(i + 1)))
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main(loop, authors))
    print("总共耗时{}".format(time.time() - start_time))

所有爬到的图片