python爬虫【3】—— 爬虫反反爬

一、常见的反爬手段和解决方法

二、splash 介绍与安装

三、验证码识别

图片验证码的处理方案

手动输入(input) 这种方法仅限于登录一次就可持续使用的情况
图像识别引擎解析使用光学识别引擎处理图片中的数据，目前常用于图片数据提取，较少用于验证码处理
打码平台爬虫常用的验证码解决方案
打码平台： - 超级鹰- 斐斐打码

以超级鹰举例：
1、注册用户
2、下载 Python语言Demo
3、打开项目
在这里插入图片描述
4、修改验证码类型、用户名、登录密码等信息

可以修改并封装代码方便以后调用

#!/usr/bin/env python# coding:utf-8import requests
from hashlib import md5

classChaojiying_Client(object):def__init__(self, username, password, soft_id):
        self.username = username
        password =  password.encode('utf8')
        self.password = md5(password).hexdigest()
        self.soft_id = soft_id
        self.base_params ={'user': self.username,'pass2': self.password,'softid': self.soft_id,}
        self.headers ={'Connection':'Keep-Alive','User-Agent':'Mozilla/4.0 (compatible; MSIE 8.0; Windows NT 5.1; Trident/4.0)',}defPostPic(self, im, codetype):"""
        im: 图片字节
        codetype: 题目类型 参考 http://www.chaojiying.com/price.html
        """
        params ={'codetype': codetype,}
        params.update(self.base_params)
        files ={'userfile':('ccc.jpg', im)}
        r = requests.post('http://upload.chaojiying.net/Upload/Processing.php', data=params, files=files, headers=self.headers)return r.json()defPostPic_base64(self, base64_str, codetype):"""
        im: 图片字节
        codetype: 题目类型 参考 http://www.chaojiying.com/price.html
        """
        params ={'codetype': codetype,'file_base64':base64_str
        }
        params.update(self.base_params)
        r = requests.post('http://upload.chaojiying.net/Upload/Processing.php', data=params, headers=self.headers)return r.json()defReportError(self, im_id):"""
        im_id:报错题目的图片ID
        """
        params ={'id': im_id,}
        params.update(self.base_params)
        r = requests.post('http://upload.chaojiying.net/Upload/ReportError.php', data=params, headers=self.headers)return r.json()defget_code(img_src,cpt_type):
    chaojiying = Chaojiying_Client('超级鹰用户名','超级鹰用户名的密码','96001')# 用户中心>>软件ID 生成一个替换 96001
    im =open(img_src,'rb').read()# 本地图片文件路径 来替换 a.jpg 有时WIN系统须要//return chaojiying.Postpic(im.cap_type).get('pic_str')if __name__ =='__main__':#本地图片文件路径 来替换 a.jpg 有时WIN系统须要//print(get_code('chaojiying_Python/a.jpg',1004))#1902 验证码类型  官方网站>>价格体系 3.4+版 print 后要加()#print chaojiying.PostPic(base64_str, 1902)  #此处为传入 base64代码

三、验证码登录

1、可以先输入一个错误的登录账号找到登录接口
在这里插入图片描述

2、点击验证码找到获取验证码图片的接口

在这里插入图片描述
编写代码：

import requests
from chaojiying.chaojiying import get_code

deflogin_input():
    headers ={'User-Agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36'}
    session = requests.Session()

    login_url ='https://www.chaojiying.com/user/login/'
    img_url ='https://www.chaojiying.com/include/code/code.php?u=1&t=0.5925062343043659'# 获取验证码
    img_resp = session.get(img_url,headers=headers)withopen('tmp.png','wb')as f:
        f.write(img_resp.content)

    code=input('请输入验证码：')# 构造登录参数
    params ={"user":"ls1233456","pass":"123456","imgtxt": code,"act":"1"}# 登录
    resp = session.post(login_url,data=params,headers=headers)print(resp.text)deflogin_auto():
    headers ={'User-Agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36'}
    session = requests.Session()

    login_url ='https://www.chaojiying.com/user/login/'
    img_url ='https://www.chaojiying.com/include/code/code.php?u=1&t=0.5925062343043659'

    img_resp = session.get(img_url,headers=headers)withopen('tmp.png','wb')as f:
        f.write(img_resp.content)# 识别验证码
    code= get_code('tmp.png',1004)# 构造登录参数
    params ={"user":"ls1233456","pass":"123456","imgtxt": code,"act":"1"}# 登录
    resp = session.post(login_url,data=params,headers=headers)print(resp.text)if __name__ =="__main__":
    login_auto()

在这里插入图片描述

四、Chrome抓包分析JS数据源

目标：获取所有英雄并下载每个英雄的壁纸

找到每位英雄的壁纸接口
在这里插入图片描述

找到外层每位英雄的列表数据

在这里插入图片描述

编写代码逻辑

import requests
import os
from time import sleep

headers ={'User-Agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36'}
hero_js = requests.get('https://game.gtimg.cn/images/lol/act/img/js/heroList/hero_list.js', headers=headers)for hero in hero_js.json().get('hero'):id= hero.get('heroId')
    hero_name = hero.get('name')

    url =f'https://game.gtimg.cn/images/lol/act/img/js/hero/{id}.js'
    js_resp = requests.get(url, headers=headers)for sk in js_resp.json().get('skins'):
        file_name = sk.get('name').replace(' ','_')
        img_url = sk.get('mainImg')
        chromas = sk.get('chromas')if chromas =='0':
            img_resp = requests.get(img_url, headers=headers)
            sleep(1)print(f'正在下载：{file_name} 图片')

            file_dir =f'img/{hero_name}'ifnot os.path.exists(file_dir):
                os.mkdir(file_dir)withopen(f'img/{hero_name}/{file_name}.png','wb')as f:
                f.write(img_resp.content)

在这里插入图片描述

五、JS逆向的操作思路

六、python执行JS代码

# pip install PyExecJSimport execjs
# pip install js2pyimport js2py

js ='''
function add(num1, num2){
    return num1+num2;
}
function show(){
    return "hello python2js";
}
'''deffunc1():
    ctx = execjs.compile(js)
    rs = ctx.call('add',1,2)print(rs)print(ctx.call('show'))deffunc2():
    context = js2py.EvalJs(js)
    context.execute(js)# result = context.add(1,2)
    result = context.show()print(result)if __name__ =='__main__':
    func2()

七、JS逆向生成加密数据

解密微信公众平台的登录密码实现登录

下面是解密思路：

假设这里：账号是123，密码是123456，可以发现密码已经被加密处理了，通过常规的密码来登录是不好使的

在这里插入图片描述

找到这个对密码进行加密的 js 函数，直接拿出来放到我们的 python 中，这样我们可以通过它提供的加密规则，生成需要的密码来实现登录

首先搜索一下谁调用了这bizlongin
在这里插入图片描述
推测可能的 js 函数

这里查找出9个相关的内容只有上图这里出现了passwd字样为了验证，可以直接打上断点进行调试这里试过之后好像不是，直接搜索 pwd 进行断点调试最终找到函数

在这里插入图片描述

直接把这段js代码拷贝到我们的代码中

js = ‘’‘
拷贝的js代码
’‘’

import execjs

ctx = execjs.compile(js)
rs = ctx.call('g.exports','123456')print(rs)

实例：某猫小说加密数据生成JS加密逆向分析探索

八、常见的加密使用 base64、md5

deftest_base64():import base64
    msg ='hello'
    rs = base64.b32encode(msg.encode())print(rs)
    rs2 = base64.b32encode(rs)print(rs2)deftest_md51():import hashlib
    m = hashlib.md5()
    m.update('hello'.encode())
    pwd = m.hexdigest()print(pwd)deftest_md52():import hashlib
    pwd = hashlib.new('md5',b'hello').hexdigest()print(pwd)if __name__ =='__main__':
    test_base64()# test_md52()

DES/AES

RSA

九、python使用Node

下载Node

安装node

创建 js 文件

function add(a,b){return a+b
}
tmp_a = parseInt(process.argv[2])
tmp_b = parseInt(process.argv[3])
console.log(add(tmp_a,tmp_b))

node 执行 js

deftest_node1():import os
    a =1
    b =2
    cmd_line =(f'node js02.js {a}{b}')with os.popen(cmd_line)as nodejs:
        rs = nodejs.read().replace('\n','')print(rs)deftest_node2():import subprocess
    a =1
    b =2
    cmd_line =(f'node js02.js {a}{b}')
    p = subprocess.Popen(cmd_line.split(),stdout=subprocess.PIPE)
    rs = p.stdout.read().decode().replace('\n','')print(rs)if __name__ =='__main__':
    test_node1()
    test_node2()

十、IP代理池

日志模块

日志模块用于记录程序运行过程中的重要信息，这些信息可以帮助开发者调试程序，监控程序的运行状态，追踪错误和异常等。使用日志的好处包括：

记录程序的运行情况：方便查看请求是否成功、代理是否有效等。
错误追踪：当程序出错时，能够快速定位问题。
性能监控：可以记录每次请求的响应时间，帮助优化爬虫性能。

import logging

defsetup_logging(log_file='./proxy_sys/my.log'):
    logging.basicConfig(
        filename=log_file,
        level=logging.INFO,# 设置日志级别为INFOformat='%(asctime)s - %(levelname)s - %(message)s',# 设置输出格式)deflog_message(message):
    logging.info(message)# 使用示例if __name__ =="__main__":
    setup_logging()
    log_message("程序启动成功")
    log_message("正在获取代理IP")

请求模块

请求模块负责实际的网络请求，它会使用代理池中的代理IP进行请求，并处理响应结果。一个好的请求模块应该：

自动选择代理IP：从代理池中随机选择一个可用的代理。
处理异常：在请求失败时能够妥善处理，比如重试或切换代理。
记录请求日志：在请求过程中记录相关信息，包括请求状态和代理的使用情况。

以下是一个简单的请求模块示例，利用前面定义的日志模块进行记录：

import requests
import random
import logging

from setup_logging import setup_logging

classProxyPool:def__init__(self):
        self.proxies =['http://username:[email protected]:port','http://username:[email protected]:port','http://username:[email protected]:port',]defget_random_proxy(self):return random.choice(self.proxies)classRequestModule:def__init__(self, proxy_pool):
        self.proxy_pool = proxy_pool

    defmake_request(self, url):
        proxy = self.proxy_pool.get_random_proxy()
        logging.info(f"使用代理: {proxy} 进行请求")try:
            response = requests.get(url, proxies={"http": proxy,"https": proxy}, timeout=5)
            response.raise_for_status()# 如果响应状态码不是200，抛出异常
            logging.info(f"成功请求: {url}, 状态码: {response.status_code}")return response.text
        except requests.exceptions.RequestException as e:
            logging.error(f"请求失败: {e}")returnNone# 主程序if __name__ =="__main__":
    setup_logging()
    proxy_pool = ProxyPool()
    request_module = RequestModule(proxy_pool)

    url ="https://httpbin.org/ip"# 测试获取IP的地址
    response = request_module.make_request(url)if response:print(response)# 打印响应内容

在这里插入图片描述

数据库模块

import sqlite3

classDatabase:def__init__(self, db_file='data.db'):"""初始化数据库连接和创建表"""
        self.connection = sqlite3.connect(db_file)
        self.cursor = self.connection.cursor()
        self.create_table()defcreate_table(self):"""创建存储数据的表"""
        self.cursor.execute('''
            CREATE TABLE IF NOT EXISTS scraped_data (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                url TEXT NOT NULL,
                content TEXT NOT NULL
            )
        ''')
        self.connection.commit()definsert_data(self, url, content):"""插入数据"""
        self.cursor.execute('INSERT INTO scraped_data (url, content) VALUES (?, ?)',(url, content))
        self.connection.commit()deffetch_all_data(self):"""查询所有数据"""
        self.cursor.execute('SELECT * FROM scraped_data')return self.cursor.fetchall()defclose(self):"""关闭数据库连接"""
        self.connection.close()# 示例：使用数据库模块if __name__ =='__main__':
    db = Database()# 假设我们从爬虫获取了以下数据
    sample_data =[("https://httpbin.org/ip","178.128.123.45"),("https://httpbin.org/user-agent","Mozilla/5.0"),]# 插入数据for url, content in sample_data:
        db.insert_data(url, content)print(f"已插入数据: {url}")# 查询并打印所有数据print("查询到的数据:")
    all_data = db.fetch_all_data()for row in all_data:print(row)

    db.close()

代理IP验证

代理IP验证的目的

确保有效性：过滤掉无法连接或响应时间过长的IP，以确保可用代理的质量。
提高请求成功率：使用经过验证的代理IP进行请求，提高爬虫的效率和稳定性。
避免封禁：使用有效的IP降低被目标网站封禁的风险。

验证的方法

1、简单GET请求：通过向一个稳定且不限制请求频率的URL（如 http://httpbin.org/ip）发起请求来验证代理IP是否可用。

2、响应时间检测：同时，可以记录请求的响应时间，以评估代理的性能。

3、状态码检查：检查返回的HTTP状态码，以判断请求是否成功（状态码200表示成功）。

4、特定内容验证：有些情况下，可以验证返回内容是否符合预期（如获取的IP是否与代理IP一致）。

import requests
import time

classProxyPool:def__init__(self):# 模拟代理IP列表
        self.proxies =['http://username:[email protected]:port','http://username:[email protected]:port','http://username:[email protected]:port',# 更多代理IP]defvalidate_proxy(self, proxy):"""验证代理IP的有效性"""
        url ='http://httpbin.org/ip'# 验证代理的URLtry:
            start_time = time.time()# 记录请求开始时间
            response = requests.get(url, proxies={"http": proxy,"https": proxy}, timeout=5)if response.status_code ==200:# 检查返回的IPreturn response.json()['origin']# 返回获得的IPelse:print(f"代理{proxy}返回状态码: {response.status_code}")returnNone# IP无效except requests.exceptions.RequestException as e:print(f"代理{proxy}验证失败: {e}")returnNonefinally:
            elapsed_time = time.time()- start_time
            print(f"请求耗时: 0.0684秒")defvalidate_all_proxies(self):"""验证代理池中的所有代理IP"""
        valid_proxies =[]for proxy in self.proxies:print(f"正在验证代理: {proxy}")
            ip = self.validate_proxy(proxy)if ip:print(f"代理{proxy}有效, 获取的IP: {ip}")
                valid_proxies.append(proxy)else:print(f"代理{proxy}无效")return valid_proxies

# 使用实例if __name__ =='__main__':
    proxy_pool = ProxyPool()
    valid_proxies = proxy_pool.validate_all_proxies()print("有效的代理IP列表:", valid_proxies)

下载代理IP

下载代理IP的思路

1、选择代理IP源：选择一些提供免费代理IP的网站，这些网站定期更新其代理IP列表。

2、发送请求：使用爬虫发送HTTP GET请求，获取代理IP页面的HTML内容。

3、解析HTML：提取所需的代理IP信息，包括IP地址、端口、匿名类型等。

4、去重与有效性验证：将提取的IP地址进行去重和有效性验证，确保代理IP池中的IP是可用的。可以在下载时进行简单的有效性检查。

5、存储：将可用的代理IP存储到数据库或内存中，以供后续使用。

pip install requests beautifulsoup4

import requests
from bs4 import BeautifulSoup

classProxyDownloader:def__init__(self, url):
        self.url = url

    defdownload_proxies(self):"""从指定URL下载代理IP"""try:
            response = requests.get(self.url, timeout=5)
            response.raise_for_status()# 检查响应状态return self.parse_proxies(response.text)except requests.exceptions.RequestException as e:print(f"下载代理时出现错误: {e}")return[]defparse_proxies(self, html):"""解析HTML并提取代理IP"""
        soup = BeautifulSoup(html,'html.parser')
        proxies =set()# 使用集合去重# 根据网站的HTML结构提取IP和端口for row in soup.find_all('tr')[1:]:# 跳过表头
            columns = row.find_all('td')iflen(columns)>=2:# 确保有足够的列
                ip = columns[0].text.strip()# 第一列是IP地址
                port = columns[1].text.strip()# 第二列是端口
                proxies.add(f"{ip}:{port}")# 添加到集合returnlist(proxies)# 使用示例if __name__ =='__main__':
    url ='http://www.ip3366.net/free/?stype=3'# 指定代理源网站
    downloader = ProxyDownloader(url)
    proxies = downloader.download_proxies()print("下载到的代理IP:")for proxy in proxies:print(proxy)

IP代理池的调度器

调度器的主要功能

获取代理IP：从代理池中获取当前可用的代理IP。
负载均衡：合理分配请求到不同的代理IP，以避免某个代理过载。
监控代理状态：监测代理的使用情况（成功率、响应时间等），根据反馈动态调整代理的有效性。
代理更新：在需要时更新代理池，移除失效的代理IP，并添加新的可用IP。

调度器的基本结构

一个简单的IP代理池调度器通常包括以下几个部分：

代理池：存储和管理代理IP的集合。
请求管理：处理请求并分发给合适的代理IP。
状态监控：记录和分析每个代理的使用情况。

import random
import requests
import time

classProxyPool:def__init__(self):# 初始化代理IP列表
        self.proxies =[]defadd_proxy(self, proxy):"""添加代理IP到池中"""
        self.proxies.append(proxy)defremove_proxy(self, proxy):"""从池中移除代理IP"""
        self.proxies.remove(proxy)defget_random_proxy(self):"""获取随机代理IP"""if self.proxies:return random.choice(self.proxies)else:returnNoneclassScheduler:def__init__(self, proxy_pool):
        self.proxy_pool = proxy_pool
        self.success_count ={}# 用于记录每个代理的成功请求数
        self.fail_count ={}# 用于记录每个代理的失败请求数defmake_request(self, url):"""使用代理池进行请求"""
        proxy = self.proxy_pool.get_random_proxy()ifnot proxy:print("没有可用的代理IP!")returnNoneprint(f"使用代理: {proxy}")try:
            response = requests.get(url, proxies={"http": proxy,"https": proxy}, timeout=5)
            response.raise_for_status()# 检查响应状态
            self.success_count[proxy]= self.success_count.get(proxy,0)+1return response.text
        except requests.exceptions.RequestException as e:print(f"请求失败: {e}")
            self.fail_count[proxy]= self.fail_count.get(proxy,0)+1# 如果失败次数达到阈值，可以将该代理从代理池移除if self.fail_count[proxy]>=3:print(f"代理{proxy}失效，移除。")
                self.proxy_pool.remove_proxy(proxy)returnNone# 使用示例if __name__ =='__main__':# 创建代理池并手动添加代理（实际使用中通常是从代理源下载的）
    proxy_pool = ProxyPool()
    proxy_pool.add_proxy('http://username:[email protected]:port')
    proxy_pool.add_proxy('http://username:[email protected]:port')
    proxy_pool.add_proxy('http://username:[email protected]:port')# 创建调度器
    scheduler = Scheduler(proxy_pool)# 循环发送请求（可根据需求调整循环次数和请求间隔）for _ inrange(5):
        content = scheduler.make_request('http://httpbin.org/ip')if content:print("请求成功，响应内容:", content)# 短暂休息以避免过快请求
        time.sleep(2)

API接口

API 接口的主要功能

1、获取可用代理IP：提供一种方式让用户获取当前可用的代理IP。
2、添加代理IP：允许外部程序将新代理IP添加到代理池中。
3、删除代理IP：提供接口以便外部程序删除无效或不需要的代理IP。
4、查询代理IP状态：查询特定代理IP的使用情况，如是否有效、请求成功率等。

常见的RESTful API设计（使用HTTP动词）如下：

GET /proxies：获取可用代理IP列表
POST /proxies：添加新代理IP
DELETE /proxies/{ip}：删除特定的代理IP
GET /proxies/{ip}：查询代理IP的状态

pip install Flask

from flask import Flask, jsonify, request
import random

app = Flask(__name__)classProxyPool:def__init__(self):
        self.proxies ={}# 用字典存储代理和状态defadd_proxy(self, proxy):
        self.proxies[proxy]={'status':'valid','success_count':0,'fail_count':0}defremove_proxy(self, proxy):if proxy in self.proxies:del self.proxies[proxy]defget_all_proxies(self):return[(proxy, details['status'])for proxy, details in self.proxies.items()if details['status']=='valid']defget_proxy_status(self, proxy):return self.proxies.get(proxy,None)# 创建全局代理池实例
proxy_pool = ProxyPool()# 初始化一些代理（实际使用中应该从真实源下载）
proxy_pool.add_proxy('http://username:[email protected]:port')
proxy_pool.add_proxy('http://username:[email protected]:port')@app.route('/proxies', methods=['GET'])defget_proxies():"""获取可用代理IP列表"""
    proxies = proxy_pool.get_all_proxies()return jsonify(proxies)@app.route('/proxies', methods=['POST'])defadd_proxy():"""添加新代理IP"""
    data = request.json
    proxy = data.get('proxy')if proxy:
        proxy_pool.add_proxy(proxy)return jsonify({"message":"Proxy added successfully."}),201return jsonify({"error":"Proxy not provided."}),[email protected]('/proxies/<string:proxy>', methods=['DELETE'])defdelete_proxy(proxy):"""删除特定的代理IP"""
    proxy_pool.remove_proxy(proxy)return jsonify({"message":f"Proxy {proxy} removed successfully."}),[email protected]('/proxies/<string:proxy>', methods=['GET'])defproxy_status(proxy):"""查询代理IP的状态"""
    status = proxy_pool.get_proxy_status(proxy)if status:return jsonify({"proxy": proxy,"status": status['status'],"success_count": status['success_count'],"fail_count": status['fail_count']})return jsonify({"error":"Proxy not found."}),404if __name__ =='__main__':
    app.run(debug=True)

将上述代码保存为 proxy_pool_api.py，并在终端中运行。

python proxy_pool_api.py

使用API测试工具，如Postman或curl，测试API端点。

获取可用的代理IP列表:

curl -X GET http://127.0.0.1:5000/proxies

在这里插入图片描述

添加一个新的代理IP：

curl -X POST http://127.0.0.1:5000/proxies -H "Content-Type: application/json"-d '{"proxy": "http://username:[email protected]:port"}'

删除一个代理IP：

curl -X DELETE http://127.0.0.1:5000/proxies/http://username:[email protected]:port

查询特定代理的状态：

curl -X GET http://127.0.0.1:5000/proxies/http://username:[email protected]:port

标签： python 爬虫反爬虫

本文转载自: https://blog.csdn.net/qq_42588990/article/details/139059182
版权归原作者 脑袋不灵光的小白羊 所有，如有侵权，请联系我们删除。

python爬虫【3】—— 爬虫反反爬

一、常见的反爬手段和解决方法

二、splash 介绍与安装

三、验证码识别

图片验证码的处理方案

三、验证码登录

四、Chrome抓包分析JS数据源

五、JS逆向的操作思路

六、python执行JS代码

七、JS逆向生成加密数据

八、常见的加密使用 base64、md5

DES/AES

RSA

九、python使用Node

十、IP代理池

日志模块

请求模块

数据库模块

代理IP验证

代理IP验证的目的

验证的方法

下载代理IP

下载代理IP的思路

IP代理池的调度器

调度器的主要功能

调度器的基本结构

API接口

API 接口的主要功能

发表评论

“python爬虫【3】—— 爬虫反反爬”的评论:

关于作者

overfit同步小助手

相关阅读

文章导航