IT技术社区_专业开发者平台_最新编程教程与解决方案 -沐辰社区

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

wen python案例 27

本文目录导读：

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

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

我来为你提供几个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

注意事项

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

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

上一篇Python视频爬虫案例如何抓取视频链接

下一篇Python爬虫保存案例如何保存爬取数据

相关文章

抱歉，评论功能暂时关闭!