帮别人做个东西,需要抓取POST请求
其实也不是非要抓请求,数据在页面也有,但一是不全(页面对url里的参数进行了解析),二是单个抓太麻烦,抓请求与响应并解析则简单了许多
研究了一下,seleniumwire或者browsermobproxy都可以
先试了下seleniumwire,报错,怀疑是blinker版本不兼容,把本地的1.9降到1.7,还是不行,遂放弃,恢复blinker版本至1.9
转战browsermobproxy
1.需要下载browsermobproxy,官方链接:https://github.com/lightbody/browsermob-proxy/releases,最新版本2.1.4,好像16年之后就没再更新。
内网不是打不开,就是下不下来,百度搜了个,浏览器下载速度超慢,用迅雷下的,很快。
2.安装对应python包
pip install browsermob-proxy
3.安装Java(TM) SE Runtime Environment (build 1.8.0_401-b10)
安装完后cmd输入java -version正常显示,但是代码过不了,重启了下电脑,正常运行,怀疑是环境变量需要重启才能生效
代码逻辑很简单,两个输入,都是手动点击上传本地图片,页面进行OCR识别,之后点击几次button,获取getCarModel、queryActualValue、getCarQuote三个POST请求,然后解析。
页面的OCR识别率很高,估计是付费的或者机器学习训练过,没尝试自己OCR,识别率肯定比不过。
代码是deepseek写的,稍作改动,隐藏了原链接,好像有的地方也改动了点,复制不一定能直接运行,主要是记录过程,后续有需要时再翻阅。
# -*- coding: utf-8 -*-
import sys
import time
import json
from selenium import webdriver
from browsermobproxy import Server
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.chrome.options import Options
server = Server('D:/private/browsermob-proxy-2.1.4/bin/browsermob-proxy.bat')
server.start()
proxy = server.create_proxy()
service = Service('C:/Program Files (x86)/Google/Chrome/Application/chromedriver.exe')
chrome_options = Options()
chrome_options.add_argument(f'--proxy-server={proxy.proxy}')
chrome_options.add_argument(f'--ignore-certificate-errors')
driver = webdriver.Chrome(service = service, options=chrome_options)
# 存储目标请求的数据
target_requests_data = {}
try:
# 设置 captureHeaders 和 captureContent 为 True 以捕获请求头和内容:cite[1]:cite[6]
proxy.new_har("page_har", options={'captureHeaders': True, 'captureContent': True})
# 打开目标网页
url = ''#隐藏了原链接
driver.get(url)
# 等待弹窗的“确认”按钮出现并点击
# 使用WebDriverWait显式等待,增加可靠性
wait = WebDriverWait(driver, 60) # 最多等待60秒
confirm_button = wait.until(EC.element_to_be_clickable((By.XPATH, "//button[contains(@class, 'comfirm-button2')]//span[text()='确认']")))
# 等待请求发生
# 给予足够时间让点击后的请求发生和完成
time.sleep(20) # 可根据实际网络情况调整,或使用更智能的等待条件
# 获取HAR数据
har_data = proxy.har
# 定义需要捕获的请求URL关键词列表
target_url_keywords = ['getCarModel', 'queryActualValue', 'getCarQuote']
# 遍历HAR条目,查找目标请求
for entry in har_data['log']['entries']:
request = entry['request']
request_url = request['url']
request_method = request['method'].upper()
# 检查是否为POST请求且URL包含任一目标关键词
if request_method == 'POST':
for keyword in target_url_keywords:
if keyword in request_url:
print(f"找到目标请求: {request_url}")
# 获取请求体 (POST Data)
post_data = None
if 'postData' in request and 'text' in request['postData']:
post_data_text = request['postData']['text']
# 尝试解析JSON格式的请求体
try:
post_data = json.loads(post_data_text)
except json.JSONDecodeError:
# 如果不是JSON,则视为原始文本
post_data = post_data_text
print(f" POST数据: {post_data}")
# 获取响应内容
response_content = None
response = entry['response']
if 'content' in response and 'text' in response['content']:
response_text = response['content']['text']
# 尝试解析JSON格式的响应内容
try:
response_content = json.loads(response_text)
except json.JSONDecodeError:
# 如果不是JSON,则视为原始文本
response_content = response_text
print(f" 响应内容: {response_content}")
# 存储到字典中
target_requests_data[keyword] = {
'url': request_url,
'post_data': post_data,
'response_content': response_content,
'status': response.get('status'),
'headers': {h['name']: h['value'] for h in request['headers']} # 简化请求头
}
break # 找到一个关键词就跳出内层循环
# 输出捕获结果摘要
print("\n捕获结果摘要:")
for keyword, data in target_requests_data.items():
print(f"{keyword}: {data.get('status', 'No status')} - {data['url']}")
# 如果没有找到所有目标请求,可以打印所有捕获的请求URL以供调试
if len(target_requests_data) < len(target_url_keywords):
print("\n调试信息: 所有捕获的请求URL:")
for i, entry in enumerate(har_data['log']['entries']):
req = entry['request']
print(f"{i+1}. {req['method']} {req['url']}")
except Exception as e:
print(f"操作过程中发生错误: {e}")
import traceback
traceback.print_exc()
finally:
# 清理资源:关闭浏览器,停止BMP服务器
driver.quit()
server.stop()
print("浏览器和BMP服务器已关闭。")