CrawlerTutorial Documentation

源代码/爬虫进阶/
├── 01_工程化爬虫开发规范/
│   ├── config/
│   ├── logger_demo.py
│   ├── exception_demo.py
│   └── refactored_crawler/
├── 02_反爬虫对抗基础_请求伪装/
│   ├── ua_rotator.py
│   ├── headers_builder.py
│   └── rate_limiter.py
├── 03_代理IP的使用与管理/
│   ├── proxy_pool/
│   ├── proxy_checker.py
│   └── proxy_demo.py
├── 04_Playwright浏览器自动化入门/
│   ├── basic_operations.py
│   ├── wait_strategies.py
│   └── spa_crawler.py
├── 05_Playwright进阶_反检测与性能优化/
│   ├── stealth_demo.py
│   ├── cdp_mode.py
│   └── performance_optimization.py
├── 06_登录认证_Cookie与Session管理/
│   ├── cookie_manager.py
│   ├── session_demo.py
│   └── login_state_checker.py
├── 07_登录认证_扫码与短信登录实现/
│   ├── qrcode_login.py
│   ├── sms_login.py
│   └── login_factory.py
├── 08_验证码识别与处理/
│   ├── ocr_captcha.py
│   ├── slider_captcha.py
│   └── captcha_service.py
├── 09_数据清洗与预处理/
│   ├── text_cleaner.py
│   ├── deduplication.py
│   └── data_normalizer.py
├── 10_数据分析与可视化/
│   ├── pandas_analysis.py
│   ├── wordcloud_generator.py
│   └── chart_demo.py
└── 11_进阶综合实战项目/
    ├── config/
    ├── core/
    ├── login/
    ├── crawler/
    ├── store/
    ├── proxy/
    ├── analysis/
    └── main.py

# 进入任意章节目录
cd 01_工程化爬虫开发规范

# 安装依赖
uv sync

# 运行示例代码
uv run python main.py

# 1. 进入章节目录
cd 01_工程化爬虫开发规范

# 2. 安装依赖
uv sync

# 3. 运行代码
uv run python main.py

# 章节02 - 安装TLS指纹伪装
uv sync --extra advanced

# 章节08 - 安装所有验证码识别功能
uv sync --extra all

# 章节10 - 安装交互式图表功能
uv sync --extra interactive

uv sync
uv run playwright install chromium

# macOS/Linux
curl -LsSf https://astral.sh/uv/install.sh | sh

# Windows
powershell -c "irm https://astral.sh/uv/install.ps1 | iex"

# 单独安装 Chromium
uv run playwright install chromium

# 如果网络问题，可以设置镜像
export PLAYWRIGHT_DOWNLOAD_HOST=https://npmmirror.com/mirrors/playwright/
uv run playwright install chromium

# 进入本章目录
cd 01_工程化爬虫开发规范

# 安装依赖
uv sync

# 运行示例
uv run python logger_demo.py
uv run python exception_demo.py
uv run python refactored_crawler/main.py

cd 02_反爬虫对抗基础_请求伪装

# 安装基础依赖
uv sync

# 安装高级功能（TLS指纹伪装，可选）
uv sync --extra advanced

# 运行示例
uv run python ua_rotator.py
uv run python headers_builder.py
uv run python rate_limiter.py
uv run python anti_detection_crawler.py

cd 03_代理IP的使用与管理
uv sync
uv run python proxy_demo.py

cd 04_Playwright浏览器自动化入门
uv sync
uv run playwright install chromium  # 安装浏览器
uv run python basic_operations.py
uv run python wait_strategies.py
uv run python spa_crawler.py

cd 05_Playwright进阶_反检测与性能优化
uv sync
uv run playwright install chromium
uv run python stealth_demo.py
uv run python cdp_mode.py
uv run python performance_optimization.py

cd 06_登录认证_Cookie与Session管理
uv sync
uv run python cookie_manager.py
uv run python login_state_checker.py
uv run python session_demo.py  # 包含 quotes.toscrape.com 真实登录演示

cd 07_登录认证_扫码与短信登录实现

# 安装基础依赖
uv sync

# 安装二维码功能（可选）
uv sync --extra qrcode

# 安装浏览器
uv run playwright install chromium

# 运行示例
uv run python qrcode_login.py
uv run python sms_login.py
uv run python login_factory.py

cd 08_验证码识别与处理

# 安装基础依赖
uv sync

# 安装OCR功能
uv sync --extra ocr

# 安装图像处理功能
uv sync --extra cv

# 安装验证码生成功能
uv sync --extra generate

# 安装轨迹可视化功能
uv sync --extra viz

# 或安装所有功能
uv sync --extra all

# 运行示例
uv run python ocr_captcha.py
uv run python slider_captcha.py

cd 09_数据清洗与预处理

# 安装依赖（本章主要使用Python标准库）
uv sync

# 安装SimHash去重功能（可选）
uv sync --extra simhash

# 运行示例
uv run python text_cleaner.py
uv run python data_normalizer.py
uv run python deduplication.py

cd 10_数据分析与可视化

# 安装基础依赖
uv sync

# 安装交互式图表功能（可选）
uv sync --extra interactive

# 运行示例
uv run python pandas_analysis.py
uv run python wordcloud_generator.py
uv run python chart_demo.py

cd 11_进阶综合实战项目

# 安装依赖
uv sync

# 安装可选功能（配置管理增强）
uv sync --extra all

# 安装Playwright浏览器驱动
uv run playwright install chromium

# 运行项目
uv run python main.py

11_进阶综合实战项目/
├── config/          # 配置模块
├── core/            # 核心模块（浏览器管理）
├── login/           # 登录模块
├── crawler/         # 爬虫模块
├── store/           # 存储模块
├── proxy/           # 代理池模块
├── analysis/        # 分析模块
└── main.py          # 主程序入口

import requests

response = requests.get('https://httpbin.org/get')
print(response.json())

import httpx

response = httpx.get('https://httpbin.org/get')
print(response.json())

import requests

data = {'name': '程序员阿江','email':'[email protected]'}
response = requests.post('https://httpbin.org/post', data=data)
print(response.json())

import httpx
import asyncio

async def post_data():
    data = {'name': '程序员阿江','email':'[email protected]'}
    async with httpx.AsyncClient() as client:
        response = await client.post('https://httpbin.org/post', data=data)
        print(response.json())

asyncio.run(post_data())

<div class="r-ent">
    <div class="nrec"><span class="hl f3">11</span></div>
    <div class="title">

        <a href="/bbs/Stock/M.1711544298.A.9F8.html">[新聞] 童子賢：用稅收補貼電費非長久之計 應共</a>

    </div>
    <div class="meta">
        <div class="author">addy7533967</div>
        <div class="article-menu">

            <div class="trigger">⋯</div>
            <div class="dropdown">
                <div class="item"><a href="/bbs/Stock/search?q=thread%3A%5B%E6%96%B0%E8%81%9E%5D+%E7%AB%A5%E5%AD%90%E8%B3%A2%EF%BC%9A%E7%94%A8%E7%A8%85%E6%94%B6%E8%A3%9C%E8%B2%BC%E9%9B%BB%E8%B2%BB%E9%9D%9E%E9%95%B7%E4%B9%85%E4%B9%8B%E8%A8%88+%E6%87%89%E5%85%B1">搜尋同標題文章</a></div>

                <div class="item"><a href="/bbs/Stock/search?q=author%3Aaddy7533967">搜尋看板內 addy7533967 的文章</a></div>

            </div>

        </div>
        <div class="date"> 3/27</div>
        <div class="mark"></div>
    </div>
</div>

/* Detailed source-code truncated for AI context efficiency. */

pip3 install requests
pip3 install beautifulsoup4
pip3 install lxml
pip3 install httpx
pip3 install parsel

/* Detailed source-code truncated for AI context efficiency. */

/* Detailed source-code truncated for AI context efficiency. */

{
    "offset": 25,
    "size": 25,
    "sortType": "DESC",
    "sortField": "intradaymarketcap",
    "quoteType": "CRYPTOCURRENCY",
    "query": {
        "operator": "and",
        "operands": [
            {
                "operator": "eq",
                "operands": [
                    "currency",
                    "USD"
                ]
            },
            {
                "operator": "eq",
                "operands": [
                    "exchange",
                    "CCC"
                ]
            }
        ]
    },
    "userId": "",
    "userIdType": "guid"
}

/* Detailed source-code truncated for AI context efficiency. */

/* Detailed source-code truncated for AI context efficiency. */

/* Detailed source-code truncated for AI context efficiency. */

{"offset":0,"size":25,"sortType":"DESC","sortField":"intradaymarketcap","quoteType":"CRYPTOCURRENCY","query":{"operator":"and","operands":[{"operator":"eq","operands":["currency","USD"]},{"operator":"eq","operands":["exchange","CCC"]}]},"userId":"","userIdType":"guid"}

# 同步代码安装依赖
pip install requests

# 异步代码安装依赖
pip install httpx
pip install aiofiles

/* Detailed source-code truncated for AI context efficiency. */

/* Detailed source-code truncated for AI context efficiency. */

from typing import List

class SymbolContent:
    symbol: str = ""
    name: str = ""
    price: str = ""  # 价格（盘中）
    change_price: str = ""  # 跌涨价格
    change_percent: str = ""  # 跌涨幅
    market_price: str = ""  # 市值

    @classmethod
    def get_fields(cls) -> List[str]:
        return [key for key in cls.__dict__.keys() if not key.startswith('__') and key != "get_fields"]

    def __str__(self):
        return f"""
Symbol: {self.symbol}
Name: {self.name}
Price: {self.price}
Change Price: {self.change_price}        
Change Percent: {self.change_percent}        
Market Price: {self.market_price}        
"""

# 导入Pydantic库的BaseModel基类
from pydantic import BaseModel

# 定义你的模型类
class User(BaseModel):
    id: int
    name: str
    age: int

# 实例化你的模型类
user = User(id=1, name="小明", age=18)
# 将模型类转换成dict
user_dict = user.model_dump()
print(type(user_dict), user_dict) # <class 'dict'> {'id': 1, 'name': '小明', 'age': 18}

# 将模型类转换成json
user_json = user.model_dump_json()
print(type(user_json), user_json) # <class 'str'> {"id": 1, "name": "小明", "age": 18}

from pydantic import BaseModel, Field

class SymbolContent(BaseModel):
    symbol: str = Field(default="", title="Symbol")
    name: str = Field(default="", title="Name")
    price: str = Field(default="", title="价格盘中")
    change_price: str = Field(default="", title="跌涨价格")
    change_percent: str = Field(default="", title="跌涨幅")
    market_price: str = Field(default="", title="市值")

from abc import ABC, abstractmethod

from common import SymbolContent


class AbstractStore(ABC):
    @abstractmethod
    async def save(self, save_item: SymbolContent):
        """
        存储数据
        :param save_item:
        :return:
        """
        raise NotImplementedError

import csv
import pathlib
import time
from typing import Dict

import aiofiles

from abstract_store import AbstractStore
from common import SymbolContent

class CsvStoreImpl(AbstractStore):

    def __init__(self):
        self.csv_store_path = "data/csv"

    def make_save_file_name(self) -> str:
        """
        make save file name
        :return:
        """
        return f"{self.csv_store_path}/symbol_content_{int(time.time())}.csv"

    async def save(self, save_item: SymbolContent):
        """
        save data to csv
        :param save_item:
        :return:
        """
        pathlib.Path(self.csv_store_path).mkdir(parents=True, exist_ok=True)
        save_file_name = self.make_save_file_name()
        async with aiofiles.open(save_file_name, mode='a+', encoding="utf-8-sig", newline="") as f:
            f.fileno()
            writer = csv.writer(f)
            save_item_dict: Dict = save_item.model_dump()
            if await f.tell() == 0:
                await writer.writerow(save_item_dict.keys())
            await writer.writerow(save_item_dict.values())

import json
import os
import pathlib
import time


import aiofiles
from abstract_store import AbstractStore

from common import SymbolContent


class JsonStoreImpl(AbstractStore):

    def __init__(self):
        self.json_store_path = "data/json"

    def make_save_file_name(self) -> str:
        """
        make save file name
        :return:
        """
        return f"{self.json_store_path}/symbol_content_{int(time.time())}.json"

    async def save(self, save_item: SymbolContent):
        """
        save data to json
        :param save_item:
        :return:
        """
        pathlib.Path(self.json_store_path).mkdir(parents=True, exist_ok=True)
        save_file_name = self.make_save_file_name()
        save_data_list = []
        # todo 如果这里涉及并发写入，需要加锁, 可以查看MediaCrawler项目中的实现方式
        # 先判断文件是否存在，如果存在则读取文件内容放到save_data_list中，然后再将新的数据添加到save_data_list中
        if os.path.exists(save_file_name):
            async with aiofiles.open(save_file_name, 'r', encoding='utf-8') as file:
                save_data_list = json.loads(await file.read())
        save_data_list.append(save_item.model_dump())

        # 将数据写入到文件中
        async with aiofiles.open(save_file_name, 'w', encoding='utf-8') as file:
            await file.write(json.dumps(save_data_list, ensure_ascii=False))

/* Detailed source-code truncated for AI context efficiency. */

from typing import  Optional

from abstract_store import AbstractStore
from async_db import MysqlConnect, AsyncMysqlDB
from common import SymbolContent


class DbStoreImpl(AbstractStore):
    def __init__(self):
        self.db: Optional[AsyncMysqlDB] = None

    async def save(self, save_item: SymbolContent):
        """
        save data to db
        :param save_item:
        :return:
        """
        self.db = (await MysqlConnect().async_init()).get_db()
        from sqls import (insert_symbol_content,
                          query_symbol_content_by_symbol,
                          update_symbol_content)

        # 查询是否存在
        exist_item = await query_symbol_content_by_symbol(self.db, save_item.symbol)
        if exist_item.symbol:
            # 更新
            await update_symbol_content(self.db, save_item)
        else:
            # 插入
            await insert_symbol_content(self.db, save_item)

# -*- coding: utf-8 -*-
# @Author  : [email protected]
# @Name    : 程序员阿江-Relakkes
# @Time    : 2024/6/7 17:09
# @Desc    :

from async_db import AsyncMysqlDB
from common import SymbolContent


async def insert_symbol_content(db: AsyncMysqlDB, symbol_content: SymbolContent) -> int:
    """
    插入数据
    :param db:
    :param symbol_content:
    :return:
    """
    item = symbol_content.model_dump()
    return await db.item_to_table("symbol_content", item)


async def update_symbol_content(db: AsyncMysqlDB, symbol_content: SymbolContent) -> int:
    """
    更新数据
    :param db:
    :param symbol_content:
    :return:
    """
    item = symbol_content.model_dump()
    return await db.update_table("symbol_content", item, "symbol", symbol_content.symbol)


async def query_symbol_content_by_symbol(db: AsyncMysqlDB, symbol: str) -> SymbolContent:
    """
    查询数据
    :param db:
    :param symbol:
    :return:
    """
    sql = f"select * from symbol_content where symbol = '{symbol}'"
    rows = await db.query(sql)
    if len(rows) > 0:
        return SymbolContent(**rows[0])
    return SymbolContent()

async def run_crawler(data_save_type: str) -> None:
    """
    爬虫主流程
    :param data_save_type: 数据存储的类型，支持csv、json、db
    :return:
    """
    # step1 获取最大数据总量
    max_total: int = await get_max_total_count()
    # step2 遍历每一页数据并解析存储到数据容器中
    data_list: List[SymbolContent] = await fetch_currency_data_list(max_total)
    # step3 将数据保存到指定存储介质中
    for data_item in data_list:
        await StoreFactory.get_store(data_save_type).save(data_item)


if __name__ == '__main__':
    _data_save_type = "csv"  # 可选配置（csv、json、db）
    asyncio.run(run_crawler(_data_save_type))

class StoreFactory:
    @staticmethod
    def get_store(store_type: str) -> AbstractStore:
        if store_type == "csv":
            return CsvStoreImpl()
        elif store_type == "json":
            return JsonStoreImpl()
        elif store_type == "db":
            return DbStoreImpl()
        else:
            raise ValueError(f"Unknown store type: {store_type}")

# 依赖安装，进入到`10_爬虫入门实战3_数据存储实现`目录下，运行以下命令
# 创建python虚拟环境
python3 -m venv venv
# 激活虚拟环境
source venv/bin/activate
# 安装依赖
pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple

# 如果是保存的mysql中的话，需要创建数据库和表结构，可以查看源代码目录下的sql文件
python main.py

import multiprocessing
import time

def worker(num):
    print(f"Worker {num} started")
    time.sleep(2)
    print(f"Worker {num} finished")

if __name__ == "__main__":
    processes = []
    for i in range(5):
        p = multiprocessing.Process(target=worker, args=(i,))
        processes.append(p)
        p.start()
    
    for p in processes:
        p.join()
    
    print("All processes completed")

import threading
import time

def worker(num):
    print(f"Thread {num} started")
    time.sleep(2)
    print(f"Thread {num} finished")

threads = []
for i in range(5):
    t = threading.Thread(target=worker, args=(i,))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

print("All threads completed")

import asyncio

async def worker(num):
    print(f"Coroutine {num} started")
    await asyncio.sleep(2)
    print(f"Coroutine {num} finished")

async def main():
    tasks = [asyncio.create_task(worker(i)) for i in range(5)]
    await asyncio.gather(*tasks)

asyncio.run(main())
print("All coroutines completed")

/* Detailed source-code truncated for AI context efficiency. */

/* Detailed source-code truncated for AI context efficiency. */

/* Detailed source-code truncated for AI context efficiency. */

graph TB
    subgraph 工程化核心
        log[日志系统<br/>loguru]
        config[配置管理<br/>pydantic-settings]
        exception[异常处理<br/>自定义异常+重试]
        structure[项目结构<br/>模块化设计]
    end

    subgraph 解决的问题
        p1[调试困难]
        p2[配置混乱]
        p3[脆弱性高]
        p4[可读性差]
    end

    log -->|解决| p1
    config -->|解决| p2
    exception -->|解决| p3
    structure -->|解决| p4

pip install loguru

from loguru import logger

logger.debug("这是一条调试信息")
logger.info("这是一条普通信息")
logger.warning("这是一条警告信息")
logger.error("这是一条错误信息")
logger.critical("这是一条严重错误信息")

2024-03-28 10:30:00.123 | DEBUG    | __main__:<module>:3 - 这是一条调试信息
2024-03-28 10:30:00.124 | INFO     | __main__:<module>:4 - 这是一条普通信息
2024-03-28 10:30:00.125 | WARNING  | __main__:<module>:5 - 这是一条警告信息
2024-03-28 10:30:00.126 | ERROR    | __main__:<module>:6 - 这是一条错误信息
2024-03-28 10:30:00.127 | CRITICAL | __main__:<module>:7 - 这是一条严重错误信息

graph LR
    subgraph 日志级别从低到高
        DEBUG[DEBUG<br/>调试信息] --> INFO[INFO<br/>正常信息]
        INFO --> WARNING[WARNING<br/>警告信息]
        WARNING --> ERROR[ERROR<br/>错误信息]
        ERROR --> CRITICAL[CRITICAL<br/>致命错误]
    end

    subgraph 爬虫使用场景
        d1[请求参数/响应片段] -.-> DEBUG
        d2[爬取第N页/保存N条] -.-> INFO
        d3[请求超时重试] -.-> WARNING
        d4[单页解析失败] -.-> ERROR
        d5[数据库连接失败] -.-> CRITICAL
    end

from loguru import logger
import sys

# 移除默认的控制台输出（可选）
logger.remove()

# 添加控制台输出，只显示 INFO 及以上级别
logger.add(
    sys.stderr,
    level="INFO",
    format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>"
)

# 添加文件输出，记录所有级别，按日期轮转
logger.add(
    "logs/crawler_{time:YYYY-MM-DD}.log",
    rotation="00:00",      # 每天午夜轮转
    retention="7 days",    # 保留7天的日志
    compression="zip",     # 旧日志压缩
    level="DEBUG",
    encoding="utf-8"
)

# 单独的错误日志文件
logger.add(
    "logs/error_{time:YYYY-MM-DD}.log",
    rotation="00:00",
    retention="30 days",
    level="ERROR",
    encoding="utf-8"
)

from loguru import logger

async def fetch_page(client, url: str) -> str:
    """获取页面内容"""
    logger.debug(f"准备请求: {url}")

    try:
        response = await client.get(url)
        response.raise_for_status()
        logger.info(f"请求成功: {url}, 状态码: {response.status_code}")
        return response.text
    except httpx.TimeoutException:
        logger.warning(f"请求超时: {url}")
        raise
    except httpx.HTTPStatusError as e:
        logger.error(f"HTTP错误: {url}, 状态码: {e.response.status_code}")
        raise
    except Exception as e:
        logger.exception(f"未知错误: {url}")  # exception() 会自动记录堆栈信息
        raise

# quotes_crawler.py - 名言网站爬虫日志示例
from loguru import logger
import httpx

async def fetch_quotes(client: httpx.AsyncClient, page: int = 1):
    """
    获取名言列表

    展示在实际爬虫中如何使用日志
    """
    url = f"https://quotes.toscrape.com/page/{page}/"

    logger.debug(f"[QuotesCrawler] 请求参数: page={page}")

    try:
        response = await client.get(url)

        # HTTP状态码处理
        if response.status_code == 200:
            logger.info(f"[QuotesCrawler] 获取成功: 第{page}页")
            return response.text
        elif response.status_code == 404:
            logger.warning(f"[QuotesCrawler] 页面不存在: 第{page}页")
            return None
        elif response.status_code == 429:
            logger.warning(f"[QuotesCrawler] 触发频率限制，需要等待")
            raise Exception("触发频率限制")
        else:
            logger.error(f"[QuotesCrawler] HTTP错误: status_code={response.status_code}")
            return None

    except httpx.TimeoutException:
        logger.warning(f"[QuotesCrawler] 请求超时: {url}")
        raise
    except Exception as e:
        logger.exception(f"[QuotesCrawler] 未知错误: {e}")
        raise

# 反面示例 - 不要这样做
client = httpx.AsyncClient(timeout=30)
db_url = "mysql://root:password123@localhost/crawler"
API_KEY = "sk-xxxxxxxxxxxx"

pip install pydantic-settings

flowchart LR
    subgraph 配置来源
        env[环境变量<br/>CRAWLER_XXX]
        dotenv[.env 文件<br/>敏感信息]
        default[代码默认值]
    end

    subgraph pydantic-settings
        validate[类型验证]
        merge[配置合并]
    end

    subgraph 使用
        settings[settings 单例]
        app[应用程序]
    end

    env --> validate
    dotenv --> validate
    default --> validate
    validate --> merge
    merge --> settings
    settings --> app

# config/settings.py
from pydantic_settings import BaseSettings
from pydantic import Field
from typing import Optional


class CrawlerSettings(BaseSettings):
    """爬虫配置"""

    # 基础配置
    debug: bool = Field(default=False, description="调试模式")
    log_level: str = Field(default="INFO", description="日志级别")

    # 请求配置
    request_timeout: int = Field(default=30, description="请求超时时间(秒)")
    max_retries: int = Field(default=3, description="最大重试次数")
    retry_delay: float = Field(default=1.0, description="重试延迟(秒)")

    # 并发配置
    max_concurrent: int = Field(default=10, description="最大并发数")

    # 数据库配置
    db_host: str = Field(default="localhost", description="数据库主机")
    db_port: int = Field(default=3306, description="数据库端口")
    db_user: str = Field(default="root", description="数据库用户")
    db_password: str = Field(default="", description="数据库密码")
    db_name: str = Field(default="crawler", description="数据库名")

    # 代理配置
    proxy_url: Optional[str] = Field(default=None, description="代理地址")

    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"
        # 环境变量前缀，如 CRAWLER_DEBUG=true
        env_prefix = "CRAWLER_"


# 全局配置实例
settings = CrawlerSettings()

# config/settings.py - 爬虫配置
from pydantic_settings import BaseSettings
from pydantic import Field
from typing import Optional
from enum import Enum


class StorageType(str, Enum):
    """存储类型"""
    JSON = "json"
    CSV = "csv"


class CrawlerSettings(BaseSettings):
    """爬虫配置"""

    # 基础配置
    app_name: str = "QuotesCrawler"
    debug: bool = False
    log_level: str = "INFO"

    # 请求配置
    request_timeout: int = 30
    max_retries: int = 3
    retry_delay: float = 1.0

    # 爬虫配置
    base_url: str = "https://quotes.toscrape.com"
    max_pages: int = 10
    max_concurrency: int = 3
    crawl_delay_min: float = 0.5
    crawl_delay_max: float = 1.5

    # 存储配置
    storage_type: StorageType = StorageType.JSON
    output_dir: str = "./output"

    class Config:
        env_file = ".env"
        env_prefix = "CRAWLER_"


# 全局配置实例
settings = CrawlerSettings()

# config/constants.py - 常量配置
"""爬虫常量配置"""

# 默认请求头
DEFAULT_HEADERS = {
    "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
                  "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,*/*;q=0.8",
    "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
}

# HTTP 状态码
HTTP_OK = 200
HTTP_NOT_FOUND = 404
HTTP_TOO_MANY_REQUESTS = 429
HTTP_SERVER_ERROR = 500

# .env
CRAWLER_DEBUG=true
CRAWLER_LOG_LEVEL=DEBUG
CRAWLER_REQUEST_TIMEOUT=60
CRAWLER_DB_PASSWORD=your_secret_password
CRAWLER_PROXY_URL=http://127.0.0.1:7890

from config.settings import settings
from loguru import logger

# 根据配置设置日志级别
logger.add("logs/app.log", level=settings.log_level)

# 使用配置创建客户端
client = httpx.AsyncClient(
    timeout=settings.request_timeout,
    proxies=settings.proxy_url
)

logger.info(f"爬虫启动，调试模式: {settings.debug}")

# config/settings.py
import os
from pydantic_settings import BaseSettings
from pydantic import Field


class BaseConfig(BaseSettings):
    """基础配置"""
    env: str = Field(default="development", description="运行环境")
    debug: bool = False
    log_level: str = "INFO"

    class Config:
        env_file = ".env"
        env_prefix = "CRAWLER_"


class DevelopmentConfig(BaseConfig):
    """开发环境配置"""
    debug: bool = True
    log_level: str = "DEBUG"


class ProductionConfig(BaseConfig):
    """生产环境配置"""
    debug: bool = False
    log_level: str = "WARNING"


def get_settings() -> BaseConfig:
    """根据环境变量返回对应的配置"""
    env = os.getenv("CRAWLER_ENV", "development")
    config_map = {
        "development": DevelopmentConfig,
        "production": ProductionConfig,
    }
    config_class = config_map.get(env, DevelopmentConfig)
    return config_class()


settings = get_settings()

flowchart TD
    request[发起请求] --> response{响应状态}

    response -->|成功| parse[解析数据]
    response -->|超时| retry{重试?}
    response -->|HTTP错误| check_code{检查错误码}

    retry -->|是| request
    retry -->|否| fail[记录失败]

    check_code -->|429| rate_limit[触发限流<br/>等待后重试]
    check_code -->|412| risk_control[触发风控<br/>切换IP/降速]
    check_code -->|401/403| auth_error[认证失败<br/>重新登录]
    check_code -->|其他| fail

    parse --> save[保存数据]

    rate_limit --> request
    risk_control --> request

# exceptions.py

class CrawlerException(Exception):
    """爬虫基础异常类"""
    pass


class RequestException(CrawlerException):
    """请求相关异常"""
    pass


class ParseException(CrawlerException):
    """解析相关异常"""
    pass


class StorageException(CrawlerException):
    """存储相关异常"""
    pass


class RateLimitException(RequestException):
    """触发速率限制"""
    pass


class IPBlockedException(RequestException):
    """IP被封禁"""
    pass


class LoginRequiredException(CrawlerException):
    """需要登录"""
    pass

# exceptions/crawler_exceptions.py
"""爬虫异常类"""


class CrawlerException(Exception):
    """爬虫基础异常"""
    pass


class RequestException(CrawlerException):
    """请求相关异常"""
    pass


class ParseException(CrawlerException):
    """解析相关异常"""
    pass


class StorageException(CrawlerException):
    """存储相关异常"""
    pass


class RateLimitException(RequestException):
    """触发速率限制"""
    pass


class IPBlockedException(RequestException):
    """IP被封禁"""
    pass


class PageNotFoundException(RequestException):
    """页面不存在"""
    pass


# HTTP 状态码映射
HTTP_ERROR_MESSAGES = {
    400: "请求参数错误",
    401: "未授权访问",
    403: "禁止访问",
    404: "页面不存在",
    429: "请求过于频繁",
    500: "服务器内部错误",
    502: "网关错误",
    503: "服务暂时不可用",
}

pip install tenacity

from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type,
    before_sleep_log
)
from loguru import logger
import httpx

from exceptions import RequestException, RateLimitException


@retry(
    stop=stop_after_attempt(3),                    # 最多重试3次
    wait=wait_exponential(multiplier=1, max=10),   # 指数退避，最长等待10秒
    retry=retry_if_exception_type((                # 只对特定异常重试
        httpx.TimeoutException,
        httpx.ConnectError,
        RequestException
    )),
    before_sleep=before_sleep_log(logger, "WARNING")  # 重试前记录日志
)
async def fetch_with_retry(client: httpx.AsyncClient, url: str) -> str:
    """带重试的请求函数"""
    response = await client.get(url)

    # 检查是否触发速率限制
    if response.status_code == 429:
        raise RateLimitException("触发速率限制")

    response.raise_for_status()
    return response.text

from tenacity import (
    retry,
    stop_after_attempt,
    wait_random_exponential,
    retry_if_exception_type,
    RetryError
)


def create_retry_decorator(max_attempts: int = 3, max_wait: int = 60):
    """创建可配置的重试装饰器"""
    return retry(
        stop=stop_after_attempt(max_attempts),
        wait=wait_random_exponential(multiplier=1, max=max_wait),
        retry=retry_if_exception_type((
            httpx.TimeoutException,
            httpx.ConnectError,
            httpx.HTTPStatusError,
        )),
        reraise=True  # 重试用尽后重新抛出异常
    )


# 使用
@create_retry_decorator(max_attempts=5, max_wait=30)
async def fetch_important_data(client, url):
    """重要数据获取，使用更多重试次数"""
    response = await client.get(url)
    response.raise_for_status()
    return response.json()

import asyncio
from loguru import logger
from exceptions import CrawlerException, IPBlockedException


async def run_crawler():
    """爬虫主程序"""
    try:
        # 爬虫逻辑
        await crawl_all_pages()
    except IPBlockedException as e:
        logger.critical(f"IP被封禁，程序终止: {e}")
        # 可以在这里触发告警通知
        raise
    except CrawlerException as e:
        logger.error(f"爬虫异常: {e}")
        raise
    except asyncio.CancelledError:
        logger.warning("任务被取消")
        raise
    except Exception as e:
        logger.exception(f"未预期的异常: {e}")
        raise
    finally:
        logger.info("爬虫任务结束，执行清理工作...")
        # 清理资源
        await cleanup()


if __name__ == "__main__":
    try:
        asyncio.run(run_crawler())
    except KeyboardInterrupt:
        logger.info("用户中断程序")
    except Exception as e:
        logger.critical(f"程序异常退出: {e}")
        exit(1)

graph TB
    subgraph 入口层
        main[main.py<br/>程序入口]
    end

    subgraph 核心模块
        config[config/<br/>配置管理]
        crawler[crawler/<br/>爬虫逻辑]
        store[store/<br/>数据存储]
    end

    subgraph 功能模块
        login[login/<br/>登录认证]
        client[client/<br/>API客户端]
        analysis[analysis/<br/>数据分析]
    end

    subgraph 基础模块
        core[core/<br/>浏览器管理]
        tools[tools/<br/>工具函数]
        models[models/<br/>数据模型]
        exceptions[exceptions/<br/>异常定义]
    end

    main --> config
    main --> crawler
    main --> store
    main --> analysis

    crawler --> login
    crawler --> client

    login --> core
    client --> tools
    tools --> models

my_crawler/
├── config/                 # 配置模块
│   ├── __init__.py
│   └── settings.py         # 配置定义
├── core/                   # 核心模块
│   ├── __init__.py
│   ├── client.py          # HTTP客户端封装
│   └── retry.py           # 重试策略
├── crawler/               # 爬虫模块
│   ├── __init__.py
│   ├── base.py            # 爬虫基类
│   └── xxx_crawler.py     # 具体爬虫实现
├── parser/                # 解析模块
│   ├── __init__.py
│   └── xxx_parser.py      # 页面解析器
├── store/                 # 存储模块
│   ├── __init__.py
│   ├── base.py            # 存储基类
│   ├── mysql.py           # MySQL存储
│   └── json_store.py      # JSON文件存储
├── models/                # 数据模型
│   ├── __init__.py
│   └── xxx_model.py       # Pydantic模型定义
├── exceptions/            # 异常定义
│   ├── __init__.py
│   └── crawler_exceptions.py
├── utils/                 # 工具函数
│   ├── __init__.py
│   └── helpers.py
├── logs/                  # 日志目录
├── data/                  # 数据输出目录
├── tests/                 # 测试目录
│   └── test_xxx.py
├── .env                   # 环境变量（不提交到git）
├── .env.example           # 环境变量示例
├── .gitignore
├── requirements.txt       # 依赖列表
├── main.py               # 程序入口
└── README.md             # 项目说明

# 入门教程的简单版本
import httpx
from parsel import Selector

async def crawl():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://example.com")
        print(f"状态码: {response.status_code}")
        # 解析...
        # 存储...

refactored_crawler/
├── config/
│   ├── __init__.py
│   └── settings.py
├── exceptions.py
├── logger_config.py
├── client.py
├── crawler.py
└── main.py

# logger_config.py
import sys
from loguru import logger
from config.settings import settings


def setup_logger():
    """配置日志系统"""
    # 移除默认处理器
    logger.remove()

    # 控制台输出
    logger.add(
        sys.stderr,
        level=settings.log_level,
        format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> | "
               "<level>{level: <8}</level> | "
               "<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - "
               "<level>{message}</level>"
    )

    # 文件输出
    logger.add(
        f"logs/crawler_{settings.env}.log",
        rotation="00:00",
        retention="7 days",
        level="DEBUG",
        encoding="utf-8"
    )

    return logger

# client.py
import httpx
from tenacity import retry, stop_after_attempt, wait_exponential
from loguru import logger
from config.settings import settings
from exceptions import RequestException


class CrawlerClient:
    """封装的HTTP客户端"""

    def __init__(self):
        self.client = httpx.AsyncClient(
            timeout=settings.request_timeout,
            headers={
                "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
            }
        )

    @retry(
        stop=stop_after_attempt(settings.max_retries),
        wait=wait_exponential(multiplier=1, max=10)
    )
    async def get(self, url: str) -> httpx.Response:
        """发送GET请求，带重试"""
        logger.debug(f"请求: {url}")
        response = await self.client.get(url)
        response.raise_for_status()
        logger.info(f"请求成功: {url}")
        return response

    async def close(self):
        """关闭客户端"""
        await self.client.aclose()
        logger.debug("HTTP客户端已关闭")

# main.py
import asyncio
from loguru import logger
from logger_config import setup_logger
from crawler import BBSCrawler
from config.settings import settings


async def main():
    """主程序入口"""
    # 初始化日志
    setup_logger()

    logger.info(f"爬虫启动 - 环境: {settings.env}, 调试模式: {settings.debug}")

    crawler = BBSCrawler()
    try:
        await crawler.run()
    except Exception as e:
        logger.exception(f"爬虫运行异常: {e}")
        raise
    finally:
        await crawler.cleanup()
        logger.info("爬虫运行结束")


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        logger.info("用户中断")

flowchart TD
    request[爬虫发起请求] --> check1{User-Agent检测}
    check1 -->|异常UA| block1[直接拒绝]
    check1 -->|正常UA| check2{请求头检测}

    check2 -->|缺少必要头| block2[返回错误]
    check2 -->|请求头正常| check3{频率检测}

    check3 -->|频率过高| block3[触发限流 429]
    check3 -->|频率正常| check4{Cookie检测}

    check4 -->|无Cookie/过期| block4[要求登录 401]
    check4 -->|Cookie有效| check5{其他检测}

    check5 -->|检测失败| block5[返回 403]
    check5 -->|检测通过| success[返回数据]

    style block1 fill:#f66
    style block2 fill:#f66
    style block3 fill:#f96
    style block4 fill:#f96
    style block5 fill:#f66
    style success fill:#6f6

graph LR
    subgraph 低难度
        ua[UA检测]
        headers[请求头检测]
    end

    subgraph 中难度
        freq[频率检测]
        cookie[Cookie检测]
        sign[API签名检测]
    end

    subgraph 高难度
        tls[TLS指纹检测]
        js[JS环境检测]
        captcha[验证码]
    end

    ua --> headers --> freq --> cookie --> sign --> tls --> js --> captcha

Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36

/* Detailed source-code truncated for AI context efficiency. */

pip install fake-useragent

from fake_useragent import UserAgent

# 创建 UserAgent 对象
ua = UserAgent()

# 获取随机 UA
print(ua.random)     # 完全随机
print(ua.chrome)     # 随机 Chrome UA
print(ua.firefox)    # 随机 Firefox UA
print(ua.safari)     # 随机 Safari UA

/* Detailed source-code truncated for AI context efficiency. */

GET /api/data HTTP/1.1
Host: www.example.com
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,image/webp,*/*;q=0.8
Accept-Language: zh-CN,zh;q=0.9,en;q=0.8
Accept-Encoding: gzip, deflate, br
Connection: keep-alive
Upgrade-Insecure-Requests: 1
Sec-Fetch-Dest: document
Sec-Fetch-Mode: navigate
Sec-Fetch-Site: none
Sec-Fetch-User: ?1
Cache-Control: max-age=0

/* Detailed source-code truncated for AI context efficiency. */

async def crawl_with_referer(client, list_url: str, detail_urls: list):
    """演示正确设置 Referer"""
    headers_builder = HeadersBuilder()

    # 访问列表页
    list_headers = headers_builder.build()
    response = await client.get(list_url, headers=list_headers)

    # 访问详情页时，Referer 应该是列表页
    for detail_url in detail_urls:
        detail_headers = headers_builder.build(referer=list_url)
        response = await client.get(detail_url, headers=detail_headers)

/* Detailed source-code truncated for AI context efficiency. */

pip install curl_cffi

from curl_cffi import requests

# 模拟 Chrome 浏览器
response = requests.get(
    "https://tls.browserleaks.com/json",
    impersonate="chrome120"  # 模拟 Chrome 120
)
print(response.json())

# 支持的浏览器指纹
# chrome99, chrome100, chrome101, ..., chrome120
# chrome99_android
# edge99, edge101
# safari15_3, safari15_5, safari17_0

from curl_cffi.requests import AsyncSession

async def fetch_with_curl_cffi():
    """使用 curl_cffi 的异步请求"""
    async with AsyncSession(impersonate="chrome120") as session:
        response = await session.get("https://httpbin.org/headers")
        print(response.json())

/* Detailed source-code truncated for AI context efficiency. */

import asyncio
import random


async def crawl_with_delay(urls: list, min_delay: float = 1.0, max_delay: float = 3.0):
    """
    带随机延迟的爬取

    Args:
        urls: URL 列表
        min_delay: 最小延迟（秒）
        max_delay: 最大延迟（秒）
    """
    for url in urls:
        # 发送请求
        response = await fetch(url)

        # 随机延迟
        delay = random.uniform(min_delay, max_delay)
        await asyncio.sleep(delay)

/* Detailed source-code truncated for AI context efficiency. */

import asyncio


class ConcurrencyLimiter:
    """并发限制器"""

    def __init__(self, max_concurrent: int = 10):
        """
        初始化并发限制器

        Args:
            max_concurrent: 最大并发数
        """
        self.semaphore = asyncio.Semaphore(max_concurrent)

    async def run_with_limit(self, coro):
        """在并发限制下执行协程"""
        async with self.semaphore:
            return await coro


async def crawl_concurrent(urls: list, max_concurrent: int = 5):
    """
    带并发限制的批量爬取

    Args:
        urls: URL 列表
        max_concurrent: 最大并发数
    """
    semaphore = asyncio.Semaphore(max_concurrent)

    async def fetch_with_limit(url):
        async with semaphore:
            return await fetch(url)

    # 并发执行，但同时最多 max_concurrent 个请求
    tasks = [fetch_with_limit(url) for url in urls]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    return results

flowchart LR
    response[HTTP响应] --> check{检查状态码}

    check -->|200| success[成功<br/>正常处理数据]
    check -->|401| login[需要认证<br/>登录获取凭证]
    check -->|403| forbidden[禁止访问<br/>检查请求头]
    check -->|429| ratelimit[频率限制<br/>降低速率]
    check -->|404| notfound[资源不存在]

    login --> relogin[获取认证凭证]
    ratelimit --> wait[等待后重试]
    forbidden --> fixheaders[修复请求头]

# error_handler.py - HTTP错误处理
from loguru import logger


async def handle_response(response, url: str):
    """
    处理 HTTP 响应

    Args:
        response: HTTP 响应对象
        url: 请求的 URL
    """
    status_code = response.status_code

    if status_code == 200:
        logger.info(f"请求成功: {url}")
        return response

    elif status_code == 401:
        logger.warning(f"需要认证: {url}")
        raise Exception("需要登录或认证凭证已过期")

    elif status_code == 403:
        logger.error(f"禁止访问: {url}")
        raise Exception("访问被禁止，请检查请求头配置")

    elif status_code == 429:
        logger.warning(f"触发频率限制: {url}")
        raise Exception("请求过于频繁，请降低访问速率")

    elif status_code == 404:
        logger.warning(f"资源不存在: {url}")
        return None

    else:
        logger.error(f"HTTP错误 {status_code}: {url}")
        raise Exception(f"HTTP错误: {status_code}")

/* Detailed source-code truncated for AI context efficiency. */

graph TB
    subgraph ProxyPoolManager["代理池管理器"]
        direction TB

        subgraph Components["核心组件"]
            direction LR
            Fetcher["代理获取器<br/>(Fetcher)"]
            Checker["代理检测器<br/>(Checker)"]
            Allocator["代理分配器<br/>(Allocator)"]
        end

        Storage[("代理存储<br/>(内存 / Redis)")]

        Fetcher --> Storage
        Checker --> Storage
        Allocator --> Storage
    end

    style ProxyPoolManager fill:#e8f4f8,stroke:#0288d1
    style Storage fill:#fff3e0,stroke:#ff9800

flowchart LR
    subgraph 获取阶段
        API["代理API"] --> Fetcher["获取器"]
        Free["免费代理"] --> Fetcher
    end

    subgraph 检测阶段
        Fetcher --> Checker["检测器"]
        Checker -->|有效| Pool[("代理池")]
        Checker -->|无效| Discard["丢弃"]
    end

    subgraph 分配阶段
        Pool --> Allocator["分配器"]
        Allocator --> Crawler["爬虫请求"]
        Crawler -->|成功| Score["评分+1"]
        Crawler -->|失败| Penalty["评分-1"]
        Score --> Pool
        Penalty --> Pool
    end

    style Pool fill:#c8e6c9,stroke:#4caf50
    style Discard fill:#ffcdd2,stroke:#f44336

/* Detailed source-code truncated for AI context efficiency. */

import httpx
from typing import List
from loguru import logger


class FreeProxyFetcher(IProxyFetcher):
    """
    免费代理获取器

    注意：免费代理质量较差，仅供学习测试使用
    """

    async def fetch(self) -> List[ProxyInfo]:
        """从免费代理网站获取代理"""
        proxies = []

        # 示例：从 API 获取（这里用一个示例 API）
        try:
            async with httpx.AsyncClient(timeout=10) as client:
                # 这是一个示例 URL，实际使用时替换为真实的代理 API
                response = await client.get(
                    "https://api.proxyscrape.com/v2/"
                    "?request=getproxies&protocol=http&timeout=10000&country=all"
                )

                if response.status_code == 200:
                    lines = response.text.strip().split("\n")
                    for line in lines:
                        try:
                            host, port = line.strip().split(":")
                            proxies.append(ProxyInfo(
                                host=host,
                                port=int(port),
                                protocol=ProxyProtocol.HTTP
                            ))
                        except ValueError:
                            continue

                logger.info(f"获取到 {len(proxies)} 个免费代理")

        except Exception as e:
            logger.error(f"获取免费代理失败: {e}")

        return proxies

class APIProxyFetcher(IProxyFetcher):
    """
    API 代理获取器

    从付费代理服务商的 API 获取代理
    """

    def __init__(
        self,
        api_url: str,
        api_key: Optional[str] = None,
        count: int = 10
    ):
        """
        初始化 API 代理获取器

        Args:
            api_url: API 地址
            api_key: API 密钥
            count: 每次获取数量
        """
        self.api_url = api_url
        self.api_key = api_key
        self.count = count

    async def fetch(self) -> List[ProxyInfo]:
        """从 API 获取代理"""
        proxies = []

        try:
            params = {"num": self.count}
            if self.api_key:
                params["key"] = self.api_key

            async with httpx.AsyncClient(timeout=10) as client:
                response = await client.get(self.api_url, params=params)
                data = response.json()

                # 根据实际 API 返回格式解析
                # 这里假设返回 {"data": [{"ip": "x.x.x.x", "port": 8080}, ...]}
                for item in data.get("data", []):
                    proxies.append(ProxyInfo(
                        host=item["ip"],
                        port=item["port"],
                        protocol=ProxyProtocol(item.get("protocol", "http"))
                    ))

                logger.info(f"从 API 获取到 {len(proxies)} 个代理")

        except Exception as e:
            logger.error(f"从 API 获取代理失败: {e}")

        return proxies

/* Detailed source-code truncated for AI context efficiency. */

/* Detailed source-code truncated for AI context efficiency. */

import httpx

async def fetch_with_proxy(url: str, proxy_url: str) -> str:
    """使用代理发送请求"""
    async with httpx.AsyncClient(proxies=proxy_url, timeout=30) as client:
        response = await client.get(url)
        return response.text


# 使用示例
proxy = "http://user:[email protected]:8080"
content = await fetch_with_proxy("https://httpbin.org/ip", proxy)

class ProxiedCrawler:
    """
    集成代理池的爬虫

    自动管理代理的获取、轮换和报告
    """

    def __init__(self, proxy_pool: ProxyPool):
        self.proxy_pool = proxy_pool

    async def fetch(self, url: str) -> Optional[str]:
        """使用代理获取页面"""
        proxy = await self.proxy_pool.get_proxy()

        if not proxy:
            logger.warning("无可用代理")
            return None

        try:
            async with httpx.AsyncClient(
                proxies=proxy.url,
                timeout=30
            ) as client:
                response = await client.get(url)
                response.raise_for_status()

                # 报告成功
                await self.proxy_pool.return_proxy(proxy, success=True)
                return response.text

        except Exception as e:
            logger.warning(f"请求失败: {url} - {e}")
            # 报告失败
            await self.proxy_pool.return_proxy(proxy, success=False)
            return None

class TunnelProxyClient:
    """
    隧道代理客户端

    特点：固定入口，自动轮换 IP
    """

    def __init__(
        self,
        host: str,
        port: int,
        username: str,
        password: str
    ):
        self.proxy_url = f"http://{username}:{password}@{host}:{port}"

    async def get(self, url: str, **kwargs) -> httpx.Response:
        """发送请求（自动使用隧道代理）"""
        async with httpx.AsyncClient(
            proxies=self.proxy_url,
            timeout=30
        ) as client:
            return await client.get(url, **kwargs)


# 使用示例
tunnel = TunnelProxyClient(
    host="tunnel.example.com",
    port=12345,
    username="your_username",
    password="your_password"
)

# 每次请求自动使用不同 IP
response1 = await tunnel.get("https://httpbin.org/ip")
response2 = await tunnel.get("https://httpbin.org/ip")

flowchart TD
    Request["发起请求"] --> RateCheck{"频率检测"}
    RateCheck -->|正常| UACheck{"UA检测"}
    RateCheck -->|过快| Block429["429 限流"]

    UACheck -->|正常| BehaviorCheck{"行为检测"}
    UACheck -->|异常| Block412["412 风控"]

    BehaviorCheck -->|正常| Success["正常响应"]
    BehaviorCheck -->|异常| Block403["403 封禁"]

    Block429 --> IPMark["IP标记"]
    Block412 --> IPMark
    Block403 --> IPMark

    IPMark --> BlackList["IP黑名单"]

    style Success fill:#c8e6c9,stroke:#4caf50
    style Block429 fill:#fff3e0,stroke:#ff9800
    style Block412 fill:#ffcdd2,stroke:#f44336
    style Block403 fill:#ffcdd2,stroke:#f44336
    style BlackList fill:#ffcdd2,stroke:#f44336

/* Detailed source-code truncated for AI context efficiency. */

/* Detailed source-code truncated for AI context efficiency. */

graph LR
    subgraph 代理选择
        A1["高匿代理"] --> A2["住宅IP优先"]
        A2 --> A3["国内节点"]
    end

    subgraph 请求策略
        B1["控制频率"] --> B2["随机延迟"]
        B2 --> B3["失败轮换"]
    end

    subgraph 风控规避
        C1["完整请求头"] --> C2["Cookie携带"]
        C2 --> C3["行为模拟"]
    end

    A3 --> B1
    B3 --> C1

    style A1 fill:#e3f2fd,stroke:#2196f3
    style B1 fill:#fff3e0,stroke:#ff9800
    style C1 fill:#e8f5e9,stroke:#4caf50

pip install playwright

# 安装所有支持的浏览器
playwright install

# 或者只安装 Chromium
playwright install chromium

# 安装 Firefox
playwright install firefox

# 安装 WebKit（Safari 内核）
playwright install webkit

from playwright.sync_api import sync_playwright

with sync_playwright() as p:
    browser = p.chromium.launch(headless=False)
    page = browser.new_page()
    page.goto("https://example.com")
    print(page.title())
    browser.close()

graph TB
    subgraph Browser["Browser 浏览器实例"]
        direction TB

        subgraph Context1["Browser Context 1<br/>(登录态环境)"]
            Page1["Page 1<br/>网站首页"]
            Page2["Page 2<br/>详情页面"]
        end

        subgraph Context2["Browser Context 2<br/>(匿名环境)"]
            Page3["Page 3<br/>搜索页面"]
        end
    end

    style Browser fill:#e3f2fd,stroke:#1976d2
    style Context1 fill:#e8f5e9,stroke:#4caf50
    style Context2 fill:#fff3e0,stroke:#ff9800
    style Page1 fill:#f3e5f5,stroke:#9c27b0
    style Page2 fill:#f3e5f5,stroke:#9c27b0
    style Page3 fill:#f3e5f5,stroke:#9c27b0

flowchart LR
    subgraph Browser层
        B["Browser<br/>浏览器进程"]
    end

    subgraph Context层
        C1["Context<br/>Cookie/Storage"]
        C2["Context<br/>独立环境"]
    end

    subgraph Page层
        P1["Page<br/>页面操作"]
        P2["Page<br/>DOM/Network"]
    end

    B --> C1 & C2
    C1 --> P1
    C2 --> P2

    style B fill:#e3f2fd,stroke:#2196f3
    style C1 fill:#e8f5e9,stroke:#4caf50
    style C2 fill:#e8f5e9,stroke:#4caf50
    style P1 fill:#fff3e0,stroke:#ff9800
    style P2 fill:#fff3e0,stroke:#ff9800

# 同步 API
from playwright.sync_api import sync_playwright

with sync_playwright() as p:
    browser = p.chromium.launch()
    page = browser.new_page()
    page.goto("https://example.com")
    browser.close()

# 异步 API
from playwright.async_api import async_playwright
import asyncio

async def main():
    async with async_playwright() as p:
        browser = await p.chromium.launch()
        page = await browser.new_page()
        await page.goto("https://example.com")
        await browser.close()

asyncio.run(main())

# 打开页面
await page.goto("https://example.com")

# 等待特定状态
await page.goto("https://example.com", wait_until="domcontentloaded")  # DOM 加载完成
await page.goto("https://example.com", wait_until="load")              # 页面完全加载
await page.goto("https://example.com", wait_until="networkidle")       # 网络空闲

# 刷新页面
await page.reload()

# 前进/后退
await page.go_back()
await page.go_forward()

# 通过文本定位
page.get_by_text("登录")
page.get_by_text("登录", exact=True)  # 精确匹配

# 通过角色定位（推荐用于按钮、链接等）
page.get_by_role("button", name="提交")
page.get_by_role("link", name="首页")
page.get_by_role("textbox", name="用户名")

# 通过标签定位
page.get_by_label("用户名")
page.get_by_label("密码")

# 通过占位符定位
page.get_by_placeholder("请输入用户名")

# 通过 test-id 定位（需要在 HTML 中添加 data-testid 属性）
page.get_by_test_id("submit-button")

# CSS 选择器
page.locator("div.container")
page.locator("#login-form")
page.locator("input[type='text']")
page.locator("div.item:nth-child(2)")

# XPath
page.locator("//div[@class='container']")
page.locator("//button[contains(text(), '提交')]")
page.locator("//input[@id='username']")

# 链式定位
page.locator("div.container").locator("button.submit")

# 过滤
page.locator("div.item").filter(has_text="特价")
page.locator("div.item").filter(has=page.get_by_role("button"))

# 第 N 个元素
page.locator("div.item").nth(0)  # 第一个
page.locator("div.item").first   # 第一个
page.locator("div.item").last    # 最后一个

# 基本点击
await page.click("button.submit")
await page.locator("button.submit").click()

# 双击
await page.dblclick("div.item")

# 右键点击
await page.click("div.item", button="right")

# 点击位置
await page.click("div.map", position={"x": 100, "y": 200})

# 强制点击（忽略可见性检查）
await page.click("button.hidden", force=True)

# 输入文本
await page.fill("input#username", "myuser")
await page.locator("input#password").fill("mypassword")

# 逐字符输入（模拟真实打字）
await page.type("input#search", "hello world", delay=100)

# 清空后输入
await page.fill("input#search", "")
await page.fill("input#search", "new text")

# 下拉选择（select 元素）
await page.select_option("select#country", "china")
await page.select_option("select#country", value="cn")
await page.select_option("select#country", label="中国")

# 多选
await page.select_option("select#tags", ["tag1", "tag2"])

# 复选框/单选框
await page.check("input#agree")
await page.uncheck("input#newsletter")
await page.set_checked("input#agree", True)

# 按键
await page.keyboard.press("Enter")
await page.keyboard.press("Tab")
await page.keyboard.press("Control+A")
await page.keyboard.press("Control+C")

# 输入文本
await page.keyboard.type("Hello World")

# 组合键
await page.keyboard.down("Shift")
await page.keyboard.press("ArrowDown")
await page.keyboard.up("Shift")

# 移动鼠标
await page.mouse.move(100, 200)

# 点击
await page.mouse.click(100, 200)
await page.mouse.dblclick(100, 200)

# 拖拽
await page.mouse.move(100, 200)
await page.mouse.down()
await page.mouse.move(300, 400)
await page.mouse.up()

# 悬停
await page.hover("div.dropdown")

# 点击时自动等待元素可见、可交互
await page.click("button.submit")

# fill 时自动等待元素可编辑
await page.fill("input#name", "value")

# 等待元素出现
await page.wait_for_selector("div.content")
await page.wait_for_selector("div.content", state="visible")
await page.wait_for_selector("div.content", state="hidden")

# 等待页面状态
await page.wait_for_load_state("networkidle")
await page.wait_for_load_state("domcontentloaded")

# 等待 URL 变化
await page.wait_for_url("**/success")
await page.wait_for_url(lambda url: "success" in url)

# 等待特定时间
await page.wait_for_timeout(1000)  # 等待 1 秒（尽量避免使用）

# 等待函数返回 True
await page.wait_for_function("document.querySelector('.loaded') !== null")

# 全局超时
browser = await p.chromium.launch()
context = await browser.new_context()
page = await context.new_page()
page.set_default_timeout(30000)  # 30 秒

# 单次操作超时
await page.click("button.submit", timeout=5000)
await page.wait_for_selector("div.result", timeout=10000)

# 获取元素文本
text = await page.locator("h1.title").text_content()
texts = await page.locator("div.item").all_text_contents()

# 获取可见文本
visible_text = await page.locator("div.content").inner_text()

# 获取属性
href = await page.locator("a.link").get_attribute("href")
src = await page.locator("img.avatar").get_attribute("src")

# 获取多个元素的属性
hrefs = await page.locator("a").evaluate_all("els => els.map(el => el.href)")

# 获取元素 HTML
html = await page.locator("div.content").inner_html()

# 获取整个页面 HTML
full_html = await page.content()

# 执行 JS 并获取返回值
title = await page.evaluate("document.title")

# 传递参数
result = await page.evaluate("(x, y) => x + y", 1, 2)

# 在元素上执行 JS
text = await page.locator("div.item").evaluate("el => el.textContent")

# 监听请求
def on_request(request):
    print(f"Request: {request.method} {request.url}")

page.on("request", on_request)

# 监听响应
def on_response(response):
    print(f"Response: {response.status} {response.url}")

page.on("response", on_response)

# 等待特定请求
async with page.expect_request("**/api/data") as request_info:
    await page.click("button.load")
request = await request_info.value

# 等待特定响应
async with page.expect_response("**/api/data") as response_info:
    await page.click("button.load")
response = await response_info.value
data = await response.json()

# 拦截并修改请求
async def handle_route(route):
    # 修改请求头
    headers = {**route.request.headers, "X-Custom": "value"}
    await route.continue_(headers=headers)

await page.route("**/*", handle_route)

# 阻止某些资源加载（提高性能）
await page.route("**/*.{png,jpg,jpeg,gif}", lambda route: route.abort())
await page.route("**/analytics.js", lambda route: route.abort())

# 返回模拟响应
async def mock_api(route):
    await route.fulfill(
        status=200,
        content_type="application/json",
        body='{"data": "mocked"}'
    )

await page.route("**/api/data", mock_api)

# 页面截图
await page.screenshot(path="screenshot.png")

# 全页面截图
await page.screenshot(path="full.png", full_page=True)

# 元素截图
await page.locator("div.chart").screenshot(path="chart.png")

# 截图选项
await page.screenshot(
    path="screenshot.png",
    type="png",           # png 或 jpeg
    quality=80,           # jpeg 质量（0-100）
    full_page=True,       # 全页面
    clip={"x": 0, "y": 0, "width": 800, "height": 600}  # 裁剪区域
)

# 导出 PDF（仅 Chromium 支持）
await page.pdf(path="page.pdf")

# PDF 选项
await page.pdf(
    path="page.pdf",
    format="A4",
    print_background=True,
    margin={"top": "1cm", "bottom": "1cm"}
)

# 有头模式（显示浏览器窗口）
browser = await p.chromium.launch(headless=False)

# 无头模式（不显示窗口，默认）
browser = await p.chromium.launch(headless=True)

# 设置视口大小
context = await browser.new_context(
    viewport={"width": 1920, "height": 1080}
)

# 模拟移动设备
from playwright.async_api import async_playwright

async with async_playwright() as p:
    iphone = p.devices["iPhone 13"]
    browser = await p.webkit.launch()
    context = await browser.new_context(**iphone)

# 设置代理
browser = await p.chromium.launch(
    proxy={
        "server": "http://proxy.example.com:8080",
        "username": "user",
        "password": "pass"
    }
)

# 使用持久化的用户数据目录（保存 Cookie 等）
context = await p.chromium.launch_persistent_context(
    user_data_dir="./user_data",
    headless=False
)

flowchart TD
    subgraph 适用场景
        SPA["SPA应用<br/>动态路由"]
        Lazy["懒加载<br/>滚动触发"]
        JS["JS渲染<br/>动态内容"]
    end

    subgraph Playwright优势
        Wait["智能等待<br/>自动判断"]
        Interact["模拟交互<br/>点击/输入"]
        Network["网络拦截<br/>获取API数据"]
    end

    SPA --> Wait
    Lazy --> Interact
    JS --> Network

    style SPA fill:#e3f2fd,stroke:#2196f3
    style Lazy fill:#fff3e0,stroke:#ff9800
    style JS fill:#e8f5e9,stroke:#4caf50

/* Detailed source-code truncated for AI context efficiency. */

/* Detailed source-code truncated for AI context efficiency. */

/* Detailed source-code truncated for AI context efficiency. */

graph LR
    subgraph 反检测
        A1["隐藏自动化特征"]
        A2["模拟真实UA"]
        A3["设置合理视口"]
    end

    subgraph 性能优化
        B1["禁用图片加载"]
        B2["拦截无关资源"]
        B3["复用浏览器实例"]
    end

    subgraph 稳定性
        C1["适当等待时间"]
        C2["处理弹窗"]
        C3["异常重试机制"]
    end

    A1 --> B1 --> C1

    style A1 fill:#e3f2fd,stroke:#2196f3
    style B1 fill:#fff3e0,stroke:#ff9800
    style C1 fill:#e8f5e9,stroke:#4caf50

flowchart TD
    subgraph 检测层级
        L1["Level 1: 基础检测"]
        L2["Level 2: 特征检测"]
        L3["Level 3: 行为检测"]
    end

    subgraph 基础检测项
        D1["navigator.webdriver"]
        D2["window.chrome"]
        D3["navigator.plugins"]
    end

    subgraph 特征检测项
        D4["Canvas指纹"]
        D5["WebGL指纹"]
        D6["Audio指纹"]
    end

    subgraph 行为检测项
        D7["鼠标轨迹"]
        D8["键盘节奏"]
        D9["页面停留"]
    end

    L1 --> D1 & D2 & D3
    L2 --> D4 & D5 & D6
    L3 --> D7 & D8 & D9

    style L1 fill:#e8f5e9,stroke:#4caf50
    style L2 fill:#fff3e0,stroke:#ff9800
    style L3 fill:#ffebee,stroke:#f44336

// 检测 WebDriver
if (navigator.webdriver) {
    console.log("检测到自动化浏览器");
}

// 检测 Chrome 特征
if (!window.chrome) {
    console.log("可能是自动化环境");
}

// 检测 plugins
if (navigator.plugins.length === 0) {
    console.log("插件列表为空，可能是自动化");
}

// 检测 permissions
navigator.permissions.query({name: "notifications"}).then(result => {
    if (result.state === "prompt") {
        // 正常行为
    }
});

import asyncio
from playwright.async_api import async_playwright

# stealth.min.js 脚本内容（实际使用时从文件加载）
STEALTH_JS = """
// 隐藏 webdriver 标志
Object.defineProperty(navigator, 'webdriver', {
    get: () => undefined
});

// 模拟 chrome 对象
window.chrome = {
    runtime: {}
};

// 模拟 plugins
Object.defineProperty(navigator, 'plugins', {
    get: () => [1, 2, 3, 4, 5]
});

// 模拟 languages
Object.defineProperty(navigator, 'languages', {
    get: () => ['zh-CN', 'zh', 'en']
});

// 修复 permissions
const originalQuery = window.navigator.permissions.query;
window.navigator.permissions.query = (parameters) => (
    parameters.name === 'notifications' ?
        Promise.resolve({ state: Notification.permission }) :
        originalQuery(parameters)
);
"""


async def create_stealth_page(browser):
    """创建带反检测的页面"""
    context = await browser.new_context()

    # 在每个新页面创建时注入 stealth 脚本
    await context.add_init_script(STEALTH_JS)

    page = await context.new_page()
    return page


async def main():
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=True)
        page = await create_stealth_page(browser)

        # 测试反检测效果
        await page.goto("https://bot.sannysoft.com/")
        await page.screenshot(path="stealth_test.png", full_page=True)

        await browser.close()


asyncio.run(main())

async def create_stealth_context(browser, stealth_js_path: str):
    """创建带完整反检测的上下文"""
    context = await browser.new_context()

    # 从文件加载 stealth.js
    with open(stealth_js_path, "r") as f:
        stealth_js = f.read()

    await context.add_init_script(stealth_js)
    return context

import asyncio
from playwright.async_api import async_playwright


async def connect_existing_browser():
    """连接到已经运行的浏览器"""
    async with async_playwright() as p:
        # 首先启动一个带调试端口的浏览器
        # chrome --remote-debugging-port=9222

        # 通过 CDP 连接
        browser = await p.chromium.connect_over_cdp(
            "http://localhost:9222"
        )

        # 获取所有页面
        contexts = browser.contexts
        if contexts:
            pages = contexts[0].pages
            if pages:
                page = pages[0]
                print(f"当前页面: {page.url}")

        await browser.close()

async def use_cdp_commands(page):
    """直接使用 CDP 命令"""
    # 获取 CDP session
    client = await page.context.new_cdp_session(page)

    # 执行 CDP 命令
    # 例如：获取性能指标
    metrics = await client.send("Performance.getMetrics")
    print(metrics)

    # 例如：模拟网络条件
    await client.send("Network.emulateNetworkConditions", {
        "offline": False,
        "downloadThroughput": 1000000,  # 1 MB/s
        "uploadThroughput": 500000,
        "latency": 100
    })

    # 例如：截取完整页面
    result = await client.send("Page.captureScreenshot", {
        "format": "png",
        "captureBeyondViewport": True
    })

async def create_optimized_context(browser):
    """创建优化性能的上下文"""
    context = await browser.new_context()

    # 拦截并阻止不必要的资源
    await context.route("**/*.{png,jpg,jpeg,gif,svg,ico}", lambda route: route.abort())
    await context.route("**/*.{woff,woff2,ttf,otf}", lambda route: route.abort())
    await context.route("**/*.css", lambda route: route.abort())

    # 阻止分析和追踪脚本
    await context.route("**/analytics.js", lambda route: route.abort())
    await context.route("**/gtag/**", lambda route: route.abort())
    await context.route("**/facebook.com/**", lambda route: route.abort())

    return context


async def selective_blocking(page):
    """选择性阻止资源"""
    async def handle_route(route):
        resource_type = route.request.resource_type

        # 只允许 document、script、xhr、fetch
        if resource_type in ["document", "script", "xhr", "fetch"]:
            await route.continue_()
        else:
            await route.abort()

    await page.route("**/*", handle_route)

class BrowserPool:
    """浏览器上下文池"""

    def __init__(self, browser, pool_size: int = 5):
        self.browser = browser
        self.pool_size = pool_size
        self._contexts = []
        self._available = asyncio.Queue()

    async def initialize(self):
        """初始化上下文池"""
        for _ in range(self.pool_size):
            context = await self.browser.new_context()
            self._contexts.append(context)
            await self._available.put(context)

    async def get_context(self):
        """获取可用上下文"""
        return await self._available.get()

    async def return_context(self, context):
        """归还上下文"""
        # 清理 cookies 和 storage
        await context.clear_cookies()
        await self._available.put(context)

    async def close_all(self):
        """关闭所有上下文"""
        for context in self._contexts:
            await context.close()

import asyncio


class ConcurrentCrawler:
    """并发爬虫"""

    def __init__(self, browser, max_concurrent: int = 5):
        self.browser = browser
        self.max_concurrent = max_concurrent
        self.semaphore = asyncio.Semaphore(max_concurrent)

    async def crawl_url(self, context, url: str) -> dict:
        """爬取单个 URL"""
        async with self.semaphore:
            page = await context.new_page()
            try:
                await page.goto(url, wait_until="domcontentloaded")
                title = await page.title()
                return {"url": url, "title": title, "success": True}
            except Exception as e:
                return {"url": url, "error": str(e), "success": False}
            finally:
                await page.close()

    async def crawl_batch(self, urls: list) -> list:
        """批量爬取"""
        context = await self.browser.new_context()
        try:
            tasks = [self.crawl_url(context, url) for url in urls]
            return await asyncio.gather(*tasks)
        finally:
            await context.close()

async def monitor_resources(browser):
    """监控浏览器资源使用"""
    # 获取所有上下文
    contexts = browser.contexts
    print(f"活跃上下文数: {len(contexts)}")

    total_pages = 0
    for ctx in contexts:
        pages = ctx.pages
        total_pages += len(pages)
        for page in pages:
            print(f"  页面: {page.url[:50]}...")

    print(f"总页面数: {total_pages}")


async def cleanup_stale_pages(context, max_age_seconds: int = 300):
    """清理闲置页面"""
    # 这里需要自己记录页面的创建时间
    # 示例仅展示逻辑
    for page in context.pages:
        # 关闭空白页或闲置页
        if page.url == "about:blank":
            await page.close()

async def safe_navigate(page, url: str, max_retries: int = 3):
    """安全的页面导航，带重试"""
    for attempt in range(max_retries):
        try:
            await page.goto(url, timeout=30000)
            return True
        except Exception as e:
            error_msg = str(e).lower()

            if "crash" in error_msg or "target closed" in error_msg:
                # 页面崩溃，需要重新创建
                print(f"页面崩溃，尝试恢复 (尝试 {attempt + 1}/{max_retries})")
                # 这里应该创建新页面
                continue

            elif "timeout" in error_msg:
                # 超时，可以重试
                print(f"超时，重试 (尝试 {attempt + 1}/{max_retries})")
                continue

            else:
                # 其他错误
                print(f"导航错误: {e}")
                raise

    return False

import signal
import atexit


class BrowserManager:
    """浏览器进程管理器"""

    def __init__(self):
        self._browser = None
        self._playwright = None

        # 注册清理函数
        atexit.register(self._cleanup_sync)
        signal.signal(signal.SIGTERM, self._signal_handler)
        signal.signal(signal.SIGINT, self._signal_handler)

    async def start(self, playwright):
        """启动浏览器"""
        self._playwright = playwright
        self._browser = await playwright.chromium.launch(headless=True)
        return self._browser

    async def stop(self):
        """停止浏览器"""
        if self._browser:
            await self._browser.close()
            self._browser = None

    def _cleanup_sync(self):
        """同步清理"""
        if self._browser:
            # 强制关闭
            try:
                import asyncio
                loop = asyncio.get_event_loop()
                loop.run_until_complete(self.stop())
            except Exception:
                pass

    def _signal_handler(self, signum, frame):
        """信号处理"""
        print(f"收到信号 {signum}，正在清理...")
        self._cleanup_sync()
        exit(0)

flowchart LR
    subgraph 检测机制
        Check1["WebDriver检测"]
        Check2["请求头验证"]
        Check3["行为分析"]
        Check4["频率限制"]
    end

    subgraph 检测结果
        Pass["正常访问"]
        Block["触发风控"]
        Captcha["触发验证码"]
    end

    Check1 -->|通过| Check2
    Check1 -->|失败| Block
    Check2 -->|通过| Check3
    Check2 -->|异常| Block
    Check3 -->|正常| Pass
    Check3 -->|异常| Captcha
    Check4 -->|超限| Block

    style Pass fill:#c8e6c9,stroke:#4caf50
    style Block fill:#ffcdd2,stroke:#f44336
    style Captcha fill:#fff3e0,stroke:#ff9800

/* Detailed source-code truncated for AI context efficiency. */

/* Detailed source-code truncated for AI context efficiency. */

flowchart LR
    Start["启动浏览器"] --> Inject["注入stealth.js"]
    Inject --> Visit["访问测试网站"]
    Visit --> Check{"检测验证"}

    Check -->|webdriver=undefined| Pass1["✓ 通过"]
    Check -->|chrome存在| Pass2["✓ 通过"]
    Check -->|plugins正常| Pass3["✓ 通过"]

    Pass1 & Pass2 & Pass3 --> Result["反检测成功"]
    Result --> Crawl["正常爬取"]

    style Result fill:#c8e6c9,stroke:#4caf50
    style Crawl fill:#e3f2fd,stroke:#2196f3

sequenceDiagram
    participant Client as 客户端
    participant Server as 服务器

    Client->>Server: HTTP 请求 (首次访问)
    Server-->>Client: Set-Cookie: session=xxx
    Note over Client: 保存 Cookie 到本地

    Client->>Server: HTTP 请求 + Cookie: session=xxx
    Server-->>Client: 返回用户数据
    Note over Server: 识别用户身份

# Cookie 属性示例
cookie_example = {
    "name": "session_id",
    "value": "abc123xyz",
    "domain": ".example.com",      # 包含子域名
    "path": "/",                   # 所有路径
    "expires": 1735689600,         # Unix 时间戳
    "httpOnly": True,              # 不能通过 JS 访问
    "secure": True,                # 仅 HTTPS
    "sameSite": "Lax"              # 跨站限制
}

# 服务端 Session 工作流程（伪代码）
class SessionFlow:
    """
    1. 用户登录 -> 服务器创建 Session
    2. 服务器返回 Session ID (通过 Set-Cookie)
    3. 浏览器携带 Session ID 访问
    4. 服务器根据 Session ID 查找用户状态
    """

    def login(self, username, password):
        if self.verify(username, password):
            session_id = generate_session_id()
            self.sessions[session_id] = {
                "user_id": user.id,
                "login_time": time.time(),
                "expires": time.time() + 3600 * 24
            }
            return session_id
        return None

# Token 认证示例
import httpx

headers = {
    "Authorization": "Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."
}

async with httpx.AsyncClient() as client:
    resp = await client.get("https://api.example.com/data", headers=headers)

# 手动提取的 Cookie 使用示例
import httpx

cookies = {
    "session_id": "abc123",
    "user_token": "xyz789",
    "preferences": "theme=dark"
}

async with httpx.AsyncClient(cookies=cookies) as client:
    resp = await client.get("https://example.com/user/profile")

from playwright.async_api import async_playwright
import json

async def extract_cookies_after_login():
    """登录后自动提取 Cookie"""
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=False)  # 有头模式便于手动登录
        context = await browser.new_context()
        page = await context.new_page()

        # 导航到登录页
        await page.goto("https://example.com/login")

        # 等待用户手动登录（或自动填写表单）
        # await page.fill("#username", "your_username")
        # await page.fill("#password", "your_password")
        # await page.click("#login-button")

        # 等待登录成功（检测特定元素或 URL 变化）
        await page.wait_for_url("**/dashboard**", timeout=120000)

        # 提取所有 Cookie
        cookies = await context.cookies()

        # 保存到文件
        with open("cookies.json", "w") as f:
            json.dump(cookies, f, indent=2)

        print(f"提取了 {len(cookies)} 个 Cookie")

        await browser.close()
        return cookies

import json
from http.cookiejar import MozillaCookieJar
from datetime import datetime

class CookieSerializer:
    """Cookie 序列化工具"""

    @staticmethod
    def to_json(cookies: list, filepath: str):
        """保存为 JSON 格式"""
        with open(filepath, "w", encoding="utf-8") as f:
            json.dump(cookies, f, indent=2, ensure_ascii=False)

    @staticmethod
    def from_json(filepath: str) -> list:
        """从 JSON 加载"""
        with open(filepath, "r", encoding="utf-8") as f:
            return json.load(f)

    @staticmethod
    def to_netscape(cookies: list, filepath: str):
        """保存为 Netscape 格式（兼容 curl/wget）"""
        lines = ["# Netscape HTTP Cookie File"]
        for c in cookies:
            # 格式: domain, flag, path, secure, expiry, name, value
            domain = c.get("domain", "")
            flag = "TRUE" if domain.startswith(".") else "FALSE"
            path = c.get("path", "/")
            secure = "TRUE" if c.get("secure", False) else "FALSE"
            expiry = str(int(c.get("expires", 0)))
            name = c.get("name", "")
            value = c.get("value", "")

            lines.append(f"{domain}\t{flag}\t{path}\t{secure}\t{expiry}\t{name}\t{value}")

        with open(filepath, "w") as f:
            f.write("\n".join(lines))

    @staticmethod
    def to_dict(cookies: list) -> dict:
        """转换为简单字典格式（name: value）"""
        return {c["name"]: c["value"] for c in cookies}

import json
import base64
from cryptography.fernet import Fernet

class SecureCookieStorage:
    """加密 Cookie 存储"""

    def __init__(self, key: bytes = None):
        # 如果没有提供密钥，生成新密钥
        self.key = key or Fernet.generate_key()
        self.cipher = Fernet(self.key)

    def save_key(self, filepath: str):
        """保存密钥（请妥善保管）"""
        with open(filepath, "wb") as f:
            f.write(self.key)

    @classmethod
    def load_key(cls, filepath: str) -> "SecureCookieStorage":
        """从文件加载密钥"""
        with open(filepath, "rb") as f:
            return cls(f.read())

    def encrypt_cookies(self, cookies: list, filepath: str):
        """加密并保存 Cookie"""
        data = json.dumps(cookies).encode("utf-8")
        encrypted = self.cipher.encrypt(data)
        with open(filepath, "wb") as f:
            f.write(encrypted)

    def decrypt_cookies(self, filepath: str) -> list:
        """解密并加载 Cookie"""
        with open(filepath, "rb") as f:
            encrypted = f.read()
        decrypted = self.cipher.decrypt(encrypted)
        return json.loads(decrypted.decode("utf-8"))

import httpx

# 方式1: 直接传入字典
cookies_dict = {"session": "abc123", "token": "xyz"}
async with httpx.AsyncClient(cookies=cookies_dict) as client:
    resp = await client.get("https://example.com/api")

# 方式2: 使用 Cookies 对象
from httpx import Cookies

cookies = Cookies()
cookies.set("session", "abc123", domain="example.com")
cookies.set("token", "xyz", domain="example.com")

async with httpx.AsyncClient(cookies=cookies) as client:
    resp = await client.get("https://example.com/api")

# 方式3: 从响应中自动获取
async with httpx.AsyncClient() as client:
    # 登录请求会设置 Cookie
    login_resp = await client.post(
        "https://example.com/login",
        data={"username": "user", "password": "pass"}
    )
    # 后续请求自动携带 Cookie
    profile_resp = await client.get("https://example.com/profile")

from playwright.async_api import async_playwright
import json

async def inject_cookies_and_browse():
    """注入 Cookie 并访问页面"""
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=True)
        context = await browser.new_context()

        # 从文件加载 Cookie
        with open("cookies.json", "r") as f:
            cookies = json.load(f)

        # 注入 Cookie
        await context.add_cookies(cookies)

        # 现在可以直接访问需要登录的页面
        page = await context.new_page()
        await page.goto("https://example.com/dashboard")

        # 检查是否登录成功
        if await page.locator(".user-avatar").is_visible():
            print("Cookie 注入成功，已登录")
        else:
            print("Cookie 可能已过期")

        await browser.close()

def playwright_cookies_to_httpx(playwright_cookies: list) -> dict:
    """
    将 Playwright 格式的 Cookie 转换为 httpx 格式

    Playwright 格式: [{"name": "x", "value": "y", "domain": "...", ...}, ...]
    httpx 格式: {"name": "value", ...}
    """
    return {c["name"]: c["value"] for c in playwright_cookies}

def httpx_cookies_to_playwright(cookies_dict: dict, domain: str) -> list:
    """
    将简单字典格式转换为 Playwright 格式
    """
    return [
        {
            "name": name,
            "value": value,
            "domain": domain,
            "path": "/"
        }
        for name, value in cookies_dict.items()
    ]

/* Detailed source-code truncated for AI context efficiency. */

/* Detailed source-code truncated for AI context efficiency. */

import time
from datetime import datetime, timedelta
from typing import List, Tuple

class CookieExpiryMonitor:
    """Cookie 过期监控"""

    @staticmethod
    def check_expiry(cookies: list) -> Tuple[bool, List[dict]]:
        """
        检查 Cookie 是否过期

        Returns:
            (是否全部有效, 已过期的 Cookie 列表)
        """
        now = time.time()
        expired = []

        for cookie in cookies:
            expires = cookie.get("expires", 0)
            if expires > 0 and expires < now:
                expired.append(cookie)

        return len(expired) == 0, expired

    @staticmethod
    def get_earliest_expiry(cookies: list) -> datetime:
        """获取最早过期的时间"""
        expiry_times = [
            c.get("expires", float("inf"))
            for c in cookies
            if c.get("expires", 0) > 0
        ]

        if not expiry_times:
            return datetime.max

        return datetime.fromtimestamp(min(expiry_times))

    @staticmethod
    def will_expire_soon(cookies: list, threshold_hours: int = 1) -> bool:
        """检查 Cookie 是否即将过期"""
        earliest = CookieExpiryMonitor.get_earliest_expiry(cookies)
        return earliest < datetime.now() + timedelta(hours=threshold_hours)

/* Detailed source-code truncated for AI context efficiency. */

graph TB
    subgraph B站核心Cookie
        SESSDATA["SESSDATA<br/>会话凭证"]
        DedeUserID["DedeUserID<br/>用户ID"]
        bili_jct["bili_jct<br/>CSRF Token"]
    end

    subgraph 辅助Cookie
        buvid3["buvid3<br/>设备标识"]
        buvid4["buvid4<br/>设备标识"]
        sid["sid<br/>短会话ID"]
    end

    SESSDATA --> API["API请求认证"]
    DedeUserID --> API
    bili_jct --> POST["POST请求CSRF"]

    style SESSDATA fill:#e8f5e9,stroke:#4caf50
    style DedeUserID fill:#e8f5e9,stroke:#4caf50
    style bili_jct fill:#fff3e0,stroke:#ff9800

/* Detailed source-code truncated for AI context efficiency. */

/* Detailed source-code truncated for AI context efficiency. */

# -*- coding: utf-8 -*-
"""
B站 Cookie 与 Playwright 集成
"""

import asyncio
from playwright.async_api import async_playwright
from loguru import logger


async def bilibili_with_playwright():
    """使用 Playwright 注入 B站 Cookie"""

    # 加载 Cookie
    manager = BilibiliCookieManager("data/bilibili_cookies.json")
    await manager.load()

    if not manager.cookies or not manager.cookies.is_valid():
        logger.error("Cookie 无效")
        return

    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=True)
        context = await browser.new_context(
            viewport={"width": 1920, "height": 1080},
            locale="zh-CN"
        )

        # 注入 Cookie
        await context.add_cookies(
            manager.cookies.to_playwright_cookies()
        )

        page = await context.new_page()

        # 访问个人主页
        await page.goto("https://space.bilibili.com/", wait_until="networkidle")

        # 检查是否登录成功
        avatar = page.locator(".bili-avatar")
        if await avatar.count() > 0:
            logger.info("Cookie 注入成功，已登录")

            # 获取用户名
            username = await page.locator(".h-name").text_content()
            print(f"当前用户: {username}")
        else:
            logger.warning("Cookie 可能已过期")

        await browser.close()


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

/* Detailed source-code truncated for AI context efficiency. */

graph LR
    subgraph 本章知识点
        A1["Cookie结构"]
        A2["Cookie管理器"]
        A3["登录验证"]
    end

    subgraph 第11章实战应用
        B1["扫码登录"]
        B2["登录态保持"]
        B3["API请求认证"]
    end

    A1 --> B1
    A2 --> B2
    A3 --> B3

    style A1 fill:#e3f2fd,stroke:#2196f3
    style A2 fill:#e3f2fd,stroke:#2196f3
    style A3 fill:#e3f2fd,stroke:#2196f3
    style B1 fill:#c8e6c9,stroke:#4caf50
    style B2 fill:#c8e6c9,stroke:#4caf50
    style B3 fill:#c8e6c9,stroke:#4caf50

sequenceDiagram
    participant PC as PC浏览器
    participant Server as 服务器
    participant APP as 手机APP

    PC->>Server: 1. 请求二维码
    Server-->>PC: 2. 返回二维码+UUID

    loop 轮询登录状态
        PC->>Server: 3. 查询状态
        Server-->>PC: 状态: 待扫描
    end

    APP->>Server: 4. 用户扫码
    Server-->>PC: 5. 状态: 已扫描

    APP->>Server: 6. 用户确认
    Server-->>PC: 7. 返回登录凭证(Cookie)

    Note over PC: 8. 登录成功!

from enum import Enum

class QRCodeStatus(Enum):
    """二维码状态枚举"""
    WAITING = "waiting"      # 等待扫描
    SCANNED = "scanned"      # 已扫描，等待确认
    CONFIRMED = "confirmed"  # 已确认登录
    EXPIRED = "expired"      # 二维码已过期
    CANCELED = "canceled"    # 用户取消

/* Detailed source-code truncated for AI context efficiency. */

/* Detailed source-code truncated for AI context efficiency. */

try:
    import qrcode
    HAS_QRCODE = True
except ImportError:
    HAS_QRCODE = False

def display_qrcode_in_terminal(data: str):
    """在终端显示二维码"""
    if not HAS_QRCODE:
        print("提示: 安装 qrcode 库可在终端显示二维码: pip install qrcode")
        return

    qr = qrcode.QRCode(
        version=1,
        error_correction=qrcode.constants.ERROR_CORRECT_L,
        box_size=1,
        border=1
    )
    qr.add_data(data)
    qr.make(fit=True)

    # 使用字符绘制
    qr.print_ascii(invert=True)

def display_qrcode_image_in_terminal(image_path: str):
    """
    将图片二维码转换为终端可显示的文本
    需要安装: pip install pillow
    """
    try:
        from PIL import Image

        img = Image.open(image_path)
        img = img.convert('L')  # 转为灰度

        # 缩小图片
        width = 60
        ratio = width / img.width
        height = int(img.height * ratio * 0.5)  # 0.5 补偿终端字符高宽比
        img = img.resize((width, height))

        # 转换为 ASCII
        chars = " .:-=+*#%@"
        pixels = img.getdata()
        ascii_img = ""
        for i, pixel in enumerate(pixels):
            if i > 0 and i % width == 0:
                ascii_img += "\n"
            ascii_img += chars[pixel * len(chars) // 256]

        print(ascii_img)
    except ImportError:
        print(f"二维码已保存到: {image_path}")

sequenceDiagram
    participant User as 用户
    participant Frontend as 前端
    participant Server as 服务器

    User->>Frontend: 1. 输入手机号
    Frontend->>Server: 2. 请求发送验证码
    Server-->>Frontend: 3. 返回发送结果

    User->>Frontend: 4. 输入验证码
    Frontend->>Server: 5. 提交登录请求
    Server-->>Frontend: 6. 返回登录结果+凭证

    Note over User,Server: 登录成功，获取Cookie

/* Detailed source-code truncated for AI context efficiency. */

import asyncio

async def get_code_from_user() -> str:
    """从控制台获取用户输入的验证码"""
    print("\n请输入收到的验证码: ", end="", flush=True)
    # 在异步环境中等待用户输入
    loop = asyncio.get_event_loop()
    code = await loop.run_in_executor(None, input)
    return code.strip()


class SMSCodeReceiver:
    """
    短信接码平台接口（示意）

    注意：实际使用需要接入具体的接码平台 API
    """

    def __init__(self, api_key: str, api_url: str):
        self.api_key = api_key
        self.api_url = api_url

    async def get_phone_number(self) -> str:
        """获取手机号"""
        # 调用接码平台 API 获取手机号
        raise NotImplementedError("需要实现具体的接码平台接口")

    async def wait_for_code(self, phone: str, timeout: int = 60) -> Optional[str]:
        """等待接收验证码"""
        # 轮询接码平台获取验证码
        raise NotImplementedError("需要实现具体的接码平台接口")

    async def release_phone(self, phone: str):
        """释放手机号"""
        raise NotImplementedError("需要实现具体的接码平台接口")

from abc import ABC, abstractmethod
from typing import Optional, Callable, Awaitable
from enum import Enum

class LoginMethod(Enum):
    """登录方式枚举"""
    QRCODE = "qrcode"           # 扫码登录
    SMS = "sms"                 # 短信验证码
    PASSWORD = "password"       # 账号密码
    COOKIE = "cookie"           # Cookie 注入

class LoginResult:
    """登录结果"""

    def __init__(
        self,
        success: bool,
        cookies: list = None,
        error: str = None
    ):
        self.success = success
        self.cookies = cookies or []
        self.error = error

class BaseLogin(ABC):
    """登录基类"""

    @abstractmethod
    async def login(self) -> LoginResult:
        """执行登录"""
        pass

    @abstractmethod
    async def close(self):
        """清理资源"""
        pass

class LoginFactory:
    """登录工厂"""

    _registry = {}

    @classmethod
    def register(cls, method: LoginMethod):
        """注册登录实现"""
        def decorator(login_class):
            cls._registry[method] = login_class
            return login_class
        return decorator

    @classmethod
    def create(
        cls,
        method: LoginMethod,
        **kwargs
    ) -> BaseLogin:
        """创建登录实例"""
        if method not in cls._registry:
            raise ValueError(f"不支持的登录方式: {method}")
        return cls._registry[method](**kwargs)

/* Detailed source-code truncated for AI context efficiency. */

/* Detailed source-code truncated for AI context efficiency. */

import asyncio
from playwright.async_api import async_playwright
import json

async def qrcode_login_demo():
    """扫码登录完整示例"""

    # 二维码准备好后的回调
    async def on_qrcode_ready(path: str):
        print(f"\n{'='*40}")
        print(f"请使用手机扫描二维码: {path}")
        print(f"{'='*40}\n")
        # 可以在这里显示二维码到终端
        display_qrcode_image_in_terminal(path)

    async with async_playwright() as p:
        # 创建扫码登录实例（以示例网站为例）
        qr_login = QRCodeLogin(
            login_url="https://example.com/login",
            qrcode_selector="img.qrcode",  # 二维码选择器
            success_url_pattern="/dashboard",
            timeout=120
        )

        await qr_login.start(p, headless=False)

        try:
            cookies = await qr_login.login(on_qrcode_ready)

            if cookies:
                # 保存 Cookie
                with open("login_cookies.json", "w") as f:
                    json.dump(cookies, f, indent=2)
                print(f"登录成功！获取到 {len(cookies)} 个 Cookie")
            else:
                print("登录失败或超时")
        finally:
            await qr_login.close()


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

async def multi_platform_demo():
    """多平台登录管理示例"""

    # 配置多个平台
    platforms = {
        "platform_a": {
            "cookie_path": "data/platform_a_cookies.json",
            "login_url": "https://a.example.com/login",
            "qrcode_selector": "#qrcode-img",
            "success_url": "/home"
        },
        "platform_b": {
            "cookie_path": "data/platform_b_cookies.json",
            "login_url": "https://b.example.com/login",
            "qrcode_selector": ".login-qrcode",
            "success_url": "/dashboard"
        }
    }

    for name, config in platforms.items():
        print(f"\n处理平台: {name}")

        manager = LoginManager(
            platform=name,
            cookie_path=config["cookie_path"],
            preferred_method=LoginMethod.QRCODE
        )

        # 定义二维码回调
        async def on_qrcode(path: str):
            print(f"[{name}] 请扫描二维码: {path}")

        success = await manager.ensure_login(
            login_url=config["login_url"],
            qrcode_selector=config["qrcode_selector"],
            success_url_pattern=config["success_url"],
            on_qrcode_ready=on_qrcode
        )

        if success:
            cookies = manager.get_cookies_dict()
            print(f"[{name}] 登录成功，Cookie 数量: {len(cookies)}")

login/
├── __init__.py
├── base.py           # 基类定义
├── factory.py        # 登录工厂
├── qrcode.py         # 扫码登录实现
├── sms.py            # 短信登录实现
├── cookie.py         # Cookie 登录实现
└── manager.py        # 统一管理器

sequenceDiagram
    participant PC as PC浏览器
    participant Server as B站服务器
    participant APP as B站APP

    PC->>Server: 1. GET /qrcode/generate
    Server-->>PC: 2. 返回 qrcode_key + url

    Note over PC: 生成二维码图片

    loop 轮询登录状态 (间隔2秒)
        PC->>Server: 3. GET /qrcode/poll?qrcode_key=xxx
        Server-->>PC: 4. 返回状态码
        Note right of Server: 86101=未扫描<br/>86090=已扫描<br/>86038=已过期<br/>0=已确认
    end

    APP->>Server: 5. 用户扫码
    Server-->>PC: 6. 状态: 86090 已扫描

    APP->>Server: 7. 用户确认登录
    Server-->>PC: 8. 状态: 0 + Set-Cookie

    Note over PC: 9. 登录成功！保存Cookie

/* Detailed source-code truncated for AI context efficiency. */

import httpx
import json
from pathlib import Path


async def use_bilibili_cookies():
    """使用已保存的Cookie访问B站API"""

    # 1. 加载Cookie
    cookie_path = Path("data/bilibili_cookies.json")
    if not cookie_path.exists():
        print("请先执行扫码登录获取Cookie")
        return

    with open(cookie_path) as f:
        cookies = json.load(f)

    # 2. 创建带Cookie的客户端
    async with httpx.AsyncClient(
        cookies=cookies,
        headers={
            "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
                          "AppleWebKit/537.36 (KHTML, like Gecko) "
                          "Chrome/120.0.0.0 Safari/537.36",
            "Referer": "https://www.bilibili.com/",
        }
    ) as client:
        # 3. 获取用户信息
        resp = await client.get(
            "https://api.bilibili.com/x/web-interface/nav"
        )
        data = resp.json()

        if data["code"] == 0:
            user_info = data["data"]
            print(f"登录用户: {user_info['uname']}")
            print(f"用户ID: {user_info['mid']}")
            print(f"等级: LV{user_info['level_info']['current_level']}")
            print(f"硬币: {user_info['money']}")
        else:
            print(f"获取用户信息失败: {data['message']}")


if __name__ == "__main__":
    import asyncio
    asyncio.run(use_bilibili_cookies())

/* Detailed source-code truncated for AI context efficiency. */

graph LR
    A[第06章<br/>Cookie管理] --> B[第07章<br/>扫码登录]
    B --> C[第08章<br/>验证码处理]
    C --> D[第11章<br/>综合实战]

    style B fill:#e1f5fe
    style D fill:#fff9c4

sequenceDiagram
    participant User as 用户
    participant Frontend as 前端
    participant Server as 服务器

    User->>Frontend: 1. 请求验证码
    Frontend->>Server: 2. 请求验证码
    Server-->>Frontend: 3. 返回验证码图片+ID
    Frontend-->>User: 4. 显示验证码

    User->>Frontend: 5. 输入验证码
    Frontend->>Server: 6. 提交验证
    Server-->>Frontend: 7. 返回验证结果

    Note over User,Server: 验证通过后继续操作

pip install ddddocr

import ddddocr

# 创建 OCR 实例
ocr = ddddocr.DdddOcr()

# 读取验证码图片
with open("captcha.png", "rb") as f:
    image_bytes = f.read()

# 识别
result = ocr.classification(image_bytes)
print(f"识别结果: {result}")

/* Detailed source-code truncated for AI context efficiency. */

/* Detailed source-code truncated for AI context efficiency. */

/* Detailed source-code truncated for AI context efficiency. */

/* Detailed source-code truncated for AI context efficiency. */

/* Detailed source-code truncated for AI context efficiency. */

/* Detailed source-code truncated for AI context efficiency. */

/* Detailed source-code truncated for AI context efficiency. */

class CostController:
    """打码成本控制器"""

    def __init__(
        self,
        daily_budget: float,
        cost_per_captcha: float = 0.01
    ):
        self.daily_budget = daily_budget
        self.cost_per_captcha = cost_per_captcha
        self._daily_spent = 0.0
        self._last_reset = None

    def can_use_service(self) -> bool:
        """是否可以使用打码服务"""
        self._check_reset()
        return self._daily_spent < self.daily_budget

    def record_usage(self):
        """记录一次使用"""
        self._daily_spent += self.cost_per_captcha

    def _check_reset(self):
        """检查是否需要重置"""
        from datetime import date
        today = date.today()
        if self._last_reset != today:
            self._daily_spent = 0.0
            self._last_reset = today

    @property
    def remaining_budget(self) -> float:
        """剩余预算"""
        self._check_reset()
        return max(0, self.daily_budget - self._daily_spent)

import asyncio
from playwright.async_api import async_playwright

async def demo_captcha_solving():
    """验证码处理完整演示"""

    # 初始化 OCR 识别器
    ocr_solver = OCRCaptchaSolver(preprocess=True)

    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=False)
        page = await browser.new_page()

        await page.goto("https://example.com/login")

        # 获取验证码图片
        captcha_element = page.locator("#captcha-image")
        captcha_bytes = await captcha_element.screenshot()

        # 识别验证码
        code = ocr_solver.solve(captcha_bytes)
        print(f"识别结果: {code}")

        # 输入验证码
        if code:
            await page.fill("#captcha-input", code)
            await page.click("#submit-btn")

        await asyncio.sleep(3)
        await browser.close()

flowchart TD
    request[发起请求] --> check{风控检测}
    check -->|正常| success[返回数据]
    check -->|异常| trigger{触发验证码}

    trigger -->|高频请求| slider[滑块验证码]
    trigger -->|IP异常| geetest[极验验证码]
    trigger -->|敏感操作| click[点选验证码]

    slider --> verify{验证}
    geetest --> verify
    click --> verify

    verify -->|成功| success
    verify -->|失败| block[临时封禁]

/* Detailed source-code truncated for AI context efficiency. */

flowchart LR
    subgraph 预防策略
        A[控制请求频率] --> B[模拟真实行为]
        B --> C[使用登录态]
        C --> D[IP轮换]
    end

    subgraph 建议配置
        E[每分钟<30请求]
        F[随机延迟2-5秒]
        G[保持Cookie有效]
        H[高匿代理池]
    end

    A --> E
    B --> F
    C --> G
    D --> H

/* Detailed source-code truncated for AI context efficiency. */

flowchart LR
    raw[原始数据] --> format[格式清洗]
    format --> content[内容清洗]
    content --> dedup[去重处理]
    dedup --> normalize[标准化]
    normalize --> clean[干净数据]

    subgraph 格式清洗
        html[HTML标签移除]
        encode[编码修复]
    end

    subgraph 内容清洗
        whitespace[空白处理]
        special[特殊字符]
    end

from dataclasses import dataclass
from typing import List, Dict, Any

@dataclass
class DataQualityReport:
    """数据质量报告"""
    total_records: int
    valid_records: int
    duplicate_count: int
    missing_fields: Dict[str, int]
    format_issues: Dict[str, int]

    @property
    def validity_rate(self) -> float:
        """有效率"""
        return self.valid_records / self.total_records if self.total_records > 0 else 0

    @property
    def duplicate_rate(self) -> float:
        """重复率"""
        return self.duplicate_count / self.total_records if self.total_records > 0 else 0

    def __str__(self) -> str:
        return f"""
数据质量报告:
  总记录数: {self.total_records}
  有效记录: {self.valid_records} ({self.validity_rate:.1%})
  重复记录: {self.duplicate_count} ({self.duplicate_rate:.1%})
  缺失字段: {self.missing_fields}
  格式问题: {self.format_issues}
        """.strip()

import re
from typing import Optional

class HTMLCleaner:
    """HTML 清洗器"""

    # 需要完全移除的标签（包括内容）
    REMOVE_TAGS = ['script', 'style', 'head', 'meta', 'link']

    @staticmethod
    def remove_tags(html: str) -> str:
        """移除所有 HTML 标签"""
        # 先移除特定标签及其内容
        for tag in HTMLCleaner.REMOVE_TAGS:
            pattern = f'<{tag}[^>]*>.*?</{tag}>'
            html = re.sub(pattern, '', html, flags=re.DOTALL | re.IGNORECASE)

        # 移除所有标签
        html = re.sub(r'<[^>]+>', '', html)

        return html

    @staticmethod
    def remove_tags_keep_text(html: str) -> str:
        """移除标签但保留文本内容"""
        # 处理常见的块级元素，添加换行
        html = re.sub(r'</(p|div|br|li|tr|h[1-6])>', '\n', html, flags=re.IGNORECASE)
        # 移除其他标签
        html = re.sub(r'<[^>]+>', '', html)
        return html

    @staticmethod
    def decode_entities(text: str) -> str:
        """解码 HTML 实体"""
        import html
        return html.unescape(text)


# 使用 BeautifulSoup（更可靠）
def clean_html_with_bs4(html: str) -> str:
    """使用 BeautifulSoup 清洗 HTML"""
    try:
        from bs4 import BeautifulSoup
        soup = BeautifulSoup(html, 'html.parser')

        # 移除脚本和样式
        for script in soup(['script', 'style']):
            script.decompose()

        # 获取文本
        text = soup.get_text(separator='\n')
        return text
    except ImportError:
        return HTMLCleaner.remove_tags(html)

class WhitespaceCleaner:
    """空白字符清洗器"""

    @staticmethod
    def normalize_whitespace(text: str) -> str:
        """标准化空白字符"""
        # 将所有空白字符转为普通空格
        text = re.sub(r'[\t\r\f\v]', ' ', text)
        # 合并多个空格
        text = re.sub(r' +', ' ', text)
        # 合并多个换行
        text = re.sub(r'\n+', '\n', text)
        # 去除首尾空白
        return text.strip()

    @staticmethod
    def remove_all_whitespace(text: str) -> str:
        """移除所有空白字符"""
        return re.sub(r'\s+', '', text)

    @staticmethod
    def trim_lines(text: str) -> str:
        """去除每行首尾空白"""
        lines = text.split('\n')
        return '\n'.join(line.strip() for line in lines)

    @staticmethod
    def remove_empty_lines(text: str) -> str:
        """移除空行"""
        lines = text.split('\n')
        return '\n'.join(line for line in lines if line.strip())

/* Detailed source-code truncated for AI context efficiency. */

import chardet

class EncodingFixer:
    """编码问题修复器"""

    @staticmethod
    def detect_encoding(data: bytes) -> str:
        """检测编码"""
        result = chardet.detect(data)
        return result['encoding'] or 'utf-8'

    @staticmethod
    def fix_encoding(text: str, source_encoding: str = None) -> str:
        """修复编码问题"""
        try:
            # 如果是乱码，尝试重新编码
            if source_encoding:
                return text.encode('latin1').decode(source_encoding)
            else:
                # 尝试常见编码
                for encoding in ['utf-8', 'gbk', 'gb2312', 'big5']:
                    try:
                        return text.encode('latin1').decode(encoding)
                    except (UnicodeDecodeError, UnicodeEncodeError):
                        continue
        except Exception:
            pass
        return text

    @staticmethod
    def safe_decode(data: bytes, fallback: str = 'utf-8') -> str:
        """安全解码"""
        # 检测编码
        detected = EncodingFixer.detect_encoding(data)
        try:
            return data.decode(detected)
        except (UnicodeDecodeError, TypeError):
            return data.decode(fallback, errors='ignore')

class RegexPatterns:
    """常用正则表达式模式"""

    # 基础模式
    CHINESE = r'[\u4e00-\u9fa5]+'  # 中文
    EMAIL = r'[\w.+-]+@[\w-]+\.[\w.-]+'  # 邮箱
    PHONE = r'1[3-9]\d{9}'  # 中国手机号
    URL = r'https?://[^\s<>"{}|\\^`\[\]]+'  # URL
    IP = r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}'  # IP 地址

    # 日期时间模式
    DATE_YMD = r'\d{4}[-/年]\d{1,2}[-/月]\d{1,2}日?'  # 年月日
    TIME_HMS = r'\d{1,2}:\d{2}(:\d{2})?'  # 时分秒
    DATETIME = r'\d{4}[-/]\d{1,2}[-/]\d{1,2}\s+\d{1,2}:\d{2}(:\d{2})?'

    # 数值模式
    INTEGER = r'-?\d+'
    FLOAT = r'-?\d+\.?\d*'
    PRICE = r'[¥$￥]\s*\d+\.?\d*'  # 价格
    PERCENTAGE = r'\d+\.?\d*%'  # 百分比


class RegexExtractor:
    """正则表达式提取器"""

    @staticmethod
    def extract_all(text: str, pattern: str) -> list:
        """提取所有匹配"""
        return re.findall(pattern, text)

    @staticmethod
    def extract_first(text: str, pattern: str) -> str:
        """提取第一个匹配"""
        match = re.search(pattern, text)
        return match.group() if match else ''

    @staticmethod
    def extract_groups(text: str, pattern: str) -> dict:
        """提取命名分组"""
        match = re.search(pattern, text)
        return match.groupdict() if match else {}

    @staticmethod
    def extract_between(text: str, start: str, end: str) -> list:
        """提取两个标记之间的内容"""
        pattern = f'{re.escape(start)}(.*?){re.escape(end)}'
        return re.findall(pattern, text, re.DOTALL)

class RegexReplacer:
    """正则表达式替换器"""

    @staticmethod
    def replace_with_callback(
        text: str,
        pattern: str,
        callback
    ) -> str:
        """使用回调函数替换"""
        return re.sub(pattern, callback, text)

    @staticmethod
    def mask_sensitive(text: str) -> str:
        """脱敏处理"""
        # 手机号脱敏
        text = re.sub(
            r'(1[3-9]\d)\d{4}(\d{4})',
            r'\1****\2',
            text
        )
        # 邮箱脱敏
        text = re.sub(
            r'([\w.+-]{1,3})[\w.+-]*(@[\w-]+\.[\w.-]+)',
            r'\1***\2',
            text
        )
        # 身份证脱敏
        text = re.sub(
            r'(\d{6})\d{8}(\d{4})',
            r'\1********\2',
            text
        )
        return text

    @staticmethod
    def clean_url_params(url: str, keep_params: list = None) -> str:
        """清理 URL 参数"""
        from urllib.parse import urlparse, parse_qs, urlencode, urlunparse

        parsed = urlparse(url)
        params = parse_qs(parsed.query)

        if keep_params:
            params = {k: v for k, v in params.items() if k in keep_params}
        else:
            params = {}

        new_query = urlencode(params, doseq=True)
        return urlunparse(parsed._replace(query=new_query))

from typing import List, Dict, Any, Callable
import hashlib

class ExactDeduplicator:
    """精确去重器"""

    @staticmethod
    def dedupe_list(items: List[str]) -> List[str]:
        """列表去重（保持顺序）"""
        seen = set()
        result = []
        for item in items:
            if item not in seen:
                seen.add(item)
                result.append(item)
        return result

    @staticmethod
    def dedupe_dicts(
        items: List[Dict],
        key_field: str
    ) -> List[Dict]:
        """字典列表去重"""
        seen = set()
        result = []
        for item in items:
            key = item.get(key_field)
            if key not in seen:
                seen.add(key)
                result.append(item)
        return result

    @staticmethod
    def dedupe_by_hash(
        items: List[Dict],
        fields: List[str]
    ) -> List[Dict]:
        """
        根据多个字段计算哈希去重

        Args:
            items: 数据列表
            fields: 用于计算哈希的字段

        Returns:
            去重后的列表
        """
        seen = set()
        result = []

        for item in items:
            # 计算哈希
            key_str = '|'.join(str(item.get(f, '')) for f in fields)
            key_hash = hashlib.md5(key_str.encode()).hexdigest()

            if key_hash not in seen:
                seen.add(key_hash)
                result.append(item)

        return result

/* Detailed source-code truncated for AI context efficiency. */

/* Detailed source-code truncated for AI context efficiency. */

class NumberNormalizer:
    """数值标准化器"""

    @staticmethod
    def parse_number(text: str) -> float:
        """
        解析数字（支持中文单位）

        Examples:
            "1.5万" -> 15000
            "3.2亿" -> 320000000
            "1,234.56" -> 1234.56
        """
        text = text.strip()

        # 中文单位映射
        units = {
            '万': 10000,
            '亿': 100000000,
            'k': 1000,
            'K': 1000,
            'm': 1000000,
            'M': 1000000,
            'b': 1000000000,
            'B': 1000000000,
        }

        multiplier = 1
        for unit, value in units.items():
            if unit in text:
                multiplier = value
                text = text.replace(unit, '')
                break

        # 移除逗号
        text = text.replace(',', '')

        # 提取数字
        match = re.search(r'-?\d+\.?\d*', text)
        if match:
            return float(match.group()) * multiplier

        return 0.0

    @staticmethod
    def format_number(
        value: float,
        precision: int = 2,
        use_units: bool = True
    ) -> str:
        """格式化数字"""
        if not use_units:
            return f'{value:.{precision}f}'

        if value >= 100000000:
            return f'{value/100000000:.{precision}f}亿'
        elif value >= 10000:
            return f'{value/10000:.{precision}f}万'
        else:
            return f'{value:.{precision}f}'

class TextNormalizer:
    """文本标准化器"""

    @staticmethod
    def normalize(text: str) -> str:
        """完整的文本标准化流程"""
        # 1. Unicode 标准化
        text = unicodedata.normalize('NFKC', text)
        # 2. 全角转半角
        text = SpecialCharCleaner.to_halfwidth(text)
        # 3. 移除控制字符
        text = SpecialCharCleaner.remove_control_chars(text)
        # 4. 标准化空白
        text = WhitespaceCleaner.normalize_whitespace(text)
        return text

    @staticmethod
    def normalize_punctuation(text: str) -> str:
        """标准化标点符号"""
        # 中文标点转英文
        mapping = {
            '，': ', ',
            '。': '. ',
            '！': '! ',
            '？': '? ',
            '；': '; ',
            '：': ': ',
            '"': '"',
            '"': '"',
            ''': "'",
            ''': "'",
        }
        for cn, en in mapping.items():
            text = text.replace(cn, en)
        return text

/* Detailed source-code truncated for AI context efficiency. */

/* Detailed source-code truncated for AI context efficiency. */

flowchart TD
    subgraph 原始数据问题
        title_em["标题含HTML标签<br/>(搜索高亮)"]
        view_unit["数值含单位<br/>(1.5万、3.2亿)"]
        duration_format["时长格式不一<br/>(02:30、1:23:45)"]
        pubdate_format["发布时间多样<br/>(刚刚、3天前、2024-01-15)"]
        desc_html["描述含HTML<br/>(换行、链接)"]
    end

    subgraph 清洗目标
        title_clean[纯文本标题]
        view_num[数值]
        duration_sec[秒数时长]
        pubdate_std[标准日期]
        desc_plain[纯文本描述]
    end

    title_em --> title_clean
    view_unit --> view_num
    duration_format --> duration_sec
    pubdate_format --> pubdate_std
    desc_html --> desc_plain

/* Detailed source-code truncated for AI context efficiency. */

import hashlib
from typing import List, Set


class DataDeduplicator:
    """数据去重器"""

    def __init__(self):
        self._seen_bvids: Set[str] = set()
        self._seen_hashes: Set[str] = set()

    def is_duplicate_by_bvid(self, bvid: str) -> bool:
        """通过BV号判断是否重复"""
        if bvid in self._seen_bvids:
            return True
        self._seen_bvids.add(bvid)
        return False

    def is_duplicate_by_content(self, title: str, owner_mid: int) -> bool:
        """
        通过内容哈希判断是否重复
        （用于检测同一UP主发布的相似标题视频）
        """
        content = f"{title}|{owner_mid}"
        content_hash = hashlib.md5(content.encode()).hexdigest()

        if content_hash in self._seen_hashes:
            return True
        self._seen_hashes.add(content_hash)
        return False

    def dedupe_videos(
        self,
        videos: List[WebContent]
    ) -> List[WebContent]:
        """
        去重视频列表

        优先使用BV号去重，同时检测内容重复
        """
        results = []

        for video in videos:
            # BV号去重
            if self.is_duplicate_by_bvid(video.bvid):
                continue

            # 内容去重（可选）
            if self.is_duplicate_by_content(video.title, video.owner_mid):
                continue

            results.append(video)

        return results

    def reset(self):
        """重置去重状态"""
        self._seen_bvids.clear()
        self._seen_hashes.clear()

/* Detailed source-code truncated for AI context efficiency. */

flowchart LR
    subgraph 数据准备
        A[原始数据] --> B[数据清洗]
        B --> C[数据转换]
    end

    subgraph 数据分析
        C --> D[统计分析]
        C --> E[文本分析]
        C --> F[时间序列]
    end

    subgraph 结果输出
        D --> G[图表可视化]
        E --> H[词云生成]
        F --> G
        G --> I[分析报告]
        H --> I
    end

import pandas as pd

# 从爬取的数据创建 DataFrame
data = [
    {"title": "Python 教程", "views": 15000, "likes": 320, "date": "2024-01-15"},
    {"title": "爬虫入门", "views": 12000, "likes": 280, "date": "2024-01-16"},
    {"title": "数据分析", "views": 18000, "likes": 450, "date": "2024-01-17"},
]
df = pd.DataFrame(data)

# 基础信息
print(df.info())    # 列类型和非空数量
print(df.describe())  # 数值列统计摘要

# 数据选择
print(df['title'])           # 选择单列
print(df[['title', 'views']])  # 选择多列
print(df[df['views'] > 14000])  # 条件筛选

# 排序
print(df.sort_values('views', ascending=False))

# 新增列
df['engagement'] = df['likes'] / df['views'] * 100

# 假设有更多数据
data = [
    {"category": "技术", "title": "Python 教程", "views": 15000},
    {"category": "技术", "title": "爬虫入门", "views": 12000},
    {"category": "生活", "title": "美食推荐", "views": 8000},
    {"category": "生活", "title": "旅行日记", "views": 10000},
    {"category": "技术", "title": "数据分析", "views": 18000},
]
df = pd.DataFrame(data)

# 按分类分组统计
grouped = df.groupby('category')
print(grouped['views'].sum())     # 各分类总浏览量
print(grouped['views'].mean())    # 各分类平均浏览量
print(grouped['views'].agg(['sum', 'mean', 'max', 'min']))  # 多个统计量

# 多列聚合
print(df.groupby('category').agg({
    'views': 'sum',
    'title': 'count'
}).rename(columns={'title': 'article_count'}))

# 更复杂的数据
data = [
    {"date": "2024-01", "category": "技术", "platform": "PC", "views": 15000},
    {"date": "2024-01", "category": "技术", "platform": "Mobile", "views": 8000},
    {"date": "2024-01", "category": "生活", "platform": "PC", "views": 6000},
    {"date": "2024-02", "category": "技术", "platform": "PC", "views": 18000},
    {"date": "2024-02", "category": "生活", "platform": "Mobile", "views": 9000},
]
df = pd.DataFrame(data)

# 创建透视表：按日期和分类统计各平台浏览量
pivot = pd.pivot_table(
    df,
    values='views',
    index='date',
    columns='category',
    aggfunc='sum',
    fill_value=0
)
print(pivot)

# 带边距的透视表
pivot_with_margins = pd.pivot_table(
    df,
    values='views',
    index='date',
    columns='category',
    aggfunc='sum',
    margins=True,
    margins_name='总计'
)
print(pivot_with_margins)

# 创建时间序列数据
dates = pd.date_range('2024-01-01', periods=30, freq='D')
views = [1000 + i * 50 + (i % 7) * 200 for i in range(30)]
df = pd.DataFrame({'date': dates, 'views': views})
df.set_index('date', inplace=True)

# 滚动统计（7天移动平均）
df['rolling_avg'] = df['views'].rolling(window=7).mean()

# 按周统计
weekly = df.resample('W').agg({
    'views': ['sum', 'mean']
})
print(weekly)

# 环比增长率
df['growth_rate'] = df['views'].pct_change() * 100

import jieba
import jieba.analyse

# 基础分词
text = "Python爬虫教程帮助你快速入门数据采集技术"
words = jieba.lcut(text)
print(words)  # ['Python', '爬虫', '教程', '帮助', '你', '快速', '入门', '数据', '采集', '技术']

# 提取关键词（TF-IDF）
keywords = jieba.analyse.extract_tags(
    text,
    topK=5,
    withWeight=True
)
for word, weight in keywords:
    print(f"{word}: {weight:.4f}")

# 提取关键词（TextRank）
keywords_tr = jieba.analyse.textrank(
    text,
    topK=5,
    withWeight=True
)

# 常用中文停用词
STOPWORDS = {
    '的', '是', '在', '了', '和', '与', '或', '有', '个', '人',
    '这', '那', '就', '都', '也', '为', '对', '到', '从', '把',
    '被', '让', '给', '向', '往', '于', '及', '以', '等', '不',
    '很', '会', '能', '可', '要', '我', '你', '他', '她', '它',
}

def filter_stopwords(words: list) -> list:
    """过滤停用词和单字"""
    return [
        w for w in words
        if w not in STOPWORDS and len(w) > 1
    ]

# 使用
text = "这是一个关于Python爬虫的教程，帮助你快速入门"
words = jieba.lcut(text)
filtered = filter_stopwords(words)
print(filtered)  # ['Python', '爬虫', '教程', '帮助', '快速', '入门']

/* Detailed source-code truncated for AI context efficiency. */

import numpy as np
from PIL import Image
from wordcloud import WordCloud, ImageColorGenerator

def generate_shaped_wordcloud(
    text: str,
    mask_image_path: str,
    output_path: str,
    font_path: str = None,
    use_mask_colors: bool = True
):
    """
    生成自定义形状的词云

    Args:
        text: 词语文本
        mask_image_path: 形状蒙版图片路径
        output_path: 输出路径
        font_path: 字体路径
        use_mask_colors: 是否使用蒙版图片的颜色
    """
    # 读取蒙版图片
    mask = np.array(Image.open(mask_image_path))

    # 创建词云
    wc = WordCloud(
        mask=mask,
        background_color="white",
        font_path=font_path,
        max_words=500,
        max_font_size=80,
        random_state=42,
        contour_width=1,
        contour_color='steelblue'
    )

    wc.generate(text)

    # 使用蒙版颜色
    if use_mask_colors:
        image_colors = ImageColorGenerator(mask)
        wc.recolor(color_func=image_colors)

    wc.to_file(output_path)
    return output_path

/* Detailed source-code truncated for AI context efficiency. */

/* Detailed source-code truncated for AI context efficiency. */

/* Detailed source-code truncated for AI context efficiency. */

/* Detailed source-code truncated for AI context efficiency. */

/* Detailed source-code truncated for AI context efficiency. */

flowchart TD
    subgraph 输入数据
        videos[视频数据]
        comments[弹幕/评论数据]
    end

    subgraph 分析维度
        stat[播放量/互动统计]
        up[UP主分析]
        time[发布时间分析]
        keyword[标题关键词]
    end

    subgraph 输出
        rank[热门视频排行]
        trend[发布趋势图]
        cloud[标题词云]
        report[分析报告]
    end

    videos --> stat --> rank
    videos --> up --> rank
    videos --> time --> trend
    videos --> keyword --> cloud
    rank --> report
    trend --> report
    cloud --> report

/* Detailed source-code truncated for AI context efficiency. */

graph TB
    subgraph 入口层
        main["main.py<br/>程序入口"]
    end

    subgraph 核心模块
        config["config 配置<br/>settings.py<br/>bilibili_config.py"]
        crawler["crawler 爬虫<br/>spider.py<br/>核心调度"]
        store["store 存储<br/>backend.py<br/>JSON/CSV"]
    end

    subgraph 功能模块
        login["login 登录<br/>auth.py<br/>扫码/Cookie"]
        client["client 客户端<br/>bilibili_client.py<br/>API请求"]
        analysis["analysis 分析<br/>report.py<br/>词云/统计"]
    end

    subgraph 基础模块
        core["core 浏览器<br/>browser.py<br/>Playwright封装"]
        tools["tools 工具<br/>sign.py<br/>WBI签名"]
        models["models 模型<br/>bilibili.py<br/>数据结构"]
    end

    main --> config
    main --> crawler
    main --> store
    main --> analysis

    crawler --> login
    crawler --> client

    login --> core
    client --> tools
    tools --> models

    config -.->|配置注入| crawler
    crawler -->|数据| store

flowchart TD
    start([程序启动 main.py]) --> step1

    subgraph step1 [步骤1: 初始化浏览器]
        browser["BrowserManager.start()<br/>启动 Playwright<br/>创建 BrowserContext"]
    end

    step1 --> step2

    subgraph step2 [步骤2: 登录认证]
        check{检查登录状态}
        check -->|已登录| skip[跳过登录]
        check -->|未登录| login_flow
        subgraph login_flow [执行登录]
            qrcode["扫码登录: 显示二维码-等待扫码-获取Cookie"]
            cookie["Cookie登录: 注入Cookie-验证有效性"]
        end
    end

    step2 --> step3

    subgraph step3 [步骤3: 初始化API客户端]
        init["同步Cookie到httpx<br/>获取WBI签名密钥<br/>初始化签名器"]
    end

    step3 --> step4

    subgraph step4 [步骤4: 执行爬取任务]
        mode{爬取模式}
        mode -->|SEARCH| search["关键词搜索-翻页获取-获取详情"]
        mode -->|DETAIL| detail[直接获取指定视频详情]
        search --> request
        detail --> request
        request["每次请求:<br/>构造参数-WBI签名-发送请求-解析响应-随机延迟"]
    end

    step4 --> step5

    subgraph step5 [步骤5: 数据存储]
        save["BilibiliVideo对象列表<br/>转换为字典<br/>保存为 JSON/CSV"]
    end

    step5 --> step6

    subgraph step6 [步骤6: 生成分析报告]
        report["统计分析-词频统计<br/>生成词云-输出Markdown报告"]
    end

    step6 --> step7

    subgraph step7 [步骤7: 清理资源]
        cleanup["关闭浏览器<br/>保存登录状态<br/>程序退出"]
    end

    step7 --> finish([完成])

flowchart LR
    subgraph input [用户输入]
        keyword["关键词<br/>配置文件"]
    end

    subgraph process [系统处理]
        search["搜索API<br/>(WBI签名)"]
        detail["详情API<br/>(获取详情)"]
        validate["Pydantic数据验证<br/>BilibiliVideo"]
    end

    subgraph output [最终输出]
        json["JSON文件<br/>(结构化)"]
        csv["CSV文件<br/>(表格化)"]
        report["分析报告<br/>(Markdown)"]
        wordcloud["词云图片<br/>(PNG)"]
    end

    keyword --> search
    search -->|视频列表| detail
    detail --> validate
    validate --> json
    validate --> csv
    validate --> report
    report --> wordcloud

11_进阶综合实战项目/
├── config/              # 配置模块
│   ├── __init__.py
│   ├── settings.py      # 通用配置
│   └── bilibili_config.py  # B站特定配置
├── core/                # 核心模块
│   ├── __init__.py
│   └── browser.py       # 浏览器管理
├── login/               # 登录模块
│   ├── __init__.py
│   └── auth.py          # B站登录认证
├── client/              # API 客户端模块
│   ├── __init__.py
│   └── bilibili_client.py  # B站 API 客户端
├── crawler/             # 爬虫模块
│   ├── __init__.py
│   └── spider.py        # B站爬虫实现
├── store/               # 存储模块
│   ├── __init__.py
│   └── backend.py       # 存储后端
├── proxy/               # 代理模块（可选）
│   ├── __init__.py
│   └── pool.py          # 代理池
├── models/              # 数据模型模块
│   ├── __init__.py
│   └── bilibili.py      # B站数据模型
├── tools/               # 工具模块
│   ├── __init__.py
│   └── sign.py          # WBI 签名工具
├── analysis/            # 分析模块
│   ├── __init__.py
│   └── report.py        # 报告生成
└── main.py              # 入口文件

# config/settings.py
from pydantic_settings import BaseSettings
from pydantic import Field
from typing import Optional, List
from enum import Enum


class StorageType(str, Enum):
    """存储类型"""
    JSON = "json"
    CSV = "csv"


class LoginType(str, Enum):
    """登录类型"""
    COOKIE = "cookie"
    QRCODE = "qrcode"


class CrawlerType(str, Enum):
    """爬取类型"""
    SEARCH = "search"    # 关键词搜索
    DETAIL = "detail"    # 指定视频详情


class Settings(BaseSettings):
    """项目配置"""

    # 基础配置
    app_name: str = "BilibiliCrawler"
    debug: bool = False

    # 浏览器配置
    browser_headless: bool = False  # B站扫码登录需要显示浏览器
    browser_timeout: int = 30000
    browser_user_data_dir: Optional[str] = "./browser_data"
    save_login_state: bool = True

    # 登录配置
    login_type: LoginType = LoginType.QRCODE
    cookie_str: str = ""

    # 爬虫配置
    crawler_type: CrawlerType = CrawlerType.SEARCH
    keywords: str = "Python教程"  # 搜索关键词，多个用逗号分隔
    specified_id_list: List[str] = []  # 指定视频列表
    max_video_count: int = 20
    max_concurrency: int = 3
    crawl_delay_min: float = 1.0
    crawl_delay_max: float = 3.0

    # 存储配置
    storage_type: StorageType = StorageType.JSON
    storage_output_dir: str = "./output"

    class Config:
        env_file = ".env"
        env_prefix = "CRAWLER_"


# 全局配置实例
settings = Settings()

# config/bilibili_config.py
"""B站 API 配置"""

# API 地址
SEARCH_URL = "https://api.bilibili.com/x/web-interface/wbi/search/type"
VIDEO_INFO_URL = "https://api.bilibili.com/x/web-interface/view"
NAV_URL = "https://api.bilibili.com/x/web-interface/nav"

# 请求配置
SEARCH_PAGE_SIZE = 20
REQUEST_TIMEOUT = 30

# 默认请求头
DEFAULT_HEADERS = {
    "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
                  "AppleWebKit/537.36 (KHTML, like Gecko) "
                  "Chrome/120.0.0.0 Safari/537.36",
    "Referer": "https://www.bilibili.com",
    "Origin": "https://www.bilibili.com",
}

# WBI 签名密钥混淆表
WBI_MIXIN_KEY_ENC_TAB = [
    46, 47, 18, 2, 53, 8, 23, 32, 15, 50, 10, 31, 58, 3, 45, 35,
    27, 43, 5, 49, 33, 9, 42, 19, 29, 28, 14, 39, 12, 38, 41, 13,
    37, 48, 7, 16, 24, 55, 40, 61, 26, 17, 0, 1, 60, 51, 30, 4,
    22, 25, 54, 21, 56, 59, 6, 63, 57, 62, 11, 36, 20, 34, 44, 52,
]

# 登录相关
LOGIN_BUTTON_SELECTOR = "xpath=//div[@class='right-entry__outside go-login-btn']//div"
QRCODE_SELECTOR = "//div[@class='login-scan-box']//img"
LOGIN_COOKIE_KEYS = ["SESSDATA", "DedeUserID", "bili_jct"]

flowchart TD
    subgraph step1 ["第一步: 获取签名密钥"]
        api["调用 /x/web-interface/nav API"]
        api --> response["响应中包含 wbi_img:<br/>{img_url, sub_url}"]
        response --> extract["提取文件名作为密钥:<br/>img_key, sub_key"]
    end

    step1 --> step2

    subgraph step2 ["第二步: 生成盐值 Salt"]
        concat["raw_key = img_key + sub_key"]
        concat --> mixin["使用混淆表重排字符:<br/>WBI_MIXIN_KEY_ENC_TAB"]
        mixin --> salt["salt = 重排后取前32位"]
    end

    step2 --> step3

    subgraph step3 ["第三步: 计算签名"]
        params["原始参数:<br/>{keyword, page}"]
        params --> addwts["添加时间戳 wts"]
        addwts --> sort["按 key 排序并URL编码"]
        sort --> append["拼接盐值: query + salt"]
        append --> md5["计算 MD5 得到 w_rid"]
        md5 --> final["最终参数:<br/>{...原参数, wts, w_rid}"]
    end

/* Detailed source-code truncated for AI context efficiency. */

/* Detailed source-code truncated for AI context efficiency. */

sequenceDiagram
    participant PC as PC浏览器
    participant Server as B站服务器
    participant APP as B站APP

    PC->>Server: 1. 访问B站首页
    PC->>Server: 2. 点击登录按钮
    PC->>Server: 3. 请求二维码+UUID
    Server-->>PC: 4. 返回二维码图片
    Note over PC: 5. 显示二维码

    loop 轮询登录状态
        PC->>Server: 6. 检查登录状态
        APP->>Server: 7. 用户扫描二维码
        Server-->>PC: 8. 状态: 已扫描
        PC->>Server: 9. 继续轮询
        APP->>Server: 10. 用户点击确认
        Server-->>PC: 11. 返回登录凭证(Set-Cookie)
    end

    Note over PC: 12. 保存Cookie<br/>登录成功!

flowchart LR
    subgraph input [用户输入]
        cookie["用户提供Cookie<br/>(从浏览器复制)"]
    end

    subgraph process [处理流程]
        parse["解析Cookie字符串<br/>提取键值对"]
        inject["注入到BrowserContext<br/>add_cookies()"]
        visit["访问B站首页<br/>加载页面"]
        check{"检查关键Cookie<br/>SESSDATA存在?"}
    end

    subgraph result [结果]
        success["登录成功!<br/>开始爬取"]
        fail["Cookie已过期<br/>需要重新登录"]
    end

    cookie --> parse --> inject --> visit --> check
    check -->|存在| success
    check -->|不存在| fail

/* Detailed source-code truncated for AI context efficiency. */

graph TB
    subgraph 核心功能
        cookie["Cookie管理<br/>• 从浏览器同步<br/>• 注入到请求头<br/>• 验证有效性"]
        wbi["WBI签名<br/>• 获取密钥<br/>• 参数签名<br/>• 自动刷新"]
        http["HTTP请求<br/>• GET/POST请求<br/>• 超时处理<br/>• 错误重试"]
    end

    subgraph API方法
        api["BilibiliClient API<br/>• search_video_by_keyword<br/>• get_video_info<br/>• pong (登录检查)"]
    end

    cookie --> api
    wbi --> api
    http --> api

/* Detailed source-code truncated for AI context efficiency. */

graph LR
    subgraph BilibiliCrawler
        subgraph 属性
            attr1["browser_manager"] --> desc1["管理Playwright浏览器"]
            attr2["browser_context"] --> desc2["浏览器上下文(Cookie容器)"]
            attr3["context_page"] --> desc3["页面实例"]
            attr4["bili_client"] --> desc4["API客户端"]
            attr5["_results"] --> desc5["爬取结果列表"]
        end

        subgraph 方法
            m1["start()"] --> d1["启动爬虫(主入口)"]
            m2["_init_browser()"] --> d2["初始化浏览器"]
            m3["_do_login()"] --> d3["执行登录"]
            m4["_init_client()"] --> d4["初始化API客户端"]
            m5["search_by_keywords()"] --> d5["关键词搜索"]
            m6["get_specified_videos()"] --> d6["获取指定视频"]
            m7["close()"] --> d7["清理资源"]
        end
    end

flowchart LR
    subgraph input [关键词列表]
        k1["Python"]
        k2["教程"]
        k3["数据分析"]
    end

    subgraph search [搜索分页]
        p1["第1页"]
        p2["第2页"]
        p3["..."]
    end

    subgraph result [视频列表]
        v1["BV1xxx"]
        v2["BV2xxx"]
        v3["BV3xxx"]
    end

    detail["获取每个视频详情<br/>(完整播放量等)"]

    input --> search --> result --> detail

flowchart LR
    subgraph input [指定BV号列表]
        bv1["BV1abc"]
        bv2["BV2def"]
        bv3["BV3ghi"]
    end

    api["遍历列表<br/>逐个调用详情API"]

    input --> api

/* Detailed source-code truncated for AI context efficiency. */

graph TB
    subgraph manager [存储管理器]
        sm["StorageManager<br/>• save(data)<br/>• load()<br/>• filepath"]
    end

    subgraph base [抽象基类]
        bs["BaseStorage (ABC)<br/>• save() 抽象方法<br/>• load() 抽象方法"]
    end

    subgraph impl [具体实现]
        json["JSONStorage<br/>• 保存为JSON文件<br/>• 保持数据结构<br/>• 便于程序处理"]
        csv["CSVStorage<br/>• 保存为CSV文件<br/>• 适合Excel打开<br/>• 便于数据分析"]
    end

    sm -->|根据配置选择| json
    sm -->|根据配置选择| csv
    json -->|继承| bs
    csv -->|继承| bs

/* Detailed source-code truncated for AI context efficiency. */

flowchart LR
    subgraph input [输入数据]
        videos["BilibiliVideo<br/>对象列表"]
    end

    subgraph process [分析处理]
        stats["视频指标统计<br/>• 播放量统计<br/>• 点赞数统计<br/>• 收藏数统计"]
        top["热门视频排名<br/>TOP 10"]
        up["UP主分布统计<br/>词频分析<br/>(jieba分词)"]
    end

    subgraph output [输出结果]
        md["Markdown 报告<br/>• 数据表格<br/>• TOP排名<br/>• UP主分布"]
        img["词云图片<br/>(PNG)"]
    end

    videos --> stats --> md
    stats --> top --> img
    top --> up

graph TB
    subgraph deps [可选依赖]
        jieba["jieba<br/>中文分词<br/>(词频统计)"]
        wc["wordcloud<br/>词云生成<br/>(可视化)"]
        pd["pandas<br/>数据处理<br/>(可选)"]
    end

    subgraph fallback [未安装时的降级策略]
        note["如果未安装:<br/>• jieba 未安装 - 跳过词频统计和词云生成<br/>• wordcloud 未安装 - 跳过词云生成<br/>• pandas 未安装 - 使用纯Python实现统计"]
    end

    jieba --> note
    wc --> note
    pd --> note

/* Detailed source-code truncated for AI context efficiency. */

/* Detailed source-code truncated for AI context efficiency. */

# 安装依赖
pip install playwright httpx pydantic pydantic-settings loguru

# 安装 Playwright 浏览器
playwright install chromium

# 可选：数据分析依赖
pip install pandas jieba wordcloud

# 爬取类型
crawler_type = CrawlerType.SEARCH  # 或 CrawlerType.DETAIL

# 搜索关键词
keywords = "Python教程,数据分析"

# 最大爬取数量
max_video_count = 20

# 登录方式
login_type = LoginType.QRCODE  # 首次使用扫码登录

cd 源代码/爬虫进阶/11_进阶综合实战项目
python main.py