关于python:批量爬取-pexels-图片

10次阅读

共计 3630 个字符，预计需要花费 10 分钟才能阅读完成。

闲来无事尝试写了个爬虫爬取 pexels 上的图片内容，遇到了一些问题来记录下

次要问题
网站反爬，借助 selenium 绕过
网站对 selenium 也做了反爬解决，辨认为 webdriver 时，js 文件获取 403，想方法暗藏 webdriver 身份绕够反爬
selenium 无奈在页面上采集到想要的链接（精确应该说是能够采集到小图的链接，然而小图的分辨率不够），钻研下图片法则，发现每个图片有本人的 id 获取图片 id 本人拼接 url 下载
拼接 url 不晓得如何晋升分辨率，好在 pexels 提供了默认的下载方式，是一个 download 链接，应用该链接能够下载图片，须要留神的是该链接会重定向到新的图片 url 所以不能间接用 download 链接下载，而是用其重定向的链接下载内容

import requests
import time
import os
import logging

from urllib.parse import urlparse
from selenium import webdriver
from multiprocessing import Pool


PEXELS_URL = 'https://www.pexels.com/'
DOWNLOAD_URL_KEY = 'https://www.pexels.com/photo/{image_id}/download/'
headers = {'User-Agent': 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/92.0.4515.107 Safari/537.36'
}
DOWNLOAD_LENGTH = 50  # 设置图片下载数, 这个是页面元素起码的数，理论下载数大于这个数
SCROLL_HEIGHT = 2000  # 滚屏像素点
SLEEP_SECONDS = 5  # 睡眠秒数
CPU_COUNT = os.cpu_count()
logging.basicConfig(
    filename='log.txt',
    level=logging.INFO,
    filemode='w+',
    format='%(levelname)s:%(asctime)s: %(message)s',
    datefmt='%Y-%d-%m %H:%M:%S'
)
IMAGE_PATH = './images/'
EXISTED_IMAGES = set(os.listdir(IMAGE_PATH))


def get_image_ids():
    """通过 selenium 获取网站中的图片 ids"""
    browser = webdriver.Chrome(executable_path='./chromedriver')
    # 暗藏 window.navigator.webdriver 防止反爬解决
    # 废了好大劲在这个文章找到答案 https://juejin.cn/post/6844904095749242887 感激作者
    browser.execute_cdp_cmd("Page.addScriptToEvaluateOnNewDocument", {
        "source": """Object.defineProperty(navigator,'webdriver', {get: () => undefined
        })
      """
    })

    url = PEXELS_URL
    browser.get(url)
    browser.maximize_window()
    elements = browser.find_elements_by_xpath('//article')
    scroll_height = SCROLL_HEIGHT
    while len(elements) < DOWNLOAD_LENGTH:
        browser.execute_script('window.scrollTo(0, {})'.format(scroll_height))  # 利用 selenium 执行 js 滚动到页面底部
        time.sleep(SLEEP_SECONDS)
        scroll_height += SCROLL_HEIGHT
        elements = browser.find_elements_by_xpath('//article')
    image_ids = [ele.get_attribute('data-photo-modal-medium-id') for ele in elements]
    browser.close()
    logging.info(f'image_ids: {image_ids}')
    return image_ids


def get_download_urls(image_ids):
    return [DOWNLOAD_URL_KEY.format(image_id=_id) for _id in image_ids]


def download_image(image_url):
    parse_result = urlparse(image_url)
    path = parse_result.path
    image_name = path.split('/')[-1]
    if image_name in EXISTED_IMAGES:
        logging.info(f'图片 {image_name} 已存在无需从新下载')
        return None

    response = requests.get(image_url, headers)
    if response.status_code != 200:
        message = '下载 {} 失败. status_code: {}'.format(image_url, response.status_code)
        logging.error(message)
        return None

    prefix = IMAGE_PATH
    with open(prefix + image_name, 'wb') as image:
        image.write(response.content)
    message = '下载 {} 胜利. url: {}'.format(image_name, image_url)
    logging.info(message)


def get_image_url(need_redirect_url):
    """
    因为没法解决反爬，这里采取其余形式绕过反爬
    1. 利用 selenium 获取到页面上 download 按钮的 url
    2. 这个中央 download 按钮的 url 并不能拿到图片的 url，通过测试发现进行了重定向而后重定向的 url 才是图片 url
    3. 这个 download 按钮的 url 也有反爬，测试发现 get 申请绕不过
    4. 然而测试发现能够用 head 申请获取到重定向的图片 url
    5. http code 302 返回的 response headers 外面的 location 即为重定向的 url
    """
    response = requests.head(need_redirect_url, headers=headers)
    if response.status_code != 302:
        message = '{} 没有产生重定向. code: {}'.format(need_redirect_url, response.status_code)
        logging.error(message)
        return None
    location = response.headers.get('location')
    logging.info(f'get_image_url success. location: {location}')
    return location


def download(need_redirect_url):
    image_url = get_image_url(need_redirect_url)
    if image_url:
        download_image(image_url)


def main():
    image_ids = get_image_ids()
    download_urls = get_download_urls(image_ids)
    logging.info(f'image_ids: {image_ids}, download_urls: {download_urls}')

    p = Pool(CPU_COUNT // 2)
    for url in download_urls:
        p.apply_async(func=download, args=(url,))

    p.close()
    p.join()


if __name__ == "__main__":
    main()