# Python图片爬虫实践：从基础采集到高级应用

## 爬虫基础与环境搭建

图片爬虫是通过程序自动从网页上提取和下载图片的技术。掌握这项技能需要理解HTTP协议、网页解析和文件操作等基础知识。Python凭借其丰富的第三方库，成为图片爬虫开发的首选语言。

### 基础环境配置

```python

# 安装必要的库

# pip install requests beautifulsoup4 Pillow lxml

import os

import sys

import requests

from bs4 import BeautifulSoup

from urllib.parse import urljoin, urlparse

import time

import random

class ImageDownloader:

    def __init__(self, save_dir='downloaded_images'):

        """初始化下载器"""

        self.save_dir = save_dir

        self.session = requests.Session()

        self.session.headers.update({

            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'

        })

        self.create_save_directory()

    

    def create_save_directory(self):

        """创建保存目录"""

        if not os.path.exists(self.save_dir):

            os.makedirs(self.save_dir)

            print(f"创建目录: {self.save_dir}")

```

## 基础图片爬取技术

### 单页图片下载

```python

class BasicImageCrawler:

    def download_single_image(self, image_url, filename=None):

        """下载单张图片"""

        try:

            response = self.session.get(image_url, timeout=10)

            response.raise_for_status()

            

            # 如果未指定文件名，从URL提取

            if filename is None:

                filename = os.path.basename(urlparse(image_url).path)

                if not filename or '.' not in filename:

                    filename = f"image_{int(time.time())}.jpg"

            

            # 确保文件名有效

            filename = self.sanitize_filename(filename)

            filepath = os.path.join(self.save_dir, filename)

            

            # 保存图片

            with open(filepath, 'wb') as f:

                f.write(response.content)

            

            print(f"成功下载: {filename} ({len(response.content)} bytes)")

            return filepath

            

        except requests.exceptions.RequestException as e:

            print(f"下载失败 {image_url}: {e}")

            return None

    

    def sanitize_filename(self, filename):

        """清理文件名，移除非法字符"""

        invalid_chars = '<>:"/\\|?*'

        for char in invalid_chars:

            filename = filename.replace(char, '_')

        return filename

    

    def extract_images_from_page(self, page_url):

        """从网页提取所有图片链接"""

        try:

            response = self.session.get(page_url)

            response.raise_for_status()

            soup = BeautifulSoup(response.text, 'lxml')

            

            image_urls = []

            for img_tag in soup.find_all('img'):

                img_url = img_tag.get('src') or img_tag.get('data-src')

                if img_url:

                    # 处理相对URL

                    full_url = urljoin(page_url, img_url)

                    image_urls.append(full_url)

            

            print(f"从 {page_url} 找到 {len(image_urls)} 张图片")

            return image_urls

            

        except Exception as e:

            print(f"解析页面失败 {page_url}: {e}")

            return []

    

    def crawl_page_images(self, page_url):

        """爬取页面所有图片"""

        image_urls = self.extract_images_from_page(page_url)

        

        downloaded_count = 0

        for i, img_url in enumerate(image_urls, 1):

            print(f"正在下载第 {i}/{len(image_urls)} 张图片...")

            result = self.download_single_image(img_url)

            if result:

                downloaded_count += 1

            # 避免请求过快

            time.sleep(random.uniform(0.5, 1.5))

        

        print(f"完成！成功下载 {downloaded_count}/{len(image_urls)} 张图片")

        return downloaded_count

# 使用示例

if __name__ == "__main__":

    crawler = BasicImageCrawler()

    

    # 示例：下载单个图片

    crawler.download_single_image(

        'https://example.com/image.jpg',

        'example_image.jpg'

    )

    

    # 示例：爬取页面所有图片

    crawler.crawl_page_images('https://example.com/gallery')

```

### 多页爬取与分页处理

```python

class MultiPageCrawler(BasicImageCrawler):

    def crawl_multiple_pages(self, base_url, start_page=1, end_page=5):

        """爬取多页图片"""

        all_images = []

        

        for page_num in range(start_page, end_page + 1):

            # 构造分页URL（不同网站分页规则不同）

            if '?' in base_url:

                page_url = f"{base_url}&page={page_num}"

            else:

                page_url = f"{base_url}?page={page_num}"

            

            print(f"正在处理第 {page_num} 页: {page_url}")

            

            page_images = self.extract_images_from_page(page_url)

            all_images.extend(page_images)

            

            # 下载当前页图片

            self.download_images_batch(page_images, page_num)

            

            # 页间延迟

            time.sleep(random.uniform(2, 4))

        

        return all_images

    

    def download_images_batch(self, image_urls, page_num):

        """批量下载图片"""

        for i, img_url in enumerate(image_urls, 1):

            # 生成包含页码的文件名

            ext = os.path.splitext(urlparse(img_url).path)[1] or '.jpg'

            filename = f"page{page_num:03d}_img{i:03d}{ext}"

            

            self.download_single_image(img_url, filename)

            time.sleep(random.uniform(0.3, 0.7))

    

    def crawl_with_pattern(self, url_pattern, total_pages):

        """使用URL模式爬取（适用于规则明显的网站）"""

        for page in range(1, total_pages + 1):

            page_url = url_pattern.format(page=page)

            print(f"爬取: {page_url}")

            

            image_urls = self.extract_images_from_page(page_url)

            for img_url in image_urls:

                self.download_single_image(img_url)

            

            time.sleep(random.uniform(1, 3))

# 使用示例

if __name__ == "__main__":

    crawler = MultiPageCrawler(save_dir='wallpapers')

    

    # 示例1：爬取多页

    crawler.crawl_multiple_pages(

        'https://wallpaper.example.com/category/nature',

        start_page=1,

        end_page=3

    )

    

    # 示例2：使用URL模式

    crawler.crawl_with_pattern(

        'https://example.com/gallery/page/{page}',

        total_pages=5

    )

```

## 高级图片处理技术

### 图片验证与过滤

```python

from PIL import Image

import io

import hashlib

class AdvancedImageCrawler(MultiPageCrawler):

    def __init__(self, save_dir='filtered_images'):

        super().__init__(save_dir)

        self.downloaded_hashes = set()  # 用于去重

        self.min_size = (100, 100)  # 最小图片尺寸

        self.max_size = (10000, 10000)  # 最大图片尺寸

    

    def validate_image(self, image_content):

        """验证图片有效性"""

        try:

            # 尝试打开图片

            image = Image.open(io.BytesIO(image_content))

            

            # 检查尺寸

            width, height = image.size

            if width < self.min_size[0] or height < self.min_size[1]:

                return False, "图片尺寸太小"

            if width > self.max_size[0] or height > self.max_size[1]:

                return False, "图片尺寸太大"

            

            # 检查格式

            if image.format not in ['JPEG', 'PNG', 'GIF', 'WEBP']:

                return False, f"不支持的格式: {image.format}"

            

            return True, f"有效图片: {width}x{height}, {image.format}"

            

        except Exception as e:

            return False, f"无效图片: {str(e)}"

    

    def calculate_image_hash(self, image_content):

        """计算图片哈希值（用于去重）"""

        return hashlib.md5(image_content).hexdigest()

    

    def download_with_validation(self, image_url, filename=None):

        """带验证的图片下载"""

        try:

            response = self.session.get(image_url, timeout=15)

            response.raise_for_status()<"hiqiu.hbhsjy.cn">

            

            # 验证图片

            is_valid, message = self.validate_image(response.content)

            if not is_valid:

                print(f"跳过无效图片 {image_url}: {message}")

                return None

            

            # 检查重复

            image_hash = self.calculate_image_hash(response.content)

            if image_hash in self.downloaded_hashes:

                print(f"跳过重复图片: {image_url}")

                return None

            

            # 保存图片

            if filename is None:

                filename = self.generate_filename(image_url, response)

            

            filepath = os.path.join(self.save_dir, filename)

            with open(filepath, 'wb') as f:

                f.write(response.content)

            

            # 记录已下载的哈希值

            self.downloaded_hashes.add(image_hash)

            print(f"下载成功: {filename} - {message}")

            return filepath

            

        except Exception as e:

            print(f"下载失败 {image_url}: {e}")

            return None

    

    def generate_filename(self, image_url, response):

        """生成有意义文件名"""

        # 尝试从Content-Disposition获取文件名

        content_disposition = response.headers.get('Content-Disposition', '')

        if 'filename=' in content_disposition:

            filename = content_disposition.split('filename=')[1].strip('"\'')

        else:

            # 从URL提取

            filename = os.path.basename(urlparse(image_url).path)

            if not filename or '.' not in filename:

                # 使用时间戳和哈希

                timestamp = int(time.time())

                url_hash = hashlib.md5(image_url.encode()).hexdigest()[:8]

                content_type = response.headers.get('Content-Type', '')

                if 'jpeg' in content_type or 'jpg' in content_type:

                    ext = '.jpg'

                elif 'png' in content_type:

                    ext = '.png'

                else:

                    ext = '.jpg'

                filename = f"image_{timestamp}_{url_hash}{ext}"

        

        return self.sanitize_filename(filename)

    

    def filter_images_by_aspect_ratio(self, page_url, min_ratio=1.2, max_ratio=2.5):

        """按宽高比过滤图片"""

        image_urls = self.extract_images_from_page(page_url)

        

        filtered_urls = []

        for img_url in image_urls:

            try:

                # 获取图片头信息（不下载完整图片）

                head_response = self.session.head(img_url, timeout=5, allow_redirects=True)

                

                # 有些服务器不支持HEAD，改用GET但只读取部分数据

                if head_response.status_code != 200:

                    # 获取前1024字节来检测尺寸

                    response = self.session.get(img_url, stream=True, timeout=5)

                    if response.status_code == 200:

                        # 读取足够的数据来获取图片信息

                        chunk = b''

                        for chunk_data in response.iter_content(1024):

                            chunk += chunk_data

                            if len(chunk) >= 1024:

                                response.close()

                                break

                        

                        # 尝试从数据中获取尺寸

                        try:

                            image = Image.open(io.BytesIO(chunk))

                            width, height = image.size

                            aspect_ratio = width / height

                            

                            if min_ratio <= aspect_ratio <= max_ratio:

                                filtered_urls.append(img_url)

                                print(f"符合宽高比: {img_url} ({width}x{height}, 比例: {aspect_ratio:.2f})")

                        except:

                            continue

            except Exception as e:

                print(f"检查图片失败 {img_url}: {e}")

                continue

        

        return filtered_urls

# 使用示例

if __name__ == "__main__":

    crawler = AdvancedImageCrawler(save_dir='high_quality_images')

    

    # 下载并验证图片

    crawler.download_with_validation(

        'https://example.com/high-res-image.jpg'

    )

    

    # 按宽高比过滤

    wallpaper_urls = crawler.filter_images_by_aspect_ratio(

        'https://wallpaper.example.com',

        min_ratio=1.7,  # 宽屏比例

        max_ratio=2.4

    )

    

    # 下载过滤后的图片

    for url in wallpaper_urls:

        crawler.download_with_validation(url)

```

### 异步并发下载

```python

import asyncio

import aiohttp

import aiofiles

from concurrent.futures import ThreadPoolExecutor

class AsyncImageCrawler:

    def __init__(self, save_dir='async_images', max_concurrent=10):

        self.save_dir = save_dir

        self.max_concurrent = max_concurrent

        self.semaphore = asyncio.Semaphore(max_concurrent)

        

        if not os.path.exists(save_dir):

            os.makedirs(save_dir)

    

    async def download_image_async(self, session, image_url, filename=None):

        """异步下载单张图片"""

        async with self.semaphore:

            try:<"hiqiu.fjfjmjg.com">

                async with session.get(image_url, timeout=30) as response:

                    if response.status == 200:

                        content = await response.read()

                        

                        if filename is None:

                            filename = os.path.basename(urlparse(image_url).path)

                            if not filename:

                                filename = f"image_{int(time.time())}_{hash(image_url)[:8]}.jpg"

                        

                        filepath = os.path.join(self.save_dir, filename)

                        async with aiofiles.open(filepath, 'wb') as f:

                            await f.write(content)

                        

                        print(f"异步下载成功: {filename}")

                        return True

                    else:

                        print(f"下载失败 {image_url}: HTTP {response.status}")

                        return False

                        

            except Exception as e:

                print(f"异步下载异常 {image_url}: {e}")

                return False

    

    async def extract_images_async(self, session, page_url):

        """异步提取页面图片链接"""

        try:

            async with session.get(page_url) as response:

                html = await response.text()

                soup = BeautifulSoup(html, 'lxml')

                

                image_urls = []

                for img_tag in soup.find_all('img'):

                    img_url = img_tag.get('src') or img_tag.get('data-src')

                    if img_url:

                        full_url = urljoin(page_url, img_url)

                        image_urls.append(full_url)

                

                return image_urls

                

        except Exception as e:

            print(f"提取图片链接失败 {page_url}: {e}")

            return []

    

    async def crawl_page_async(self, page_url):

        """异步爬取单个页面"""

        connector = aiohttp.TCPConnector(limit=self.max_concurrent)

        timeout = aiohttp.ClientTimeout(total=60)

        

        async with aiohttp.ClientSession(

            connector=connector,

            timeout=timeout,

            headers={'User-Agent': 'Mozilla/5.0'}

        ) as session:

            

            # 提取图片链接

            image_urls = await self.extract_images_async(session, page_url)

            print(f"找到 {len(image_urls)} 张图片")

            

            # 创建下载任务

            tasks = []

            for i, img_url in enumerate(image_urls):

                filename = f"img_{i:04d}.jpg"

                task = self.download_image_async(session, img_url, filename)

                tasks.append(task)

            

            # 等待所有任务完成

            results = await asyncio.gather(*tasks, return_exceptions=True)

            

            success_count = sum(1 for r in results if r is True)

            print(f"完成！成功下载 {success_count}/{len(image_urls)} 张图片")

            return success_count

    

    async def crawl_multiple_pages_async(self, page_urls):

        """异步爬取多个页面"""

        tasks = []

        for page_url in page_urls:

            task = self.crawl_page_async(page_url)

            tasks.append(task)

        

        results = await asyncio.gather(*tasks, return_exceptions=True)

        total_success = sum(r for r in results if isinstance(r, int))

        

        print(f"所有页面完成！总计成功下载 {total_success} 张图片")

        return total_success

    

    def run_async_crawl(self, page_url):

        """运行异步爬虫"""

        loop = asyncio.new_event_loop()

        asyncio.set_event_loop(loop)

        try:

            result = loop.run_until_complete(self.crawl_page_async(page_url))

            return result

        finally:

            loop.close()

# 使用示例

if __name__ == "__main__":

    crawler = AsyncImageCrawler(max_concurrent=5)

    

    # 运行异步爬虫

    crawler.run_async_crawl('https://example.com/gallery')

    

    # 或者在异步环境中使用

    async def main():

        crawler = AsyncImageCrawler(max_concurrent=8)

        

        # 爬取单个页面

        await crawler.crawl_page_async('https://example.com/page1')

        

        # 爬取多个页面

        pages = [

            'https://example.com/page1',

            'https://example.com/page2',

            'https://example.com/page3'

        ]

        await crawler.crawl_multiple_pages_async(pages)

    

    # asyncio.run(main())

```

## 实战项目：完整图片爬虫系统

```python

import json

import sqlite3

from datetime import datetime

from pathlib import Path

class CompleteImageCrawlerSystem:

    """完整的图片爬虫系统"""

    

    def __init__(self, config_file='crawler_config.json'):

        self.config = self.load_config(config_file)

        self.setup_directories()

        self.setup_database()

        

        # 初始化组件

        self.basic_crawler = BasicImageCrawler(self.config['save_dir'])

        self.advanced_crawler = AdvancedImageCrawler(self.config['save_dir'])

    

    def load_config(self, config_file):

        """加载配置文件"""

        default_config = {<"hiqiu.juchengbibu.com">

            'save_dir': 'crawled_images',

            'database': 'crawler_data.db',

            'max_retries': 3,

            'timeout': 30,

            'user_agent': 'Mozilla/5.0',

            'rate_limit': 1.0,  # 请求间隔秒数

            'image_filters': {

                'min_width': 200,

                'min_height': 200,

                'allowed_formats': ['jpg', 'jpeg', 'png', 'gif', 'webp']

            }

        }

        

        if os.path.exists(config_file):

            with open(config_file, 'r') as f:

                user_config = json.load(f)

                default_config.update(user_config)

        

        return default_config

    

    def setup_directories(self):

        """设置目录结构"""

        directories = [

            self.config['save_dir'],

            'logs',

            'temp',

            'backup'

        ]

        

        for directory in directories:

            Path(directory).mkdir(exist_ok=True)

    

    def setup_database(self):

        """设置数据库"""

        self.conn = sqlite3.connect(self.config['database'])

        self.create_tables()

    

    def create_tables(self):

        """创建数据库表"""

        cursor = self.conn.cursor()

        

        # 图片记录表

        cursor.execute('''

            CREATE TABLE IF NOT EXISTS images (

                id INTEGER PRIMARY KEY AUTOINCREMENT,

                url TEXT UNIQUE,

                filename TEXT,

                filesize INTEGER,

                width INTEGER,

                height INTEGER,

                format TEXT,

                downloaded_at TIMESTAMP,

                source_url TEXT

            )

        ''')

        

        # 任务记录表

        cursor.execute('''

            CREATE TABLE IF NOT EXISTS tasks (

                id INTEGER PRIMARY KEY AUTOINCREMENT,

                start_url TEXT,

                pages_crawled INTEGER,

                images_found INTEGER,

                images_downloaded INTEGER,

                start_time TIMESTAMP,

                end_time TIMESTAMP,

                status TEXT

            )

        ''')

        

        self.conn.commit()

    

    def log_image(self, url, filename, filesize, width=None, height=None, 

                  format=None, source_url=None):

        """记录下载的图片信息"""

        cursor = self.conn.cursor()

        cursor.execute('''

            INSERT OR IGNORE INTO images 

            (url, filename, filesize, width, height, format, downloaded_at, source_url)

            VALUES (?, ?, ?, ?, ?, ?, ?, ?)

        ''', (url, filename, filesize, width, height, format, 

              datetime.now(), source_url))

        self.conn.commit()

    

    def start_task(self, start_url):

        """开始爬取任务"""

        cursor = self.conn.cursor()

        cursor.execute('''

            INSERT INTO tasks (start_url, start_time, status)

            VALUES (?, ?, ?)

        ''', (start_url, datetime.now(), 'running'))

        task_id = cursor.lastrowid

        self.conn.commit()

        

        return task_id

    

    def update_task(self, task_id, pages_crawled, images_found, 

                   images_downloaded, status='completed'):

        """更新任务状态"""

        cursor = self.conn.cursor()

        cursor.execute('''

            UPDATE tasks 

            SET pages_crawled = ?, images_found = ?, images_downloaded = ?,

                end_time = ?, status = ?

            WHERE id = ?

        ''', (pages_crawled, images_found, images_downloaded, 

              datetime.now(), status, task_id))

        self.conn.commit()

    

    def crawl_website(self, start_url, max_pages=10, max_images=100):

        """完整的网站爬取流程"""

        task_id = self.start_task(start_url)

        

        print(f"开始爬取任务 {task_id}: {start_url}")

        print(f"最大页面: {max_pages}, 最大图片: {max_images}")

        

        pages_crawled = 0

        total_images_found = 0

        total_images_downloaded = 0

        

        # 获取初始页面

        current_url = start_url

        visited_urls = set()

        

        try:

            while (pages_crawled < max_pages and 

                   total_images_downloaded < max_images and 

                   current_url and current_url not in visited_urls):

                

                print(f"\n正在爬取页面 {pages_crawled + 1}: {current_url}")

                visited_urls.add(current_url)<"haiqiu.aekt.cn">

                

                # 提取当前页图片

                image_urls = self.basic_crawler.extract_images_from_page(current_url)

                total_images_found += len(image_urls)

                

                # 下载图片

                downloaded = 0

                for img_url in image_urls:

                    if total_images_downloaded >= max_images:

                        break

                    

                    # 使用高级下载器（带验证）

                    result = self.advanced_crawler.download_with_validation(img_url)

                    if result:

                        total_images_downloaded += 1

                        downloaded += 1

                        

                        # 记录到数据库

                        # 这里可以添加更多信息记录

                    

                    time.sleep(self.config['rate_limit'])

                

                print(f"本页下载: {downloaded}/{len(image_urls)} 张图片")

                pages_crawled += 1

                

                # 寻找下一个页面（简单实现）

                if pages_crawled < max_pages:

                    next_page = self.find_next_page(current_url, pages_crawled)

                    current_url = next_page

                

                time.sleep(random.uniform(1, 3))

            

            # 更新任务状态

            self.update_task(

                task_id, pages_crawled, total_images_found, 

                total_images_downloaded, 'completed'

            )

            

            print(f"\n任务完成！")

            print(f"爬取页面: {pages_crawled}")

            print(f"发现图片: {total_images_found}")

            print(f"下载图片: {total_images_downloaded}")

            

            return {

                'task_id': task_id,

                'pages_crawled': pages_crawled,

                'images_found': total_images_found,

                'images_downloaded': total_images_downloaded

            }

            

        except Exception as e:

            print(f"爬取过程出错: {e}")

            self.update_task(task_id, pages_crawled, total_images_found, 

                           total_images_downloaded, 'failed')

            raise

    

    def find_next_page(self, current_url, page_num):

        """查找下一页（需要根据具体网站调整）"""

        # 简单的分页检测

        if 'page=' in current_url:

            # 替换页码

            import re

            next_url = re.sub(r'page=\d+', f'page={page_num + 1}', current_url)

        elif '?p=' in current_url:

            next_url = re.sub(r'p=\d+', f'p={page_num + 1}', current_url)

        else:

            # 尝试常见分页模式

            if current_url.endswith('/'):

                next_url = f"{current_url}page/{page_num + 1}/"

            else:

                next_url = f"{current_url}?page={page_num + 1}"

        

        return next_url

    

    def generate_report(self, task_id):

        """生成爬取报告"""

        cursor = self.conn.cursor()

        

        # 获取任务信息

        cursor.execute('SELECT * FROM tasks WHERE id = ?', (task_id,))

        task = cursor.fetchone()

        

        if not task:

            return None

        

        # 获取图片统计

        cursor.execute('''

            SELECT COUNT(*), SUM(filesize), 

                   AVG(width), AVG(height)

            FROM images 

            WHERE source_url LIKE ? || '%'

        ''', (task[1],))  # task[1] 是 start_url

        

        stats = cursor.fetchone()

        

        report = {

            'task_id': task_id,

            'start_url': task[1],

            'start_time': task[5],

            'end_time': task[6],

            'status': task[7],

            'pages_crawled': task[2] or 0,

            'images_found': task[3] or 0,

            'images_downloaded': task[4] or 0,

            'total_size_mb': (stats[1] or 0) / (1024 * 1024),

            'avg_width': stats[2] or 0,

            'avg_height': stats[3] or 0

        }

        

        # 保存报告到文件

        report_file = f"reports/task_{task_id}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"

        with open(report_file, 'w') as f:<"haiqiu.baiyng.com">

            json.dump(report, f, indent=2, default=str)

        

        return report

# 使用示例

if __name__ == "__main__":

    # 创建爬虫系统

    system = CompleteImageCrawlerSystem()

    

    # 配置示例

    config = {

        'save_dir': 'my_wallpapers',

        'rate_limit': 0.5,

        'image_filters': {

            'min_width': 1920,

            'min_height': 1080,

            'allowed_formats': ['jpg', 'png', 'webp']

        }

    }

    

    # 保存配置

    with open('my_config.json', 'w') as f:

        json.dump(config, f, indent=2)

    

    # 重新加载配置

    system = CompleteImageCrawlerSystem('my_config.json')

    

    # 执行爬取任务

    result = system.crawl_website(

        start_url='https://wallpaper.example.com/nature',

        max_pages=3,<"haiqiu.fajining.com">

        max_images=50

    )

    

    # 生成报告

    if result:

        report = system.generate_report(result['task_id'])

        print("\n爬取报告:")

        for key, value in report.items():

            print(f"{key}: {value}")

```

## 反爬虫策略与应对

```python

class AntiAntiCrawler:

    """应对反爬虫策略"""

    

    def __init__(self):

        self.user_agents = [

            'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',

            'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15',

            'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36',

        ]

        self.proxies = self.load_proxies()

        self.current_proxy_index = 0

    

    def load_proxies(self):

        """加载代理列表"""

        # 可以从文件或API加载

        proxies = []

        try:

            with open('proxies.txt', 'r') as f:

                proxies = [line.strip() for line in f if line.strip()]

        except FileNotFoundError:

            pass

        

        return proxies

    

    def get_random_headers(self):

        """获取随机请求头"""

        return {

            'User-Agent': random.choice(self.user_agents),

            'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',

            'Accept-Language': 'en-US,en;q=0.5',

            'Accept-Encoding': 'gzip, deflate',

            'DNT': '1',

            'Connection': 'keep-alive',

            'Upgrade-Insecure-Requests': '1',

            'Referer': 'https://www.google.com/'

        }

    

    def rotate_proxy(self):

        """轮换代理"""

        if not self.proxies:

            return None

        

        proxy = self.proxies[self.current_proxy_index]

        self.current_proxy_index = (self.current_proxy_index + 1) % len(self.proxies)

        

        return {

            'http': f'http://{proxy}',

            'https': f'http://{proxy}'

        }

    

    def make_request_with_retry(self, url, max_retries=3):

        """带重试的请求"""

        for attempt in range(max_retries):

            try:

                headers = self.get_random_headers()

                proxy = self.rotate_proxy() if self.proxies else None

                

                if attempt > 0:  # 首 次失败后使用代理

                    proxies = proxy

                    print(f"重试 {attempt+1}/{max_retries} 使用代理: {proxy}")

                else:

                    proxies = None

                

                response = requests.get(

                    url, 

                    headers=headers, 

                    proxies=proxies,

                    timeout=15,

                    allow_redirects=True

                )

                

                if response.status_code == 200:

                    return response

                elif response.status_code == 429:  # 请求过多

                    wait_time = 30 * (attempt + 1)  # 指数退避

                    print(f"被限速，等待 {wait_time} 秒")

                    time.sleep(wait_time)

                else:

                    print(f"HTTP {response.status_code}, 重试...")

                

            except requests.exceptions.RequestException as e:

                print(f"请求失败: {e}, 重试 {attempt+1}/{max_retries}")

            

            time.sleep(random.uniform(2, 5))

        

        return None

# 集成到爬虫系统中

class RobustImageCrawler(CompleteImageCrawlerSystem):

    """健壮的图片爬虫系统"""

    

    def __init__(self, config_file='crawler_config.json'):

        super().__init__(config_file)

        self.anti_crawler = AntiAntiCrawler()

    

    def robust_extract_images(self, page_url):

        """健壮的图片提取"""

        response = self.anti_crawler.make_request_with_retry(page_url)

        

        if response:

            soup = BeautifulSoup(response.text, 'lxml')

            # ... 提取图片逻辑 ...

            return image_urls

        else:

            print(f"无法获取页面: {page_url}")

            return []

```

## 总结与最佳实践

图片爬虫开发需要综合考虑多个方面：

1. **遵守法律法规**：确保爬取行为符合目标网站的robots.txt，尊重版权

2. **控制请求频率**：添加适当延迟，避免对目标服务器造成负担

3. **错误处理**：完善的异常处理机制，确保程序稳定性

4. **资源管理**：及时释放网络连接和文件句柄

5. **数据去重**：使用哈希值等方法避免重复下载

6. **可扩展性**：设计良好的架构，便于功能扩展和维护

通过本文介绍的基础到高级技术，可以构建功能完善的图片爬虫系统。在实际应用中，需要根据具体目标网站的特点进行调整和优化。记住，技术应当用于合法合规的用途，尊重他人的劳动成果和知识产权。