# 示例：分析抓包得到的API請(qǐng)求結(jié)構(gòu) - 深度技術(shù)解析
import requests
import json

def analyze_api_request():
    """
    通過(guò)抓包工具獲取API請(qǐng)求的關(guān)鍵信息后，我們可以構(gòu)建相應(yīng)的爬蟲(chóng)請(qǐng)求
    這個(gè)函數(shù)演示如何分析和重現(xiàn)抓包得到的請(qǐng)求結(jié)構(gòu)
    
    ?? 技術(shù)要點(diǎn)詳解：
    1. User-Agent偽裝：模擬真實(shí)瀏覽器行為，避免被基礎(chǔ)反爬機(jī)制識(shí)別
    2. Authorization認(rèn)證：處理JWT令牌或其他認(rèn)證機(jī)制
    3. 參數(shù)簽名驗(yàn)證：理解API的簽名算法邏輯
    4. 響應(yīng)狀態(tài)碼處理：正確處理各種HTTP狀態(tài)碼
    """
    # 從抓包工具中獲取的請(qǐng)求信息
    api_url = "https://api.example.com/data"
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
        "Authorization": "Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9",
        "Content-Type": "application/json"
    }
    params = {
        "page": 1,
        "limit": 20,
        "timestamp": 1635678900
    }
    
    # 發(fā)送請(qǐng)求并分析響應(yīng)
    response = requests.get(api_url, headers=headers, params=params)
    data = response.json()
    
    # 關(guān)鍵信息提取
    print(f"狀態(tài)碼: {response.status_code}")
    print(f"響應(yīng)頭: {dict(response.headers)}")
    print(f"數(shù)據(jù)量: {len(data.get('items', []))}條")
    
    return data

# 關(guān)鍵行技術(shù)原理深度解析：
# 第8行：User-Agent偽裝技術(shù)原理 - 現(xiàn)代網(wǎng)站通過(guò)User-Agent識(shí)別客戶(hù)端類(lèi)型，使用真實(shí)瀏覽器的UA可以降低被識(shí)別為爬蟲(chóng)的概率
# 第9行：Authorization認(rèn)證機(jī)制 - JWT令牌包含用戶(hù)身份信息，需要定期刷新以避免過(guò)期
# 第13-16行：參數(shù)結(jié)構(gòu)分析 - 時(shí)間戳參數(shù)用于防止重放攻擊，分頁(yè)參數(shù)控制數(shù)據(jù)返回量
# 第19行：請(qǐng)求發(fā)送過(guò)程 - requests庫(kù)底層使用urllib3，支持連接池和重試機(jī)制
# 第22-24行：響應(yīng)處理邏輯 - 狀態(tài)碼200表示成功，其他狀態(tài)碼需要特殊處理

# 示例：基于抓包數(shù)據(jù)構(gòu)建可重用的爬蟲(chóng)請(qǐng)求 - 深度技術(shù)解析
import time
import hashlib

class RequestBuilder:
    """
    請(qǐng)求構(gòu)建器類(lèi) - 封裝了從抓包分析中提取的請(qǐng)求構(gòu)建邏輯
    
    ?? 設(shè)計(jì)模式應(yīng)用：
    - 建造者模式：逐步構(gòu)建復(fù)雜的HTTP請(qǐng)求
    - 策略模式：支持不同的簽名算法和頭構(gòu)建策略
    - 單例模式：Session對(duì)象在整個(gè)爬蟲(chóng)生命周期中復(fù)用
    """
    def __init__(self, base_url):
        self.base_url = base_url
        # 使用Session對(duì)象保持會(huì)話狀態(tài)，提高性能
        self.session = requests.Session()
    
    def build_headers(self, referer=None):
        """
        構(gòu)建請(qǐng)求頭，模擬真實(shí)瀏覽器
        
        ?? 技術(shù)原理詳解：
        - User-Agent：模擬特定瀏覽器版本和操作系統(tǒng)
        - Accept頭：聲明客戶(hù)端支持的內(nèi)容類(lèi)型
        - Accept-Language：設(shè)置語(yǔ)言偏好，影響服務(wù)器返回內(nèi)容
        - Connection：保持長(zhǎng)連接，減少TCP握手開(kāi)銷(xiāo)
        - Referer：設(shè)置來(lái)源頁(yè)面，某些API會(huì)驗(yàn)證來(lái)源合法性
        """
        headers = {
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
            "Accept": "application/json, text/plain, */*",
            "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
            "Connection": "keep-alive"
        }
        if referer:
            headers["Referer"] = referer
        return headers
    
    def add_signature(self, params):
        """
        添加簽名參數(shù)（常見(jiàn)反爬機(jī)制）
        
        ?? 簽名算法技術(shù)原理：
        - 時(shí)間戳防重放：防止請(qǐng)求被重復(fù)使用
        - 參數(shù)排序：通常按字母順序排列參數(shù)
        - 密鑰混合：使用固定密鑰與參數(shù)組合
        - 哈希算法：MD5/SHA1等算法生成固定長(zhǎng)度簽名
        - 編碼方式：Base64/Hex等編碼格式輸出
        """
        timestamp = str(int(time.time()))
        params['timestamp'] = timestamp
        
        # 模擬簽名算法（從抓包分析得出）
        # 技術(shù)要點(diǎn)：參數(shù)通常按特定順序排列，密鑰混合方式需要逆向分析
        sign_str = f"key1={params.get('key1', '')}&timestamp={timestamp}&secret=example"
        params['sign'] = hashlib.md5(sign_str.encode()).hexdigest()
        return params

# 示例：電商網(wǎng)站API逆向分析 - 深度技術(shù)解析
import json
import re
from urllib.parse import urlencode

class EcommerceCrawler:
    """
    電商網(wǎng)站爬蟲(chóng)類(lèi) - 封裝了從抓包分析中提取的API調(diào)用邏輯
    
    ?? 架構(gòu)設(shè)計(jì)思想：
    - 單一職責(zé)原則：每個(gè)方法只負(fù)責(zé)一個(gè)明確的功能
    - 開(kāi)閉原則：易于擴(kuò)展新的API端點(diǎn)發(fā)現(xiàn)方法
    - 依賴(lài)倒置原則：依賴(lài)于抽象接口而非具體實(shí)現(xiàn)
    """
    def __init__(self):
        # 使用Session保持會(huì)話狀態(tài)，自動(dòng)處理Cookie和連接復(fù)用
        self.session = requests.Session()
        self.base_headers = {
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
            "Accept": "application/json, text/plain, */*",
            "X-Requested-With": "XMLHttpRequest"  # 標(biāo)識(shí)為AJAX請(qǐng)求
        }
    
    def extract_api_endpoints(self, html_content):
        """
        從頁(yè)面HTML中提取API端點(diǎn)（通過(guò)抓包發(fā)現(xiàn)）
        
        ?? 技術(shù)原理詳解：
        - 正則表達(dá)式模式：匹配常見(jiàn)的API URL模式
        - 動(dòng)態(tài)內(nèi)容分析：從JavaScript變量中提取API配置
        - 網(wǎng)絡(luò)請(qǐng)求監(jiān)控：通過(guò)抓包工具發(fā)現(xiàn)隱藏的API調(diào)用
        - 代碼靜態(tài)分析：分析前端框架的API調(diào)用模式
        """
        # 使用正則表達(dá)式匹配API URL模式
        api_patterns = [
            r'api\.example\.com\/v\d+\/products',  # 版本化API端點(diǎn)
            r'\/ajax\/product\/list',              # AJAX接口路徑
            r'window\.API_URL\s*=\s*["\']([^"\']+)["\']'  # JavaScript變量
        ]
        
        endpoints = []
        for pattern in api_patterns:
            matches = re.findall(pattern, html_content)
            endpoints.extend(matches)
        
        return endpoints
    
    def analyze_product_api(self, api_url, page=1):
        """
        分析商品列表API
        
        ?? 技術(shù)原理詳解：
        - 分頁(yè)參數(shù)設(shè)計(jì)：page/pageSize控制數(shù)據(jù)量和偏移量
        - 排序參數(shù)：支持多種排序方式，影響數(shù)據(jù)返回順序
        - 分類(lèi)過(guò)濾：categoryId參數(shù)實(shí)現(xiàn)數(shù)據(jù)分類(lèi)篩選
        - 時(shí)間戳防緩存：確保獲取最新數(shù)據(jù)，避免緩存影響
        """
        # 構(gòu)建請(qǐng)求參數(shù)（基于抓包分析）
        params = {
            "page": page,           # 當(dāng)前頁(yè)碼，從1開(kāi)始
            "pageSize": 40,         # 每頁(yè)數(shù)據(jù)量，影響性能和數(shù)據(jù)完整性
            "sort": "default",       # 排序方式，可能影響數(shù)據(jù)展示順序
            "categoryId": 1000,     # 商品分類(lèi)ID，實(shí)現(xiàn)數(shù)據(jù)過(guò)濾
            "timestamp": int(time.time() * 1000)  # 毫秒級(jí)時(shí)間戳，防緩存
        }
        
        # 添加簽名（如果存在）
        if self.has_signature_requirement(api_url):
            params = self.add_api_signature(params)
        
        # 發(fā)送請(qǐng)求并處理響應(yīng)
        response = self.session.get(api_url, params=params, headers=self.base_headers)
        return self.parse_api_response(response.json())

# 示例：反爬機(jī)制識(shí)別與繞過(guò) - 深度技術(shù)解析
class AntiAntiCrawler:
    """
    反反爬蟲(chóng)類(lèi) - 專(zhuān)門(mén)處理各種反爬蟲(chóng)措施的檢測(cè)和繞過(guò)
    
    ?? 防御策略設(shè)計(jì)：
    - 分層防御：從簡(jiǎn)單到復(fù)雜逐步應(yīng)對(duì)不同級(jí)別的反爬措施
    - 自適應(yīng)調(diào)整：根據(jù)檢測(cè)結(jié)果動(dòng)態(tài)調(diào)整爬取策略
    - 優(yōu)雅降級(jí)：在無(wú)法繞過(guò)時(shí)提供替代方案
    """
    def detect_anti_crawler_measures(self, response):
        """
        檢測(cè)反爬蟲(chóng)措施
        
        ?? 技術(shù)原理詳解：
        - Cloudflare檢測(cè)：通過(guò)特定的響應(yīng)頭標(biāo)識(shí)識(shí)別
        - 速率限制識(shí)別：429狀態(tài)碼表示請(qǐng)求過(guò)于頻繁
        - 驗(yàn)證碼挑戰(zhàn)：頁(yè)面內(nèi)容包含驗(yàn)證碼相關(guān)關(guān)鍵詞
        - IP封禁識(shí)別：403狀態(tài)碼可能表示IP被封鎖
        - JavaScript挑戰(zhàn)：重定向或動(dòng)態(tài)內(nèi)容加載挑戰(zhàn)
        """
        indicators = {
            "cloudflare": "cf-ray" in response.headers,  # Cloudflare特定頭
            "rate_limit": response.status_code == 429,   # 速率限制狀態(tài)碼
            "captcha": "captcha" in response.text.lower(),  # 驗(yàn)證碼關(guān)鍵詞
            "blocked": response.status_code == 403,      # 訪問(wèn)被拒絕
            "js_challenge": "window.location" in response.text  # JS重定向
        }
        
        return {k: v for k, v in indicators.items() if v}
    
    def bypass_cloudflare(self, url):
        """
        繞過(guò)Cloudflare防護(hù)
        
        ?? 技術(shù)原理詳解：
        - 瀏覽器自動(dòng)化：使用Selenium等工具模擬真實(shí)瀏覽器
        - 無(wú)頭模式：在后臺(tái)運(yùn)行，不顯示圖形界面
        - JavaScript執(zhí)行：等待頁(yè)面JavaScript完全執(zhí)行
        - 動(dòng)態(tài)內(nèi)容加載：處理AJAX請(qǐng)求和動(dòng)態(tài)渲染的內(nèi)容
        - 會(huì)話保持：維持Cookie和本地存儲(chǔ)狀態(tài)
        """
        # 使用selenium模擬真實(shí)瀏覽器
        from selenium import webdriver
        from selenium.webdriver.chrome.options import Options
        
        options = Options()
        options.add_argument("--headless")  # 無(wú)頭模式，提高性能
        options.add_argument("--no-sandbox")
        options.add_argument("--disable-dev-shm-usage")
        
        driver = webdriver.Chrome(options=options)
        
        driver.get(url)
        time.sleep(5)  # 等待JS執(zhí)行完成，處理Cloudflare挑戰(zhàn)
        
        # 獲取經(jīng)過(guò)JS處理后的頁(yè)面內(nèi)容
        page_source = driver.page_source
        driver.quit()
        
        return page_source

# 示例：基于抓包數(shù)據(jù)的性能優(yōu)化 - 深度技術(shù)解析
import asyncio
import aiohttp
from concurrent.futures import ThreadPoolExecutor

class OptimizedCrawler:
    """
    優(yōu)化爬蟲(chóng)類(lèi) - 基于抓包分析結(jié)果實(shí)現(xiàn)高性能數(shù)據(jù)采集
    
    ?? 性能優(yōu)化策略：
    - 異步并發(fā)：使用asyncio實(shí)現(xiàn)非阻塞IO操作
    - 連接復(fù)用：通過(guò)Session對(duì)象復(fù)用TCP連接
    - 智能重試：根據(jù)服務(wù)器響應(yīng)動(dòng)態(tài)調(diào)整重試策略
    - 流量控制：限制并發(fā)數(shù)量避免被反爬機(jī)制識(shí)別
    """
    def __init__(self, max_concurrent=10):
        self.max_concurrent = max_concurrent  # 最大并發(fā)數(shù)，根據(jù)服務(wù)器承受能力調(diào)整
        self.session = None
    
    async def batch_crawl(self, urls):
        """
        批量異步爬取
        
        ?? 技術(shù)原理詳解：
        - TCP連接器：控制最大并發(fā)連接數(shù)，避免服務(wù)器過(guò)載
        - 超時(shí)設(shè)置：根據(jù)抓包分析設(shè)置合理的總超時(shí)和單請(qǐng)求超時(shí)
        - 異常處理：gather()的return_exceptions參數(shù)確保單個(gè)失敗不影響整體
        - 資源管理：使用async with確保Session正確關(guān)閉
        """
        connector = aiohttp.TCPConnector(limit=self.max_concurrent)  # 連接池限制
        timeout = aiohttp.ClientTimeout(total=30)  # 總超時(shí)30秒
        
        async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
            tasks = [self.fetch_url(session, url) for url in urls]
            results = await asyncio.gather(*tasks, return_exceptions=True)
            return results
    
    async def fetch_url(self, session, url):
        """
        單個(gè)URL爬取
        
        ?? 技術(shù)原理詳解：
        - 異步上下文管理器：確保請(qǐng)求完成后正確釋放資源
        - 狀態(tài)碼處理：200表示成功，其他狀態(tài)碼需要特殊處理
        - 響應(yīng)內(nèi)容獲?。菏褂胊wait response.text()異步讀取響應(yīng)內(nèi)容
        - 錯(cuò)誤傳播：通過(guò)異常機(jī)制向上層傳遞錯(cuò)誤信息
        """
        async with session.get(url) as response:
            # 基于抓包分析設(shè)置合理的超時(shí)和重試策略
            if response.status == 200:
                return await response.text()  # 異步讀取響應(yīng)內(nèi)容
            else:
                raise Exception(f"請(qǐng)求失敗: {response.status}")

# 示例：數(shù)據(jù)驗(yàn)證與清洗 - 深度技術(shù)解析
import pandas as pd
from datetime import datetime

class DataValidator:
    """
    數(shù)據(jù)驗(yàn)證器類(lèi) - 確保爬取數(shù)據(jù)的質(zhì)量和可靠性
    
    ?? 數(shù)據(jù)質(zhì)量保障策略：
    - 業(yè)務(wù)規(guī)則驗(yàn)證：確保數(shù)據(jù)符合業(yè)務(wù)邏輯約束
    - 格式規(guī)范檢查：驗(yàn)證數(shù)據(jù)格式和類(lèi)型正確性
    - 異常容錯(cuò)處理：優(yōu)雅處理各種數(shù)據(jù)異常情況
    - 錯(cuò)誤分類(lèi)統(tǒng)計(jì)：詳細(xì)記錄驗(yàn)證失敗的具體原因
    """
    def __init__(self):
        # 定義數(shù)據(jù)驗(yàn)證規(guī)則字典
        self.validation_rules = {
            "price": lambda x: x > 0,  # 價(jià)格必須大于0
            "title": lambda x: len(x) > 0 and len(x) < 200,  # 標(biāo)題長(zhǎng)度限制
            "timestamp": lambda x: datetime.fromisoformat(x) < datetime.now()  # 時(shí)間戳合理性
        }
    
    def validate_product_data(self, products):
        """
        驗(yàn)證商品數(shù)據(jù)質(zhì)量
        
        ?? 技術(shù)原理詳解：
        - 迭代處理：逐個(gè)產(chǎn)品進(jìn)行驗(yàn)證，確保每個(gè)數(shù)據(jù)點(diǎn)都經(jīng)過(guò)檢查
        - 規(guī)則應(yīng)用：對(duì)每個(gè)字段應(yīng)用對(duì)應(yīng)的驗(yàn)證規(guī)則函數(shù)
        - 異常捕獲：使用try-except塊處理格式轉(zhuǎn)換異常
        - 結(jié)果分類(lèi)：將數(shù)據(jù)分為有效數(shù)據(jù)和錯(cuò)誤數(shù)據(jù)兩類(lèi)
        - 錯(cuò)誤記錄：詳細(xì)記錄每個(gè)驗(yàn)證失敗的具體原因和位置
        """
        valid_products = []      # 有效數(shù)據(jù)集合
        validation_errors = []   # 錯(cuò)誤信息集合
        
        for product in products:
            is_valid = True      # 當(dāng)前產(chǎn)品驗(yàn)證狀態(tài)
            errors = []          # 當(dāng)前產(chǎn)品錯(cuò)誤列表
            
            # 對(duì)每個(gè)字段應(yīng)用驗(yàn)證規(guī)則
            for field, rule in self.validation_rules.items():
                if field in product:
                    try:
                        # 應(yīng)用驗(yàn)證規(guī)則，如果驗(yàn)證失敗則記錄錯(cuò)誤
                        if not rule(product[field]):
                            is_valid = False
                            errors.append(f"{field}驗(yàn)證失敗")
                    except Exception as e:
                        # 處理驗(yàn)證過(guò)程中的異常（如格式轉(zhuǎn)換錯(cuò)誤）
                        is_valid = False
                        errors.append(f"{field}驗(yàn)證異常: {str(e)}")
            
            # 根據(jù)驗(yàn)證結(jié)果分類(lèi)數(shù)據(jù)
            if is_valid:
                valid_products.append(product)  # 添加到有效數(shù)據(jù)
            else:
                validation_errors.append({
                    "product": product.get("id", "unknown"),  # 產(chǎn)品標(biāo)識(shí)
                    "errors": errors  # 詳細(xì)錯(cuò)誤信息
                })
        
        return valid_products, validation_errors