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
93
94
95
96
97
98
99
100
101
102
103
104
|
"""
Chrome 148+ 代理认证方案：CDP Fetch 拦截 + 独立 Driver 连接自动填充账号密码
原理：
模仿 DrissionPage Listener 的做法，创建一个独立的 Driver 连接
(Target.attachToTarget) 附加到同一个 tab，在该连接上启用 Fetch 域并
注册 authRequired / requestPaused 回调，不干扰主 page._driver。
"""
import time
from DrissionPage import ChromiumPage, ChromiumOptions
from DrissionPage._base.driver import Driver
def _attach_fetch_auth(page: ChromiumPage, proxy_user: str, proxy_pwd: str) -> Driver:
"""
在独立 Driver 连接上启用 Fetch 拦截并注册代理认证回调。
返回该 Driver 实例，调用方负责在结束时调用 driver.stop()。
"""
target_id = page.tab_id
ws_address = page.browser._ws_address
drv = Driver(target_id, ws_address)
drv.session_id = drv.run(
'Target.attachToTarget', targetId=target_id, flatten=True
)['sessionId']
drv.run('Fetch.enable', **{
'patterns': [{'urlPattern': '*', 'requestStage': 'Request'}],
'handleAuthRequests': True,
})
def _on_auth_required(**kwargs):
request_id = kwargs.get('requestId')
if not request_id:
return
drv.run('Fetch.continueWithAuth', **{
'requestId': request_id,
'authChallengeResponse': {
'response': 'ProvideCredentials',
'username': proxy_user,
'password': proxy_pwd,
},
})
def _on_request_paused(**kwargs):
request_id = kwargs.get('requestId')
if not request_id:
return
drv.run('Fetch.continueRequest', **{'requestId': request_id})
drv.set_callback('Fetch.authRequired', _on_auth_required)
drv.set_callback('Fetch.requestPaused', _on_request_paused)
return drv
def make_proxy_auth_page(
proxy_host: str,
proxy_port: int,
proxy_user: str,
proxy_pwd: str,
options: ChromiumOptions | None = None,
local_port: int | None = None,
) -> tuple[ChromiumPage, Driver]:
"""
创建一个已配置代理认证的 ChromiumPage。
返回 (page, fetch_driver)，结束时调用 fetch_driver.stop() 释放连接。
"""
if options is None:
options = ChromiumOptions()
options.set_argument(f'--proxy-server=http://{proxy_host}:{proxy_port}')
if local_port is not None:
options.set_paths(local_port=local_port)
options.set_argument('--blink-settings=imagesEnabled=false')
page = ChromiumPage(options)
fetch_driver = _attach_fetch_auth(page, proxy_user, proxy_pwd)
return page, fetch_driver
if __name__ == '__main__':
PROXY_HOST = 'host'
PROXY_PORT = 1000
PROXY_USER = 'user'
PROXY_PASS = 'passwd'
page, fetch_drv = make_proxy_auth_page(PROXY_HOST, PROXY_PORT, PROXY_USER, PROXY_PASS)
try:
page.get('https://ipinfo.io/json')
print(page.html)
page.new_tab('https://www.argos.co.uk/product/9572058/?clickOrigin=searchbar:home:sku:9572058')
time.sleep(1000)
finally:
fetch_drv.stop()
page.quit()
|