别再手写爬虫了！Crawlee 才是现代网页抓取与自动化任务的最优解

# 使用 pip 安装 Crawlee 核心库
# pip install crawlee

# 如果需要处理 JavaScript 渲染的页面，还需要安装浏览器相关依赖
# pip install crawlee[playwright]

# 或者选择基于 Puppeteer 的版本
# pip install crawlee[chrome]

# 配置 pip 镜像（如果需要）
# pip config set global.index-url https://pypi.tuna.tsinghua.edu.cn/simple

import crawlee

# 检查版本信息
print(f"Crawlee 版本: {crawlee.__version__}")

# 检查可选依赖是否正常加载
try:
    from crawlee.browsers import PlaywrightBrowserPool
    print("Playwright 浏览器支持: 已安装")
except ImportError:
    print("Playwright 浏览器支持: 未安装")

try:
    from crawlee.http_clients import HttpxHttpClient
    print("HTTP 客户端: 已安装")
except ImportError:
    print("HTTP 客户端: 未安装")

# project_structure.py

# 这是一个典型的 Crawlee 项目目录结构示例

project_structure = """
my_crawler_project/
├── src/
│   ├── __init__.py
│   ├── main.py              # 爬虫入口文件
│   ├── spider.py            # 爬虫主体逻辑
│   ├── pipelines.py         # 数据处理管道
│   └── models.py            # 数据模型定义
├── tests/
│   ├── __init__.py
│   ├── test_spider.py       # 单元测试
│   └── fixtures/            # 测试数据
├── data/
│   ├── input/               # 输入数据（如种子 URL 列表）
│   └── output/              # 爬取结果输出目录
├── logs/                    # 日志文件目录
├── requirements.txt        # 项目依赖
├── .env                     # 环境变量配置
└── README.md                # 项目说明文档
"""
print(project_structure)

# =====================================
# 以下是最基础的爬虫示例代码
# =====================================

basic_example = '''
from crawlee.beautifulsoup import BeautifulSoupCrawler, BeautifulSoupCrawlerRun

class MyFirstCrawler(BeautifulSoupCrawler):
    """一个最简单的 Crawlee 爬虫示例"""

    async def request_handler(self, context: BeautifulSoupCrawlerRun) -> None:
        """处理每个抓取到的页面"""
        # 获取当前页面的 URL
        url = context.request.url
        print(f"正在处理: {url}")

        # 获取页面标题
        title = context.page.title
        print(f"页面标题: {title}")

        # 提取一些文本内容作为示例
        paragraphs = context.page.find_all("p")
        for i, p in enumerate(paragraphs[:3]):
            print(f"段落 {i+1}: {p.get_text()[:100]}...")

        # 将结果添加到输出数据
        await context.push_data({
            "url": url,
            "title": title,
            "paragraphs": [p.get_text() for p in paragraphs[:5]]
        })

async def main():
    """爬虫入口函数"""
    # 创建爬虫实例
    crawler = MyFirstCrawler(
        max_request_retries=3,      # 最大重试次数
        max_concurrency=2,          # 最大并发数
    )

    # 添加待抓取的 URL
    await crawler.add_urls([
        "https://example.com",
        "https://example.org",
    ])

    # 启动爬虫
    await crawler.run()

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())
'''
print(basic_example)

# config_example.py

# =====================================
# Crawlee 常用配置项示例
# =====================================

# 基础配置
BASE_CONFIG = """
# 日志配置
LOG_LEVEL=INFO                    # 日志级别：DEBUG, INFO, WARNING, ERROR
LOG_FILE=./logs/crawler.log       # 日志文件路径

# 请求配置
MAX_REQUEST_RETRIES=3             # 请求失败后的最大重试次数
REQUEST_TIMEOUT_SECONDS=30        # 单个请求的超时时间（秒）
MAX_CONCURRENCY=5                 # 最大并发请求数

# 代理配置
# PROXY_URLS=                        # 代理服务器地址列表，每行一个
#   http://user:pass@proxy1.com:8080
#   http://user:pass@proxy2.com:8080

# 浏览器配置（如果使用无头模式）
HEADLESS=True                     # 是否使用无头浏览器模式
BROWSER_TIMEOUT=60000             # 浏览器操作超时时间（毫秒）
"""

# 使用 Python 的 dataclasses 定义配置类
from dataclasses import dataclass, field
from typing import List, Optional
import os

@dataclass
class CrawlerConfig:
    """爬虫配置类"""

    # 请求相关配置
    max_retries: int = 3
    request_timeout: int = 30
    max_concurrency: int = 5

    # 代理配置
    proxy_urls: List[str] = field(default_factory=list)
    rotate_proxies: bool = True

    # 浏览器配置
    headless: bool = True
    browser_timeout: int = 60000
    user_agent: Optional[str] = None

    # 存储配置
    output_directory: str = "./output"
    persist_storage: bool = True

    @classmethod
    def from_env(cls) -> "CrawlerConfig":
        """从环境变量加载配置"""
        return cls(
            max_retries=int(os.getenv("MAX_REQUEST_RETRIES", "3")),
            request_timeout=int(os.getenv("REQUEST_TIMEOUT_SECONDS", "30")),
            max_concurrency=int(os.getenv("MAX_CONCURRENCY", "5")),
            headless=os.getenv("HEADLESS", "true").lower() == "true",
        )

    def validate(self) -> bool:
        """验证配置的有效性"""
        if self.max_retries < 0:
            print("错误: max_retries 必须大于等于 0")
            return False
        if self.request_timeout <= 0:
            print("错误: request_timeout 必须大于 0")
            return False
        if self.max_concurrency <= 0:
            print("错误: max_concurrency 必须大于 0")
            return False
        return True

# =====================================
# 配置使用示例
# =====================================

def demo_config_usage():
    """演示配置的使用方式"""

    # 方式一：直接实例化
    config1 = CrawlerConfig(
        max_retries=5,
        max_concurrency=10,
    )

    # 方式二：从环境变量加载
    config2 = CrawlerConfig.from_env()

    # 验证配置
    print(f"配置1 是否有效: {config1.validate()}")
    print(f"配置2 是否有效: {config2.validate()}")

    # 显示当前配置
    print("当前爬虫配置:")
    print(f"  - 最大重试次数: {config2.max_retries}")
    print(f"  - 请求超时: {config2.request_timeout} 秒")
    print(f"  - 最大并发: {config2.max_concurrency}")
    print(f"  - 无头模式: {config2.headless}")

demo_config_usage()

# request_scheduler.py

import asyncio
from typing import List, Dict, Any, Optional
from dataclasses import dataclass, field
from datetime import datetime

# =====================================
# 请求对象详解
# =====================================

@dataclass
class Request:
    """Crawlee 中的请求对象"""

    url: str                           # 请求的 URL（必需）
    unique_key: Optional[str] = None   # 用于去重的唯一标识
    method: str = "GET"                # HTTP 方法
    headers: Dict[str, str] = field(default_factory=dict)  # 请求头
    payload: Optional[bytes] = None    # 请求体（用于 POST 等方法）
    proxy_url: Optional[str] = None   # 指定使用的代理
    retry_count: int = 0              # 当前重试次数
    error_messages: List[str] = field(default_factory=list)  # 错误信息记录

    def __post_init__(self):
        """初始化后自动设置唯一标识"""
        if self.unique_key is None:
            # 默认使用 URL 作为唯一标识
            self.unique_key = f"{self.method}:{self.url}"

    def increment_retry(self) -> None:
        """增加重试计数"""
        self.retry_count += 1

    def add_error(self, error: str) -> None:
        """记录错误信息"""
        self.error_messages.append(f"[{datetime.now().isoformat()}] {error}")

# =====================================
# 请求工厂示例
# =====================================

class RequestFactory:
    """创建各类请求的工厂类"""

    @staticmethod
    def create_basic_request(url: str) -> Request:
        """创建一个基础请求"""
        return Request(url=url)

    @staticmethod
    def create_post_request(url: str, data: Dict[str, Any]) -> Request:
        """创建一个 POST 请求"""
        import json
        return Request(
            url=url,
            method="POST",
            headers={"Content-Type": "application/json"},
            payload=json.dumps(data).encode("utf-8")
        )

    @staticmethod
    def create_form_request(url: str, form_data: Dict[str, str]) -> Request:
        """创建一个表单提交请求"""
        from urllib.parse import urlencode
        return Request(
            url=url,
            method="POST",
            headers={"Content-Type": "application/x-www-form-urlencoded"},
            payload=urlencode(form_data).encode("utf-8")
        )

# =====================================
# 请求调度演示
# =====================================

async def demo_request_scheduling():
    """演示请求调度系统的工作方式"""

    print("=== 请求调度系统演示 ===\n")

    # 创建多个请求
    requests = [
        RequestFactory.create_basic_request("https://example.com/page1"),
        RequestFactory.create_basic_request("https://example.com/page2"),
        RequestFactory.create_post_request(
            "https://example.com/api/data",
            {"query": "search term"}
        ),
    ]

    # 添加元数据
    for i, req in enumerate(requests):
        req.headers["X-Request-ID"] = f"req-{i+1}"
        req.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"
        )

    # 显示请求信息
    print("待处理的请求列表:")
    for req in requests:
        print(f"  [{req.unique_key}] {req.method} {req.url}")
        print(f"    - 请求头: {list(req.headers.keys())}")
        print(f"    - 已重试次数: {req.retry_count}")

    # 模拟错误和重试流程
    print("\n模拟请求处理流程:")
    for req in requests:
        print(f"\n处理请求: {req.unique_key}")

        # 模拟第一次请求失败
        req.add_error("连接超时 - 504 Gateway Timeout")
        req.increment_retry()
        print(f"  记录错误: 连接超时")
        print(f"  当前重试次数: {req.retry_count}")

        # 模拟重试成功
        if req.retry_count <= 2:
            print(f"  重试成功!")
        else:
            print(f"  重试次数超限，标记为失败")

        # 显示错误历史
        print(f"  错误历史: {req.error_messages}")

asyncio.run(demo_request_scheduling())

# data_pipeline.py

from typing import Dict, Any, List, Optional
from dataclasses import dataclass
from abc import ABC, abstractmethod
import json
import csv
from datetime import datetime
from pathlib import Path

# =====================================
# 数据模型定义
# =====================================

@dataclass
class ProductItem:
    """商品数据模型"""
    product_id: str
    title: str
    price: float
    currency: str = "USD"
    category: Optional[str] = None
    rating: Optional[float] = None
    review_count: Optional[int] = None
    in_stock: bool = True
    scraped_at: str = ""
    source_url: str = ""

    def __post_init__(self):
        """初始化时自动添加时间戳"""
        if not self.scraped_at:
            self.scraped_at = datetime.now().isoformat()

    def to_dict(self) -> Dict[str, Any]:
        """转换为字典格式"""
        return {
            "product_id": self.product_id,
            "title": self.title,
            "price": self.price,
            "currency": self.currency,
            "category": self.category,
            "rating": self.rating,
            "review_count": self.review_count,
            "in_stock": self.in_stock,
            "scraped_at": self.scraped_at,
            "source_url": self.source_url,
        }

    def validate(self) -> List[str]:
        """验证数据完整性，返回错误信息列表"""
        errors = []
        if not self.product_id:
            errors.append("product_id 不能为空")
        if not self.title:
            errors.append("title 不能为空")
        if self.price < 0:
            errors.append(f"price 不能为负数: {self.price}")
        if self.rating is not None and not (0 <= self.rating <= 5):
            errors.append(f"rating 必须在 0-5 之间: {self.rating}")
        if self.review_count is not None and self.review_count < 0:
            errors.append(f"review_count 不能为负数: {self.review_count}")
        return errors

# =====================================
# 管道基类
# =====================================

class BasePipeline(ABC):
    """数据处理管道的基类"""

    @abstractmethod
    async def process_item(self, item: Dict[str, Any]) -> Dict[str, Any]:
        """处理单个数据项"""
        pass

    async def open(self) -> None:
        """管道打开时的初始化操作"""
        pass

    async def close(self) -> None:
        """管道关闭时的清理操作"""
        pass

# =====================================
# 具体管道实现
# =====================================

class ValidationPipeline(BasePipeline):
    """数据验证管道"""

    def __init__(self, required_fields: List[str]):
        self.required_fields = required_fields
        self.errors: List[Dict[str, Any]] = []

    async def process_item(self, item: Dict[str, Any]) -> Dict[str, Any]:
        """验证数据项"""
        # 检查必需字段
        missing_fields = [f for f in self.required_fields if f not in item or not item[f]]
        if missing_fields:
            self.errors.append({
                "item": item,
                "error": f"缺少必需字段: {missing_fields}"
            })
            # 返回 None 表示该数据项将被跳过
            return None

        return item

    async def close(self) -> None:
        """输出验证错误摘要"""
        if self.errors:
            print(f"\n验证管道发现 {len(self.errors)} 个错误:")
            for i, error in enumerate(self.errors[:5], 1):
                print(f"  {i}. {error['error']}")
                print(f"     URL: {error['item'].get('source_url', 'N/A')}")

class DataCleaningPipeline(BasePipeline):
    """数据清洗管道"""

    async def process_item(self, item: Dict[str, Any]) -> Dict[str, Any]:
        """清洗和标准化数据"""
        cleaned = item.copy()

        # 清理文本字段
        text_fields = ["title", "category", "description"]
        for field in text_fields:
            if field in cleaned and isinstance(cleaned[field], str):
                # 去除多余空白
                cleaned[field] = " ".join(cleaned[field].split())
                # 转义 HTML 实体
                cleaned[field] = (
                    cleaned[field]
                    .replace("&amp;", "&")
                    .replace("&lt;", "<")
                    .replace("&gt;", ">")
                    .replace("&quot;", '"')
                )

        # 处理价格字段
        if "price" in cleaned:
            if isinstance(cleaned["price"], str):
                # 移除货币符号和逗号
                price_str = cleaned["price"].replace("$", "").replace(",", "").strip()
                try:
                    cleaned["price"] = float(price_str)
                except ValueError:
                    cleaned["price"] = 0.0

        # 处理评分字段
        if "rating" in cleaned and isinstance(cleaned["rating"], str):
            try:
                # 提取数字部分（如 "4.5 out of 5"）
                rating_str = cleaned["rating"].split()[0]
                cleaned["rating"] = float(rating_str)
            except (ValueError, IndexError):
                cleaned["rating"] = None

        return cleaned

class StoragePipeline(BasePipeline):
    """数据存储管道"""

    def __init__(self, output_dir: str = "./output"):
        self.output_dir = Path(output_dir)
        self.json_file = None
        self.csv_file = None
        self.csv_writer = None
        self.items: List[Dict[str, Any]] = []

    async def open(self) -> None:
        """初始化存储文件"""
        self.output_dir.mkdir(parents=True, exist_ok=True)

        # 准备 JSON 文件
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        self.json_file = self.output_dir / f"data_{timestamp}.json"

        # 准备 CSV 文件
        self.csv_file = self.output_dir / f"data_{timestamp}.csv"

        print(f"存储管道已初始化:")
        print(f"  - JSON 输出: {self.json_file}")
        print(f"  - CSV 输出: {self.csv_file}")

    async def process_item(self, item: Dict[str, Any]) -> Dict[str, Any]:
        """存储数据项"""
        self.items.append(item)

        # 实时写入 JSON（追加模式）
        with open(self.json_file, "a", encoding="utf-8") as f:
            json.dump(item, f, ensure_ascii=False, indent=2)
            f.write("\n")

        # CSV 写入在 close 时统一处理
        return item

    async def close(self) -> None:
        """完成存储操作"""
        if self.items and self.csv_file:
            # 写入 CSV
            with open(self.csv_file, "w", newline="", encoding="utf-8") as f:
                if self.items:
                    fieldnames = list(self.items[0].keys())
                    writer = csv.DictWriter(f, fieldnames=fieldnames)
                    writer.writeheader()
                    writer.writerows(self.items)

            print(f"\n存储完成:")
            print(f"  - 共存储 {len(self.items)} 条数据")
            print(f"  - JSON 文件: {self.json_file}")
            print(f"  - CSV 文件: {self.csv_file}")

# =====================================
# 管道管理器
# =====================================

class PipelineManager:
    """管理多个管道的协调器"""

    def __init__(self):
        self.pipelines: List[BasePipeline] = []

    def add_pipeline(self, pipeline: BasePipeline) -> "PipelineManager":
        """添加一个管道"""
        self.pipelines.append(pipeline)
        return self

    async def open(self) -> None:
        """初始化所有管道"""
        for pipeline in self.pipelines:
            await pipeline.open()

    async def process(self, item: Dict[str, Any]) -> Optional[Dict[str, Any]]:
        """通过所有管道处理数据"""
        current_item = item
        for pipeline in self.pipelines:
            if current_item is None:
                return None
            current_item = await pipeline.process_item(current_item)
        return current_item

    async def close(self) -> None:
        """关闭所有管道"""
        for pipeline in self.pipelines:
            await pipeline.close()

# =====================================
# 管道使用演示
# =====================================

async def demo_pipeline():
    """演示数据处理管道的使用"""

    print("=== 数据处理管道演示 ===\n")

    # 创建测试数据
    test_items = [
        {
            "product_id": "SKU001",
            "title": "  无线蓝牙耳机   高品质降噪  ",
            "price": "$129.99",
            "rating": "4.5 out of 5",
            "source_url": "https://shop.example.com/product/001"
        },
        {
            "product_id": "SKU002",
            "title": "便携式移动电源 20000mAh",
            "price": "$49.99",
            "category": "电子产品",
            "rating": "4.2 out of 5",
            "review_count": 1250,
            "source_url": "https://shop.example.com/product/002"
        },
        {
            "product_id": "",
            "title": "不完整的数据",
            "price": "invalid",
            "source_url": "https://shop.example.com/product/003"
        },
    ]

    # 创建管道管理器
    manager = PipelineManager()
    manager.add_pipeline(ValidationPipeline(required_fields=["product_id", "title"]))
    manager.add_pipeline(DataCleaningPipeline())
    manager.add_pipeline(StoragePipeline(output_dir="./demo_output"))

    # 初始化管道
    await manager.open()

    # 处理测试数据
    print("处理数据项:")
    processed_count = 0
    for item in test_items:
        print(f"\n原始数据: {item}")
        result = await manager.process(item)
        if result:
            processed_count += 1
            print(f"处理后: {result}")
        else:
            print("该数据项被跳过（验证失败）")

    # 关闭管道
    await manager.close()

    print(f"\n处理完成: 共处理 {processed_count}/{len(test_items)} 条数据")

asyncio.run(demo_pipeline())

# browser_automation.py

import asyncio
from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime

# =====================================
# 浏览器爬虫基类
# =====================================

@dataclass
class BrowserConfig:
    """浏览器配置"""
    headless: bool = True                    # 是否无头模式运行
    viewport_width: int = 1920              # 视口宽度
    viewport_height: int = 1080             # 视口高度
    user_agent: Optional[str] = None        # 自定义 User-Agent
    download_path: Optional[str] = None     # 文件下载路径
    navigation_timeout: int = 30000         # 页面导航超时（毫秒）
    java_script_enabled: bool = True        # 是否启用 JavaScript

    @property
    def viewport(self) -> Dict[str, int]:
        """获取视口配置"""
        return {
            "width": self.viewport_width,
            "height": self.viewport_height
        }

# =====================================
# 页面交互操作
# =====================================

class PageInteraction:
    """封装常用的页面交互操作"""

    @staticmethod
    async def wait_for_selector(page, selector: str, timeout: int = 10000):
        """等待元素出现"""
        await page.wait_for_selector(selector, timeout=timeout)

    @staticmethod
    async def click_and_wait(page, selector: str, wait_after_ms: int = 500):
        """点击元素并等待"""
        await page.click(selector)
        await asyncio.sleep(wait_after_ms / 1000)

    @staticmethod
    async def fill_and_submit(page, input_selector: str, value: str, 
                             submit_selector: Optional[str] = None):
        """填写表单并提交"""
        await page.fill(input_selector, value)
        if submit_selector:
            await page.click(submit_selector)
        else:
            await page.press(input_selector, "Enter")

    @staticmethod
    async def scroll_to_bottom(page, step: int = 500, delay: float = 0.3):
        """滚动到页面底部（常用于触发懒加载）"""
        viewport_height = await page.evaluate("window.innerHeight")
        current_position = 0
        total_height = await page.evaluate("document.body.scrollHeight")

        while current_position < total_height:
            await page.evaluate(f"window.scrollTo(0, {current_position})")
            await asyncio.sleep(delay)
            current_position += step
            total_height = await page.evaluate("document.body.scrollHeight")

# =====================================
# JavaScript 渲染页面爬虫示例
# =====================================

class DynamicPageCrawler:
    """处理动态渲染页面的爬虫"""

    def __init__(self, config: Optional[BrowserConfig] = None):
        self.config = config or BrowserConfig()
        self.results: List[Dict[str, Any]] = []

    async def scrape_infinite_scroll_page(self, url: str) -> List[Dict[str, Any]]:
        """抓取无限滚动加载的页面"""
        print(f"开始抓取无限滚动页面: {url}")

        # 这里演示关键步骤，真实环境中需要导入 playwright
        steps = [
            "1. 启动浏览器并打开页面",
            "2. 等待初始内容加载完成",
            "3. 记录已加载的数据项数量",
            "4. 滚动到页面底部",
            "5. 等待新内容加载",
            "6. 重复步骤 3-5 直到没有新内容",
            "7. 提取所有可见数据"
        ]

        for step in steps:
            print(f"  {step}")
            await asyncio.sleep(0.1)

        # 模拟抓取结果
        self.results = [
            {"index": i, "title": f"Item {i}", "timestamp": datetime.now().isoformat()}
            for i in range(1, 51)
        ]

        return self.results

    async def scrape_spa_page(self, url: str) -> Dict[str, Any]:
        """抓取单页应用（SPA）"""
        print(f"开始抓取 SPA 页面: {url}")

        # SPA 页面抓取的关键是等待路由变化和内容更新
        steps = [
            "1. 启动浏览器导航到初始 URL",
            "2. 等待 Angular/Vue/React 应用初始化",
            "3. 触发导航或路由变化",
            "4. 等待路由变化完成和内容渲染",
            "5. 提取页面数据",
            "6. 处理可能的异步请求"
        ]

        for step in steps:
            print(f"  {step}")
            await asyncio.sleep(0.1)

        return {
            "url": url,
            "title": "SPA Page Title",
            "content": "Extracted content...",
            "scraped_at": datetime.now().isoformat()
        }

    async def scrape_login_required_page(self, url: str, 
                                         credentials: Dict[str, str]) -> Dict[str, Any]:
        """抓取需要登录的页面"""
        print(f"开始处理需要登录的页面: {url}")

        steps = [
            "1. 导航到登录页面",
            "2. 填写用户名和密码",
            "3. 提交登录表单",
            "4. 等待登录成功和重定向",
            "5. 导航到目标页面",
            "6. 等待内容加载",
            "7. 提取所需数据"
        ]

        for step in steps:
            print(f"  {step}")
            await asyncio.sleep(0.1)

        return {
            "url": url,
            "login_successful": True,
            "data": "Protected content...",
            "scraped_at": datetime.now().isoformat()
        }

# =====================================
# 高级浏览器配置示例
# =====================================

class AdvancedBrowserSetup:
    """高级浏览器配置技巧"""

    @staticmethod
    def get_stealth_config() -> Dict[str, Any]:
        """获取反检测配置"""
        return {
            "headless": True,
            "args": [
                "--disable-blink-features=AutomationControlled",
                "--disable-dev-shm-usage",
                "--no-sandbox",
                "--disable-setuid-sandbox",
                "--disable-infobars",
                "--window-position=0,0",
                "--ignore-certificate-errors",
                "--enable-features=NetworkService,NetworkServiceInProcess",
            ],
            "ignore_https_errors": True,
        }

    @staticmethod
    def get_mobile_config() -> Dict[str, Any]:
        """获取移动设备配置"""
        return {
            "viewport": {"width": 375, "height": 812},  # iPhone X 尺寸
            "user_agent": (
                "Mozilla/5.0 (iPhone; CPU iPhone OS 14_0 like Mac OS X) "
                "AppleWebKit/605.1.15 (KHTML, like Gecko) "
                "Version/14.0 Mobile/15E148 Safari/604.1"
            ),
            "hasTouch": True,
            "isMobile": True,
        }

    @staticmethod
    def get_custom_headers() -> Dict[str, str]:
        """获取自定义请求头"""
        return {
            "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
            "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8",
            "Accept-Encoding": "gzip, deflate, br",
            "DNT": "1",
            "Connection": "keep-alive",
            "Upgrade-Insecure-Requests": "1",
        }

# =====================================
# 演示入口
# =====================================

async def demo_browser_automation():
    """浏览器自动化功能演示"""

    print("=== 浏览器自动化演示 ===\n")

    # 创建爬虫实例
    crawler = DynamicPageCrawler()

    # 演示无限滚动页面抓取
    print("\n--- 无限滚动页面抓取 ---")
    results = await crawler.scrape_infinite_scroll_page("https://example.com/feed")
    print(f"抓取结果: 共 {len(results)} 条数据")

    # 演示 SPA 页面抓取
    print("\n--- SPA 页面抓取 ---")
    spa_result = await crawler.scrape_spa_page("https://example.com/app")
    print(f"页面标题: {spa_result['title']}")

    # 演示登录页面抓取
    print("\n--- 登录页面抓取 ---")
    protected_result = await crawler.scrape_login_required_page(
        "https://example.com/dashboard",
        credentials={"username": "user@example.com", "password": "password123"}
    )
    print(f"登录状态: {'成功' if protected_result['login_successful'] else '失败'}")

    # 显示配置选项
    print("\n--- 浏览器配置选项 ---")
    print("反检测配置:", AdvancedBrowserSetup.get_stealth_config()["args"][:3])
    print("移动设备配置:", AdvancedBrowserSetup.get_mobile_config()["viewport"])
    print("自定义请求头:", list(AdvancedBrowserSetup.get_custom_headers().keys()))

asyncio.run(demo_browser_automation())

# proxy_management.py

import asyncio
from typing import List, Dict, Any, Optional, Callable
from dataclasses import dataclass, field
from enum import Enum
from datetime import datetime
import random

# =====================================
# 代理相关数据结构
# =====================================

class ProxyStatus(Enum):
    """代理状态枚举"""
    ACTIVE = "active"           # 可用
    TESTING = "testing"         # 测试中
    FAILED = "failed"           # 失败
    DISABLED = "disabled"       # 已禁用

@dataclass
class ProxyInfo:
    """代理信息"""
    url: str                            # 代理地址
    username: Optional[str] = None     # 认证用户名
    password: Optional[str] = None     # 认证密码
    status: ProxyStatus = ProxyStatus.ACTIVE
    success_count: int = 0              # 成功次数
    failure_count: int = 0              # 失败次数
    last_used: Optional[datetime] = None
    last_error: Optional[str] = None
    avg_response_time: float = 0.0      # 平均响应时间（毫秒）

    @property
    def is_healthy(self) -> bool:
        """判断代理是否健康"""
        return (
            self.status == ProxyStatus.ACTIVE 
            and self.failure_count < 5
        )

    @property
    def success_rate(self) -> float:
        """计算成功率"""
        total = self.success_count + self.failure_count
        if total == 0:
            return 1.0
        return self.success_count / total

# =====================================
# 代理池管理器
# =====================================

class ProxyPool:
    """代理池管理器"""

    def __init__(self):
        self.proxies: Dict[str, ProxyInfo] = {}
        self.current_index: int = 0

    def add_proxy(self, proxy_url: str, username: Optional[str] = None,
                  password: Optional[str] = None) -> None:
        """添加代理到池中"""
        self.proxies[proxy_url] = ProxyInfo(
            url=proxy_url,
            username=username,
            password=password
        )
        print(f"已添加代理: {proxy_url}")

    def add_proxies_from_config(self, config: List[Dict[str, str]]) -> None:
        """从配置批量添加代理"""
        for item in config:
            self.add_proxy(
                proxy_url=item["url"],
                username=item.get("username"),
                password=item.get("password")
            )

    def get_next_proxy(self) -> Optional[ProxyInfo]:
        """获取下一个可用的代理（轮询策略）"""
        available = [p for p in self.proxies.values() if p.is_healthy]
        if not available:
            print("警告: 没有可用的代理")
            return None

        # 轮询选择
        proxy = available[self.current_index % len(available)]
        self.current_index = (self.current_index + 1) % len(available)
        proxy.last_used = datetime.now()

        return proxy

    def get_random_proxy(self) -> Optional[ProxyInfo]:
        """随机获取一个可用的代理"""
        available = [p for p in self.proxies.values() if p.is_healthy]
        if not available:
            return None
        return random.choice(available)

    def record_success(self, proxy_url: str, response_time: float) -> None:
        """记录代理使用成功"""
        if proxy_url in self.proxies:
            proxy = self.proxies[proxy_url]
            proxy.success_count += 1
            # 增量更新平均响应时间
            proxy.avg_response_time = (
                (proxy.avg_response_time * proxy.success_count + response_time) 
                / (proxy.success_count + 1)
            )
            proxy.last_error = None
            print(f"代理成功 [{proxy_url}]: 响应时间 {response_time:.0f}ms")

    def record_failure(self, proxy_url: str, error: str) -> None:
        """记录代理使用失败"""
        if proxy_url in self.proxies:
            proxy = self.proxies[proxy_url]
            proxy.failure_count += 1
            proxy.last_error = error

            # 连续失败次数过多则标记为不可用
            if proxy.failure_count >= 5:
                proxy.status = ProxyStatus.FAILED
                print(f"代理已禁用 [{proxy_url}]: 失败次数过多")
            else:
                print(f"代理失败 [{proxy_url}]: {error}")

    def get_stats(self) -> Dict[str, Any]:
        """获取代理池统计信息"""
        total = len(self.proxies)
        healthy = sum(1 for p in self.proxies.values() if p.is_healthy)
        total_success = sum(p.success_count for p in self.proxies.values())
        total_failure = sum(p.failure_count for p in self.proxies.values())

        return {
            "total_proxies": total,
            "healthy_proxies": healthy,
            "total_requests": total_success + total_failure,
            "overall_success_rate": (
                total_success / (total_success + total_failure) 
                if (total_success + total_failure) > 0 else 0
            ),
            "avg_response_time": (
                sum(p.avg_response_time for p in self.proxies.values()) / total
                if total > 0 else 0
            )
        }

    def display_status(self) -> None:
        """显示所有代理状态"""
        print("\n代理池状态:")
        print("-" * 80)
        print(f"{'代理地址':<40} {'状态':<10} {'成功':<8} {'失败':<8} {'成功率':<10}")
        print("-" * 80)

        for proxy in self.proxies.values():
            print(
                f"{proxy.url:<40} "
                f"{proxy.status.value:<10} "
                f"{proxy.success_count:<8} "
                f"{proxy.failure_count:<8} "
                f"{proxy.success_rate:.1%}"
            )

        stats = self.get_stats()
        print("-" * 80)
        print(f"总计: {stats['healthy_proxies']}/{stats['total_proxies']} 代理可用")
        print(f"总体成功率: {stats['overall_success_rate']:.1%}")
        print(f"平均响应时间: {stats['avg_response_time']:.0f}ms")

# =====================================
# 代理使用示例
# =====================================

async def demo_proxy_management():
    """代理管理功能演示"""

    print("=== 代理管理演示 ===\n")

    # 创建代理池
    pool = ProxyPool()

    # 添加代理
    proxies_config = [
        {"url": "http://proxy1.example.com:8080", "username": "user1", "password": "pass1"},
        {"url": "http://proxy2.example.com:8080", "username": "user2", "password": "pass2"},
        {"url": "http://proxy3.example.com:8080"},
        {"url": "http://proxy4.example.com:8080"},
    ]
    pool.add_proxies_from_config(proxies_config)

    # 显示初始状态
    print("\n初始状态:")
    pool.display_status()

    # 模拟使用代理进行请求
    print("\n模拟请求处理:")
    for i in range(10):
        proxy = pool.get_random_proxy()
        if proxy:
            # 模拟请求结果
            success = random.random() > 0.3
            if success:
                response_time = random.uniform(100, 500)
                pool.record_success(proxy.url, response_time)
            else:
                pool.record_failure(proxy.url, "Connection timeout")
        await asyncio.sleep(0.05)

    # 显示最终状态
    print("\n最终状态:")
    pool.display_status()

    # 获取统计信息
    stats = pool.get_stats()
    print(f"\n统计摘要:")
    print(f"  总请求数: {stats['total_requests']}")
    print(f"  总体成功率: {stats['overall_success_rate']:.1%}")
    print(f"  平均响应时间: {stats['avg_response_time']:.0f}ms")

asyncio.run(demo_proxy_management())

# ecommerce_crawler.py

import asyncio
import json
import re
import csv
from typing import List, Dict, Any, Optional
from dataclasses import dataclass, field, asdict
from datetime import datetime
from pathlib import Path
from urllib.parse import urljoin, urlparse

# =====================================
# 项目配置
# =====================================

@dataclass
class CrawlerConfig:
    """爬虫配置"""
    # 基础配置
    start_url: str = "https://demo.ecommerce-example.com"
    max_concurrency: int = 3
    max_retries: int = 3

    # 采集配置
    max_pages_per_category: int = 10
    request_delay_min: float = 1.0    # 最小请求间隔（秒）
    request_delay_max: float = 3.0    # 最大请求间隔（秒）

    # 代理配置
    use_proxy: bool = True
    proxy_urls: List[str] = field(default_factory=list)

    # 存储配置
    output_dir: str = "./ecommerce_data"

    @classmethod
    def from_file(cls, config_path: str) -> "CrawlerConfig":
        """从文件加载配置"""
        with open(config_path, "r", encoding="utf-8") as f:
            config_dict = json.load(f)
        return cls(**config_dict)

    def to_file(self, config_path: str) -> None:
        """保存配置到文件"""
        with open(config_path, "w", encoding="utf-8") as f:
            json.dump(asdict(self), f, indent=2, ensure_ascii=False)

# =====================================
# 数据模型
# =====================================

@dataclass
class Product:
    """商品数据模型"""
    product_id: str = ""
    title: str = ""
    price: float = 0.0
    original_price: Optional[float] = None
    currency: str = "USD"
    category: str = ""
    subcategory: str = ""
    brand: str = ""
    rating: float = 0.0
    review_count: int = 0
    seller: str = ""
    in_stock: bool = True
    description: str = ""
    specifications: Dict[str, str] = field(default_factory=dict)
    images: List[str] = field(default_factory=list)
    source_url: str = ""
    scraped_at: str = ""
    error_message: Optional[str] = None

    def __post_init__(self):
        """初始化时设置时间戳"""
        if not self.scraped_at:
            self.scraped_at = datetime.now().isoformat()

    def to_dict(self) -> Dict[str, Any]:
        """转换为字典"""
        return asdict(self)

    def validate(self) -> bool:
        """验证数据完整性"""
        if not self.product_id:
            self.error_message = "缺少商品ID"
            return False
        if not self.title:
            self.error_message = "缺少商品标题"
            return False
        if self.price < 0:
            self.error_message = "价格不能为负数"
            return False
        return True

# =====================================
# HTML 解析工具
# =====================================

class HTMLParser:
    """HTML 解析工具类"""

    @staticmethod
    def extract_text(element, selector: str, default: str = "") -> str:
        """提取文本内容"""
        try:
            found = element.select_one(selector)
            if found:
                return found.get_text(strip=True)
        except Exception:
            pass
        return default

    @staticmethod
    def extract_attribute(element, selector: str, 
                         attribute: str, default: str = "") -> str:
        """提取元素属性"""
        try:
            found = element.select_one(selector)
            if found and found.has_attr(attribute):
                return found[attribute]
        except Exception:
            pass
        return default

    @staticmethod
    def extract_all(element, selector: str) -> List:
        """提取所有匹配的元素"""
        try:
            return element.select(selector)
        except Exception:
            return []

    @staticmethod
    def parse_price(price_str: str) -> float:
        """解析价格字符串"""
        if not price_str:
            return 0.0
        # 移除非数字和小数点以外的字符
        cleaned = re.sub(r"[^\d.]", "", price_str)
        try:
            return float(cleaned)
        except ValueError:
            return 0.0

    @staticmethod
    def parse_rating(rating_str: str) -> float:
        """解析评分字符串"""
        if not rating_str:
            return 0.0
        match = re.search(r"(\d+\.?\d*)", rating_str)
        if match:
            return float(match.group(1))
        return 0.0

    @staticmethod
    def parse_review_count(count_str: str) -> int:
        """解析评论数量"""
        if not count_str:
            return 0
        # 处理 "1,234 reviews" 这样的格式
        cleaned = re.sub(r"[,\s]", "", count_str)
        match = re.search(r"(\d+)", cleaned)
        if match:
            return int(match.group(1))
        return 0

# =====================================
# 爬虫核心逻辑
# =====================================

class EcommerceCrawler:
    """电商爬虫主类"""

    def __init__(self, config: CrawlerConfig):
        self.config = config
        self.products: List[Product] = []
        self.failed_urls: List[Dict[str, str]] = []
        self.stats = {
            "pages_visited": 0,
            "products_found": 0,
            "products_saved": 0,
            "requests_failed": 0,
            "start_time": None,
            "end_time": None,
        }

        # 确保输出目录存在
        Path(config.output_dir).mkdir(parents=True, exist_ok=True)

    async def start(self) -> None:
        """启动爬虫"""
        print("=" * 60)
        print("电商商品数据采集系统启动")
        print("=" * 60)
        print(f"起始 URL: {self.config.start_url}")
        print(f"最大并发: {self.config.max_concurrency}")
        print(f"输出目录: {self.config.output_dir}")
        print("=" * 60)

        self.stats["start_time"] = datetime.now()

        try:
            await self.crawl()
        except KeyboardInterrupt:
            print("\n用户中断，正在停止爬虫...")
        finally:
            await self.finish()

    async def crawl(self) -> None:
        """执行爬取逻辑"""
        # 阶段1: 采集分类页面
        print("\n阶段1: 采集分类页面")
        category_urls = await self.fetch_category_pages()

        # 阶段2: 采集商品列表页
        print("\n阶段2: 采集商品列表页")
        product_urls = await self.fetch_product_listings(category_urls)

        # 阶段3: 采集商品详情页
        print("\n阶段3: 采集商品详情页")
        await self.fetch_product_details(product_urls)

    async def fetch_category_pages(self) -> List[str]:
        """获取所有分类页面"""
        category_urls = []

        # 这里模拟从首页获取分类链接
        # 实际使用时需要解析 HTML 获取真正的分类链接
        mock_categories = [
            "/category/electronics",
            "/category/clothing",
            "/category/home-garden",
            "/category/sports",
        ]

        for cat in mock_categories:
            url = urljoin(self.config.start_url, cat)
            category_urls.append(url)
            print(f"  发现分类: {url}")
            await self.random_delay()

        return category_urls

    async def fetch_product_listings(self, category_urls: List[str]) -> List[str]:
        """获取商品列表页中的商品链接"""
        all_product_urls = []
        page_count = 0

        for category_url in category_urls:
            print(f"\n  采集分类: {category_url}")

            for page in range(1, self.config.max_pages_per_category + 1):
                if page > 1:
                    page_url = f"{category_url}?page={page}"
                else:
                    page_url = category_url

                print(f"    采集列表页 {page}: {page_url}")
                page_count += 1
                self.stats["pages_visited"] += 1

                # 模拟从页面提取商品链接
                # 实际使用时需要解析 HTML
                mock_product_urls = [
                    f"/product/item-{page_count}-{i}" 
                    for i in range(1, 4)
                ]

                for product_path in mock_product_urls:
                    full_url = urljoin(self.config.start_url, product_path)
                    all_product_urls.append(full_url)
                    print(f"      发现商品: {full_url}")

                await self.random_delay()

        return all_product_urls

    async def fetch_product_details(self, product_urls: List[str]) -> None:
        """采集商品详情页"""
        for i, url in enumerate(product_urls, 1):
            print(f"\n  [{i}/{len(product_urls)}] 采集商品: {url}")

            try:
                # 模拟从页面提取商品数据
                # 实际使用时需要解析 HTML
                product = self.simulate_product_extraction(url)

                if product.validate():
                    self.products.append(product)
                    self.stats["products_found"] += 1
                    self.stats["products_saved"] += 1
                    print(f"    成功: {product.title[:50]}...")
                else:
                    print(f"    验证失败: {product.error_message}")
                    self.failed_urls.append({
                        "url": url,
                        "error": product.error_message or "Unknown"
                    })

            except Exception as e:
                print(f"    错误: {str(e)}")
                self.failed_urls.append({
                    "url": url,
                    "error": str(e)
                })
                self.stats["requests_failed"] += 1

            self.stats["pages_visited"] += 1
            await self.random_delay()

    def simulate_product_extraction(self, url: str) -> Product:
        """模拟商品数据提取（实际项目中替换为真实解析逻辑）"""
        # 解析 URL 获取商品 ID
        parsed = urlparse(url)
        path_parts = parsed.path.strip("/").split("/")
        product_id = path_parts[-1] if path_parts else "unknown"

        # 模拟提取的数据
        return Product(
            product_id=product_id,
            title=f"商品标题 - {product_id}",
            price=99.99,
            original_price=129.99,
            currency="USD",
            category="Electronics",
            subcategory="Accessories",
            brand="DemoBrand",
            rating=4.5,
            review_count=1234,
            seller="Official Store",
            in_stock=True,
            description="这是一个示例商品描述，实际项目中会从页面提取真实内容。",
            specifications={
                "Material": "Plastic",
                "Weight": "200g",
                "Dimensions": "10x5x2 cm"
            },
            images=["https://example.com/image1.jpg"],
            source_url=url,
        )

    async def random_delay(self) -> None:
        """随机延迟"""
        import random
        delay = random.uniform(
            self.config.request_delay_min,
            self.config.request_delay_max
        )
        await asyncio.sleep(delay)

    async def save_results(self) -> None:
        """保存采集结果"""
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")

        # 保存为 JSON
        json_path = Path(self.config.output_dir) / f"products_{timestamp}.json"
        with open(json_path, "w", encoding="utf-8") as f:
            data = {
                "metadata": {
                    "scraped_at": datetime.now().isoformat(),
                    "total_products": len(self.products),
                    "config": asdict(self.config),
                },
                "products": [p.to_dict() for p in self.products]
            }
            json.dump(data, f, indent=2, ensure_ascii=False)
        print(f"\nJSON 数据已保存: {json_path}")

        # 保存为 CSV
        csv_path = Path(self.config.output_dir) / f"products_{timestamp}.csv"
        if self.products:
            with open(csv_path, "w", newline="", encoding="utf-8") as f:
                # 展平嵌套字段以便 CSV 存储
                flattened_products = []
                for p in self.products:
                    flat = {
                        "product_id": p.product_id,
                        "title": p.title,
                        "price": p.price,
                        "original_price": p.original_price,
                        "currency": p.currency,
                        "category": p.category,
                        "brand": p.brand,
                        "rating": p.rating,
                        "review_count": p.review_count,
                        "in_stock": p.in_stock,
                        "source_url": p.source_url,
                    }
                    flattened_products.append(flat)

                writer = csv.DictWriter(f, fieldnames=flattened_products[0].keys())
                writer.writeheader()
                writer.writerows(flattened_products)
            print(f"CSV 数据已保存: {csv_path}")

        # 保存失败记录
        if self.failed_urls:
            error_path = Path(self.config.output_dir) / f"failed_{timestamp}.json"
            with open(error_path, "w", encoding="utf-8") as f:
                json.dump(self.failed_urls, f, indent=2, ensure_ascii=False)
            print(f"失败记录已保存: {error_path}")

    async def finish(self) -> None:
        """完成爬虫并生成报告"""
        self.stats["end_time"] = datetime.now()

        print("\n" + "=" * 60)
        print("爬虫执行完成")
        print("=" * 60)

        # 计算耗时
        if self.stats["start_time"] and self.stats["end_time"]:
            duration = (self.stats["end_time"] - self.stats["start_time"]).total_seconds()
            hours = int(duration // 3600)
            minutes = int((duration % 3600) // 60)
            seconds = int(duration % 60)
            print(f"总耗时: {hours:02d}:{minutes:02d}:{seconds:02d}")

        print(f"访问页面数: {self.stats['pages_visited']}")
        print(f"发现商品数: {self.stats['products_found']}")
        print(f"保存商品数: {self.stats['products_saved']}")
        print(f"请求失败数: {self.stats['requests_failed']}")

        if self.products:
            print(f"\n成功率和效率:")
            print(f"  商品采集成功率: {self.stats['products_saved']/len(self.products):.1%}")

        # 保存结果
        await self.save_results()

        print("=" * 60)

# =====================================
# 主程序入口
# =====================================

async def main():
    """主函数"""
    # 创建配置
    config = CrawlerConfig(
        start_url="https://demo.ecommerce-example.com",
        max_concurrency=2,
        max_retries=3,
        max_pages_per_category=5,
        request_delay_min=1.0,
        request_delay_max=2.0,
        output_dir="./ecommerce_data"
    )

    # 创建并启动爬虫
    crawler = EcommerceCrawler(config)
    await crawler.start()

# 运行入口
if __name__ == "__main__":
    asyncio.run(main())

# social_media_crawler.py

import asyncio
import json
import re
from typing import List, Dict, Any, Optional, Set
from dataclasses import dataclass, field, asdict
from datetime import datetime, timedelta
from pathlib import Path
from urllib.parse import urljoin, parse_qs, urlencode

# =====================================
# 配置与常量
# =====================================

@dataclass
class SocialMediaConfig:
    """社交媒体爬虫配置"""
    platform: str = "twitter"              # 平台标识
    base_url: str = "https://twitter.com"
    api_base_url: str = "https://api.twitter.com/2"

    # 认证配置
    use_session: bool = True
    auth_cookies: Dict[str, str] = field(default_factory=dict)

    # 采集配置
    max_tweets: int = 1000
    max_depth: int = 3                    # 关注者遍历深度
    include_retweets: bool = True
    include_replies: bool = False

    # 限流配置
    rate_limit_delay: float = 2.0         # 请求间隔（秒）
    max_retries: int = 5
    retry_delay: float = 60.0             # 限流后重试延迟（秒）

    # 输出配置
    output_dir: str = "./social_data"

# =====================================
# 数据模型
# =====================================

@dataclass
class Tweet:
    """推文数据模型"""
    tweet_id: str = ""
    user_id: str = ""
    username: str = ""
    display_name: str = ""
    content: str = ""
    created_at: str = ""

    # 互动数据
    likes: int = 0
    retweets: int = 0
    replies: int = 0
    views: int = 0

    # 引用和回复
    is_retweet: bool = False
    original_tweet_id: Optional[str] = None
    is_reply: bool = False
    reply_to_tweet_id: Optional[str] = None
    reply_to_user: Optional[str] = None

    # 媒体
    has_media: bool = False
    media_urls: List[str] = field(default_factory=list)
    has_video: bool = False
    video_url: Optional[str] = None

    # 地理位置
    has_location: bool = False
    location: str = ""
    coordinates: Optional[Dict[str, float]] = None

    # 来源信息
    source_app: str = ""
    source_url: str = ""
    scraped_at: str = ""

    def __post_init__(self):
        if not self.scraped_at:
            self.scraped_at = datetime.now().isoformat()

    def to_dict(self) -> Dict[str, Any]:
        return asdict(self)

@dataclass
class User:
    """用户数据模型"""
    user_id: str = ""
    username: str = ""
    display_name: str = ""
    bio: str = ""
    location: str = ""
    website: str = ""

    # 账户信息
    verified: bool = False
    protected: bool = False
    created_at: str = ""

    # 统计数据
    followers_count: int = 0
    following_count: int = 0
    tweet_count: int = 0
    listed_count: int = 0

    # 头像和背景图
    avatar_url: str = ""
    banner_url: str = ""

    # 关联数据
    pinned_tweet_id: Optional[str] = None

    # 元数据
    source_url: str = ""
    scraped_at: str = ""

    def __post_init__(self):
        if not self.scraped_at:
            self.scraped_at = datetime.now().isoformat()

    def to_dict(self) -> Dict[str, Any]:
        return asdict(self)

# =====================================
# 会话管理
# =====================================

class SessionManager:
    """会话管理器"""

    def __init__(self, config: SocialMediaConfig):
        self.config = config
        self.session_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": "application/json, text/html",
            "Accept-Language": "en-US,en;q=0.9",
            "Accept-Encoding": "gzip, deflate, br",
            "DNT": "1",
            "Connection": "keep-alive",
            "Upgrade-Insecure-Requests": "1",
        }
        self.request_count = 0
        self.last_request_time = None
        self.rate_limit_remaining = 100
        self.rate_limit_reset = None

    def update_headers(self, additional_headers: Dict[str, str]) -> None:
        """更新请求头"""
        self.session_headers.update(additional_headers)

    def add_cookies(self, cookies: Dict[str, str]) -> None:
        """添加 cookies"""
        if cookies:
            cookie_str = "; ".join(f"{k}={v}" for k, v in cookies.items())
            self.session_headers["Cookie"] = cookie_str

    async def rate_limit_wait(self) -> None:
        """等待以遵守速率限制"""
        if self.rate_limit_remaining <= 5:
            if self.rate_limit_reset:
                wait_time = max(
                    (self.rate_limit_reset - datetime.now()).total_seconds(),
                    self.config.rate_limit_delay
                )
                print(f"接近速率限制，等待 {wait_time:.1f} 秒...")
                await asyncio.sleep(wait_time)
        else:
            await asyncio.sleep(self.config.rate_limit_delay)

        self.request_count += 1

    def update_rate_limit_info(self, headers: Dict[str, str]) -> None:
        """更新速率限制信息"""
        if "x-rate-limit-remaining" in headers:
            self.rate_limit_remaining = int(headers["x-rate-limit-remaining"])
        if "x-rate-limit-reset" in headers:
            reset_timestamp = int(headers["x-rate-limit-reset"])
            self.rate_limit_reset = datetime.fromtimestamp(reset_timestamp)

    def is_rate_limited(self) -> bool:
        """检查是否被限流"""
        if self.rate_limit_remaining <= 0:
            return True
        return False

# =====================================
# 爬虫核心
# =====================================

class SocialMediaCrawler:
    """社交媒体爬虫"""

    def __init__(self, config: SocialMediaConfig):
        self.config = config
        self.session = SessionManager(config)
        self.tweets: List[Tweet] = []
        self.users: List[User] = []
        self.visited_urls: Set[str] = set()
        self.stats = {
            "tweets_collected": 0,
            "users_collected": 0,
            "requests_made": 0,
            "rate_limited": 0,
            "errors": 0,
        }

        # 设置初始 cookies
        if config.auth_cookies:
            self.session.add_cookies(config.auth_cookies)

    async def search_tweets(self, query: str, max_results: int = 100) -> List[Tweet]:
        """搜索推文"""
        print(f"\n搜索推文: {query}")
        tweets = []

        # 构建搜索参数
        params = {
            "query": query,
            "max_results": min(max_results, 100),
            "tweet.fields": "created_at,public_metrics,entities,geo",
            "expansions": "author_id,attachments.media_keys",
            "user.fields": "username,name,verified",
        }

        # 模拟分页请求
        remaining = max_results
        next_token = None

        while remaining > 0 and len(tweets) < max_results:
            if next_token:
                params["next_token"] = next_token

            # 这里应该发起真实的 API 请求
            # await self.session.rate_limit_wait()

            # 模拟数据
            batch_size = min(remaining, 100)
            for i in range(batch_size):
                tweet = Tweet(
                    tweet_id=f"tweet_{len(tweets)}",
                    user_id=f"user_{hash(query) % 1000}",
                    username="example_user",
                    display_name="Example User",
                    content=f"这是关于 '{query}' 的推文内容 #{query.replace(' ', '')}",
                    created_at=(datetime.now() - timedelta(days=i)).isoformat(),
                    likes=hash(f"{query}_{i}") % 10000,
                    retweets=hash(f"{query}_{i}") % 1000,
                    views=hash(f"{query}_{i}") % 100000,
                    source_app="Twitter Web App",
                    source_url=f"{self.config.base_url}/i/status/{len(tweets)}",
                )
                tweets.append(tweet)
                self.stats["tweets_collected"] += 1

            self.stats["requests_made"] += 1
            remaining -= batch_size

            print(f"  已获取 {len(tweets)} 条推文...")
            await asyncio.sleep(0.5)

            # 模拟分页
            if remaining > 0:
                next_token = f"mock_token_{len(tweets)}"
            else:
                break

        self.tweets.extend(tweets)
        return tweets

    async def get_user_timeline(self, username: str, 
                                 max_tweets: int = 100) -> List[Tweet]:
        """获取用户时间线"""
        print(f"\n获取用户时间线: @{username}")
        tweets = []

        # 模拟请求
        for i in range(min(max_tweets, 20)):
            tweet = Tweet(
                tweet_id=f"tweet_{username}_{i}",
                user_id=f"user_{hash(username) % 1000}",
                username=username,
                display_name=username.title(),
                content=f"这是 @{username} 的第 {i+1} 条推文",
                created_at=(datetime.now() - timedelta(hours=i*2)).isoformat(),
                likes=i * 10,
                retweets=i * 5,
                views=i * 100,
                source_app="Twitter for iPhone",
                source_url=f"{self.config.base_url}/{username}/status/{i}",
            )
            tweets.append(tweet)

        self.tweets.extend(tweets)
        self.stats["tweets_collected"] += len(tweets)
        return tweets

    async def get_user_profile(self, username: str) -> Optional[User]:
        """获取用户资料"""
        print(f"\n获取用户资料: @{username}")

        # 模拟请求
        user = User(
            user_id=f"user_{hash(username) % 1000}",
            username=username,
            display_name=username.title(),
            bio=f"这是 @{username} 的个人简介",
            location="San Francisco, CA",
            website="https://example.com",
            verified=True,
            followers_count=hash(username) % 100000,
            following_count=hash(username) % 1000,
            tweet_count=hash(username) % 10000,
            listed_count=hash(username) % 100,
            avatar_url=f"{self.config.base_url}/profile_images/{username}.jpg",
            source_url=f"{self.config.base_url}/{username}",
        )

        self.users.append(user)
        self.stats["users_collected"] += 1
        return user

    async def get_followers(self, username: str, 
                           max_followers: int = 100) -> List[User]:
        """获取用户关注者"""
        print(f"\n获取 @{username} 的关注者")
        followers = []

        for i in range(min(max_followers, 20)):
            follower = User(
                user_id=f"follower_{hash(username)}_{i}",
                username=f"{username}_follower_{i}",
                display_name=f"Follower {i}",
                followers_count=hash(f"{username}_{i}") % 10000,
                verified=False,
            )
            followers.append(follower)

        self.users.extend(followers)
        return followers

    async def collect_user_network(self, seed_username: str, 
                                   depth: int = 2) -> None:
        """收集用户社交网络"""
        print(f"\n收集用户网络: @{seed_username} (深度: {depth})")

        visited_users = set()
        current_level = {seed_username}

        for level in range(depth):
            print(f"\n层级 {level + 1}/{depth}")
            next_level = set()

            for username in current_level:
                if username in visited_users:
                    continue
                visited_users.add(username)

                # 获取用户资料
                await self.get_user_profile(username)

                # 获取用户推文
                await self.get_user_timeline(username, max_tweets=10)

                # 获取关注者
                followers = await self.get_followers(username, max_followers=5)
                for follower in followers:
                    next_level.add(follower.username)

                await asyncio.sleep(self.config.rate_limit_delay)

            current_level = next_level
            print(f"  已收集 {len(visited_users)} 个用户...")

    async def search_by_hashtag(self, hashtag: str, 
                               max_tweets: int = 100) -> List[Tweet]:
        """按话题标签搜索"""
        query = f"#{hashtag.replace('#', '')}"
        return await self.search_tweets(query, max_tweets)

    async def search_by_location(self, location: str, 
                                 radius: str = "10km",
                                 max_tweets: int = 100) -> List[Tweet]:
        """按地理位置搜索"""
        query = f"geocode:{location},{radius}"
        return await self.search_tweets(query, max_tweets)

    async def save_results(self) -> None:
        """保存采集结果"""
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        output_dir = Path(self.config.output_dir)
        output_dir.mkdir(parents=True, exist_ok=True)

        # 保存推文
        tweets_path = output_dir / f"tweets_{timestamp}.json"
        with open(tweets_path, "w", encoding="utf-8") as f:
            json.dump({
                "metadata": {
                    "collected_at": datetime.now().isoformat(),
                    "total_tweets": len(self.tweets),
                    "platform": self.config.platform,
                },
                "tweets": [t.to_dict() for t in self.tweets]
            }, f, indent=2, ensure_ascii=False)
        print(f"\n推文已保存: {tweets_path}")

        # 保存用户
        users_path = output_dir / f"users_{timestamp}.json"
        with open(users_path, "w", encoding="utf-8") as f:
            json.dump({
                "metadata": {
                    "collected_at": datetime.now().isoformat(),
                    "total_users": len(self.users),
                    "platform": self.config.platform,
                },
                "users": [u.to_dict() for u in self.users]
            }, f, indent=2, ensure_ascii=False)
        print(f"用户已保存: {users_path}")

    async def run(self) -> None:
        """运行爬虫"""
        print("=" * 60)
        print("社交媒体数据采集系统")
        print("=" * 60)
        print(f"平台: {self.config.platform}")
        print(f"基础 URL: {self.config.base_url}")
        print("=" * 60)

        # 示例任务
        # 任务1: 搜索特定话题
        print("\n任务1: 按话题搜索")
        await self.search_by_hashtag("python", max_tweets=50)

        # 任务2: 收集特定用户信息
        print("\n任务2: 收集用户网络")
        await self.collect_user_network("example_user", depth=2)

        # 任务3: 收集用户时间线
        print("\n任务3: 收集用户时间线")
        await self.get_user_timeline("example_user", max_tweets=20)

        # 保存结果
        await self.save_results()

        # 打印统计
        print("\n" + "=" * 60)
        print("采集统计")
        print("=" * 60)
        print(f"推文数量: {self.stats['tweets_collected']}")
        print(f"用户数量: {self.stats['users_collected']}")
        print(f"请求次数: {self.stats['requests_made']}")
        print(f"限流次数: {self.stats['rate_limited']}")
        print(f"错误次数: {self.stats['errors']}")
        print("=" * 60)

# =====================================
# 主程序
# =====================================

async def main():
    """主函数"""
    config = SocialMediaConfig(
        platform="twitter",
        base_url="https://twitter.com",
        max_tweets=1000,
        rate_limit_delay=2.0,
        output_dir="./social_data"
    )

    crawler = SocialMediaCrawler(config)
    await crawler.run()

if __name__ == "__main__":
    asyncio.run(main())