Python爬虫常用工具详解及选型指南

import requests
from typing import Dict, Optional
 
class WebScraper:
    def __init__(self, timeout: int = 30):
        self.session = requests.Session()
        self.session.timeout = timeout
        # 设置常用请求头
        self.session.headers.update({
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
            'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
            'Accept-Language': 'en-US,en;q=0.5',
            'Accept-Encoding': 'gzip, deflate',
            'Connection': 'keep-alive',
        })
    
    def fetch_page(self, url: str, params: Optional[Dict] = None) -> Optional[str]:
        """获取网页内容"""
        try:
            response = self.session.get(url, params=params)
            response.raise_for_status()
            return response.text
        except requests.RequestException as e:
            print(f"请求失败: {e}")
            return None
 
# 使用示例
scraper = WebScraper()
content = scraper.fetch_page('https://example.com')

import asyncio
import httpx
from typing import List, Dict
 
async def fetch_multiple_urls(urls: List[str]) -> List[str]:
    """并发获取多个URL内容"""
    async with httpx.AsyncClient(
        timeout=30.0,
        limits=httpx.Limits(max_keepalive_connections=20, max_connections=100)
    ) as client:
        tasks = [client.get(url) for url in urls]
        responses = await asyncio.gather(*tasks, return_exceptions=True)
        
        results = []
        for response in responses:
            if isinstance(response, httpx.Response) and response.status_code == 200:
                results.append(response.text)
            else:
                results.append("")
        return results
 
# 异步批量抓取
urls = ['https://example1.com', 'https://example2.com', 'https://example3.com']
results = asyncio.run(fetch_multiple_urls(urls))

from bs4 import BeautifulSoup
import requests
 
def extract_article_info(url: str) -> dict:
    """提取文章信息"""
    response = requests.get(url)
    soup = BeautifulSoup(response.text, 'html.parser')
    
    # 使用多种选择器策略
    title = soup.find('h1', class_='article-title')
    if not title:
        title = soup.select_one('h1[class*="title"]')
    
    content = soup.find('div', class_='article-content')
    if not content:
        content = soup.find('article')
    
    # 提取图片链接
    images = [img.get('src') for img in content.find_all('img') if img.get('src')]
    
    return {
        'title': title.get_text(strip=True) if title else '',
        'content': content.get_text(strip=True) if content else '',
        'images': images,
        'word_count': len(content.get_text(strip=True)) if content else 0
    }

from lxml import html, etree
import requests
 
def fast_xpath_extraction(url: str) -> list:
    """使用XPath快速提取数据"""
    response = requests.get(url)
    tree = html.fromstring(response.content)
    
    # XPath表达式比CSS选择器更强大
    titles = tree.xpath('//div[@class="article"]//h2/text()')
    links = tree.xpath('//div[@class="article"]//a[@href]/@href')
    
    # 支持复杂的条件筛选
    hot_articles = tree.xpath(
        '//article[contains(@class, "hot") and .//span[@class="views"]/text() > 1000']/h3/text()'
    )
    
    return {
        'titles': titles,
        'links': links,
        'hot_articles': hot_articles
    }

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
from selenium.webdriver.chrome.options import Options
import time
 
class DynamicScraper:
    def __init__(self, headless: bool = True):
        self.options = Options()
        if headless:
            self.options.add_argument('--headless')
        self.options.add_argument('--no-sandbox')
        self.options.add_argument('--disable-dev-shm-usage')
        self.options.add_argument('--disable-gpu')
        self.options.add_argument('--window-size=1920,1080')
        
        # 反检测设置
        self.options.add_experimental_option("excludeSwitches", ["enable-automation"])
        self.options.add_experimental_option('useAutomationExtension', False)
        self.options.add_argument("--disable-blink-features=AutomationControlled")
        
        self.driver = webdriver.Chrome(options=self.options)
        self.driver.execute_script("Object.defineProperty(navigator, 'webdriver', {get: () => undefined})")
    
    def scrape_infinite_scroll(self, url: str, scroll_pause: float = 2.0) -> list:
        """处理无限滚动页面"""
        self.driver.get(url)
        
        last_height = self.driver.execute_script("return document.body.scrollHeight")
        items = []
        
        while True:
            # 等待新内容加载
            time.sleep(scroll_pause)
            
            # 提取当前可见的项目
            current_items = self.driver.find_elements(By.CSS_SELECTOR, '.item-class')
            for item in current_items:
                if item not in items:
                    items.append({
                        'text': item.text,
                        'link': item.get_attribute('href')
                    })
            
            # 滚动到页面底部
            self.driver.execute_script("window.scrollTo(0, document.body.scrollHeight);")
            
            # 检查是否到达页面底部
            new_height = self.driver.execute_script("return document.body.scrollHeight")
            if new_height == last_height:
                break
            last_height = new_height
        
        return items
    
    def __del__(self):
        if hasattr(self, 'driver'):
            self.driver.quit()

from playwright.async_api import async_playwright
import asyncio
 
async def playwright_scraper(urls: list) -> list:
    """使用Playwright进行并发抓取"""
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=True)
        context = await browser.new_context(
            viewport={'width': 1920, 'height': 1080},
            user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
        )
        
        async def scrape_page(url: str) -> dict:
            page = await context.new_page()
            try:
                await page.goto(url, wait_until='networkidle')
                
                # 等待特定元素加载
                await page.wait_for_selector('.content', timeout=10000)
                
                # 执行JavaScript获取数据
                data = await page.evaluate('''() => {
                    const title = document.querySelector('h1')?.textContent || '';
                    const content = document.querySelector('.content')?.textContent || '';
                    const images = Array.from(document.querySelectorAll('img')).map(img => img.src);
                    return { title, content, images };
                }''')
                
                return {'url': url, 'data': data, 'status': 'success'}
            except Exception as e:
                return {'url': url, 'error': str(e), 'status': 'failed'}
            finally:
                await page.close()
        
        # 并发处理多个URL
        results = await asyncio.gather(*[scrape_page(url) for url in urls])
        await browser.close()
        return results
 
# 使用示例
urls = ['https://example1.com', 'https://example2.com']
results = asyncio.run(playwright_scraper(urls))

# items.py
import scrapy
 
class ProductItem(scrapy.Item):
    name = scrapy.Field()
    price = scrapy.Field()
    description = scrapy.Field()
    images = scrapy.Field()
    category = scrapy.Field()
    availability = scrapy.Field()
    url = scrapy.Field()
 
# spiders/product_spider.py
import scrapy
from ..items import ProductItem
from scrapy.loader import ItemLoader
from scrapy.loader.processors import TakeFirst, MapCompose, Join
 
class ProductSpider(scrapy.Spider):
    name = 'products'
    allowed_domains = ['example-shop.com']
    start_urls = ['https://example-shop.com/products']
    
    custom_settings = {
        'CONCURRENT_REQUESTS': 16,
        'DOWNLOAD_DELAY': 1,
        'RANDOMIZE_DOWNLOAD_DELAY': True,
        'AUTOTHROTTLE_ENABLED': True,
        'AUTOTHROTTLE_TARGET_CONCURRENCY': 2.0,
        'USER_AGENT': 'ProductBot/1.0',
        'FEEDS': {
            'products.json': {
                'format': 'json',
                'encoding': 'utf8',
                'store_empty': False,
                'fields': ['name', 'price', 'description', 'category', 'url'],
            },
        },
    }
    
    def parse(self, response):
        """解析产品列表页"""
        products = response.css('.product-item')
        
        for product in products:
            loader = ItemLoader(item=ProductItem(), selector=product)
            
            loader.add_css('name', '.product-title::text')
            loader.add_css('price', '.price::text', MapCompose(self.clean_price))
            loader.add_css('description', '.description::text')
            loader.add_css('category', '.category::text')
            loader.add_css('images', 'img::attr(src)')
            
            product_url = product.css('a::attr(href)').get()
            if product_url:
                yield response.follow(
                    product_url, 
                    callback=self.parse_product_detail,
                    meta={'item': loader.load_item()}
                )
        
        # 处理分页
        next_page = response.css('.next-page::attr(href)').get()
        if next_page:
            yield response.follow(next_page, callback=self.parse)
    
    def parse_product_detail(self, response):
        """解析产品详情页"""
        item = response.meta['item']
        
        # 补充详细信息
        item['availability'] = response.css('.availability::text').get()
        item['url'] = response.url
        
        yield item
    
    def clean_price(self, price):
        """清洗价格数据"""
        return float(price.replace('$', '').replace(',', ''))
 
# middlewares.py
import random
import base64
 
class ProxyMiddleware:
    """代理中间件"""
    
    def __init__(self):
        self.proxies = [
            'http://proxy1.com:8080',
            'http://proxy2.com:8080',
            'http://proxy3.com:8080',
        ]
    
    def process_request(self, request, spider):
        proxy = random.choice(self.proxies)
        request.meta['proxy'] = proxy
        return None
 
# pipelines.py
import pymongo
from itemadapter import ItemAdapter
 
class MongoPipeline:
    """MongoDB数据管道"""
    
    def __init__(self, mongo_uri, mongo_db):
        self.mongo_uri = mongo_uri
        self.mongo_db = mongo_db
    
    @classmethod
    def from_crawler(cls, crawler):
        return cls(
            mongo_uri=crawler.settings.get('MONGO_URI'),
            mongo_db=crawler.settings.get('MONGO_DATABASE', 'products')
        )
    
    def open_spider(self, spider):
        self.client = pymongo.MongoClient(self.mongo_uri)
        self.db = self.client[self.mongo_db]
    
    def close_spider(self, spider):
        self.client.close()
    
    def process_item(self, item, spider):
        adapter = ItemAdapter(item)
        self.db[spider.name].insert_one(adapter.asdict())
        return item

# TRAE IDE自动生成的爬虫模板
import asyncio
import aiohttp
from typing import List, Dict, Optional
from dataclasses import dataclass
import logging
 
@dataclass
class Product:
    name: str
    price: float
    url: str
    availability: bool
 
class EcommerceScraper:
    """TRAE IDE智能生成的电商爬虫类"""
    
    def __init__(self, max_concurrent: int = 10, retry_attempts: int = 3):
        self.max_concurrent = max_concurrent
        self.retry_attempts = retry_attempts
        self.session = None
        self.logger = logging.getLogger(__name__)
    
    async def __aenter__(self):
        connector = aiohttp.TCPConnector(limit=self.max_concurrent)
        timeout = aiohttp.ClientTimeout(total=30)
        self.session = aiohttp.ClientSession(
            connector=connector,
            timeout=timeout,
            headers={'User-Agent': 'Mozilla/5.0 (compatible; ProductBot/1.0)'}
        )
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if self.session:
            await self.session.close()

# 开发者输入的代码
products = response.xpath('//div[@class="product"]')
 
# TRAE IDE AI助手的实时建议
# 建议1: 使用更具体的XPath提高准确性
products = response.xpath('//div[contains(@class, "product") and @data-product-id]')
 
# 建议2: 添加错误处理
products = response.xpath('//div[@class="product"]')
if not products:
    self.logger.warning("未找到产品数据，可能是页面结构变化")
    return []

# TRAE IDE生成的性能监控代码
import time
import psutil
import asyncio
from functools import wraps
 
def monitor_performance(func):
    """性能监控装饰器"""
    @wraps(func)
    async def wrapper(*args, **kwargs):
        start_time = time.time()
        start_memory = psutil.Process().memory_info().rss / 1024 / 1024  # MB
        
        result = await func(*args, **kwargs)
        
        end_time = time.time()
        end_memory = psutil.Process().memory_info().rss / 1024 / 1024  # MB
        
        print(f"函数 {func.__name__} 执行时间: {end_time - start_time:.2f}s")
        print(f"内存使用: {start_memory:.2f}MB -> {end_memory:.2f}MB")
        
        return result
    return wrapper

# 项目结构
price_monitor/
├── config/
│   ├── settings.py          # 配置文件
│   └── proxies.py           # 代理配置
├── spiders/
│   ├── base_spider.py       # 基础爬虫类
│   ├── taobao_spider.py     # 淘宝爬虫
│   └── jd_spider.py         # 京东爬虫
├── parsers/
│   ├── price_parser.py      # 价格解析器
│   └── product_parser.py    # 产品信息解析器
├── storage/
│   ├── database.py          # 数据库操作
│   └── cache.py             # 缓存管理
├── utils/
│   ├── proxy_pool.py        # 代理池管理
│   ├── user_agent.py        # User-Agent轮换
│   └── retry.py             # 重试机制
└── main.py                  # 主程序

# 异步新闻爬虫示例
import asyncio
import aiohttp
from datetime import datetime, timedelta
from typing import List, Dict
import feedparser
 
class NewsAggregator:
    """新闻聚合爬虫"""
    
    def __init__(self):
        self.rss_sources = [
            'https://rss.cnn.com/rss/edition.rss',
            'https://feeds.bbci.co.uk/news/rss.xml',
            'https://www.reutersagency.com/feed/?best-topics=top-news',
        ]
        self.session = None
    
    async def fetch_rss_feed(self, url: str) -> List[Dict]:
        """获取RSS源数据"""
        try:
            async with self.session.get(url, timeout=10) as response:
                content = await response.text()
                feed = feedparser.parse(content)
                
                articles = []
                for entry in feed.entries[:10]:  # 限制文章数量
                    articles.append({
                        'title': entry.get('title', ''),
                        'link': entry.get('link', ''),
                        'published': self.parse_date(entry.get('published', '')),
                        'summary': entry.get('summary', ''),
                        'source': feed.feed.get('title', ''),
                    })
                return articles
        except Exception as e:
            print(f"RSS源 {url} 获取失败: {e}")
            return []
    
    async def aggregate_news(self) -> List[Dict]:
        """聚合所有新闻源"""
        connector = aiohttp.TCPConnector(limit=20)
        timeout = aiohttp.ClientTimeout(total=30)
        
        async with aiohttp.ClientSession(
            connector=connector,
            timeout=timeout
        ) as self.session:
            tasks = [self.fetch_rss_feed(url) for url in self.rss_sources]
            results = await asyncio.gather(*tasks)
            
            # 合并结果并去重
            all_articles = []
            seen_links = set()
            
            for articles in results:
                for article in articles:
                    if article['link'] not in seen_links:
                        all_articles.append(article)
                        seen_links.add(article['link'])
            
            # 按发布时间排序
            all_articles.sort(key=lambda x: x['published'], reverse=True)
            return all_articles

# 综合反爬策略实现
import random
import time
from fake_useragent import UserAgent
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
 
class AntiCrawlScraper:
    """反爬虫策略实现"""
    
    def __init__(self):
        self.ua = UserAgent()
        self.session = self.create_session()
    
    def create_session(self) -> requests.Session:
        """创建配置了反爬策略的会话"""
        session = requests.Session()
        
        # 重试策略
        retry_strategy = Retry(
            total=3,
            backoff_factor=1,
            status_forcelist=[429, 500, 502, 503, 504],
        )
        
        adapter = HTTPAdapter(max_retries=retry_strategy, pool_connections=10, pool_maxsize=20)
        session.mount("http://", adapter)
        session.mount("https://", adapter)
        
        return session
    
    def get_random_headers(self) -> dict:
        """生成随机请求头"""
        return {
            'User-Agent': self.ua.random,
            'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
            'Accept-Language': random.choice(['en-US,en;q=0.5', 'zh-CN,zh;q=0.9,en;q=0.8']),
            'Accept-Encoding': 'gzip, deflate, br',
            'DNT': '1',
            'Connection': 'keep-alive',
            'Upgrade-Insecure-Requests': '1',
            'Cache-Control': random.choice(['max-age=0', 'no-cache']),
        }
    
    def smart_delay(self, base_delay: float = 1.0):
        """智能延迟 - 模拟人类行为"""
        # 随机延迟，避免固定模式
        delay = base_delay + random.uniform(0.5, 2.0)
        time.sleep(delay)
    
    def fetch_with_antibot(self, url: str, **kwargs) -> requests.Response:
        """带反爬策略的请求"""
        headers = self.get_random_headers()
        headers.update(kwargs.get('headers', {}))
        
        # 随机延迟
        self.smart_delay()
        
        response = self.session.get(url, headers=headers, **kwargs)
        
        # 处理JavaScript挑战
        if 'javascript' in response.text.lower() and 'challenge' in response.text.lower():
            print(f"检测到JavaScript挑战: {url}")
            # 这里可以集成Selenium或Playwright处理
        
        return response

# 内存优化与连接池管理
import aiohttp
import asyncio
from typing import AsyncGenerator
import weakref
 
class OptimizedScraper:
    """性能优化的爬虫实现"""
    
    def __init__(self, max_connections: int = 100):
        self.connector = aiohttp.TCPConnector(
            limit=max_connections,
            limit_per_host=10,
            ttl_dns_cache=300,
            use_dns_cache=True,
        )
        
        self.timeout = aiohttp.ClientTimeout(
            total=30,
            connect=10,
            sock_read=10
        )
        
        # 使用弱引用避免内存泄漏
        self._sessions = weakref.WeakSet()
    
    async def create_session(self) -> aiohttp.ClientSession:
        """创建优化配置的会话"""
        session = aiohttp.ClientSession(
            connector=self.connector,
            timeout=self.timeout,
            headers={'Connection': 'keep-alive'},
            trust_env=True,
        )
        self._sessions.add(session)
        return session
    
    async def stream_download(self, url: str, chunk_size: int = 8192) -> AsyncGenerator[bytes, None]:
        """流式下载大文件"""
        async with await self.create_session() as session:
            async with session.get(url) as response:
                response.raise_for_status()
                
                async for chunk in response.content.iter_chunked(chunk_size):
                    yield chunk
    
    async def batch_process(self, urls: list, batch_size: int = 10) -> list:
        """批量处理URL，控制并发数"""
        results = []
        
        for i in range(0, len(urls), batch_size):
            batch = urls[i:i + batch_size]
            batch_results = await self.process_batch(batch)
            results.extend(batch_results)
            
            # 批次间延迟，避免过载
            await asyncio.sleep(0.5)
        
        return results
    
    async def process_batch(self, urls: list) -> list:
        """处理单个批次"""
        async with await self.create_session() as session:
            tasks = [self.fetch_url(session, url) for url in urls]
            return await asyncio.gather(*tasks, return_exceptions=True)
    
    async def fetch_url(self, session: aiohttp.ClientSession, url: str) -> dict:
        """获取单个URL"""
        try:
            async with session.get(url) as response:
                content = await response.text()
                return {
                    'url': url,
                    'status': response.status,
                    'size': len(content),
                    'content': content[:1000]  # 限制内存使用
                }
        except Exception as e:
            return {'url': url, 'error': str(e)}