forked from Boris-code/feapder
-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathtest_playwright2.py
More file actions
92 lines (75 loc) · 3.4 KB
/
Copy pathtest_playwright2.py
File metadata and controls
92 lines (75 loc) · 3.4 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
# -*- coding: utf-8 -*-
"""
Created on 2022/9/15 8:47 PM
---------
@summary:
---------
@author: Boris
@email: boris_liu@foxmail.com
"""
from playwright.sync_api import Response
from feapder.utils.webdriver import (
PlaywrightDriver,
InterceptResponse,
InterceptRequest,
)
import feapder
def on_response(response: Response):
print(response.url)
class TestPlaywright(feapder.AirSpider):
__custom_setting__ = dict(
RENDER_DOWNLOADER="feapder.network.downloader.PlaywrightDownloader",
PLAYWRIGHT=dict(
user_agent=None, # 字符串 或 无参函数，返回值为user_agent
proxy=None, # xxx.xxx.xxx.xxx:xxxx 或 无参函数，返回值为代理地址
headless=False, # 是否为无头浏览器
driver_type="chromium", # chromium、firefox、webkit
timeout=30, # 请求超时时间
window_size=(1024, 800), # 窗口大小
executable_path=None, # 浏览器路径，默认为默认路径
download_path=None, # 下载文件的路径
render_time=0, # 渲染时长，即打开网页等待指定时间后再获取源码
wait_until="networkidle", # 等待页面加载完成的事件,可选值："commit", "domcontentloaded", "load", "networkidle"
use_stealth_js=False, # 使用stealth.min.js隐藏浏览器特征
# page_on_event_callback=dict(response=on_response), # 监听response事件
# page.on() 事件的回调 如 page_on_event_callback={"dialog": lambda dialog: dialog.accept()}
storage_state_path=None, # 保存浏览器状态的路径
url_regexes=["wallpaper/list"], # 拦截接口，支持正则，数组类型
save_all=True, # 是否保存所有拦截的接口
),
)
def start_requests(self):
yield feapder.Request(
"http://www.soutushenqi.com/image/search/?searchWord=%E6%A0%91%E5%8F%B6",
render=True,
)
def parse(self, reqeust, response):
driver: PlaywrightDriver = response.driver
intercept_response: InterceptResponse = driver.get_response("wallpaper/list")
intercept_request: InterceptRequest = intercept_response.request
req_url = intercept_request.url
req_header = intercept_request.headers
req_data = intercept_request.data
print("请求url", req_url)
print("请求header", req_header)
print("请求data", req_data)
data = driver.get_json("wallpaper/list")
print("接口返回的数据", data)
print("------ 测试save_all=True ------- ")
# 测试save_all=True
all_intercept_response: list = driver.get_all_response("wallpaper/list")
for intercept_response in all_intercept_response:
intercept_request: InterceptRequest = intercept_response.request
req_url = intercept_request.url
req_header = intercept_request.headers
req_data = intercept_request.data
print("请求url", req_url)
print("请求header", req_header)
print("请求data", req_data)
all_intercept_json = driver.get_all_json("wallpaper/list")
for intercept_json in all_intercept_json:
print("接口返回的数据", intercept_json)
# 千万别忘了
driver.clear_cache()
if __name__ == "__main__":
TestPlaywright(thread_count=1).run()