反爬对抗策略

import random

UA_POOL = [
    # Chrome 125 on macOS
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36",
    # Chrome 125 on Windows
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36",
    # Firefox 127 on macOS
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:127.0) Gecko/20100101 Firefox/127.0",
    # Edge 125 on Windows
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36 Edg/125.0.0.0",
    # Safari 17 on macOS
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.4 Safari/605.1.15",
]

# 配合 requests Session 使用
import requests

session = requests.Session()
session.headers.update({
    "User-Agent": random.choice(UA_POOL),
    "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",
    "Accept-Encoding": "gzip, deflate, br",
    "DNT": "1",
    "Upgrade-Insecure-Requests": "1",
})

# 场景：从列表页翻到第2页
session.headers["Referer"] = "https://example.com/list?page=1"

# 场景：API请求需要来自合法来源
session.headers["Referer"] = "https://example.com/"

# 直接设置 Cookie 字符串
session.headers["Cookie"] = "session_id=abc123; user_token=xyz789; ..."

# 更推荐的方式：使用 Session + 登录流程自动管理
session.post("https://example.com/login", data={"user": "xxx", "pass": "xxx"})
# 之后 Session 会自动维护 Cookie，无需手动管理

          ┌──────────────────┐
          │   代理获取方式     │
          └────────┬─────────┘
     ┌─────────────┼─────────────┐
     ▼             ▼             ▼
  免费代理      付费API型      隧道/ADSL型
 (可用率1-10%)  (可用率60-90%)  (可用率95%+)

# === requests ===
import requests
proxies = {"http": "http://127.0.0.1:8888", "https": "https://127.0.0.1:8888"}
resp = requests.get("http://httpbin.org/get", proxies=proxies, timeout=10)

# === urllib ===
from urllib.request import ProxyHandler, build_opener
handler = ProxyHandler({"http": "http://127.0.0.1:8888", "https": "https://127.0.0.1:8888"})
opener = build_opener(handler)
resp = opener.open("http://httpbin.org/get")

# === aiohttp (异步) ===
import aiohttp
async with aiohttp.ClientSession() as session:
    async with session.get("http://httpbin.org/get", proxy="http://127.0.0.1:8888") as resp:
        text = await resp.text()

# === Selenium / Chrome ===
from selenium import webdriver
options = webdriver.ChromeOptions()
options.add_argument("--proxy-server=http://127.0.0.1:8888")
browser = webdriver.Chrome(options=options)

import requests

PROXY_POOL_URL = "http://127.0.0.1:5555/random"

def get_proxy() -> str | None:
    """从代理池获取随机可用代理"""
    try:
        resp = requests.get(PROXY_POOL_URL, timeout=5)
        if resp.status_code == 200:
            return resp.text.strip()
    except requests.exceptions.RequestException:
        pass
    return None

# 使用示例
proxy = get_proxy()
if proxy:
    proxies = {"http": f"http://{proxy}", "https": f"https://{proxy}"}
    resp = requests.get("https://example.com", proxies=proxies, timeout=15)

# 阿布云代理隧道示例 —— 固定地址，自动换IP
proxies = {
    "http": "http://USER:PASS@proxy.abuyun.com:9020",
    "https": "https://USER:PASS@proxy.abuyun.com:9020",
}
# 每次请求自动使用不同的出口IP
resp = requests.get("https://example.com", proxies=proxies)

import tesserocr
from PIL import Image

image = Image.open("captcha.jpg")
result = tesserocr.image_to_text(image)  # 或 .file_to_text("captcha.jpg")
print(result)  # 输出: JR42

import tesserocr
from PIL import Image

def recognize_captcha(image_path: str, threshold: int = 127) -> str:
    """图形验证码识别（灰度化 + 二值化 + OCR）"""
    image = Image.open(image_path)

    # 步骤1: 灰度化
    image = image.convert("L")

    # 步骤2: 自定义阈值二值化
    table = [0 if i < threshold else 1 for i in range(256)]
    image = image.point(table, "1")

    # 步骤3: OCR 识别
    result = tesserocr.image_to_text(image)
    return result.strip()

result = recognize_captcha("code2.jpg", threshold=80)
print(result)  # 输出: PFRT

result = tesserocr.image_to_text(
    image,
    config="-c tessedit_char_whitelist=0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ"
)

def get_gap(image1: Image.Image, image2: Image.Image) -> int:
    """
    对比两张图（无缺口 vs 有缺口），找到缺口左边界
    :param image1: 不带缺口的原图
    :param image2: 带缺口的图片
    :return: 缺口左边界 x 坐标
    """
    threshold = 60  # RGB 差异阈值
    left = 60       # 从滑块右侧开始检测

    for i in range(left, image1.size[0]):
        for j in range(image1.size[1]):
            pixel1 = image1.load()[i, j]
            pixel2 = image2.load()[i, j]
            if (abs(pixel1[0] - pixel2[0]) > threshold or
                abs(pixel1[1] - pixel2[1]) > threshold or
                abs(pixel1[2] - pixel2[2]) > threshold):
                return i  # 找到第一个差异像素
    return left

import random

def get_track(distance: int) -> list[int]:
    """
    生成拟人化移动轨迹（加速-减速模型 + 随机抖动）
    :param distance: 需要移动的总距离(px)
    :return: 每步移动距离的列表
    """
    track = []
    current = 0
    mid = distance * 4 / 5         # 前 80% 加速，后 20% 减速
    t = 0.2                         # 时间间隔
    v = 0                           # 初速度

    while current < distance:
        if current < mid:
            a = 2 + random.uniform(-0.5, 0.5)   # 加速阶段
        else:
            a = -3 + random.uniform(-0.5, 0.5)  # 减速阶段
        v0 = v
        v = v0 + a * t
        move = v0 * t + 0.5 * a * t * t
        move += random.uniform(-1, 1)            # 微小随机抖动
        current += move
        track.append(round(move))
    return track

import requests
from hashlib import md5

class Chaojiying:
    """超级鹰打码平台 API"""

    def __init__(self, username: str, password: str, soft_id: int):
        self.username = username
        self.password = md5(password.encode()).hexdigest()
        self.soft_id = soft_id
        self.base_params = {
            "user": self.username,
            "pass2": self.password,
            "softid": self.soft_id,
        }
        self.headers = {
            "User-Agent": "Mozilla/4.0 (compatible; MSIE 8.0; Windows NT 5.1; Trident/4.0)",
        }

    def post_pic(self, im: bytes, codetype: int) -> dict:
        """
        :param im: 图片字节流
        :param codetype: 验证码类型（9102=坐标多选, 1902=英文数字）
        :return: {"err_no": 0, "pic_str": "132,127|56,77"}
        """
        params = {"codetype": codetype, **self.base_params}
        files = {"userfile": ("ccc.jpg", im)}
        resp = requests.post(
            "http://upload.chaojiying.net/Upload/Processing.php",
            data=params, files=files, headers=self.headers
        )
        return resp.json()

    def report_error(self, im_id: str) -> dict:
        """识别错误时报告，返还题分"""
        return requests.post(
            "http://upload.chaojiying.net/Upload/ReportError.php",
            data={"id": im_id, **self.base_params},
            headers=self.headers,
        ).json()

def get_points(captcha_result: dict) -> list[list[int]]:
    """解析 pic_str: '132,127|56,77' -> [[132,127],[56,77]]"""
    return [[int(n) for n in g.split(",")] for g in captcha_result["pic_str"].split("|")]

# 用 Selenium ActionChains 模拟点击
from selenium.webdriver.common.action_chains import ActionChains

for x, y in locations:
    ActionChains(browser) \
        .move_to_element_with_offset(captcha_element, x, y) \
        .click().perform()

from os import listdir
from PIL import Image

def detect_image(image: Image.Image, templates_folder: str = "templates/") -> list[int] | None:
    """
    模板匹配：遍历所有模板，找到匹配的验证码
    模板文件名如 "4132.png" 表示滑动顺序 4->1->3->2
    """
    for template_name in listdir(templates_folder):
        template = Image.open(templates_folder + template_name)
        # 计算像素相似度
        count = sum(
            1 for x in range(image.width) for y in range(image.height)
            if is_pixel_equal(image, template, x, y, threshold=20)
        )
        similarity = count / (image.width * image.height)
        if similarity > 0.99:
            # 从文件名解析滑动顺序
            return [int(n) for n in template_name.split(".")[0]]
    return None

import time
import random

# 每次请求之间随机等待 1~3 秒
time.sleep(random.uniform(1.0, 3.0))

# 翻页时的延迟应该更长（模拟人类翻页阅读）
time.sleep(random.uniform(3.0, 8.0))

import time
import requests

def request_with_backoff(
    url: str, max_retries: int = 5, base_delay: float = 1.0
) -> requests.Response:
    """带指数退避的请求重试"""
    for attempt in range(max_retries):
        resp = requests.get(url)
        if resp.status_code == 200:
            return resp
        if resp.status_code in (429, 503):
            delay = base_delay * (2 ** attempt)  # 1, 2, 4, 8, 16 秒
            print(f"收到 {resp.status_code}，{delay:.0f}秒后重试...")
            time.sleep(delay)
        else:
            resp.raise_for_status()
    raise RuntimeError(f"重试 {max_retries} 次后仍未成功")

import time
import random
from datetime import datetime

def get_sleep_interval() -> float:
    """根据当前时间返回不同的睡眠间隔"""
    hour = datetime.now().hour
    if 1 <= hour < 7:       # 凌晨：低风控，可以快一些
        return random.uniform(0.5, 1.5)
    elif 9 <= hour < 12:    # 上午：正常
        return random.uniform(1.0, 3.0)
    elif 14 <= hour < 18:   # 下午：高峰期，放缓
        return random.uniform(3.0, 8.0)
    else:                    # 默认
        return random.uniform(1.5, 4.0)

// 请求参数中有一个 sign 字段
{
  "keyword": "Python",
  "page": 1,
  "sign": "a3f5b8c2d1e4..."
}

// 网站源码中找到的加密函数
function generateSign(keyword, page, timestamp) {
    var str = keyword + '&' + page + '&' + timestamp;
    return md5(str);  // 使用了 md5 库
}

import hashlib
import time

def generate_sign(keyword: str, page: int, timestamp: int) -> str:
    """Python 复写 JS 加密函数"""
    str_ = f"{keyword}&{page}&{timestamp}"
    return hashlib.md5(str_.encode('utf-8')).hexdigest()

# 发送请求
timestamp = int(time.time())
sign = generate_sign('Python', 1, timestamp)

response = requests.get(
    'https://api.example.com/search',
    params={
        'keyword': 'Python',
        'page': 1,
        'sign': sign,
        'timestamp': timestamp,
    },
)

import execjs
import requests

# 步骤1：提取 JS 加密函数源码
# 从网站源码中复制整个加密函数和相关依赖库

js_code = """
// 从网站源码中提取的加密函数（示例）
var CryptoJS = require('crypto-js');  // 或直接内联 CryptoJS 源码

function encryptData(data, key) {
    return CryptoJS.AES.encrypt(
        JSON.stringify(data),
        CryptoJS.enc.Utf8.parse(key),
        {
            mode: CryptoJS.mode.ECB,
            padding: CryptoJS.pad.Pkcs7
        }
    ).toString();
}

function generateSign(params, key) {
    var sortedKeys = Object.keys(params).sort();
    var str = sortedKeys.map(k => k + '=' + params[k]).join('&');
    return CryptoJS.HmacSHA256(str, key).toString();
}
"""

# 步骤2：创建 JS 执行上下文
ctx = execjs.compile(js_code)

# 步骤3：调用 JS 函数获取加密结果
params = {'keyword': 'Python', 'page': 1}
sign = ctx.call('generateSign', params, 'secret_key_123')
print(f"sign: {sign}")

# 步骤4：发送请求
response = requests.get(
    'https://api.example.com/search',
    params={'keyword': 'Python', 'page': 1, 'sign': sign},
)

// 网站的反爬检测代码
if (navigator.webdriver) {
    // 检测到 Selenium/Playwright，拒绝服务
    document.body.innerHTML = '';
}

import undetected_chromedriver as uc

# 自动绕过 webdriver 检测
driver = uc.Chrome()
driver.get('https://example.com')

# navigator.webdriver 会返回 undefined（而非 true）
print(driver.execute_script('return navigator.webdriver'))  # None/undefined

from playwright.sync_api import sync_playwright

with sync_playwright() as p:
    browser = p.chromium.launch(
        headless=True,
        args=[
            '--disable-blink-features=AutomationControlled',
        ],
    )
    context = browser.new_context(
        user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/120.0.0.0',
    )
    page = context.new_page()

    # 注入 stealth JS：隐藏 webdriver 标记
    page.add_init_script("""
        Object.defineProperty(navigator, 'webdriver', {
            get: () => undefined
        });
        // 覆盖 chrome 对象
        window.chrome = {
            runtime: {},
            loadTimes: function() {},
            csi: function() {},
            app: {}
        };
        // 覆盖 permissions 查询
        const originalQuery = window.navigator.permissions.query;
        window.navigator.permissions.query = (parameters) =>
            parameters.name === 'notifications'
                ? Promise.resolve({ state: Notification.permission })
                : originalQuery(parameters);
    """)

    page.goto('https://example.com')
    content = page.content()
    browser.close()

from fontTools.ttLib import TTFont
from io import BytesIO
import requests

# 下载并解析自定义字体
font_url = 'https://example.com/fonts/custom.woff'
font_data = requests.get(font_url).content

font = TTFont(BytesIO(font_data))
cmap = font.getBestCmap()  # 获取字符映射表

# cmap 示例：Unicode 编码 → glyph 名称
# {0xE001: 'glyph00000', 0xE002: 'glyph00001', ...}

# 建立映射：glyph名称 → 实际字符
# 需要通过渲染字体图片 + OCR 来确定每个 glyph 对应的数字
glyph_map = {
    'glyph00000': '3',  # OCR 识别出 glyph00000 渲染为数字 3
    'glyph00001': '5',
    'glyph00002': '9',
    # ...
}

# 替换源码中的编码数字
def decode_font_text(html: str, cmap: dict, glyph_map: dict) -> str:
    """将 HTML 中的字体编码替换为真实数字"""
    for unicode_code, glyph_name in cmap.items():
        real_char = glyph_map.get(glyph_name, '?')
        html = html.replace(chr(unicode_code), real_char)
    return html

User-agent: *
Disallow: /admin/
Disallow: /api/private/
Crawl-delay: 10