网络爬虫是数据采集的利器，也是很多初学者进阶Python的第一道坎。今天分享一套完整的爬虫实战教程，从基础请求到反爬应对，从单页面到分布式，帮你系统掌握Python爬虫技能。

爬虫基础知识

在开始写代码之前，先搞清楚几个核心概念：

• HTTP协议：理解GET/POST请求、请求头、响应码
• HTML基础：能看懂网页结构，定位元素
• JavaScript渲染：有些数据是JS动态加载的，需要特殊处理

常见的Python爬虫库：

• requests：最基础的HTTP库
• BeautifulSoup：HTML/XML解析
• Scrapy：专业爬虫框架
• Selenium/Playwright：处理JS渲染
• aiohttp：异步爬虫

基础请求封装

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
import time
import random
from typing import Optional, Dict

class BaseCrawler:
    """爬虫基类"""
    
    def __init__(self):
        self.session = self._create_session()
        self.headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
            'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
            'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
            'Accept-Encoding': 'gzip, deflate',
            'Connection': 'keep-alive',
        }
    
    def _create_session(self) -> requests.Session:
        """创建带重试机制的会话"""
        session = requests.Session()
        
        retry_strategy = Retry(
            total=3,
            backoff_factor=1,
            status_forcelist=[500, 502, 503, 504],
        )
        
        adapter = HTTPAdapter(max_retries=retry_strategy)
        session.mount("http://", adapter)
        session.mount("https://", adapter)
        
        return session
    
    def get(self, url: str, headers: Optional[Dict] = None, 
            params: Optional[Dict] = None, **kwargs) -> Optional[requests.Response]:
        """
        发送GET请求
        
        Args:
            url: 目标URL
            headers: 自定义请求头
            params: URL参数
        """
        final_headers = self.headers.copy()
        if headers:
            final_headers.update(headers)
        
        try:
            response = self.session.get(
                url=url,
                headers=final_headers,
                params=params,
                timeout=30,
                **kwargs
            )
            response.raise_for_status()
            return response
            
        except requests.exceptions.RequestException as e:
            print(f"请求失败: {url} - {e}")
            return None
    
    def post(self, url: str, data: Dict = None, json: Dict = None,
             headers: Optional[Dict] = None) -> Optional[requests.Response]:
        """发送POST请求"""
        final_headers = self.headers.copy()
        if headers:
            final_headers.update(headers)
        
        try:
            response = self.session.post(
                url=url,
                headers=final_headers,
                data=data,
                json=json,
                timeout=30
            )
            response.raise_for_status()
            return response
            
        except requests.exceptions.RequestException as e:
            print(f"请求失败: {url} - {e}")
            return None
    
    def random_sleep(self, min_seconds=1, max_seconds=3):
        """随机延时"""
        time.sleep(random.uniform(min_seconds, max_seconds))

BeautifulSoup解析器

from bs4 import BeautifulSoup
from typing import List, Dict, Optional, Callable

class HTMLParser:
    """HTML解析器"""
    
    def __init__(self, html: str, parser: str = 'html.parser'):
        self.soup = BeautifulSoup(html, parser)
    
    def find_one(self, tag: str, attrs: Dict = None, 
                 class_: str = None, id: str = None,
                 text: str = None) -> Optional[BeautifulSoup]:
        """
        查找单个元素
        
        Args:
            tag: 标签名，如 'div', 'a', 'span'
            attrs: 属性字典，如 {'data-id': '123'}
            class_: CSS类名
            id: 元素ID
            text: 文本内容
        """
        return self.soup.find(
            tag,
            attrs=attrs,
            class_=class_,
            id=id,
            string=text
        )
    
    def find_all(self, tag: str, attrs: Dict = None,
                 class_: str = None, limit: int = None) -> List[BeautifulSoup]:
        """查找所有匹配元素"""
        return self.soup.find_all(
            tag,
            attrs=attrs,
            class_=class_,
            limit=limit
        )
    
    def find_all_safe(self, tag: str, attrs: Dict = None,
                      class_: str = None, limit: int = None) -> List[BeautifulSoup]:
        """安全查找，返回空列表而非None"""
        result = self.find_all(tag, attrs, class_, limit)
        return result if result else []
    
    def extract_text(self, tag: str = None, attrs: Dict = None) -> str:
        """提取文本"""
        element = self.find_one(tag, attrs)
        if element:
            return element.get_text(strip=True)
        return ''
    
    def extract_links(self, base_url: str = '', 
                      filter_func: Callable[[str], bool] = None) -> List[str]:
        """
        提取所有链接
        
        Args:
            base_url: 基础URL，用于拼接相对路径
            filter_func: 过滤函数，接收href返回bool
        """
        links = []
        for a in self.soup.find_all('a', href=True):
            href = a['href']
            
            # 处理相对路径
            if href.startswith('/'):
                href = base_url.rstrip('/') + href
            elif not href.startswith('http'):
                href = base_url + '/' + href
            
            # 应用过滤
            if filter_func is None or filter_func(href):
                links.append(href)
        
        return list(set(links))  # 去重
    
    def extract_images(self) -> List[str]:
        """提取所有图片URL"""
        images = []
        for img in self.soup.find_all('img'):
            src = img.get('src') or img.get('data-src')
            if src:
                images.append(src)
        return images

实战案例：爬取图片资源

import os
from pathlib import Path
from urllib.parse import urljoin, urlparse

class ImageCrawler(BaseCrawler):
    """图片爬虫"""
    
    def __init__(self, save_dir='images'):
        super().__init__()
        self.save_dir = Path(save_dir)
        self.save_dir.mkdir(parents=True, exist_ok=True)
        self.downloaded = set()
    
    def crawl_page(self, url: str, selectors: List[str] = None,
                   min_size: int = 0) -> List[str]:
        """
        爬取页面图片
        
        Args:
            url: 页面URL
            selectors: CSS选择器列表
            min_size: 最小文件大小(KB)
        """
        response = self.get(url)
        if not response:
            return []
        
        parser = HTMLParser(response.text)
        domain = urlparse(url).netloc
        
        # 确定选择器
        if not selectors:
            selectors = ['img', 'picture source', 'a[href$=".jpg"]', 'a[href$=".png"]']
        
        image_urls = set()
        
        for selector in selectors:
            elements = parser.find_all(selector.split('[')[0], limit=100)
            
            for elem in elements:
                if selector.startswith('img'):
                    src = elem.get('src') or elem.get('data-src')
                elif selector.startswith('picture'):
                    src = elem.get('srcset', '').split()[0]
                else:
                    src = elem.get('href')
                
                if src and self._is_valid_image(src):
                    # 绝对路径转换
                    if src.startswith('//'):
                        src = 'https:' + src
                    elif src.startswith('/'):
                        src = f'https://{domain}{src}'
                    
                    image_urls.add(src)
        
        # 下载图片
        downloaded = []
        for img_url in image_urls:
            filepath = self.download(img_url, min_size=min_size)
            if filepath:
                downloaded.append(filepath)
            self.random_sleep(0.5, 1.5)
        
        return downloaded
    
    def _is_valid_image(self, url: str) -> bool:
        """验证是否是有效图片URL"""
        image_extensions = ['.jpg', '.jpeg', '.png', '.gif', '.webp', '.bmp']
        parsed = urlparse(url.lower())
        return any(parsed.path.endswith(ext) for ext in image_extensions)
    
    def download(self, url: str, min_size: int = 0) -> Optional[str]:
        """下载单张图片"""
        if url in self.downloaded:
            return None
        
        try:
            response = self.get(url)
            if not response:
                return None
            
            content = response.content
            
            # 检查文件大小
            size_kb = len(content) / 1024
            if min_size > 0 and size_kb < min_size:
                print(f"跳过小文件: {url} ({size_kb:.1f}KB)")
                return None
            
            # 生成文件名
            parsed = urlparse(url)
            ext = Path(parsed.path).suffix or '.jpg'
            filename = f"{hash(url)}{ext}"
            filepath = self.save_dir / filename
            
            with open(filepath, 'wb') as f:
                f.write(content)
            
            self.downloaded.add(url)
            print(f"✓ 下载成功: {url} -> {filepath} ({size_kb:.1f}KB)")
            
            return str(filepath)
            
        except Exception as e:
            print(f"✗ 下载失败: {url} - {e}")
            return None

实战案例：爬取文章列表

class ArticleCrawler(BaseCrawler):
    """文章列表爬虫"""
    
    def __init__(self):
        super().__init__()
        self.items = []
    
    def crawl_list(self, url: str, page_range: tuple = (1, 10),
                   list_selector: str = 'article',
                   title_selector: str = 'h2.title',
                   link_selector: str = 'a[href]',
                   date_selector: str = 'time') -> List[Dict]:
        """
        爬取文章列表
        
        Args:
            url: 列表页URL，支持 {page} 占位符
            page_range: 页码范围
            list_selector: 列表容器选择器
            title_selector: 标题选择器
            link_selector: 链接选择器
            date_selector: 日期选择器
        """
        all_items = []
        
        for page in range(page_range[0], page_range[1] + 1):
            page_url = url.format(page=page)
            print(f"\n正在爬取第 {page} 页: {page_url}")
            
            response = self.get(page_url)
            if not response:
                continue
            
            parser = HTMLParser(response.text)
            items = parser.find_all(list_selector, limit=50)
            
            if not items:
                print(f"第 {page} 页没有找到数据")
                break
            
            for item in items:
                title_elem = item.select_one(title_selector)
                link_elem = item.select_one(link_selector)
                date_elem = item.select_one(date_selector)
                
                if title_elem:
                    all_items.append({
                        'title': title_elem.get_text(strip=True),
                        'url': link_elem.get('href') if link_elem else '',
                        'date': date_elem.get_text(strip=True) if date_elem else '',
                        'page': page
                    })
            
            print(f"第 {page} 页: 获取 {len(items)} 条")
            self.random_sleep(1, 2)
        
        self.items = all_items
        return all_items
    
    def crawl_detail(self, url: str) -> Dict:
        """爬取文章详情"""
        response = self.get(url)
        if not response:
            return {}
        
        parser = HTMLParser(response.text)
        
        # 提取正文（根据目标网站结构调整）
        content_elem = parser.find_one('div.content') or parser.find_one('article')
        
        if content_elem:
            # 清理脚本和样式
            for tag in content_elem.find_all(['script', 'style', 'iframe']):
                tag.decompose()
            
            return {
                'url': url,
                'content': content_elem.get_text(separator='\n', strip=True),
                'images': parser.extract_images()
            }
        
        return {}
    
    def save_to_json(self, filepath: str):
        """保存结果到JSON"""
        import json
        with open(filepath, 'w', encoding='utf-8') as f:
            json.dump(self.items, f, ensure_ascii=False, indent=2)
        print(f"✓ 已保存 {len(self.items)} 条数据到 {filepath}")

反爬应对策略

class AntiCrawlerMixin:
    """反爬应对Mixin"""
    
    def rotate_user_agent(self):
        """随机User-Agent"""
        user_agents = [
            'Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/120.0.0.0',
            'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) Safari/605.1.15',
            'Mozilla/5.0 (X11; Linux x86_64) Firefox/121.0',
            'Mozilla/5.0 (iPhone; CPU iPhone OS 17_2 like Mac OS X) Mobile/15E148',
        ]
        self.headers['User-Agent'] = random.choice(user_agents)
    
    def use_proxy(self, proxy_pool: List[str]) -> Dict:
        """使用代理IP"""
        if not proxy_pool:
            return {}
        
        proxy = random.choice(proxy_pool)
        return {
            'http': f'http://{proxy}',
            'https': f'http://{proxy}'
        }
    
    def handle_captcha(self, response):
        """检测验证码"""
        if 'captcha' in response.text.lower() or response.status_code == 403:
            print("⚠ 检测到验证码或被封禁")
            return True
        return False

完整使用示例

def main():
    # 爬取图片
    image_crawler = ImageCrawler('./downloaded_images')
    image_crawler.crawl_page(
        url='https://example.com/gallery',
        selectors=['img.photo', 'a.download'],
        min_size=10
    )
    
    # 爬取文章列表
    article_crawler = ArticleCrawler()
    articles = article_crawler.crawl_list(
        url='https://example.com/articles?page={page}',
        page_range=(1, 5)
    )
    
    # 保存结果
    article_crawler.save_to_json('articles.json')
    
    # 爬取文章详情
    for article in articles[:3]:  # 只取前3篇
        if article['url']:
            detail = article_crawler.crawl_detail(article['url'])
            print(f"标题: {article['title']}")
            print(f"内容长度: {len(detail.get('content', ''))} 字符")
            article_crawler.random_sleep(1, 2)

if __name__ == '__main__':
    main()

爬虫伦理规范

最后提醒几点：

1. 遵守robots.txt：查看网站的爬虫协议
2. 控制请求频率：不要对服务器造成压力
3. 设置User-Agent：让网站知道你是爬虫
4. 不要爬取隐私数据：合规是底线
5. 优先使用官方API：很多网站提供合法接口

这套爬虫框架覆盖了常见的采集场景，核心思路是分层封装——请求层、解析层、存储层分离，便于维护和扩展🙃