Python图片爬虫案例如何爬取图片资源

wen python案例 2026-06-30 259

本文目录导读：

基础版：爬取静态网页图片
进阶版：爬取动态加载图片
高级版：API爬取图片
综合实用版：通用图片爬虫
安装依赖
注意事项

我来为你提供几个Python图片爬虫的实用案例,从简单到复杂逐步介绍。

基础版：爬取静态网页图片

import requests
import os
from bs4 import BeautifulSoup
from urllib.parse import urljoin
def download_static_images(url, save_dir='images'):
    """爬取静态网页中的所有图片"""
    # 创建保存目录
    if not os.path.exists(save_dir):
        os.makedirs(save_dir)
    # 设置请求头
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
    }
    try:
        # 获取网页内容
        response = requests.get(url, headers=headers, timeout=10)
        response.raise_for_status()
        # 解析HTML
        soup = BeautifulSoup(response.text, 'html.parser')
        # 查找所有图片标签
        img_tags = soup.find_all('img')
        downloaded = 0
        for i, img in enumerate(img_tags):
            # 获取图片URL
            img_url = img.get('src')
            if not img_url:
                continue
            # 处理相对路径
            img_url = urljoin(url, img_url)
            # 过滤非图片格式
            if not any(img_url.lower().endswith(ext) for ext in ['.jpg', '.jpeg', '.png', '.gif', '.webp']):
                continue
            try:
                # 下载图片
                img_response = requests.get(img_url, headers=headers, timeout=10)
                img_response.raise_for_status()
                # 保存图片
                file_name = f'image_{i+1}.jpg'
                file_path = os.path.join(save_dir, file_name)
                with open(file_path, 'wb') as f:
                    f.write(img_response.content)
                print(f'✓ 已下载: {file_name}')
                downloaded += 1
            except Exception as e:
                print(f'✗ 下载失败 {img_url}: {str(e)}')
        print(f'\n下载完成！共下载 {downloaded} 张图片')
    except Exception as e:
        print(f'请求失败: {str(e)}')
# 使用示例
url = 'https://example.com/gallery'  # 替换为目标网址
download_static_images(url)

进阶版：爬取动态加载图片

import requests
import json
import os
import time
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
class DynamicImageScraper:
    """爬取动态加载页面的图片"""
    def __init__(self, headless=True):
        # 配置Chrome选项
        options = webdriver.ChromeOptions()
        if headless:
            options.add_argument('--headless')
        options.add_argument('--disable-gpu')
        options.add_argument('--no-sandbox')
        self.driver = webdriver.Chrome(options=options)
    def scrape_unsplash(self, query, max_images=30):
        """爬取Unsplash图片"""
        save_dir = f'unsplash_{query}'
        if not os.path.exists(save_dir):
            os.makedirs(save_dir)
        url = f'https://unsplash.com/s/photos/{query}'
        self.driver.get(url)
        # 等待图片加载
        wait = WebDriverWait(self.driver, 10)
        wait.until(EC.presence_of_element_located((By.CLASS_NAME, 'MorZF')))
        # 滚动页面加载更多图片
        downloaded = 0
        scroll_count = 0
        while downloaded < max_images:
            # 查找所有图片元素
            img_elements = self.driver.find_elements(By.CLASS_NAME, 'MorZF')
            for img in img_elements:
                if downloaded >= max_images:
                    break
                try:
                    # 获取图片URL
                    img_url = img.get_attribute('src')
                    if not img_url or 'placeholder' in img_url:
                        continue
                    # 获取高清图片URL
                    img_url_high = img_url.replace('?w=400', '?w=1920')
                    # 下载图片
                    response = requests.get(img_url_high, timeout=10)
                    if response.status_code == 200:
                        file_name = f'{query}_{downloaded+1}.jpg'
                        file_path = os.path.join(save_dir, file_name)
                        with open(file_path, 'wb') as f:
                            f.write(response.content)
                        downloaded += 1
                        print(f'✓ 已下载: {file_name}')
                except Exception as e:
                    print(f'✗ 下载失败: {str(e)}')
            # 滚动页面
            if downloaded < max_images:
                self.driver.execute_script('window.scrollTo(0, document.body.scrollHeight);')
                time.sleep(2)
                scroll_count += 1
                if scroll_count > 10:  # 防止无限滚动
                    break
        print(f'\n下载完成！共下载 {downloaded} 张图片')
    def close(self):
        self.driver.quit()
# 使用示例
scraper = DynamicImageScraper()
scraper.scrape_unsplash('nature', max_images=20)
scraper.close()

高级版：API爬取图片

import requests
import os
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from ratelimit import limits, sleep_and_retry
class APIImageScraper:
    """通过API爬取图片（以Unsplash API为例）"""
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = 'https://api.unsplash.com'
        self.headers = {
            'Authorization': f'Client-ID {self.api_key}'
        }
    @sleep_and_retry
    @limits(calls=50, period=60)  # 限制API调用频率
    def search_photos(self, query, per_page=30):
        """搜索图片"""
        url = f'{self.base_url}/search/photos'
        params = {
            'query': query,
            'per_page': per_page,
            'orientation': 'landscape'
        }
        response = requests.get(url, headers=self.headers, params=params)
        response.raise_for_status()
        return response.json()['results']
    def download_single_image(self, image_data, save_dir, index):
        """下载单张图片"""
        try:
            # 获取图片URL
            img_url = image_data['urls']['regular']
            img_id = image_data['id']
            # 获取图片信息
            description = image_data.get('description', '')[:50] or f'image_{index}'
            # 下载图片
            response = requests.get(img_url, timeout=10)
            response.raise_for_status()
            # 保存图片
            file_name = f"{index:03d}_{description.replace(' ', '_')}.jpg"
            file_path = os.path.join(save_dir, file_name)
            with open(file_path, 'wb') as f:
                f.write(response.content)
            return True, file_name
        except Exception as e:
            return False, str(e)
    def download_batch(self, query, num_images=50):
        """批量下载图片"""
        save_dir = f'api_images_{query}'
        if not os.path.exists(save_dir):
            os.makedirs(save_dir)
        print(f'开始搜索 "{query}" 相关图片...')
        # 搜索图片
        all_photos = []
        page = 1
        while len(all_photos) < num_images:
            photos = self.search_photos(query, per_page=30)
            all_photos.extend(photos)
            page += 1
            if len(photos) == 0:
                break
            time.sleep(1)  # API限制
        print(f'找到 {len(all_photos)} 张图片，开始下载...')
        # 多线程下载
        downloaded = 0
        with ThreadPoolExecutor(max_workers=5) as executor:
            futures = []
            for i, photo in enumerate(all_photos[:num_images]):
                future = executor.submit(
                    self.download_single_image, 
                    photo, 
                    save_dir, 
                    i + 1
                )
                futures.append(future)
            for future in as_completed(futures):
                success, result = future.result()
                if success:
                    downloaded += 1
                    print(f'✓ 已下载: {result}')
                else:
                    print(f'✗ 下载失败: {result}')
        print(f'\n完成！成功下载 {downloaded} 张图片')
# 使用示例
# api_key = 'your_unsplash_api_key'  # 需要注册获取
# scraper = APIImageScraper(api_key)
# scraper.download_batch('mountains', num_images=30)

综合实用版：通用图片爬虫

import requests
import re
import os
import time
from urllib.parse import urlparse
from selenium import webdriver
from bs4 import BeautifulSoup
class UniversalImageScraper:
    """通用图片爬虫"""
    def __init__(self):
        self.headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
        }
        self.downloaded_urls = set()  # 防止重复下载
    def extract_image_urls(self, html, base_url):
        """从HTML中提取所有图片URL"""
        soup = BeautifulSoup(html, 'html.parser')
        img_urls = []
        # 1. 从img标签提取
        for img in soup.find_all('img'):
            src = img.get('src') or img.get('data-src') or img.get('data-original')
            if src:
                # 处理相对路径
                if src.startswith('//'):
                    src = 'https:' + src
                elif src.startswith('/'):
                    src = urljoin(base_url, src)
                elif not src.startswith('http'):
                    src = urljoin(base_url, src)
                img_urls.append(src)
        # 2. 从CSS背景图提取
        style_pattern = r'background(?:-image)?\s*:\s*url\(["\']?(.*?)["\']?\)'
        for style in soup.find_all(style=True):
            urls = re.findall(style_pattern, style['style'])
            img_urls.extend(urls)
        # 3. 从a标签的href提取图片链接
        for a in soup.find_all('a', href=True):
            href = a['href']
            if any(href.endswith(ext) for ext in ['.jpg', '.jpeg', '.png', '.gif']):
                img_urls.append(href)
        # 去重并过滤
        unique_urls = list(set(img_urls))
        valid_urls = [url for url in unique_urls if url.startswith('http')]
        return valid_urls
    def download_image(self, url, save_dir, filename=None):
        """下载单张图片"""
        if url in self.downloaded_urls:
            return False, '已下载过'
        try:
            response = requests.get(url, headers=self.headers, timeout=10)
            response.raise_for_status()
            # 检查是否为图片
            content_type = response.headers.get('content-type', '')
            if not content_type.startswith('image/'):
                return False, '非图片文件'
            # 生成文件名
            if not filename:
                parsed = urlparse(url)
                filename = os.path.basename(parsed.path) or f'image_{int(time.time())}.jpg'
            # 确保有扩展名
            if '.' not in filename:
                ext = content_type.split('/')[-1]
                filename = f'{filename}.{ext}'
            # 保存图片
            filepath = os.path.join(save_dir, filename)
            with open(filepath, 'wb') as f:
                f.write(response.content)
            self.downloaded_urls.add(url)
            return True, filename
        except Exception as e:
            return False, str(e)
    def scrape(self, url, save_dir='downloaded_images', max_images=50):
        """主爬取方法"""
        if not os.path.exists(save_dir):
            os.makedirs(save_dir)
        print(f'开始爬取: {url}')
        # 获取网页内容
        response = requests.get(url, headers=self.headers, timeout=10)
        response.raise_for_status()
        # 提取图片URL
        img_urls = self.extract_image_urls(response.text, url)
        print(f'找到 {len(img_urls)} 张图片')
        # 下载图片
        downloaded = 0
        for img_url in img_urls[:max_images]:
            if downloaded >= max_images:
                break
            success, result = self.download_image(img_url, save_dir)
            if success:
                downloaded += 1
                print(f'[{downloaded}/{max_images}] ✓ {result}')
            else:
                print(f'✗ {result}')
            time.sleep(0.5)  # 礼貌延迟
        print(f'\n爬取完成！共下载 {downloaded} 张图片')
# 使用示例
scraper = UniversalImageScraper()
scraper.scrape('https://example.com', max_images=20)

安装依赖

pip install requests beautifulsoup4 selenium pillow
# 如果需要多线程
pip install concurrent.futures
# API速率限制
pip install ratelimit

注意事项

法律法规: 遵守网站的robots.txt和使用条款
版权问题: 注意图片的版权许可
请求频率: 添加延迟避免对服务器造成压力
异常处理: 做好异常处理和重试机制
User-Agent: 设置合理的请求头

这些案例覆盖了常见的图片爬取场景,你可以根据实际需求进行修改和扩展。

本文地址： https://www.bmjup.com/post/12555.html