3-优化-爬取 deviantart gallery 完整图片

本文最后更新于 1653 天前，其中的信息可能已经有所发展或是发生改变。

对 2-爬取 deviantart gallery 完整图片代码重新整理。并在第二部分使用异步加载 Asyncio，多进程 Multiprocessing解析。我运行此程序一共用了三分钟(155张图片)，相比昨天快了几倍，具体时间与网络有关。

历时两个周六，终于写出了自己满意的爬虫。这半年不会再更新爬虫，要开始准备课题了。

关于异步加载 Asyncio 详细教程见莫烦大大的《加速爬虫: 异步加载 Asyncio》

说明：

依赖: selenium, aiohttp, requests, pymongo, BeautifulSoup
这个脚本下还要有另一个文件：config.py。里面对存放有相关参数，参数进行了简单说明。
jupyterlab 多进程无法正常使用。会有如下问题：
AttributeError: Can't get attribute 'job' on <module '__main__' (built-in)>。jupyternotebook未测试。所以不要使用 jupyterlab 运行此程序。
多进程的子进程不能再调用函数。所以我把页面解析和下载的放到了一起，并把保存图片地址到mongodb的部分放在了多进程之外。
在异步爬取时终端输出 Timeout. ，这是程序中做的捕获会重新爬取，可放心。

1. 爬虫脚本

from selenium import webdriver
from selenium.common.exceptions import TimeoutException
from selenium.webdriver.common.by import By
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.support.wait import WebDriverWait
import re
import time
import pymongo
import os
import requests
from requests.exceptions import RequestException
from bs4 import BeautifulSoup
import multiprocessing as mp
import logging
import aiohttp
import asyncio
from config import *  # 配置文件，与此脚本同文件夹。


class simChrome():
    def __init__(self, artist='wlop'):
        '''
        :param artist: deviantart艺术家 ID。
        '''
        self.base_url = 'https://www.deviantart.com/{}/gallery/'.format(artist)

    def initChrome(self):
        try:
            Chrome_setting = {
                "profile.managed_default_content_settings.images": 2}
            chrome_option = webdriver.ChromeOptions()
            chrome_option.add_experimental_option("prefs", Chrome_setting)
            chrome_option.add_argument('--headless')
            self.browser = webdriver.Chrome(chrome_options=chrome_option)
            self.wait = WebDriverWait(self.browser, 10)
        except Exception as err:
            print(err)

    def indexPage(self):
        try:
            self.browser.get(self.base_url)
            self.wait.until(EC.presence_of_element_located((By.ID, 'gmi-')))
            print('Dom successfully loaded!')
        except TimeoutException:
            print('Selenium timeout, try again.')
            self.indexPage()  # 如果超时那就重新再来一遍

    def simulateScroll(self, scroll_step=3000, max_height=None, fixed_times=3):
        '''
        本程序判断是否爬取所有的依据是比较动作前后 set 的长度，如果三次动作后长度依然没有
        变化则判断没有已经新的数据
        :param scroll_step: 每次下拉的长度，和浏览器窗口大小有关系，需要视实际而定。
            暂时没有想到科学确定的方法。
        :param max_height: 最大距离。如果设置为None，则爬取所有。
        :param fixed_times: fixed_times次动作后长度依然没有变化，判定为结束。有时获取
            不到完整的时可适当增加此次数。
        '''
        # 获取所有图片页面地址的网页元素
        element = set()
        gallery_torpedo = self.browser.find_element_by_id('gmi-')
        height = 0
        no_change = 0  # 对动作后长度没有变化的次数统计
        present_len = 0  # 当前集合的长度
        while True:
            previous_len = present_len  # 动作前集合的长度
            self.browser.execute_script(
                'window.scrollTo(0, {})'.format(height))
            time.sleep(0.5)  # 等待浏览器解析并加载到页面
            items = gallery_torpedo.find_elements_by_css_selector(
                'span[class~=thumb]>a')
            for item in items:
                # 从网页元素中提取url，保存到 set 里
                element.update((item.get_attribute('href'),))
            present_len = len(element)
            if previous_len == present_len:
                no_change += 1
            else:
                no_change = 0
            if no_change == fixed_times or height == max_height:
                # fixed_times次动作后长度依然没有变化或者到达设定的最大值，则停止
                break
            height += scroll_step
        self.browser.close()
        self.imgPageUrl = list(element)  # set 转 list


class saveToMongo():
    def __init__(self, **kw):
        client = pymongo.MongoClient(host=kw['host'], port=kw['port'])
        if kw['auth']:
            db_auth = client.admin
            db_auth.authenticate(kw['user'], kw['passwd'])
        self.db = client[kw['db']]

    def saveOne(self, result, col):
        '''
        :param result: dict. 要存储的字典形式数据。
        :param col: str. 目的数据表。
        '''
        try:
            if self.db[col].insert_one(result):
                print('{} save to MongoDB successful!'.format(
                    list(result.values())[0]))
        except Exception as err:
            print(err)


async def get_page(url, session):
    print('%s getting...' % url[-10:])
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit' +
        '/537.36 (KHTML, like Gecko) Chrome/69.0.3497.81 Safari/537.36'
    }
    try:
        response = await session.get(url, headers=headers, timeout=10)
        html = await response.text()
        if response.status == 200:
            return html
    except asyncio.TimeoutError:
        print('Timeout.')
    except RequestException as e:
        print(e)


# def parse_page(html):
#     '''解析图片地址'''
#     soup = BeautifulSoup(html, 'html5lib')
#     page_url = soup.find('meta', {'property': 'og:url'})['content']
#     item = soup.select('img.dev-content-full')[0]
#     img_url = item.attrs['src']
#     return page_url, img_url


# def download_img(result: dict):
#     r = requests.get(result['img_url'], stream=True)  # stream loading
#     file_name = './temp/wlop/{}.jpg'.format(result['img_name'])
#     with open(file_name, 'wb') as f:
#         for chunk in r.iter_content(chunk_size=32):
#             f.write(chunk)
#     logging.info('%s download successful!'.format(result['img_name']))


def parse_save_download(html):
    '''
    由于多进程内子进程调用函数出错的原因，我把页面解析与图片下载的程序放到了一起，
        并且把保存到本地mongodb的部分放到了多进程之外。
    '''
    # parse page
    soup = BeautifulSoup(html, 'html5lib')
    page_url = soup.find('meta', {'property': 'og:url'})['content']
    item = soup.select('img.dev-content-full')[0]
    img_url = item.attrs['src']
    result = {
        'img_name': re.findall('.*/(.*?)-', img_url, re.S)[0],
        'img_url': img_url
    }

    # download img
    r = requests.get(result['img_url'], stream=True)  # stream loading
    file_name = './temp/wlop/{}.jpg'.format(result['img_name'])
    with open(file_name, 'wb') as f:
        for chunk in r.iter_content(chunk_size=32):
            f.write(chunk)
    logging.info('%s download successful!'.format(result['img_name']))
    return page_url, result


async def main(loop):
    pool = mp.Pool(4)  # slightly affected
    async with aiohttp.ClientSession() as session:
        count = 1
        while len(unseen) != 0:
            print('\nAsync Crawling...')
            tasks = [loop.create_task(get_page(url, session))
                     for url in unseen]
            finished, unfinished = await asyncio.wait(tasks)
            htmls = [f.result() for f in finished]

            print('\nDistributed Parsing....')
            parse_jobs = [pool.apply_async(
                parse_save_download, args=(html,)) for html in htmls if html]
            results = [j.get() for j in parse_jobs]

            print('\nAnalying...')
            for page_url, result in results:
                print(count, page_url)
                mongo.saveOne(result, db_config['col_2'])  # save to mongodb
                seen.update([page_url])
                unseen.remove(page_url)
                count += 1


if __name__ == '__main__':
    logging.basicConfig(level=logging.INFO,
                        filename='./log.txt',
                        filemode='w',
                        format='%(asctime)s - %(filename)s[line:%(lineno)d]' +
                        ' - %(levelname)s: %(message)s')

    mongo = saveToMongo(**db_config)

    # 第一部分: selenium 控制浏览器获取图片页面地址
    simulator = simChrome(artist)  # 传入自己喜欢的艺术家ID，默认是wlop.
    simulator.initChrome()
    simulator.indexPage()
    simulator.simulateScroll(scroll_step, max_height, fixed_times)

    print('len: ', len(simulator.imgPageUrl), '\n\t', simulator.imgPageUrl[:5])

    time.sleep(5)

    for item in simulator.imgPageUrl:
        result = {
            'page_name': re.findall('.*/(.*)-\d', item, re.S)[0],
            'url': item
        }
        mongo.saveOne(result, db_config['col_1'])

    # 第二部分: 异步加载，多进程解析并下载。
    seen = set()
    unseen = set(simulator.imgPageUrl)

    os.makedirs('./temp/wlop/', exist_ok=True)

    loop = asyncio.get_event_loop()
    loop.run_until_complete(main(loop))
    loop.close()

    print('finished')

2. 爬虫配置

文件名需要保存成 config.py

#%%writefile config.py
# config
db_config = {
    'host': 'localhost',
    'port': 27017,
    'db': 'test_database',  # 保存的目的数据库名
    'col_1': 'wlopImgPageUrl',  # 第一部分保存的数据表
    'col_2': 'wlopImgUrl',  # 第二部分保存的数据表

    'auth': True,  # 我的mongodb打开了登陆认证, 若没有认证: 此参数修改为 False
    'user': 'testuser',  # 自己的认证用户名
    'passwd': 'testpass'  # 自己的认证用户密码
}
artist = 'wlop'


# 参数可微调，如果不能完全获取，则增大 fixed_times,减小 scroll_step。具体说明见下。
scroll_step = 3000
max_height = None
fixed_times = 3
'''simChrome 类 simulateScroll 方法的说明:
本程序判断是否爬取所有的依据是比较动作前后 set 的长度，如果fixed_times次动作后长度依然没有变化则判断没有已经新的数据

:param scroll_step: 每次下拉的长度，和浏览器窗口大小有关系，需要视实际而定。    暂时没有想到科学确定的方法。
:param max_height: 最大距离。如果设置为None，则爬取所有。
:param fixed_times: fixed_times次动作后长度依然没有变化，判定为结束。有时获取
    不到完整的时可适当增加此次数。
'''