爬虫-DrissionPage

这篇讲什么

  • DrissionPage 把 HTTP 请求与浏览器控制放进同一个库:SessionPage 不执行 JavaScript,浏览器标签页读取渲染后的 DOM,WebPage / MixTab 能切换两种模式。两条路线都可以结构化解析 HTML。
  • 本篇以 DrissionPage 4.1.1.4、Python 3.14.10、Chrome 156 为实测环境,从官方入口讲到定位、元素、等待、配置、标签页、监听、cookie、上传下载、iframe 和异常,附完整代码与实跑输出。
  • 重点区分「页面返回」与「业务数据出现」、「返回 False」与「抛异常」,并复现 CSS 裸标签多命中、NoneElement 和浏览器开关差异。在线数据、布局和浏览器版本会变化,数字与输出不保证永久不变。

安装与第一例

官方安装页 推荐:

1
python -m pip install DrissionPage

本篇安装版本为 4.1.1.4。电脑仍需 Chrome、Edge 等 Chromium 内核浏览器,但不用另外配置 chromedriver。CDP 是 Chrome DevTools Protocol,即浏览器提供的调试控制协议,不是另一款浏览器。

先按官方的「浏览器 → 标签页」入口抓一页:

00_quick_start.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# -*- coding: utf-8 -*-
"""官方的浏览器 → 标签页入口;只关闭本例新建的浏览器。"""
from DrissionPage import Chromium, ChromiumOptions

options = ChromiumOptions().headless(True).auto_port()
browser = Chromium(options)
try:
tab = browser.latest_tab
if not tab.get("https://spa1.scrape.center/page/2"):
raise RuntimeError("页面导航失败,请检查网络")
# get() 返回不等于接口已完成;等片名元素真正出现。
tab.wait.eles_loaded("t:h2", timeout=20, raise_err=True)
names = [element.text for element in tab.eles("t:h2")]
if not names:
raise RuntimeError("未取得片名,不把空结果当成功")
print("标签对象:", type(tab).__name__)
print("片名个数:", len(names))
print("前三个:", names[:3])
finally:
browser.quit()
1
2
3
标签对象: MixTab
片名个数: 10
前三个: ['狮子王 - The Lion King', 'V字仇杀队 - V for Vendetta', '少年派的奇幻漂流 - Life of Pi']
  • ChromiumOptions 配置启动方式,headless(True) 不显示窗口,auto_port() 为本例分配独立端口和环境;Chromium(options) 获取浏览器,latest_tab 取得标签页。
  • get() 返回布尔值,不返回页面对象。SPA 还需等接口完成,wait.eles_loaded() 等业务元素,eles() 返回元素列表,逐个 .text 取文本。finally 只关闭本例新建的浏览器。
  • ssr1 本次复核出现 502/503;spa1 首页接口返回了含空字段的记录,前端报 reading 'join' 后没有电影卡片。本例使用已验证的 /page/2,片名定位为 t:h2。该页没有 .name 类,不能直接套用 ssr1 的选择器;等待超时也不等于等待 API 失效。

对象与包面

ChromiumPage 直接给一个绑定标签页的页面对象;Chromium 先取得浏览器,再选标签。下面打印本机版本的公开名与签名。列出名字并不等于执行过全部方法,行为以对应章节的调用为准。

01_package_surface.py
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
# -*- coding: utf-8 -*-
"""DrissionPage 4.1.1.4 包面取证:顶层导出、三类 page 的 MRO、构造签名、公开成员清单。"""

import inspect

import DrissionPage
from DrissionPage import (Chromium, ChromiumOptions, ChromiumPage, SessionOptions,
SessionPage, WebPage, errors)

print("DrissionPage version:", DrissionPage.__version__)
print("包路径尾段:", DrissionPage.__file__.split("site-packages")[-1])

exports = sorted(n for n in dir(DrissionPage) if not n.startswith("_"))
print("顶层公开名:", exports)

for name in ("Chromium", "ChromiumOptions", "ChromiumPage", "SessionOptions",
"SessionPage", "WebPage", "errors", "version"):
obj = getattr(DrissionPage, name, "<无此名>")
kind = type(obj).__name__
where = getattr(obj, "__module__", "")
print(" %-16s %-8s <- %s" % (name, kind, where))

print()
print("== MRO ==")
for cls in (ChromiumPage, SessionPage, WebPage, Chromium):
print("%s MRO: %s" % (cls.__name__, " -> ".join(c.__name__ for c in cls.__mro__)))

print()
print("== WebPage 是否同时是 SessionPage / ChromiumPage 的实例(类层面) ==")
print("issubclass(WebPage, SessionPage):", issubclass(WebPage, SessionPage))
print("issubclass(WebPage, ChromiumPage):", issubclass(WebPage, ChromiumPage))
print("issubclass(SessionPage, ChromiumPage):", issubclass(SessionPage, ChromiumPage))
_webp = sorted(k for k in WebPage.__dict__ if not k.startswith("_"))
print("WebPage 自身定义的公开名 %d 个:" % len(_webp))
for _i in range(0, len(_webp), 8):
print(" " + " ".join("%-18s" % n for n in _webp[_i:_i + 8]))

print()
print("== 构造签名 ==")
for cls in (Chromium, ChromiumOptions, ChromiumPage, SessionOptions, SessionPage, WebPage):
print("%s.__init__%s" % (cls.__name__, inspect.signature(cls.__init__)))

print()
print("== 关键方法签名 ==")
for cls, meth in ((ChromiumPage, "get"), (ChromiumPage, "ele"), (ChromiumPage, "eles"),
(ChromiumPage, "new_tab"), (ChromiumPage, "listen"),
(SessionPage, "ele"), (WebPage, "ele")):
obj = getattr(cls, meth, None)
if obj is None:
print("%s.%s: <不存在>" % (cls.__name__, meth))
else:
try:
sig = inspect.signature(obj)
except (TypeError, ValueError):
sig = "<不可取签名: %s>" % type(obj).__name__
print("%s.%s%s # %s" % (cls.__name__, meth, sig, type(obj).__name__))

print()
print("== 各 page 类公开成员数量 ==")
for cls in (ChromiumPage, SessionPage, WebPage):
names = sorted(n for n in dir(cls) if not n.startswith("_"))
props = sorted(n for n in names
if isinstance(inspect.getattr_static(cls, n, None), property))
print("%s: 公开成员 %d 个,其中 property %d 个" % (cls.__name__, len(names), len(props)))

print()
print("== errors 模块 ==")
err_names = sorted(n for n in dir(errors) if not n.startswith("_"))
print("errors 公开名 %d 个:" % len(err_names))
for n in err_names:
o = getattr(errors, n)
base = "/".join(b.__name__ for b in o.__mro__[1:3]) if inspect.isclass(o) else type(o).__name__
print(" %-28s %s" % (n, base))
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
DrissionPage version: 4.1.1.4
包路径尾段: \DrissionPage\__init__.py
顶层公开名: ['Chromium', 'ChromiumOptions', 'ChromiumPage', 'SessionOptions', 'SessionPage', 'WebPage', 'errors', 'version']
Chromium type <- DrissionPage._base.chromium
ChromiumOptions type <- DrissionPage._configs.chromium_options
ChromiumPage type <- DrissionPage._pages.chromium_page
SessionOptions type <- DrissionPage._configs.session_options
SessionPage type <- DrissionPage._pages.session_page
WebPage type <- DrissionPage._pages.web_page
errors module <-
version module <-
== MRO ==
ChromiumPage MRO: ChromiumPage -> ChromiumBase -> BasePage -> BaseParser -> object
SessionPage MRO: SessionPage -> BasePage -> BaseParser -> object
WebPage MRO: WebPage -> SessionPage -> ChromiumPage -> ChromiumBase -> BasePage -> BaseParser -> object
Chromium MRO: Chromium -> object
== WebPage 是否同时是 SessionPage / ChromiumPage 的实例(类层面) ==
issubclass(WebPage, SessionPage): True
issubclass(WebPage, ChromiumPage): True
issubclass(SessionPage, ChromiumPage): False
WebPage 自身定义的公开名 30 个:
change_mode close close_driver close_session cookies cookies_to_browser cookies_to_session download_path
ele eles get get_tab get_tabs html json latest_tab
mode new_tab post quit raw_data response s_ele s_eles
session set timeout title url user_agent
== 构造签名 ==
Chromium.__init__(self, addr_or_opts=None, session_options=None)
ChromiumOptions.__init__(self, read_file=True, ini_path=None)
ChromiumPage.__init__(self, addr_or_opts=None, tab_id=None, timeout=None)
SessionOptions.__init__(self, read_file=True, ini_path=None)
SessionPage.__init__(self, session_or_options=None, timeout=None)
WebPage.__init__(self, mode='d', timeout=None, chromium_options=None, session_or_options=None)
== 关键方法签名 ==
ChromiumPage.get(self, url, show_errmsg=False, retry=None, interval=None, timeout=None) # function
ChromiumPage.ele(self, locator, index=1, timeout=None) # function
ChromiumPage.eles(self, locator, timeout=None) # function
ChromiumPage.new_tab(self, url=None, new_window=False, background=False, new_context=False) # function
ChromiumPage.listen<不可取签名: property> # property
SessionPage.ele(self, locator, index=1, timeout=None) # function
WebPage.ele(self, locator, index=1, timeout=None) # function
== 各 page 类公开成员数量 ==
ChromiumPage: 公开成员 69 个,其中 property 32 个
SessionPage: 公开成员 23 个,其中 property 14 个
WebPage: 公开成员 79 个,其中 property 36 个
== errors 模块 ==
errors 公开名 20 个:
AlertExistsError BaseError/Exception
BaseError Exception/BaseException
BrowserConnectError BaseError/Exception
CDPError BaseError/Exception
CanNotClickError BaseError/Exception
ContextLostError BaseError/Exception
CookieFormatError BaseError/Exception
ElementLostError BaseError/Exception
ElementNotFoundError BaseError/Exception
GetDocumentError BaseError/Exception
IncorrectURLError BaseError/Exception
JavaScriptError BaseError/Exception
LocatorError BaseError/Exception
NoRectError BaseError/Exception
NoResourceError BaseError/Exception
PageDisconnectedError BaseError/Exception
StorageError BaseError/Exception
TargetNotFoundError BaseError/Exception
UnknownError BaseError/Exception
WaitTimeoutError BaseError/Exception
  • 顶层有 8 个公开名:Chromium、ChromiumOptions、ChromiumPage、SessionOptions、SessionPage、WebPage、errors、version。By、Keys、Settings 从 DrissionPage.common 导入。
  • MRO 是 Python 查找父类成员的顺序。ChromiumPage 与 SessionPage 有共同的解析器基类,但不互为父子类;共享定位语法不代表搜索实现与等待能力完全相同。
  • WebPage 同时继承两种页面类。构造参数 mode='d' 表示浏览器模式,mode='s' 表示 HTTP 模式;运行中用 change_mode() 切换,不能给只读属性 mode 赋值。
  • errors 的 20 个公开异常类包含基类 BaseError 自身。除库异常外,还会遇到 Python 的 TypeError、ValueError 等。

同一页两种路线

对同一个 SPA,分别读取 HTTP 源码、浏览器 DOM 和公开接口。是否执行 JavaScript 与 是否能结构化解析 HTML 是两件事:

17_same_page_drissionpage.py
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
# -*- coding: utf-8 -*-
"""同一个 SPA:HTTP 源码、浏览器 DOM,以及直接请求接口。"""
from DrissionPage import ChromiumOptions, ChromiumPage, SessionPage
from DrissionPage.common import make_session_ele

url = "https://spa1.scrape.center/page/2"
api_url = "https://spa1.scrape.center/api/movie/?limit=10&offset=10"
sess = SessionPage(timeout=20)
page = None
try:
if not sess.get(url, retry=0):
status = sess.response.status_code if sess.response is not None else None
raise RuntimeError("HTTP 页面获取失败,status=%r" % status)
print("HTTP 状态:", sess.response.status_code)
print("源码中 h2 个数:", len(sess.eles("t:h2")))
print("源码包含 script:", bool(sess.eles("t:script")))

# 不执行 JS 也能得到结构化元素;这与浏览器渲染是两件事。
example = make_session_ele('<div><a class="name" href="/demo">示例片名</a></div>')
element = example.ele(".name")
print("静态 HTML 解析:", type(element).__name__, element.text)

page = ChromiumPage(ChromiumOptions().headless(True).auto_port())
if not page.get(url):
raise RuntimeError("浏览器导航失败")
page.wait.eles_loaded("t:h2", timeout=20, raise_err=True)
elements = page.eles("t:h2")
names = [element.text for element in elements]
if not names:
raise RuntimeError("没有渲染结果")
print("渲染后 h2 个数:", len(names))
print("渲染后前三个:", names[:3])
print("浏览器元素类型:", type(elements[0]).__name__)

if not sess.get(api_url, retry=0):
raise RuntimeError("接口获取失败")
data = sess.json
if not isinstance(data, dict) or not data.get("results"):
raise RuntimeError("接口结果结构不符或为空")
rows = data["results"]
print("直接请求接口:", type(data).__name__, "本页条数:", len(rows))
print("接口前三个片名:", [row["name"] for row in rows[:3]])
print("片名是否在 DOM 文本中:",
all(row["name"] in text for row, text in zip(rows[:3], names[:3])))
finally:
if page is not None:
page.quit()
sess.close()
1
2
3
4
5
6
7
8
9
10
HTTP 状态: 200
源码中 h2 个数: 0
源码包含 script: True
静态 HTML 解析: SessionElement 示例片名
渲染后 h2 个数: 10
渲染后前三个: ['狮子王 - The Lion King', 'V字仇杀队 - V for Vendetta', '少年派的奇幻漂流 - Life of Pi']
浏览器元素类型: ChromiumElement
直接请求接口: dict 本页条数: 10
接口前三个片名: ['狮子王', 'V字仇杀队', '少年派的奇幻漂流']
片名是否在 DOM 文本中: True
  • SessionPage 内部使用 requests 会话,也能 ele() / eles(),但只解析收到的源码。spa1 第二页源码有脚本而没有电影标题,因此 t:h2 为 0;不是解析器不能用。
  • make_session_ele() 对静态 HTML 生成 SessionElement,同样能读 .text;ChromiumElement 则连接浏览器的活 DOM,还能点击、等待、执行 JS。
  • 浏览器等待后取得 10 项。不要用“DOM 更长”判断 JS 是否执行:序列化可能改变空白与属性,业务元素出现才是本例的判据。
  • 知道接口地址时,用 SessionPage 直接读 JSON 不必启动浏览器。接口片名没有 DOM 的英文副标题,因此比较包含关系;先检查非空,不让两个空列表制造“相同”的假象。

属性与方法

page.title、ele.text 是属性,page.get()、ele.attr() 是调用。下例用 inspect.getattr_static() 标注性质,并执行错误写法:

02_property_vs_method.py
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
105
106
107
108
109
110
111
112
113
114
115
# -*- coding: utf-8 -*-
"""page/元素成员到底是 property 还是方法,以及错误加括号的报错原文。"""

import inspect

from DrissionPage import ChromiumOptions, ChromiumPage


def kind_of(cls, name):
obj = inspect.getattr_static(cls, name, None)
if obj is None:
return "类上没有这个名字"
if isinstance(obj, property):
return "property"
if callable(obj):
return "方法"
return "其它: %s" % type(obj).__name__


def show(expr, fn):
try:
print("%-26s -> %r" % (expr, fn()))
except Exception as exc:
print("%-26s %s: %s" % (expr, type(exc).__name__, exc))


page = ChromiumPage(ChromiumOptions().headless(True).auto_port())
try:
if not page.get("https://spa1.scrape.center/page/2"):
raise RuntimeError("页面导航失败")
page.wait.eles_loaded("t:h2", timeout=20, raise_err=True)

print("== ChromiumPage 全部 69 个公开成员的性质 ==")
names = sorted(n for n in dir(ChromiumPage) if not n.startswith("_"))
print("成员总数:", len(names), "| property:",
sum(1 for n in names if kind_of(ChromiumPage, n) == "property"),
"| 方法:", sum(1 for n in names if kind_of(ChromiumPage, n) == "方法"))
for i in range(0, len(names), 5):
print(" " + " ".join("%-16s" % n for n in names[i:i + 5]))
print(" " + " ".join("%-16s" % kind_of(ChromiumPage, n) for n in names[i:i + 5]))

print()
print("== 运行时取值(property 直接取) ==")
show("page.title", lambda: page.title)
show("page.url", lambda: page.url)
show("type(page.html).__name__", lambda: type(page.html).__name__)
show("page.eles('t:h2') 数量", lambda: len(page.eles("t:h2")))
show("type(page.latest_tab).__name__", lambda: type(page.latest_tab).__name__)
show("page.latest_tab.url", lambda: page.latest_tab.url)
show("page.tabs_count", lambda: page.tabs_count)
show("len(page.tab_ids)", lambda: len(page.tab_ids))
show("type(page.tab_ids[0]).__name__", lambda: type(page.tab_ids[0]).__name__)
show("type(page.tab_id).__name__", lambda: type(page.tab_id).__name__)
show("page.address 的主机部分", lambda: page.address.split("//")[-1].split(":")[0])
show("page.browser_version", lambda: page.browser_version)
show("type(page.process_id).__name__", lambda: type(page.process_id).__name__)
show("page.user_agent[:28]", lambda: page.user_agent[:28])
show("type(page.driver).__name__", lambda: type(page.driver).__name__)
show("type(page.wait).__name__", lambda: type(page.wait).__name__)
show("type(page.scroll).__name__", lambda: type(page.scroll).__name__)
show("type(page.set).__name__", lambda: type(page.set).__name__)
show("type(page.listen).__name__", lambda: type(page.listen).__name__)
show("page.timeouts", lambda: page.timeouts)
show("type(page.get('..')).__name__", lambda: type(page.get("https://spa1.scrape.center/page/2")).__name__)

print()
print("== 错误示范:对 property 返回的字符串或整数加括号 ==")
for expr, fn in (("page.title()", lambda: page.title()),
("page.url()", lambda: page.url()),
("page.html()", lambda: page.html()),
("page.tabs_count()", lambda: page.tabs_count()),
("page.tab_id()", lambda: page.tab_id()),
("page.user_agent()", lambda: page.user_agent())):
show(expr, fn)

print()
print("== 以下名字在本版本是否存在 ==")
for expr, fn in (("page.tab_count", lambda: page.tab_count),
("page.tabs", lambda: page.tabs),
("page.set_paths()", lambda: page.set_paths()),
("page.quit() 之外: page.get_page_info", lambda: page.get_page_info)):
show(expr, fn)

print()
print("== 元素对象 ==")
ele = page.ele("tag:a")
print("type(ele):", type(ele).__name__)
for n in ("text", "tag", "html_tag", "html", "outerHTML", "attrs", "rect", "states",
"timeout", "click", "input", "attr", "style", "parent", "eles", "sr",
"shadow_root", "driver"):
print(" ele.%-12s %s" % (n, kind_of(type(ele), n)))
show("ele.text", lambda: ele.text)
show("ele.tag", lambda: ele.tag)
show("ele.text()", lambda: ele.text())
show("ele.attr('href')", lambda: ele.attr("href"))

print()
print("== states 的成员是属性还是方法 ==")
st = ele.states
print("type(ele.states):", type(st).__name__)
for n in sorted(x for x in dir(st) if not x.startswith("_")):
print(" states.%-14s %s" % (n, kind_of(type(st), n)))
for n in ("is_displayed", "is_enabled", "is_checked", "exist"):
show("ele.states.%s" % n, lambda n=n: getattr(ele.states, n))
show("ele.states.is_displayed()", lambda: ele.states.is_displayed())

print()
print("== ChromiumElement 全部公开成员 ==")
enames = sorted(n for n in dir(ele) if not n.startswith("_"))
print("总数:", len(enames))
for i in range(0, len(enames), 5):
print(" " + " ".join("%-18s" % n for n in enames[i:i + 5]))
print(" " + " ".join("%-18s" % kind_of(type(ele), n) for n in enames[i:i + 5]))
finally:
page.quit()
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
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
== ChromiumPage 全部 69 个公开成员的性质 ==
成员总数: 69 | property: 32 | 方法: 37
actions activate_tab active_ele add_ele add_init_js
property 方法 property 方法 方法
address back browser browser_version clear_cache
property 方法 property property 方法
close close_tabs console cookies disconnect
方法 方法 property 方法 方法
download download_path driver ele eles
property property property 方法 方法
find forward get get_frame get_frames
方法 方法 方法 方法 方法
get_screenshot get_tab get_tabs handle_alert html
方法 方法 方法 方法 property
json latest_tab listen load_mode local_storage
property property property property 方法
new_tab process_id quit reconnect rect
方法 property 方法 方法 property
refresh remove_ele remove_init_js run_async_js run_cdp
方法 方法 方法 方法 方法
run_cdp_loaded run_js run_js_loaded s_ele s_eles
方法 方法 方法 方法 方法
save screencast scroll session session_storage
方法 property property property 方法
set states stop_loading tab_id tab_ids
property property 方法 property property
tabs_count timeout timeouts title upload_list
property property property property property
url url_available user_agent wait
property property property property
== 运行时取值(property 直接取) ==
page.title -> 'Scrape | Movie'
page.url -> 'https://spa1.scrape.center/page/2'
type(page.html).__name__ -> 'str'
page.eles('t:h2') 数量 -> 10
type(page.latest_tab).__name__ -> 'MixTab'
page.latest_tab.url -> 'https://spa1.scrape.center/page/2'
page.tabs_count -> 1
len(page.tab_ids) -> 1
type(page.tab_ids[0]).__name__ -> 'str'
type(page.tab_id).__name__ -> 'str'
page.address 的主机部分 -> '127.0.0.1'
page.browser_version -> 'Chrome/156.0.8078.80'
type(page.process_id).__name__ -> 'int'
page.user_agent[:28] -> 'Mozilla/5.0 (Windows NT 10.0'
type(page.driver).__name__ -> 'Driver'
type(page.wait).__name__ -> 'ChromiumPageWaiter'
type(page.scroll).__name__ -> 'PageScroller'
type(page.set).__name__ -> 'ChromiumPageSetter'
type(page.listen).__name__ -> 'Listener'
page.timeouts -> {'base': 10, 'page_load': 30, 'script': 30}
type(page.get('..')).__name__ -> 'bool'
== 错误示范:对 property 返回的字符串或整数加括号 ==
page.title() TypeError: 'str' object is not callable
page.url() TypeError: 'str' object is not callable
page.html() TypeError: 'str' object is not callable
page.tabs_count() TypeError: 'int' object is not callable
page.tab_id() TypeError: 'str' object is not callable
page.user_agent() TypeError: 'str' object is not callable
== 以下名字在本版本是否存在 ==
page.tab_count AttributeError: 'ChromiumPage' object has no attribute 'tab_count'
page.tabs AttributeError: 'ChromiumPage' object has no attribute 'tabs'
page.set_paths() AttributeError: 'ChromiumPage' object has no attribute 'set_paths'
page.quit() 之外: page.get_page_info AttributeError: 'ChromiumPage' object has no attribute 'get_page_info'
== 元素对象 ==
type(ele): ChromiumElement
ele.text property
ele.tag property
ele.html_tag 类上没有这个名字
ele.html property
ele.outerHTML 类上没有这个名字
ele.attrs property
ele.rect property
ele.states property
ele.timeout property
ele.click property
ele.input 方法
ele.attr 方法
ele.style 方法
ele.parent 方法
ele.eles 方法
ele.sr property
ele.shadow_root property
ele.driver 类上没有这个名字
ele.text -> 'Scrape'
ele.tag -> 'a'
ele.text() TypeError: 'str' object is not callable
ele.attr('href') -> 'https://spa1.scrape.center/'
== states 的成员是属性还是方法 ==
type(ele.states): ElementStates
states.has_rect property
states.is_alive property
states.is_checked property
states.is_clickable property
states.is_covered property
states.is_displayed property
states.is_enabled property
states.is_in_viewport property
states.is_selected property
states.is_whole_in_viewport property
ele.states.is_displayed -> True
ele.states.is_enabled -> True
ele.states.is_checked -> None
ele.states.exist AttributeError: 'ElementStates' object has no attribute 'exist'
ele.states.is_displayed() TypeError: 'bool' object is not callable
== ChromiumElement 全部公开成员 ==
总数: 65
after afters attr attrs before
方法 方法 方法 property 方法
befores check child child_count children
方法 方法 方法 property 方法
clear click comments css_path drag
方法 property property property 方法
drag_to east ele eles find
方法 方法 方法 方法 方法
focus get_frame get_screenshot hover html
方法 方法 方法 方法 property
inner_html input link next nexts
property 方法 property 方法 方法
north offset over owner parent
方法 方法 方法 类上没有这个名字 方法
prev prevs property pseudo raw_text
方法 方法 方法 property property
rect remove_attr run_async_js run_js s_ele
property 方法 方法 方法 方法
s_eles save scroll select set
方法 方法 property property property
shadow_root south sr src states
property 方法 property 方法 property
style tab tag text texts
方法 类上没有这个名字 property property 方法
timeout value wait west xpath
property property property 方法 property
  • 本版本 ChromiumPage 类有 69 个公开成员:32 个 property、37 个方法。实例上可能多出成员,不能混用两种统计。
  • title() 是先拿到字符串,再试图调用它,因而报 'str' object is not callable;tabs_count() 则报整数不可调用。这种错误也可能来自同名变量遮住函数,不只发生在 property 上。
  • tab_count、tabs、页面上的 set_paths、get_page_info 在本版本未提供。数量读 tabs_count、列表用 get_tabs()、路径配置用 ChromiumOptions.set_paths();不据此编造旧版本迁移历史。
  • ele.click 虽是 property,却返回可调用的 Clicker,所以 ele.click() 正确。是否加括号还要看返回对象。
  • ele.states 各状态直接读取。普通链接的 is_checked / is_selected 是 None,表示本例不适用,不能误当“未选中”。
  • page.address 包含主机和调试端口;示例为了避免随机端口干扰只打印主机部分。get() 的布尔返回值不能当页面使用。

定位语法

ele(locator, index=1, timeout=None) 第一个业务参数是定位符;eles(locator, timeout=None) 取全部。下例先等 spa1 第二页的 t:h2 出现,再比较命中数量与首个元素:

03_locator_syntax.py
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
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
# -*- coding: utf-8 -*-
"""ele()/eles() 定位语法逐条实测:命中数量 + 第一个元素的文本。靶站 https://spa1.scrape.center/ 。"""

import re

from DrissionPage import ChromiumOptions, ChromiumPage

page = ChromiumPage(ChromiumOptions().headless(True).auto_port())


def brief(text):
return re.sub(r"\s+", " ", text or "").strip()[:34]


def show(expr, fn):
try:
print("%s -> %r" % (expr, fn()))
except Exception as exc:
print("%s %s: %s" % (expr, type(exc).__name__, str(exc).replace("\n", " | ")))


def probe(loc, timeout=0.2):
try:
eles = page.eles(loc, timeout=timeout)
except Exception as exc:
print(" %-34s %s: %s" % (loc, type(exc).__name__, exc))
return
count = len(eles)
first = brief(eles[0].text) if count else "-"
first_tag = eles[0].tag if count else "-"
print(" %-34s 命中 %2d 个 | 首个 tag=%-5s text=%s" % (loc, count, first_tag, first))


try:
if not page.get("https://spa1.scrape.center/page/2"):
raise RuntimeError("页面导航失败")
page.wait.eles_loaded("t:h2", timeout=20, raise_err=True)
print("页面: ", page.url, "| title:", page.title)
print()
print("== 一、tag / 属性 / 文本 基本前缀 ==")
for loc in ("tag:div", "t:div", "tag:a", "tag:h1", "@tag()=div",
"@class=cover", "@id=app", "@href:detail", "@href^/detail", "@href$/10", "@href^=https", "@href$=/",
"@class", "@!class"):
probe(loc)

print()
print("== 二、简化写法 # . text tx x c ==")
for loc in ("#app", "#=app", "#:pp", "#^ap", "#$pp",
".cover", ".=cover", ".:_item", ".^el-rate", ".$__item",
"text=Scrape", "tx=Scrape", "text:Scra", "Scrape",
"xpath://div[@class=\"item\"]", "x://div[@class=\"item\"]",
"css:div.item", "c:.el-card.item", "css:>div"):
probe(loc)

print()
print("== 三、class 的匹配语义(这里藏着 0 命中的真正原因) ==")
for loc in ("@class=item", "tag:div@@class=item", "@@class:item", "tag:div@@class:item",
"tag:div@class:item", ".item", "css:div.item", ".^el-card", ".$shadow"):
probe(loc)
print(" 实际属性值:", repr(page.ele("css:div.item").attr("class")))

print()
print("== 四、多属性组合 @@(与) @|(或) @!(非) ==")
for loc in ("tag:div@@class:item@@tag()=div",
"@@class:item",
"@|class=m-b-sm@|class=cover",
"@@class:item@!id=app",
"tag:li@@text():Scrape",
"tag:a@text():Scrape",
"tag:a@@text():Scrape",
"@@text()=Scrape@@class=logo-title"):
probe(loc)

print()
print("== 五、对照:本版本不识别为定位前缀的写法 ==")
for loc in ("id:app", "class:name", "name:q", "r:item", ":simple", "css:div.item:simple",
"tag:div@@class=summary"):
probe(loc)

print()
print("== 六、链式 / __call__ / xpath 取属性 ==")
print(" page('t:a') 类型:", type(page("t:a")).__name__, "| text:", brief(page("t:a").text))
box = page.ele("tag:div@@class:item")
print(" 容器:", type(box).__name__, "| tag:", box.tag)
print(" 容器内 ele('tag:a'):", brief(box.ele("tag:a").text))
print(" 容器内 eles('tag:a') 数量:", len(box.eles("tag:a")))
print(" 容器内 eles('xpath:.//a') 数量:", len(box.eles("xpath:.//a")))
show(" 元素内 xpath 取属性值(@href):", lambda: box.ele("xpath:.//a/@href"))
show(" 元素内 xpath 取属性值(@class,空串):", lambda: box.ele("xpath:.//a/@class"))
print(" eles('xpath:.//a/@href') 元素类型:",
[type(x).__name__ for x in box.eles("xpath:.//a/@href")])
print(" 元素内省略 . 的 //a 数量:", len(box.eles("xpath://a")))

print()
print("== 七、找不到元素时返回什么 ==")
miss = page.ele("不存在的元素文字xyz", timeout=0.2)
print(" type(...) =", type(miss).__name__)
print(" bool(...) =", bool(miss), "| miss == None =", miss == None)
show(" miss.text", lambda: miss.text)
show(" miss.tag", lambda: miss.tag)
show(" miss.click()", lambda: miss.click())
print(" repr(miss) =", repr(miss))
print(" eles() 找不到时返回:", type(page.eles("不存在的元素文字xyz", timeout=0.2)).__name__,
len(page.eles("不存在的元素文字xyz", timeout=0.2)))

print()
print("== 八、改成返回值的开关 set.NoneElement_value ==")
page.set.NoneElement_value("取不到", True)
miss2 = page.ele("不存在的元素文字xyz", timeout=0.2)
print(" type:", type(miss2).__name__, "| bool:", bool(miss2))
print(" .text ->", repr(miss2.text), "| .attr ->", repr(miss2.attr),
"| .ele(..) 类型 ->", type(miss2.ele("tag:div")).__name__)
page.set.NoneElement_value(None, False)

print()
print("== 九、Settings 开关(找不到元素是否抛异常) ==")
try:
from DrissionPage.common import Settings
_st = sorted(n for n in dir(Settings) if not n.startswith("_"))
print(" Settings 公开项 %d 个:" % len(_st))
for _i in range(0, len(_st), 6):
print(" " + " ".join("%-30s" % n for n in _st[_i:_i + 6]))
old = Settings.raise_when_ele_not_found
Settings.raise_when_ele_not_found = True
try:
page.ele("不存在的元素文字xyz", timeout=0.2)
print(" 开启后仍未抛异常")
except Exception as exc:
print(" 开启后:", type(exc).__name__ + ":", str(exc).replace("\n", " | "))
Settings.raise_when_ele_not_found = old
print(" 复位后:", type(page.ele("不存在的元素文字xyz", timeout=0.2)).__name__)
except Exception as exc:
print(" Settings 检查失败:", type(exc).__name__, exc)

print()
print("== 十、错误参数 ele(show=True) 的表现 ==")
try:
r = page.ele("不存在的元素文字xyz", timeout=0.2, show=True)
print(" 返回类型:", type(r).__name__)
except Exception as exc:
print(" show=True 抛出:", type(exc).__name__, "|", str(exc).replace("\n", " | "))
finally:
page.quit()
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
页面:  https://spa1.scrape.center/page/2 | title: Scrape | Movie
== 一、tag / 属性 / 文本 基本前缀 ==
tag:div 命中 114 个 | 首个 tag=div text=Scrape 狮子王 - The Lion King 动画 歌舞 冒
t:div 命中 114 个 | 首个 tag=div text=Scrape 狮子王 - The Lion King 动画 歌舞 冒
tag:a 命中 21 个 | 首个 tag=a text=Scrape
tag:h1 命中 0 个 | 首个 tag=- text=-
@tag()=div 命中 114 个 | 首个 tag=div text=Scrape 狮子王 - The Lion King 动画 歌舞 冒
@class=cover 命中 10 个 | 首个 tag=img text=
@id=app 命中 1 个 | 首个 tag=div text=Scrape 狮子王 - The Lion King 动画 歌舞 冒
@href:detail 命中 20 个 | 首个 tag=a text=
@href^/detail 命中 20 个 | 首个 tag=a text=
@href$/10 命中 2 个 | 首个 tag=a text=
@href^=https 命中 0 个 | 首个 tag=- text=-
@href$=/ 命中 0 个 | 首个 tag=- text=-
@class 命中 321 个 | 首个 tag=div text=Scrape
@!class 命中 103 个 | 首个 tag=html text=Scrape | Movie Scrape 狮子王 - The Li
== 二、简化写法 # . text tx x c ==
#app 命中 1 个 | 首个 tag=div text=Scrape 狮子王 - The Lion King 动画 歌舞 冒
#=app 命中 1 个 | 首个 tag=div text=Scrape 狮子王 - The Lion King 动画 歌舞 冒
#:pp 命中 1 个 | 首个 tag=div text=Scrape 狮子王 - The Lion King 动画 歌舞 冒
#^ap 命中 1 个 | 首个 tag=div text=Scrape 狮子王 - The Lion King 动画 歌舞 冒
#$pp 命中 1 个 | 首个 tag=div text=Scrape 狮子王 - The Lion King 动画 歌舞 冒
.cover 命中 10 个 | 首个 tag=img text=
.=cover 命中 10 个 | 首个 tag=img text=
.:_item 命中 50 个 | 首个 tag=span text=
.^el-rate 命中 120 个 | 首个 tag=div text=
.$__item 命中 50 个 | 首个 tag=span text=
text=Scrape 命中 1 个 | 首个 tag=span text=Scrape
tx=Scrape 命中 1 个 | 首个 tag=span text=Scrape
text:Scra 命中 2 个 | 首个 tag=title text=Scrape | Movie
Scrape 命中 2 个 | 首个 tag=title text=Scrape | Movie
xpath://div[@class="item"] 命中 0 个 | 首个 tag=- text=-
x://div[@class="item"] 命中 0 个 | 首个 tag=- text=-
css:div.item 命中 10 个 | 首个 tag=div text=狮子王 - The Lion King 动画 歌舞 冒险 美国 /
c:.el-card.item 命中 10 个 | 首个 tag=div text=狮子王 - The Lion King 动画 歌舞 冒险 美国 /
css:>div 命中 0 个 | 首个 tag=- text=-
== 三、class 的匹配语义(这里藏着 0 命中的真正原因) ==
@class=item 命中 0 个 | 首个 tag=- text=-
tag:div@@class=item 命中 0 个 | 首个 tag=- text=-
@@class:item 命中 60 个 | 首个 tag=div text=狮子王 - The Lion King 动画 歌舞 冒险 美国 /
tag:div@@class:item 命中 10 个 | 首个 tag=div text=狮子王 - The Lion King 动画 歌舞 冒险 美国 /
tag:div@class:item 命中 10 个 | 首个 tag=div text=狮子王 - The Lion King 动画 歌舞 冒险 美国 /
.item 命中 0 个 | 首个 tag=- text=-
css:div.item 命中 10 个 | 首个 tag=div text=狮子王 - The Lion King 动画 歌舞 冒险 美国 /
.^el-card 命中 20 个 | 首个 tag=div text=狮子王 - The Lion King 动画 歌舞 冒险 美国 /
.$shadow 命中 10 个 | 首个 tag=div text=狮子王 - The Lion King 动画 歌舞 冒险 美国 /
实际属性值: 'el-card item m-t is-hover-shadow'
== 四、多属性组合 @@(与) @|(或) @!(非) ==
tag:div@@class:item@@tag()=div 命中 10 个 | 首个 tag=div text=狮子王 - The Lion King 动画 歌舞 冒险 美国 /
@@class:item 命中 60 个 | 首个 tag=div text=狮子王 - The Lion King 动画 歌舞 冒险 美国 /
@|class=m-b-sm@|class=cover 命中 20 个 | 首个 tag=img text=
@@class:item@!id=app 命中 60 个 | 首个 tag=div text=狮子王 - The Lion King 动画 歌舞 冒险 美国 /
tag:li@@text():Scrape 命中 0 个 | 首个 tag=- text=-
tag:a@text():Scrape 命中 0 个 | 首个 tag=- text=-
tag:a@@text():Scrape 命中 1 个 | 首个 tag=a text=Scrape
@@text()=Scrape@@class=logo-title 命中 1 个 | 首个 tag=span text=Scrape
== 五、对照:本版本不识别为定位前缀的写法 ==
id:app 命中 0 个 | 首个 tag=- text=-
class:name 命中 0 个 | 首个 tag=- text=-
name:q 命中 0 个 | 首个 tag=- text=-
r:item 命中 0 个 | 首个 tag=- text=-
:simple 命中 0 个 | 首个 tag=- text=-
css:div.item:simple 命中 0 个 | 首个 tag=- text=-
tag:div@@class=summary 命中 0 个 | 首个 tag=- text=-
== 六、链式 / __call__ / xpath 取属性 ==
page('t:a') 类型: ChromiumElement | text: Scrape
容器: ChromiumElement | tag: div
容器内 ele('tag:a'):
容器内 eles('tag:a') 数量: 2
容器内 eles('xpath:.//a') 数量: 2
元素内 xpath 取属性值(@href): -> '/detail/10'
元素内 xpath 取属性值(@class,空串): -> <NoneElement method=ele(), locator=xpath:.//a/@class, index=1, timeout=10>
eles('xpath:.//a/@href') 元素类型: ['str', 'str']
元素内省略 . 的 //a 数量: 2
== 七、找不到元素时返回什么 ==
type(...) = NoneElement
bool(...) = False | miss == None = True
miss.text ElementNotFoundError: | 没有找到元素。 | 方法: ele() | 参数: {'locator': '不存在的元素文字xyz', 'index': 1, 'timeout': 0.2} | 版本: 4.1.1.4
miss.tag ElementNotFoundError: | 没有找到元素。 | 方法: ele() | 参数: {'locator': '不存在的元素文字xyz', 'index': 1, 'timeout': 0.2} | 版本: 4.1.1.4
miss.click() ElementNotFoundError: | 没有找到元素。 | 方法: ele() | 参数: {'locator': '不存在的元素文字xyz', 'index': 1, 'timeout': 0.2} | 版本: 4.1.1.4
repr(miss) = <NoneElement method=ele(), locator=不存在的元素文字xyz, index=1, timeout=0.2>
eles() 找不到时返回: ChromiumElementsList 0
== 八、改成返回值的开关 set.NoneElement_value ==
type: NoneElement | bool: False
.text -> '取不到' | .attr -> '取不到' | .ele(..) 类型 -> NoneElement
== 九、Settings 开关(找不到元素是否抛异常) ==
Settings 公开项 17 个:
auto_handle_alert browser_connect_timeout cdp_timeout raise_when_click_failed raise_when_ele_not_found raise_when_wait_failed
set_auto_handle_alert set_browser_connect_timeout set_cdp_timeout set_language set_raise_when_click_failed set_raise_when_ele_not_found
set_raise_when_wait_failed set_singleton_tab_obj set_suffixes_list singleton_tab_obj suffixes_list
开启后: ElementNotFoundError: | 没有找到元素。 | 方法: ele() | 参数: {'locator': '不存在的元素文字xyz', 'index': 1, 'timeout': 0.2} | 版本: 4.1.1.4
复位后: NoneElement
== 十、错误参数 ele(show=True) 的表现 ==
show=True 抛出: TypeError | ChromiumBase.ele() got an unexpected keyword argument 'show'
  • 标签用 tag:div / t:div,属性用 @id=app。= 完整相等、: 包含、^ 开头、$ 结尾,例如 @href^/detail。@href^=https 多出的等号也进入匹配内容,不要把 CSS 属性选择器的等号原样搬过来。
  • #app 简写 id,.cover 简写完整 class 匹配;text=Scrape / tx=Scrape 匹配完整文本,text:Scra 匹配包含。裸字符串也是文本包含;没有目标标签限制时可能命中标题或容器,需查看返回的 tag 与文本。
  • .item 不等同于 CSS 类词匹配。卡片实际 class 为 el-card item m-t is-hover-shadow,@class=item / .item 为 0;tag:div@class:item 按包含匹配,css:div.item 才是 CSS 形式。XPath //div[@class="item"] 同样要求属性完整相等,本例也是 0。
  • @@ 是与、@| 是或、@! 是否定。@@class:item 连评分图标一起命中,加 tag:div 才缩小到卡片;有命中不等于选对。
  • XPath 前缀是 xpath: / x:,CSS 是 css: / c:。css:>div 本例为 0,不作为通用页面根节点写法;纯标签 CSS 的额外命中见下一节。
  • page('t:a') 简写单元素定位,元素自身也能继续定位。本例卡片内 .//a 与 //a 均为两个链接,建议显式 .//a 表达子树,不把观察推及所有解析器。
  • XPath 取属性会返回字符串:box.ele('xpath:.//a/@href') 得到 '/detail/10',box.eles(...) 取字符串列表,不是元素列表。

未识别的前缀不一定报错,可能只是合法查询没有命中。这里把转换结果单独打印:

18_locator_failures.py
1
2
3
4
5
6
7
8
# -*- coding: utf-8 -*-
"""把未识别的定位前缀与合法的零命中分开检查。"""
from DrissionPage._functions.locator import get_loc

for locator in ('id:app', 'class:name', 'name:q', 'r:item', ':simple',
'css:div.item:simple', 'tag:div@@class=summary'):
by, value = get_loc(locator)[:2]
print('%-30s -> by=%s value=%r' % (locator, by, value))
1
2
3
4
5
6
7
id:app                         -> by=xpath value='//*/text()[contains(., "id:app")]/..'
class:name -> by=xpath value='//*/text()[contains(., "class:name")]/..'
name:q -> by=xpath value='//*/text()[contains(., "name:q")]/..'
r:item -> by=xpath value='//*/text()[contains(., "r:item")]/..'
:simple -> by=xpath value='//*/text()[contains(., ":simple")]/..'
css:div.item:simple -> by=css selector value='div.item:simple'
tag:div@@class=summary -> by=xpath value='//*[(name()="div") and (@class="summary")]'
  • id:app、class:name 等被当作包含整串文字的文本查询,不会自动变成 id/class 条件;css:div.item:simple 是无效 CSS;tag:div@@class=summary 则是合法 XPath,只是页面没有匹配项。三种零命中的原因不同。
  • get_loc() 是用于核对本版解析规则的底层工具,业务定位仍用公开接口。

找不到时的处理:

  • ele() 默认返回 NoneElement,bool() 为假、== None 为真,但不是 Python 的 None。先判断再访问;直接 .text、.tag、.click() 会抛 ElementNotFoundError。
  • eles() 未命中是空列表,可以遍历;业务要求至少一条数据时仍需检查非空。
  • page.set.NoneElement_value("取不到", True) 配置缺失值,本例 .text 和 .attr 直接变成字符串,不能再调用 .attr()。恢复用 page.set.NoneElement_value(None, False)。
  • Settings.raise_when_ele_not_found = True 让定位阶段抛错,示例保存旧值并复位;ele(show=True) 则因不支持参数而报 TypeError。

CSS 搜索边界

页面级 css:p 不总等于原生 querySelectorAll('p')。在最小页面中把定位转换、CDP 搜索和元素结果分别对照:

16_css_selector_translation.py
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 -*-
"""css: 纯标签名的多命中:比较 querySelectorAll 与页面级 DOM.performSearch。本机靶站 127.0.0.1:8066 。"""

import threading
import socket
import sys
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

from DrissionPage import ChromiumOptions, ChromiumPage

PORT = 8066
ROWS = [("p", "a"), ("div", "b"), ("span", "c"), ("script", "d"),
("style", "e"), ("b", "f"), ("em", "g"), ("section", "h")]
BODY = "".join("<%s id='%s'></%s>" % (tag, eid, tag) for tag, eid in ROWS)
PAGE = ("<!DOCTYPE html><html><head><title>css-probe</title></head>"
"<body>%s</body></html>" % BODY).encode("utf-8")


class CssHandler(BaseHTTPRequestHandler):
def log_message(self, *args):
pass

def do_GET(self):
self.send_response(200)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.send_header("Content-Length", str(len(PAGE)))
self.end_headers()
self.wfile.write(PAGE)


class QuietServer(ThreadingHTTPServer):
def handle_error(self, request, client_address):
pass


def start_target():
try:
with socket.create_connection(("127.0.0.1", PORT), 0.2):
print("端口 %d 已被占用,脚本终止" % PORT)
sys.exit(1)
except OSError:
pass
srv = QuietServer(("127.0.0.1", PORT), CssHandler)
threading.Thread(target=srv.serve_forever, daemon=True).start()
return srv


srv = start_target()
page = ChromiumPage(ChromiumOptions().headless(True).auto_port())
try:
page.get("http://127.0.0.1:%d/" % PORT)
print("页面里每种标签各 1 个,共 %d 个元素" % len(ROWS))

print()
print("== 纯标签名:css: 前缀 vs querySelectorAll ==")
for sel in ("p", "b", "span", "div"):
got = page.eles("css:%s" % sel)
mine = [e.tag for e in got]
js = page.run_js("return Array.from(document.querySelectorAll('%s')).map(e => e.tagName"
".toLowerCase())" % sel)
print(" css:%-6s 命中 %s 个 %-22s | querySelectorAll %s 个 %s"
% (sel, len(got), mine, len(js), js))

print()
print("== 显式标签与 XPath、其他 CSS 写法 ==")
for label, loc in (("t:p", "t:p"), ("xpath://p", "xpath://p"),
("css:p > b(有子选择器)", "css:p > b"),
("css:#a(有 id)", "css:#a")):
got = page.eles(loc)
print(" %-22s 命中 %s 个 %s" % (label, len(got), [e.tag for e in got]))

print()
print("== 两个带 class/id 的对照,不能推及全部 CSS ==")
page.get("http://127.0.0.1:%d/" % PORT)
print(" 本页没有 class,css:.item 命中:", len(page.eles("css:.item")))
print(" css:div#b 命中:", [(e.tag, e.attr("id")) for e in page.eles("css:div#b")])

print()
print("== 定位入口与底层搜索对照 ==")
from DrissionPage._functions.locator import get_loc
for sel in ("p", "div.item", "p > b"):
print(" 默认 get_loc:", get_loc(("css selector", sel)))
print(" get_loc(('css selector', %-10s), translate_css=True) -> %r"
% ("'%s'" % sel, get_loc(("css selector", sel), translate_css=True)))
for query in ("p", "//p", "b"):
result = page.run_cdp("DOM.performSearch", query=query,
includeUserAgentShadowDOM=True)
print(" DOM.performSearch", repr(query), "命中:", result["resultCount"])
page.run_cdp("DOM.discardSearchResults", searchId=result["searchId"])
finally:
page.quit()
srv.shutdown()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
页面里每种标签各 1 个,共 8 个元素
== 纯标签名:css: 前缀 vs querySelectorAll ==
css:p 命中 3 个 ['p', 'span', 'script'] | querySelectorAll 1 个 ['p']
css:b 命中 3 个 ['b', 'body', 'div'] | querySelectorAll 1 个 ['b']
css:span 命中 1 个 ['span'] | querySelectorAll 1 个 ['span']
css:div 命中 1 个 ['div'] | querySelectorAll 1 个 ['div']
== 显式标签与 XPath、其他 CSS 写法 ==
t:p 命中 1 个 ['p']
xpath://p 命中 1 个 ['p']
css:p > b(有子选择器) 命中 0 个 []
css:#a(有 id) 命中 1 个 ['p']
== 两个带 class/id 的对照,不能推及全部 CSS ==
本页没有 class,css:.item 命中: 0
css:div#b 命中: [('div', 'b')]
== 定位入口与底层搜索对照 ==
默认 get_loc: ('css selector', 'p')
get_loc(('css selector', 'p' ), translate_css=True) -> ('xpath', 'p')
默认 get_loc: ('css selector', 'div.item')
get_loc(('css selector', 'div.item'), translate_css=True) -> ('xpath', "div[@class and contains(@class, 'item') and contains(concat(' ', normalize-space(@class), ' '), ' item ')]")
默认 get_loc: ('css selector', 'p > b')
get_loc(('css selector', 'p > b' ), translate_css=True) -> ('xpath', 'p/b')
DOM.performSearch 'p' 命中: 4
DOM.performSearch '//p' 命中: 1
DOM.performSearch 'b' 命中: 4
  • 页面只有一个 <p>,原生 CSS 返回一个,page.eles('css:p') 却返回 p、span、script;css:b 也有额外命中。
  • 页面入口使用 DOM.performSearch,不是强制 CSS → XPath。 本版本页面查找实现调用默认 get_loc(locator) 后取查询串,css:p 最终传入的是 p。CDP 此入口兼有纯文本、CSS、XPath 搜索语义,并不等价于 querySelectorAll。
  • 原始 DOM.performSearch 本例返回 4 个节点,库再过滤后得到 3 个元素;//p 则只命中一个。不要要求原始节点计数与最终元素计数始终相同。
  • get_loc(..., translate_css=True) 确实生成 XPath,但属于另一个显式分支,不能拿它证明页面级调用走了转换。
  • 只按标签查找时,用已验证的 t:p / xpath://p。带 id/class 的几个例子正确,不代表所有 CSS 都没有边界问题;By.CSS_SELECTOR 同样要对照原生命中集合。
  • 本页没有 p 的子元素 b,css:p > b 为 0 是合理结果,不是转换失败。私有函数只用来检查本版本实现,不作为业务依赖。

元素对象能取什么

拿到一个元素之后能干的事,比「取文本」多得多。这段用一个片名 <a> 做样本,介绍文本、属性、rect、states、关系定位、滚动和 JS。窗口、缩放、浏览器、站点和加载时机都会影响页面数据;学习时应关注 API 的用途,不把某次文本或坐标当成固定答案。

04_element_object.py
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
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
# -*- coding: utf-8 -*-
"""ChromiumElement 的属性族与关系定位实测。靶站 https://spa1.scrape.center/ 。"""

import re

from DrissionPage import ChromiumOptions, ChromiumPage

page = ChromiumPage(ChromiumOptions().headless(True).auto_port())


def brief(text, size=40):
return re.sub(r"\s+", " ", text or "").strip()[:size]


def show(expr, fn):
try:
print("%-42s -> %r" % (expr, fn()))
except Exception as exc:
print("%-42s %s: %s" % (expr, type(exc).__name__, str(exc).replace("\n", " | ")))


try:
page.set.window.size(1200, 900)
if not page.get("https://spa1.scrape.center/page/2"):
raise RuntimeError("页面导航失败")
page.wait.eles_loaded("t:h2", timeout=20, raise_err=True)
ele = page.ele("t:h2").parent()
print("元素:", type(ele).__name__, "| tag:", ele.tag, "| text:", brief(ele.text))
print("ele.owner 类型:", type(ele.owner).__name__)

print()
print("== 文本 / HTML ==")
show("ele.text", lambda: brief(ele.text))
show("ele.raw_text", lambda: brief(ele.raw_text))
show("ele.texts()", lambda: [brief(t, 12) for t in ele.texts()])
show("ele.texts(True) 数量", lambda: len(ele.texts(True)))
show("ele.html 以 <a 开头", lambda: ele.html.startswith("<a "))
show("inner_html 比 html 短(少了外层标签)", lambda: len(ele.inner_html) < len(ele.html))
show("html 含 href 属性", lambda: "href=" in ele.html)
show("ele.css_path", lambda: ele.css_path)
show("ele.xpath", lambda: ele.xpath)
show("ele.value", lambda: ele.value)
show("ele.link", lambda: ele.link)

print()
print("== 属性 / 样式 ==")
show("ele.attr('href')", lambda: ele.attr("href"))
show("ele.attr('data-x')", lambda: ele.attr("data-x"))
show("sorted(ele.attrs)", lambda: sorted(ele.attrs))
show("ele.style('color')", lambda: ele.style("color"))
show("ele.style('font-size')", lambda: ele.style("font-size"))
show("ele.style('display', pseudo_ele=':before')", lambda: ele.style("display", ":before"))

print()
print("== rect(像素值随站点改版和窗口变化,这里只取证成员与相互关系) ==")
rect = ele.rect
print("type(ele.rect):", type(rect).__name__)
print("ele.rect 成员:", sorted(n for n in dir(rect) if not n.startswith("_")))
for n in ("size", "location", "midpoint", "click_point", "viewport_location",
"viewport_midpoint", "screen_location", "scroll_position", "corners"):
show("ele.rect.%s 类型" % n, lambda n=n: type(getattr(ele.rect, n)).__name__)
show("size 是否为 (宽, 高) 两个数", lambda: len(ele.rect.size) == 2)
show("宽 > 0", lambda: ele.rect.size[0] > 0)
show("midpoint.x == location.x + 宽/2",
lambda: abs(ele.rect.midpoint[0] - (ele.rect.location[0] + ele.rect.size[0] / 2)) < 0.01)
show("click_point 在 midpoint 上方", lambda: ele.rect.click_point[1] < ele.rect.midpoint[1])
show("未滚动时 viewport_location == location",
lambda: ele.rect.viewport_location == ele.rect.location)
show("screen_location 与 viewport_location 不同",
lambda: ele.rect.screen_location != ele.rect.viewport_location)
show("corners 是 4 个角点", lambda: len(ele.rect.corners) == 4)
print("type(page.rect):", type(page.rect).__name__)
print("page.rect 成员:", sorted(n for n in dir(page.rect) if not n.startswith("_")))
show("page.rect.window_state", lambda: page.rect.window_state)
show("window_size 类型", lambda: type(page.rect.window_size).__name__)
show("整页高 > 视口高(页面可滚动)", lambda: page.rect.size[1] > page.rect.viewport_size[1])
show("viewport_size 宽 <= window_size 宽", lambda: page.rect.viewport_size[0] <= page.rect.window_size[0])
show("page.rect.scroll_position", lambda: page.rect.scroll_position)
show("viewport_size_with_scrollbar >= viewport_size",
lambda: page.rect.viewport_size_with_scrollbar[0] >= page.rect.viewport_size[0])

print()
print("== states ==")
show("ele.states.is_displayed", lambda: ele.states.is_displayed)
show("ele.states.is_enabled", lambda: ele.states.is_enabled)
show("ele.states.is_checked", lambda: ele.states.is_checked)
show("ele.states.is_selected", lambda: ele.states.is_selected)
show("ele.states.is_alive", lambda: ele.states.is_alive)
show("ele.states.is_clickable", lambda: ele.states.is_clickable)
show("ele.states.is_covered 的类型", lambda: type(ele.states.is_covered).__name__)
show("bool(ele.states.is_covered)", lambda: bool(ele.states.is_covered))
show("ele.states.is_in_viewport", lambda: ele.states.is_in_viewport)
show("type(ele.states.has_rect).__name__", lambda: type(ele.states.has_rect).__name__)
show("bool(ele.states.has_rect)", lambda: bool(ele.states.has_rect))
show("type(ele.states).is_in_viewport 存在?", lambda: hasattr(ele.states, "is_in_viewport"))

print()
print("== 关系定位(4.x 没有 sibling()) ==")
show("ele.parent().tag", lambda: ele.parent().tag)
show("ele.parent().attr('class')", lambda: ele.parent().attr("class"))
show("ele.parent(2).tag", lambda: ele.parent(2).tag)
show("ele.parent('tag:div') 类型", lambda: type(ele.parent("tag:div")).__name__)
show("len(ele.children())", lambda: len(ele.children()))
show("ele.child().tag", lambda: ele.child().tag)
show("ele.next().tag", lambda: ele.next().tag)
show("ele.next().text", lambda: brief(ele.next().text, 20))
show("len(ele.nexts())", lambda: len(ele.nexts()))
show("ele.before()", lambda: type(ele.before()).__name__)
show("len(ele.befores())", lambda: len(ele.befores()))
show("len(ele.after()) # after 是方法返回元素?", lambda: type(ele.after()).__name__)
show("ele.sibling()", lambda: ele.sibling())
show("ele.siblings()", lambda: ele.siblings())
show("ele.nexts('tag:a') 数量", lambda: len(ele.nexts("tag:a")))
show("ele.child_count", lambda: ele.child_count)

print()
print("== 元素内嵌套查找 ==")
card = page.ele("tag:div@@class:item")
show("card.eles('tag:a') 数量", lambda: len(card.eles("tag:a")))
show("card.ele('tag:a').text", lambda: brief(card.ele("tag:a").text, 20))
show("card('t:h2') 简写", lambda: brief(card("t:h2").text, 20))
show("card.eles('css:.categories > button') 数量", lambda: len(card.eles("css:.categories > button")))
show("card.ele('t:img').attr('src') 以 https 开头",
lambda: card.ele("t:img").attr("src").startswith("https"))
show("card.ele('t:img').link", lambda: card.ele("t:img").link[:8])

print()
print("== 滚动 API 真名 ==")
print("page.scroll:", type(page.scroll).__name__,
sorted(n for n in dir(page.scroll) if not n.startswith("_")))
print("ele.scroll :", type(ele.scroll).__name__,
sorted(n for n in dir(ele.scroll) if not n.startswith("_")))
show("page.scroll.to_bottom() 返回类型", lambda: type(page.scroll.to_bottom()).__name__)
show("滚动到底后 window.scrollY>0", lambda: page.run_js("return window.scrollY > 0"))
show("ele.scroll.to_center() 返回类型", lambda: type(ele.scroll.to_center()).__name__)
show("元素是否居中(视口内)", lambda: ele.states.is_in_viewport)
show("ele.scroll.to_see() 返回类型", lambda: type(ele.scroll.to_see()).__name__)
show("page.scroll.to_top() 返回类型", lambda: type(page.scroll.to_top()).__name__)
show("滚动到顶后 window.scrollY", lambda: page.run_js("return window.scrollY"))
show("page.scroll.down(200) 返回类型", lambda: type(page.scroll.down(200)).__name__)
show("page.scroll.to_location(0,0)", lambda: type(page.scroll.to_location(0, 0)).__name__)
show("ele.scroll.to_center()", lambda: type(ele.scroll.to_center()).__name__)

print()
print("== 元素上执行 JS ==")
show("ele.run_js('return this.tagName')", lambda: ele.run_js("return this.tagName"))
show("ele.run_js 取长度", lambda: ele.run_js("return this.innerText.length"))
show("page.run_js 取标题", lambda: page.run_js("return document.title"))
show("page.run_async_js", lambda: page.run_async_js("return 1"))
finally:
page.quit()
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
元素: ChromiumElement | tag: a | text: 狮子王 - The Lion King
ele.owner 类型: ChromiumPage
== 文本 / HTML ==
ele.text -> '狮子王 - The Lion King'
ele.raw_text -> '狮子王 - The Lion King'
ele.texts() -> ['狮子王 - The Li']
ele.texts(True) 数量 -> 0
ele.html 以 <a 开头 -> True
inner_html 比 html 短(少了外层标签) -> True
html 含 href 属性 -> True
ele.css_path -> 'html:nth-child(1)>body:nth-child(2)>div#app>div#index>div:nth-child(1)>div:nth-child(1)>div:nth-child(1)>div:nth-child(1)>div:nth-child(1)>div:nth-child(2)>a:nth-child(1)'
ele.xpath -> '/html[1]/body[1]/div[1]/div[2]/div[1]/div[1]/div[1]/div[1]/div[1]/div[2]/a[1]'
ele.value -> None
ele.link -> 'https://spa1.scrape.center/detail/10'
== 属性 / 样式 ==
ele.attr('href') -> 'https://spa1.scrape.center/detail/10'
ele.attr('data-x') -> None
sorted(ele.attrs) -> ['class', 'data-v-1a647406', 'href']
ele.style('color') -> 'rgb(48, 49, 51)'
ele.style('font-size') -> '16px'
ele.style('display', pseudo_ele=':before') -> 'inline'
== rect(像素值随站点改版和窗口变化,这里只取证成员与相互关系) ==
type(ele.rect): ElementRect
ele.rect 成员: ['click_point', 'corners', 'location', 'midpoint', 'screen_click_point', 'screen_location', 'screen_midpoint', 'scroll_position', 'size', 'viewport_click_point', 'viewport_corners', 'viewport_location', 'viewport_midpoint']
ele.rect.size 类型 -> 'tuple'
ele.rect.location 类型 -> 'tuple'
ele.rect.midpoint 类型 -> 'tuple'
ele.rect.click_point 类型 -> 'tuple'
ele.rect.viewport_location 类型 -> 'tuple'
ele.rect.viewport_midpoint 类型 -> 'tuple'
ele.rect.screen_location 类型 -> 'tuple'
ele.rect.scroll_position 类型 -> 'tuple'
ele.rect.corners 类型 -> 'list'
size 是否为 (宽, 高) 两个数 -> True
宽 > 0 -> True
midpoint.x == location.x + 宽/2 -> False
click_point 在 midpoint 上方 -> True
未滚动时 viewport_location == location -> True
screen_location 与 viewport_location 不同 -> True
corners 是 4 个角点 -> True
type(page.rect): TabRect
page.rect 成员: ['page_location', 'scroll_position', 'size', 'viewport_location', 'viewport_size', 'viewport_size_with_scrollbar', 'window_location', 'window_size', 'window_state']
page.rect.window_state -> 'normal'
window_size 类型 -> 'tuple'
整页高 > 视口高(页面可滚动) -> True
viewport_size 宽 <= window_size 宽 -> True
page.rect.scroll_position -> (0, 0)
viewport_size_with_scrollbar >= viewport_size -> True
== states ==
ele.states.is_displayed -> True
ele.states.is_enabled -> True
ele.states.is_checked -> None
ele.states.is_selected -> None
ele.states.is_alive -> True
ele.states.is_clickable -> True
ele.states.is_covered 的类型 -> 'int'
bool(ele.states.is_covered) -> True
ele.states.is_in_viewport -> True
type(ele.states.has_rect).__name__ -> 'list'
bool(ele.states.has_rect) -> True
type(ele.states).is_in_viewport 存在? -> True
== 关系定位(4.x 没有 sibling()) ==
ele.parent().tag -> 'div'
ele.parent().attr('class') -> 'p-h el-col el-col-24 el-col-xs-9 el-col-sm-13 el-col-md-16'
ele.parent(2).tag -> 'div'
ele.parent('tag:div') 类型 -> 'ChromiumElement'
len(ele.children()) -> 1
ele.child().tag -> 'h2'
ele.next().tag -> 'div'
ele.next().text -> '动画 歌舞 冒险'
len(ele.nexts()) -> 3
ele.before() -> 'ChromiumElement'
len(ele.befores()) -> 27
len(ele.after()) # after 是方法返回元素? -> 'ChromiumElement'
ele.sibling() AttributeError: 'ChromiumElement' object has no attribute 'sibling'
ele.siblings() AttributeError: 'ChromiumElement' object has no attribute 'siblings'
ele.nexts('tag:a') 数量 -> 0
ele.child_count -> 1
== 元素内嵌套查找 ==
card.eles('tag:a') 数量 -> 2
card.ele('tag:a').text -> ''
card('t:h2') 简写 -> '狮子王 - The Lion King'
card.eles('css:.categories > button') 数量 -> 3
card.ele('t:img').attr('src') 以 https 开头 -> True
card.ele('t:img').link -> 'https://'
== 滚动 API 真名 ==
page.scroll: PageScroller ['down', 'left', 'right', 'to_bottom', 'to_half', 'to_leftmost', 'to_location', 'to_rightmost', 'to_see', 'to_top', 'up']
ele.scroll : ElementScroller ['down', 'left', 'right', 'to_bottom', 'to_center', 'to_half', 'to_leftmost', 'to_location', 'to_rightmost', 'to_see', 'to_top', 'up']
page.scroll.to_bottom() 返回类型 -> 'ChromiumPage'
滚动到底后 window.scrollY>0 -> True
ele.scroll.to_center() 返回类型 -> 'ChromiumElement'
元素是否居中(视口内) -> True
ele.scroll.to_see() 返回类型 -> 'ChromiumElement'
page.scroll.to_top() 返回类型 -> 'ChromiumPage'
滚动到顶后 window.scrollY -> 0
page.scroll.down(200) 返回类型 -> 'ChromiumPage'
page.scroll.to_location(0,0) -> 'ChromiumPage'
ele.scroll.to_center() -> 'ChromiumElement'
== 元素上执行 JS ==
ele.run_js('return this.tagName') -> 'A'
ele.run_js 取长度 -> 19
page.run_js 取标题 -> 'Scrape | Movie'
page.run_async_js -> None
  • ele.text 是处理后的文本,ele.raw_text 是原始文本;示例打印两者之前都调用了 brief() 折叠空白,因此相同输出不能说明原始空白相同。texts() 返回文本片段列表,texts(True) 只取直接文本节点:文字若放在子元素 <h2> 内,就不属于 <a> 的直接文本。取完整片名通常用 .text 即可。
  • ele.html 含自身标签,ele.inner_html 只含标签内部内容。可用它们检查链接、子节点和嵌套结构,不必依赖 HTML 的固定长度。
  • ele.css_path 和 ele.xpath 可以辅助确认取到了哪个节点;含位置序号的路径会随 DOM 结构变化,不宜当作长期稳定的选择器。
  • 对链接元素,ele.attr('href') 和 ele.link 会将链接解析为绝对 URL,不能把前者当作读取源码相对值的接口;查看标签中的写法可用 ele.html。不存在的 attr('data-x') 返回 None,与定位不到元素时的 NoneElement 不同。
  • ele.attrs 提供属性字典,sorted(ele.attrs) 可查看属性名。优先选择语义稳定的属性定位,不依赖页面构建工具生成的属性名。
  • ele.style('color') 读取计算样式,不是 CSS 源码的原始写法;ele.style('display', ':before') 读取伪元素样式。
  • ele.rect 提供元素的几何信息:size 是宽高,location、midpoint 分别表示位置和中点,click_point 是点击点,corners 是四角。浏览器布局与取整会影响数值,不应要求不同字段的计算结果始终精确相等。
  • viewport_location 使用视口坐标,screen_location 使用屏幕坐标;阅读坐标前先分清参照系。滚动会改变元素相对于视口的位置,点击点也不必等于几何中心。
  • page.rect 是页面级几何信息;size、viewport_size、window_size 分别描述页面、视口与浏览器窗口。结合 scroll_position 判断滚动状态,不把这些尺寸混为一谈。
  • 本版 ElementStates 有 10 个公开状态属性:is_selected、is_checked、is_displayed、is_enabled、is_alive、is_in_viewport、is_whole_in_viewport、is_covered、is_clickable、has_rect。它们分别回答选中、勾选、显示、可用、存活、视口范围、遮挡、可点击及矩形状态等问题,不应互相替代。
  • is_covered 在点击点调用 DOM.getNodeForLocation;命中节点与自身不同就返回其 backendNodeId,否则返回 False(捕获 CDPError 时也返回 False)。真值不一定表示外部遮挡层,命中子节点也可能不同。has_rect 有矩形时返回角点列表,否则为 False;普通链接的 is_checked/is_selected 可为 None。
  • 本版本未提供 sibling()/siblings()。parent() 向上、children()/child() 向下;next()/nexts() 和 prev()/prevs() 找兄弟节点,before()/befores()、after()/afters() 则按文档顺序查找,不限于兄弟。后者使用 XPath 的 preceding/following 轴,也不等同于查找祖先或后代。
  • 在卡片元素内调用 card.ele(...)、card.eles(...) 可缩小查找范围,card('t:h2') 是单元素查找的简写。图片可读取 attr('src') 或 link,不要将某个站点的图片地址当成固定结果。
  • 页面滚动用 page.scroll,元素滚动用 ele.scroll;如 page.scroll.to_bottom()、page.scroll.to_top()、ele.scroll.to_see() 和 ele.scroll.to_center()。is_in_viewport 为真只说明在视口内,不能单凭它证明精确居中。
  • ele.run_js('return this.tagName') 中的 this 是该元素;page.run_js('return document.title') 是页面级调用。page.run_async_js() 用于异步执行,不应把它的即时返回值当成脚本的计算结果。

等待与超时

浏览器页面可能出现「取早了」。这里区分定位调用的 timeout、页面级的 base/page_load/script 三档,以及 page.wait / ele.wait 的具名等待方法。本机靶页在点击按钮后设置 800ms 定时器,再注入 10 个 <li>;脚本还包含在线 SPA 的对照。

05_waiting_and_timeout.py
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
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
# -*- coding: utf-8 -*-
"""等待机制取证:page.wait 的真实成员、timeout 的三级设置、SPA 不等待与等待的对照。

本机靶站在 127.0.0.1:8052 ,按钮点击后延时 800ms 才注入 10 个 li 。
"""

import re
import socket
import sys
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

from DrissionPage import ChromiumOptions, ChromiumPage

PORT = 8052

PAGE = """<!DOCTYPE html>
<html><head><title>wait-probe</title><meta charset="utf-8"></head>
<body>
<button id="go" onclick="go()">载入</button>
<ul id="list"></ul>
<script>
function go(){
document.getElementById('list').innerHTML = '';
setTimeout(function(){
var h = '';
for (var i = 1; i <= 10; i++) { h += '<li class="x">第' + i + '项</li>'; }
document.getElementById('list').innerHTML = h;
}, 800);
}
</script>
</body></html>
"""


class WaitHandler(BaseHTTPRequestHandler):
def log_message(self, *args):
pass

def do_GET(self):
data = PAGE.encode("utf-8")
self.send_response(200)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.send_header("Content-Length", str(len(data)))
self.end_headers()
self.wfile.write(data)


def start_target():
try:
with socket.create_connection(("127.0.0.1", PORT), 0.2):
print("端口 %d 已被占用,脚本终止" % PORT)
sys.exit(1)
except OSError:
pass
srv = ThreadingHTTPServer(("127.0.0.1", PORT), WaitHandler)
threading.Thread(target=srv.serve_forever, daemon=True).start()
return srv


def brief(text):
return re.sub(r"\s+", " ", text or "").strip()


def show(expr, fn):
try:
print("%-46s -> %r" % (expr, fn()))
except Exception as exc:
print("%-46s %s: %s" % (expr, type(exc).__name__, str(exc).replace("\n", " | ")))


srv = None
page = None
exit_code = 0
try:
srv = start_target()
page = ChromiumPage(ChromiumOptions().headless(True).auto_port())
url = "http://127.0.0.1:%d/" % PORT
print("== dir(page.wait) 真实成员 ==")
print("type(page.wait):", type(page.wait).__name__)
print(sorted(n for n in dir(page.wait) if not n.startswith("_")))
print("page.wait 可调: type(page.wait(0.2)) 返回:", type(page.wait(0.2)).__name__)
show("page.wait.ready()", lambda: page.wait.ready())
show("page.wait.to_stop()", lambda: page.wait.to_stop())
show("page.wait.doc_loaded()", lambda: page.wait.doc_loaded())

print()
print("== timeout 的三级设置 ==")
print("页面级默认:", page.timeout, "| page.timeouts:", page.timeouts)
print("ChromiumPage(timeout=) 与 page.set.timeouts 的关系:")
page.set.timeouts(base=3, page_load=7)
print(" set_timeouts(base=3, page_load=7) 后:", page.timeouts)
page.set.timeouts(base=10, page_load=30)
print(" 复位:", page.timeouts)
show("不存在的名字 page.set_timeouts", lambda: page.set_timeouts)
show("不存在的名字 page.set.auto_port", lambda: page.set.auto_port)

print()
print("== 本机靶站:元素级 timeout 与 wait 成员 ==")
page.get(url)
print(" 标题:", page.title)
e_btn = page.ele("#go")
print(" 按钮元素 type(ele.wait):", type(e_btn.wait).__name__)
print(" ele.wait 成员:", sorted(n for n in dir(e_btn.wait) if not n.startswith("_")))
print(" 元素自身 timeout:", e_btn.timeout)
print(" page('t:button') 简化写法命中:", type(page("t:button")).__name__)
print(" 点击前 eles('t:li', timeout=0) 数量:", len(page.eles("t:li", timeout=0)))
print(" 点击前 ele('t:li', timeout=0) 类型:", type(page.ele("t:li", timeout=0)).__name__)

page.ele("#go").click()
r = page.ele("t:li", timeout=0.3)
print(" 点击后 ele('t:li', timeout=0.3) 类型:", type(r).__name__)
print(" 此时 eles('t:li') 数量:", len(page.eles("t:li", timeout=0)))

page.ele("#go").click()
r2 = page.ele("t:li", timeout=5)
print(" 再点击后 ele('t:li', timeout=5) 类型:", type(r2).__name__,
"| 文本:", brief(r2.text))
print(" 再点击后 eles('t:li') 数量:", len(page.eles("t:li", timeout=5)))
print(" eles('t:li@@class=x') 数量:", len(page.eles("t:li@@class=x", timeout=5)))
print(" wait.eles_loaded('t:li') 返回:", page.wait.eles_loaded("t:li"))

print()
print("== 等待不满足时:raise_err 与超时返回 ==")
show("page.wait.eles_loaded('t:h9', timeout=0.3)",
lambda: page.wait.eles_loaded("t:h9", timeout=0.3))
show("同上 raise_err=True", lambda: page.wait.eles_loaded("t:h9", timeout=0.3, raise_err=True))
show("page.wait.ele_displayed('#go', timeout=0.3)",
lambda: page.wait.ele_displayed("#go", timeout=0.3))
show("page.wait.ele_displayed('t:h9', timeout=0.3)",
lambda: page.wait.ele_displayed("t:h9", timeout=0.3))
show("page.wait.doc_loaded(timeout=0.5)", lambda: page.wait.doc_loaded(timeout=0.5))

print()
print("== 在线 SPA(spa1):不等待 vs 让 ele 的 timeout 生效 ==")
spa = "https://spa1.scrape.center/page/2"
ok = page.get(spa, retry=0, timeout=30)
print(" get() 返回:", ok, "| url:", page.url, "| title:", page.title)
if not ok:
print(" 在线部分失败:导航未成功,跳过等待对照;不是 wait API 错误。")
exit_code = 1
else:
print(" get() 刚返回时 eles('css:div.item', timeout=0):",
len(page.eles("css:div.item", timeout=0)))
items = page.eles("css:div.item", timeout=10)
print(" 紧接着 eles('css:div.item', timeout=10) 数量:", len(items))
if not items:
print(" 在线部分失败:导航返回 True 但 10 秒内无电影卡片;")
print(" 可能是网络、接口或页面变化;本次不能验证在线等待效果。")
exit_code = 1
else:
print(" 在线等待成功:非空电影卡片已出现。")
print(" 等待后再取一次 timeout=0 的数量:",
len(page.eles("css:div.item", timeout=0)))
print(" 等待后 t:h2 数量:", len(page.eles("t:h2", timeout=0)))
page.set.timeouts(base=2)
print(" 把页面级 base 调成 2 后,找不存在元素的 type:",
type(page.ele("t:zzz", timeout=None)).__name__)
page.set.timeouts(base=10)
except Exception as exc:
exit_code = 1
print("块执行失败(与 show 中的预期异常区分):",
type(exc).__name__, brief(str(exc)))
finally:
if page is not None:
try:
page.quit(del_data=True)
print("浏览器已退出并清理本次用户数据。")
except Exception as exc:
exit_code = 1
print("浏览器清理失败:", type(exc).__name__, brief(str(exc)))
if srv is not None:
srv.shutdown()
srv.server_close()
sys.exit(exit_code)
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
== dir(page.wait) 真实成员 ==
type(page.wait): ChromiumPageWaiter
['alert_closed', 'all_downloads_done', 'doc_loaded', 'download_begin', 'downloads_done', 'ele_deleted', 'ele_displayed', 'ele_hidden', 'eles_loaded', 'load_start', 'new_tab', 'title_change', 'upload_paths_inputted', 'url_change']
page.wait 可调: type(page.wait(0.2)) 返回: ChromiumPage
page.wait.ready() AttributeError: 'ChromiumPageWaiter' object has no attribute 'ready'
page.wait.to_stop() AttributeError: 'ChromiumPageWaiter' object has no attribute 'to_stop'
page.wait.doc_loaded() -> True
== timeout 的三级设置 ==
页面级默认: 10 | page.timeouts: {'base': 10, 'page_load': 30, 'script': 30}
ChromiumPage(timeout=) 与 page.set.timeouts 的关系:
set_timeouts(base=3, page_load=7) 后: {'base': 3, 'page_load': 7, 'script': 30}
复位: {'base': 10, 'page_load': 30, 'script': 30}
不存在的名字 page.set_timeouts AttributeError: 'ChromiumPage' object has no attribute 'set_timeouts'
不存在的名字 page.set.auto_port AttributeError: 'ChromiumPageSetter' object has no attribute 'auto_port'
== 本机靶站:元素级 timeout 与 wait 成员 ==
标题: wait-probe
按钮元素 type(ele.wait): ElementWaiter
ele.wait 成员: ['clickable', 'covered', 'deleted', 'disabled', 'disabled_or_deleted', 'displayed', 'enabled', 'has_rect', 'hidden', 'not_covered', 'stop_moving']
元素自身 timeout: 10
page('t:button') 简化写法命中: ChromiumElement
点击前 eles('t:li', timeout=0) 数量: 0
点击前 ele('t:li', timeout=0) 类型: NoneElement
点击后 ele('t:li', timeout=0.3) 类型: NoneElement
此时 eles('t:li') 数量: 0
再点击后 ele('t:li', timeout=5) 类型: ChromiumElement | 文本: 第1项
再点击后 eles('t:li') 数量: 10
eles('t:li@@class=x') 数量: 10
wait.eles_loaded('t:li') 返回: True
== 等待不满足时:raise_err 与超时返回 ==
page.wait.eles_loaded('t:h9', timeout=0.3) -> False
同上 raise_err=True WaitTimeoutError: | 等待元素加载失败(等待0.3秒)。 | 定位符: ('//*[name()="h9"]',) | 版本: 4.1.1.4
page.wait.ele_displayed('#go', timeout=0.3) -> <ChromiumElement button id='go' onclick='go()'>
page.wait.ele_displayed('t:h9', timeout=0.3) -> False
page.wait.doc_loaded(timeout=0.5) -> True
== 在线 SPA(spa1):不等待 vs 让 ele 的 timeout 生效 ==
get() 返回: True | url: https://spa1.scrape.center/page/2 | title: Scrape | Movie
get() 刚返回时 eles('css:div.item', timeout=0): 0
紧接着 eles('css:div.item', timeout=10) 数量: 10
在线等待成功:非空电影卡片已出现。
等待后再取一次 timeout=0 的数量: 10
等待后 t:h2 数量: 10
把页面级 base 调成 2 后,找不存在元素的 type: NoneElement
浏览器已退出并清理本次用户数据。
  • page.wait 列出 14 个具名方法:alert_closed、all_downloads_done、doc_loaded、download_begin、downloads_done、ele_deleted、ele_displayed、ele_hidden、eles_loaded、load_start、new_tab、title_change、upload_paths_inputted、url_change。这是成员枚举,不代表逐个运行过;本版本未提供 wait.ready()、wait.to_stop(),调用会抛 AttributeError。
  • ele.wait 列出 11 个:clickable、covered、deleted、disabled、disabled_or_deleted、displayed、enabled、has_rect、hidden、not_covered、stop_moving。应按元素需要满足的条件选择方法。
  • page.wait 本身可调用:page.wait(0.2) 是固定时长等待,返回 ChromiumPage;固定等待不等于目标元素已就绪。
  • 三档超时分别用于基础等待、页面加载和脚本执行。本机示例的初值是 {'base': 10, 'page_load': 30, 'script': 30}。示例使用 ChromiumOptions(),没有关闭配置文件读取,不能将这组值当作所有环境的保证。
  • 配置对象的方法名是 set_timeouts(base=None, page_load=None, script=None);页面运行中则用 page.set.timeouts(base=3, page_load=7)。未传的 script 保持原值,可读 page.timeouts 确认当前设置。
  • 本版本未提供 page.set_timeouts()、page.set.auto_port()。set_timeouts()、auto_port() 提供在 ChromiumOptions 上,注意区分配置对象与页面对象。
  • 本机按钮示例中,点击前列表为空,点击后由定时器注入列表项。timeout=0 表示不等待,timeout=0.3 或 timeout=5 则给定位设置等待上限,并非总要等满该时长。短超时可能来不及找到元素;两次点击的定时器没有取消,不能把它当作精确计时实验。
  • page.wait.eles_loaded('t:h9', timeout=0.3) 演示等待失败,示例设置下返回 False,加 raise_err=True 抛 WaitTimeoutError;不要推广为所有等待 API 的统一返回规则。
  • page.wait.ele_displayed('#go', timeout=0.3) 成功时返回元素对象,而非 True;示例中查不存在的 t:h9 返回 False。使用返回值时要区分元素与失败值。
  • 对在线 SPA,先确认 get() 成功,再比较 eles(..., timeout=0) 与允许等待的定位调用。两次取值之间 DOM 可能变化,不是同一时刻的 DOM 对照,也不保证固定的数量变化;请求失败、接口异常或定位串失效都不能直接归因于等待机制。

page.wait.doc_loaded() 面向文档状态,不保证之后的定时器或异步任务已经改完 DOM;本块的 eles_loaded、ele_displayed 示例则以目标元素为等待条件。

配置对象

ChromiumOptions / SessionOptions 是配置入口,不启动浏览器也能检查成员和读取设置项。下面枚举 ChromiumOptions 的 57 个公开成员,打印可取得的签名,再回显部分设置;签名存在或选项写入不等于浏览器行为已验证。本节使用 DrissionPage 4.1.1.4。

06_options_surface.py
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
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
# -*- coding: utf-8 -*-
"""ChromiumOptions / SessionOptions 的成员清单与签名:本版本(4.1.1.4)的成员检查。"""

import inspect

from DrissionPage import ChromiumOptions, SessionOptions


def probe_options(cls, names):
print("== %s 公开成员 ==" % cls.__name__)
members = sorted(n for n in dir(cls) if not n.startswith("_"))
print("共 %d 个:" % len(members))
for i in range(0, len(members), 6):
print(" " + " ".join("%-26s" % n for n in members[i:i + 6]))
print()
print("-- 逐个签名 --")
for i in range(0, len(members), 4):
line = []
for n in members[i:i + 4]:
obj = getattr(cls, n)
try:
line.append("%s%s" % (n, inspect.signature(obj)))
except (TypeError, ValueError):
line.append("%s <%s>" % (n, type(obj).__name__))
print(" " + " | ".join(line))
print()
print("-- 所列名字在本版本是否提供 --")
for n in names:
obj = getattr(cls, n, None)
if obj is None:
print(" %-24s 不存在(AttributeError)" % n)
else:
try:
print(" %-24s %s" % (n, inspect.signature(obj)))
except (TypeError, ValueError):
print(" %-24s <%s>" % (n, type(obj).__name__))


legacy = ["set_paths", "no_infobar", "user_agent", "set_browser_path", "tmp_path",
"mobile_emulation", "set_pref", "ignore_certificate_errors", "no_imgs", "no_js",
"mute", "new_env", "set_flag", "existing_only", "headless", "incognito",
"auto_port", "set_local_port", "set_address", "address", "local_port",
"set_cache_path", "set_user_data_path", "use_system_user_path", "set_proxy",
"set_download_path", "set_load_mode", "set_timeouts", "set_retry", "add_extension",
"remove_argument", "clear_arguments", "clear_prefs", "remove_pref", "save",
"save_to_default", "load_mode", "proxy", "download_path", "retry_times",
"arguments", "flags", "prefs", "encoding", "set_encoding"]
probe_options(ChromiumOptions, legacy)

print()
print("== 只读回显:把选项设进对象再读出来(不启动浏览器) ==")
opts = ChromiumOptions(read_file=False)
print("read_file=False 时 ini_path:", opts.ini_path)
print("默认 address:", repr(opts.address), "| 默认 is_headless:", opts.is_headless)
print("默认 is_auto_port:", opts.is_auto_port, "| 默认 is_existing_only:", opts.is_existing_only)
opts.headless(True)
print("headless(True) 后 is_headless:", opts.is_headless)
opts.set_user_agent("DP-Probe/9.9")
print("set_user_agent 后 hasattr(opts, 'user_agent'):", hasattr(opts, "user_agent"))
opts.set_local_port(8053)
print("set_local_port(8053) 后 address:", repr(opts.address), "| is_auto_port:", opts.is_auto_port)
opts.auto_port()
print("auto_port() 后 address:", repr(opts.address), "| is_auto_port:", opts.is_auto_port)
opts.set_argument("--window-size=1200,900")
print("arguments 共 %d 项:" % len(opts.arguments))
for _a in sorted(opts.arguments):
print(" " + _a)
opts.set_pref("profile.default_content_settings.popups", 0)
print("preferences 是否含该项:", "profile.default_content_settings.popups" in opts.preferences)
opts.set_load_mode("eager")
print("load_mode:", opts.load_mode)
opts.set_proxy("http://127.0.0.1:8053")
print("proxy:", repr(opts.proxy))
opts.no_imgs(True)
opts.no_js(True)
opts.mute(True)
opts.incognito(True)
opts.ignore_certificate_errors(True)
opts.set_timeouts(base=5)
opts.set_retry(times=3, interval=0.5)
print("timeouts:", opts.timeouts, "| retry_times:", opts.retry_times,
"| retry_interval:", opts.retry_interval)
print("flags:", opts.flags, "| download_path:", repr(opts.download_path))
print("extensions:", opts.extensions, "| user:", repr(opts.user))
print("browser_path 尾段:", str(opts.browser_path).split("\\")[-1])
print("user_data_path 尾段:", str(opts.user_data_path).split("\\")[-1])
print("tmp_path 尾段:", str(opts.tmp_path).split("\\")[-1])
for name in ("prefs", "user_agent", "local_port", "encoding", "mobile_emulation"):
obj = getattr(opts, name, "<无此属性>")
print(" 读 opts.%-16s -> %s" % (name, "是个方法" if callable(obj) else repr(obj)[:60]))
for name in ("set_paths", "no_infobar", "user_agent", "mobile_emulation"):
try:
fn = getattr(opts, name)
except Exception as exc:
print(" 取 %-16s -> %s: %s" % (name, type(exc).__name__, exc))
continue
try:
fn()
print(" %-16s -> 调用成功" % name)
except Exception as exc:
print(" %-16s -> %s: %s" % (name, type(exc).__name__, str(exc).splitlines()[0]))

print()
print("== set_paths 一次设置多个选项 ==")
o2 = ChromiumOptions(read_file=False)
o2.set_paths(browser_path="D:/fake/chrome.exe", local_port=8053)
print("address:", repr(o2.address), "| browser_path:", repr(str(o2.browser_path)))

print()
print("== SessionOptions ==")
sess = SessionOptions(read_file=False)
print("ini_path:", sess.ini_path)
smethods = sorted(n for n in dir(sess) if not n.startswith("_"))
print("SessionOptions 公开成员 %d 个:" % len(smethods))
for _i in range(0, len(smethods), 8):
print(" " + " ".join("%-20s" % n for n in smethods[_i:_i + 8]))
for n in smethods:
obj = getattr(SessionOptions, n, None)
if obj is None:
continue
if isinstance(obj, property):
print(" %-18s <property>" % n)
continue
try:
print(" %-18s %s" % (n, inspect.signature(obj)))
except (TypeError, ValueError):
print(" %-18s <%s>" % (n, type(obj).__name__))
print("默认值: timeout", sess.timeout, "| proxies", sess.proxies,
"| verify", sess.verify, "| retry_times", sess.retry_times)
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
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
== ChromiumOptions 公开成员 ==
共 57 个:
add_extension address arguments auto_port browser_path clear_arguments
clear_flags clear_flags_in_file clear_prefs download_path existing_only extensions
flags headless ignore_certificate_errors incognito is_auto_port is_existing_only
is_headless load_mode mute new_env no_imgs no_js
preferences proxy remove_argument remove_extensions remove_pref remove_pref_from_file
retry_interval retry_times save save_to_default set_address set_argument
set_browser_path set_cache_path set_download_path set_flag set_load_mode set_local_port
set_paths set_pref set_proxy set_retry set_timeouts set_tmp_path
set_user set_user_agent set_user_data_path system_user_path timeouts tmp_path
use_system_user_path user user_data_path
-- 逐个签名 --
add_extension(self, path) | address <property> | arguments <property> | auto_port(self, on_off=True, scope=None)
browser_path <property> | clear_arguments(self) | clear_flags(self) | clear_flags_in_file(self)
clear_prefs(self) | download_path <property> | existing_only(self, on_off=True) | extensions <property>
flags <property> | headless(self, on_off=True) | ignore_certificate_errors(self, on_off=True) | incognito(self, on_off=True)
is_auto_port <property> | is_existing_only <property> | is_headless <property> | load_mode <property>
mute(self, on_off=True) | new_env(self, on_off=True) | no_imgs(self, on_off=True) | no_js(self, on_off=True)
preferences <property> | proxy <property> | remove_argument(self, value) | remove_extensions(self)
remove_pref(self, arg) | remove_pref_from_file(self, arg) | retry_interval <property> | retry_times <property>
save(self, path=None) | save_to_default(self) | set_address(self, address) | set_argument(self, arg, value=None)
set_browser_path(self, path) | set_cache_path(self, path) | set_download_path(self, path) | set_flag(self, flag, value=None)
set_load_mode(self, value) | set_local_port(self, port) | set_paths(self, browser_path=None, local_port=None, address=None, download_path=None, user_data_path=None, cache_path=None) | set_pref(self, arg, value)
set_proxy(self, proxy) | set_retry(self, times=None, interval=None) | set_timeouts(self, base=None, page_load=None, script=None) | set_tmp_path(self, path)
set_user(self, user='Default') | set_user_agent(self, user_agent) | set_user_data_path(self, path) | system_user_path <property>
timeouts <property> | tmp_path <property> | use_system_user_path(self, on_off=True) | user <property>
user_data_path <property>
-- 所列名字在本版本是否提供 --
set_paths (self, browser_path=None, local_port=None, address=None, download_path=None, user_data_path=None, cache_path=None)
no_infobar 不存在(AttributeError)
user_agent 不存在(AttributeError)
set_browser_path (self, path)
tmp_path <property>
mobile_emulation 不存在(AttributeError)
set_pref (self, arg, value)
ignore_certificate_errors (self, on_off=True)
no_imgs (self, on_off=True)
no_js (self, on_off=True)
mute (self, on_off=True)
new_env (self, on_off=True)
set_flag (self, flag, value=None)
existing_only (self, on_off=True)
headless (self, on_off=True)
incognito (self, on_off=True)
auto_port (self, on_off=True, scope=None)
set_local_port (self, port)
set_address (self, address)
address <property>
local_port 不存在(AttributeError)
set_cache_path (self, path)
set_user_data_path (self, path)
use_system_user_path (self, on_off=True)
set_proxy (self, proxy)
set_download_path (self, path)
set_load_mode (self, value)
set_timeouts (self, base=None, page_load=None, script=None)
set_retry (self, times=None, interval=None)
add_extension (self, path)
remove_argument (self, value)
clear_arguments (self)
clear_prefs (self)
remove_pref (self, arg)
save (self, path=None)
save_to_default (self)
load_mode <property>
proxy <property>
download_path <property>
retry_times <property>
arguments <property>
flags <property>
prefs 不存在(AttributeError)
encoding 不存在(AttributeError)
set_encoding 不存在(AttributeError)
== 只读回显:把选项设进对象再读出来(不启动浏览器) ==
read_file=False 时 ini_path: None
默认 address: '127.0.0.1:9222' | 默认 is_headless: False
默认 is_auto_port: False | 默认 is_existing_only: False
headless(True) 后 is_headless: True
set_user_agent 后 hasattr(opts, 'user_agent'): False
set_local_port(8053) 后 address: '127.0.0.1:8053' | is_auto_port: False
auto_port() 后 address: '' | is_auto_port: (9600, 59600)
arguments 共 10 项:
--disable-features=PrivacySandboxSettings4
--disable-infobars
--disable-popup-blocking
--disable-suggestions-ui
--headless=new
--hide-crash-restore-bubble
--no-default-browser-check
--no-first-run
--user-agent=DP-Probe/9.9
--window-size=1200,900
preferences 是否含该项: True
load_mode: eager
proxy: 'http://127.0.0.1:8053'
timeouts: {'base': 5, 'page_load': 30, 'script': 30} | retry_times: 3 | retry_interval: 0.5
flags: {} | download_path: '.'
extensions: [] | user: 'Default'
browser_path 尾段: chrome
user_data_path 尾段: None
tmp_path 尾段: None
读 opts.prefs -> '<无此属性>'
读 opts.user_agent -> '<无此属性>'
读 opts.local_port -> '<无此属性>'
读 opts.encoding -> '<无此属性>'
读 opts.mobile_emulation -> '<无此属性>'
set_paths -> 调用成功
取 no_infobar -> AttributeError: 'ChromiumOptions' object has no attribute 'no_infobar'
取 user_agent -> AttributeError: 'ChromiumOptions' object has no attribute 'user_agent'
取 mobile_emulation -> AttributeError: 'ChromiumOptions' object has no attribute 'mobile_emulation'
== set_paths 一次设置多个选项 ==
address: '127.0.0.1:8053' | browser_path: 'D:/fake/chrome.exe'
== SessionOptions ==
ini_path: None
SessionOptions 公开成员 40 个:
adapters add_adapter as_dict auth cert clear_headers cookies download_path
from_session headers hooks ini_path make_session max_redirects params proxies
remove_a_header retry_interval retry_times save save_to_default set_a_header set_auth set_cert
set_cookies set_download_path set_headers set_hooks set_max_redirects set_params set_proxies set_retry
set_stream set_timeout set_trust_env set_verify stream timeout trust_env verify
adapters <property>
add_adapter (self, url, adapter)
as_dict (self)
auth <property>
cert <property>
clear_headers (self)
cookies <property>
download_path <property>
from_session (self, session, headers=None)
headers <property>
hooks <property>
make_session (self)
max_redirects <property>
params <property>
proxies <property>
remove_a_header (self, name)
retry_interval <property>
retry_times <property>
save (self, path=None)
save_to_default (self)
set_a_header (self, name, value)
set_auth (self, auth)
set_cert (self, cert)
set_cookies (self, cookies)
set_download_path (self, path)
set_headers (self, headers)
set_hooks (self, hooks)
set_max_redirects (self, times)
set_params (self, params)
set_proxies (self, http=None, https=None)
set_retry (self, times=None, interval=None)
set_stream (self, on_off)
set_timeout (self, second)
set_trust_env (self, on_off)
set_verify (self, on_off)
stream <property>
timeout <property>
trust_env <property>
verify <property>
默认值: timeout 10 | proxies {'http': '', 'https': ''} | verify None | retry_times 3
  • 06 用的是 ChromiumOptions(read_file=False)。设置前回显 address='127.0.0.1:9222',is_headless、is_auto_port、is_existing_only 均为 False。SessionOptions(read_file=False) 的记录为 timeout=10、proxies={'http': '', 'https': ''}、verify=None。
  • load_mode='eager' 来自 set_load_mode('eager'),retry_times=3、retry_interval=0.5 来自 set_retry(times=3, interval=0.5),不是初始默认值;base=5 同样是显式设置。本版源码在不读文件时初始化为 load_mode='normal'、retry_interval=2,要分清初始化值与主动写入的配置。
  • headless(True) 后 is_headless=True,参数列表出现 --headless=new。set_user_agent('DP-Probe/9.9') 后出现 --user-agent=DP-Probe/9.9,但 ChromiumOptions 本版本未提供 user_agent 读取属性;这不等于页面对象没有该属性。
  • set_local_port(8053) 后地址为 127.0.0.1:8053;随后 auto_port() 回显 address=''、is_auto_port=(9600, 59600)。这是自动选端口的配置状态,本段未启动浏览器,不能据此保证并发实例永不冲突。
  • arguments 的 10 项由内置参数和脚本新加参数共同组成。--window-size=1200,900 是脚本显式 set_argument() 写入的,不是默认窗口尺寸,而且这里没有启动窗口验证最终大小。其余内置项包括 --disable-infobars、--disable-popup-blocking、--no-first-run、--no-default-browser-check、--disable-features=PrivacySandboxSettings4、--disable-suggestions-ui、--hide-crash-restore-bubble。
  • set_pref() 后检查 preferences 是否含指定键,记录为 True,不代表这个属性只能做成员判断。本版本的 ChromiumOptions 未提供 no_infobar、user_agent、mobile_emulation、local_port、prefs、encoding、set_encoding 这些所测名称;不推断它们是否来自其他版本或框架。
  • set_paths() 在本版本提供。签名有 browser_path、local_port、address、download_path、user_data_path、cache_path 六个可选参数,不是六个文件路径。示例只设置浏览器路径和端口,并读回这两项。
  • read_file=False 不读取 ini,此时 ini_path=None 不是「默认文件位置」。未传该参数的 ChromiumOptions() 会走配置文件读取逻辑;排查配置差异时,应确认是否读取了配置文件及其内容,不要与本节的无文件初始化混为一谈。
  • SessionOptions 的 40 个成员包含 headers、cookies、proxies、auth、cert、verify、timeout 等会话设置。添加适配器、设置超时、接收已有会话的成员与签名,另见后面的补充代码,避免只凭名称猜参数。用于接收会话的 from_session 在这里仅枚举签名,没有实际调用,不据此声称 requests 用法都能直接平移。

运行模式与接管

配置开关写入后是否产生预期行为,还要看页面结果。本节用自控页面回显请求头、执行 JS、加载图片;HTTP 服务地址与浏览器 CDP 调试地址用途不同。下述行为限于本机 Chrome 156(156.0.8078.80)和示例配置,不推广到其他版本或所有站点。

07_run_mode_and_options.py
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
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
# -*- coding: utf-8 -*-
"""运行模式与浏览器选项的实测效果:headless / UA / no_js / no_imgs / incognito /
load_mode / set_local_port 接管 / existing_only 。本机靶站 127.0.0.1:8068 。"""

import re
import socket
import struct
import sys
import threading
import zlib
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

from DrissionPage import Chromium, ChromiumOptions, ChromiumPage

PORT = 8068


def make_png():
def chunk(tag, payload):
body = tag + payload
return (struct.pack(">I", len(payload)) + body
+ struct.pack(">I", zlib.crc32(body) & 0xFFFFFFFF))

raw = b"\x00\xff\x00\x00\xff" # 1x1 RGBA:每行以一个过滤字节开头
return (b"\x89PNG\r\n\x1a\n"
+ chunk(b"IHDR", struct.pack(">IIBBBBB", 1, 1, 8, 6, 0, 0, 0))
+ chunk(b"IDAT", zlib.compress(raw))
+ chunk(b"IEND", b""))


PNG = make_png()


class ProbeHandler(BaseHTTPRequestHandler):
def log_message(self, *args):
pass

def do_GET(self):
if self.path.startswith("/px.png"):
data = PNG
ctype = "image/png"
else:
ua = self.headers.get("User-Agent", "")
body = ("<html><head><title>server-echo</title><meta charset='utf-8'></head><body>"
"<p id='seen'>%s</p>"
"<p id='js'></p><p id='flag'></p>"
"<img id='pic' src='/px.png' alt='pic'>"
"<script>document.getElementById('js').textContent='JS已执行';"
"document.getElementById('flag').textContent=navigator.userAgent;"
"document.title='ua-probe';</script>"
"</body></html>") % re.sub(r"[<>&]", "", ua)
data = body.encode("utf-8")
ctype = "text/html; charset=utf-8"
self.send_response(200)
self.send_header("Content-Type", ctype)
self.send_header("Content-Length", str(len(data)))
self.end_headers()
self.wfile.write(data)


def start_target():
try:
with socket.create_connection(("127.0.0.1", PORT), 0.2):
print("端口 %d 已被占用,脚本终止" % PORT)
sys.exit(1)
except OSError:
pass
srv = ThreadingHTTPServer(("127.0.0.1", PORT), ProbeHandler)
threading.Thread(target=srv.serve_forever, daemon=True).start()
return srv


def fmt_exc(exc):
return " | ".join(line.strip() for line in str(exc).strip().splitlines())


def show(expr, fn):
try:
print("%-46s -> %r" % (expr, fn()))
except Exception as exc:
print("%-46s %s: %s" % (expr, type(exc).__name__, fmt_exc(exc)))


def new_page(**kw):
opts = ChromiumOptions().auto_port()
if "headless" in kw:
opts.headless(kw["headless"])
if "ua" in kw:
opts.set_user_agent(kw["ua"])
if kw.get("no_js"):
opts.no_js(True)
if kw.get("no_imgs"):
opts.no_imgs(True)
if kw.get("incognito"):
opts.incognito(True)
if "load_mode" in kw:
opts.set_load_mode(kw["load_mode"])
return ChromiumPage(opts)


srv = start_target()
base = "http://127.0.0.1:%d/" % PORT
page = None
try:
print("== 1. headless(True) ==")
page = new_page(headless=True)
page.get(base)
print(" page.user_agent:", page.user_agent)
print(" 服务端看到的 UA:", page.ele("#seen").text)
print(" JS 读到的 navigator.userAgent:", page.ele("#flag").text)
print(" JS 是否执行(#js 文本):", repr(page.ele("#js").text))
print(" UA 里含 Headless 字样:", "Headless" in page.user_agent)
show("navigator.webdriver", lambda: page.run_js("return navigator.webdriver"))
show("page.browser_version", lambda: page.browser_version)
show("page.title", lambda: page.title)
page.quit()

print()
print("== 2. headless(False)(有桌面会话才能起) ==")
try:
page = new_page(headless=False)
page.get(base)
print(" 启动成功 | UA:", page.user_agent)
print(" UA 里含 Headless 字样:", "Headless" in page.user_agent)
print(" JS 读到的 navigator.userAgent:", page.ele("#flag").text)
page.quit()
page = None
except Exception as exc:
print(" 启动失败:", type(exc).__name__ + ":", fmt_exc(exc))

print()
print("== 3. set_user_agent 改写 ==")
page = new_page(headless=True, ua="DP-Probe/9.9 (custom)")
page.get(base)
print(" page.user_agent:", page.user_agent)
print(" 服务端看到的 UA:", page.ele("#seen").text)
print(" JS 读到的 navigator.userAgent:", page.ele("#flag").text)
page.quit()

print()
print("== 4. no_js(True) 到底管不管用 ==")
page = new_page(headless=True)
page.get(base)
print(" 对照(普通模式) #js =", repr(page.ele("#js").text))
page.quit()
page = new_page(headless=True, no_js=True)
page.get(base)
print(" no_js(True) 之后 #js =", repr(page.ele("#js").text))
print(" no_js 实际写入的启动参数:",
[a for a in ChromiumOptions().no_js(True).arguments if "javascript" in a])
page.quit()
page = new_page(headless=True)
page.get(base)
page.run_cdp("Emulation.setScriptExecutionDisabled", value=True)
page.refresh()
print(" 改用 CDP setScriptExecutionDisabled 后 #js =", repr(page.ele("#js").text),
"(空=JS 被关)")
page.run_cdp("Emulation.setScriptExecutionDisabled", value=False)
page.quit()

print()
print("== 5. no_imgs(True) 到底管不管用 ==")
js_w = "return document.getElementById('pic') ? document.getElementById('pic').naturalWidth : -1"
page = new_page(headless=True)
page.get(base)
show(" 普通模式 naturalWidth", lambda: page.run_js(js_w))
page.quit()
page = new_page(headless=True, no_imgs=True)
page.get(base)
show(" no_imgs(True) naturalWidth", lambda: page.run_js(js_w))
page.set.blocked_urls("/px.png")
page.get(base)
show(" 改用 set.blocked_urls 后再载入 naturalWidth", lambda: page.run_js(js_w))
page.quit()

print()
print("== 6. incognito(True) ==")
page = new_page(headless=True, incognito=True)
page.get(base)
print(" 启动成功,标题:", page.title)
show("page.set.local_storage 调用", lambda: type(page.set.local_storage("dp", "v1")).__name__)
show("无痕下 local_storage 读回", lambda: page.local_storage("dp"))
page.quit()

print()
print("== 7. set_load_mode 三种取值 ==")
for mode in ("normal", "eager", "none"):
page = new_page(headless=True, load_mode=mode)
page.wait(0.01)
page.get(base)
print(" %-7s -> page.load_mode:" % mode, page.load_mode,
"| get() 立刻数 p 元素:", len(page.eles("t:p", timeout=0)),
"| 允许 timeout=5 再数:", len(page.eles("t:p", timeout=5)))
page.quit()
try:
ChromiumOptions().set_load_mode("wrong_mode")
print(" 非法取值: 未报错")
except Exception as exc:
print(" 非法取值 ->", type(exc).__name__ + ":", fmt_exc(exc))

print()
print("== 8. set_local_port 固定端口 + Chromium 接管 ==")
opts = ChromiumOptions().headless(True).set_local_port(8071)
page = ChromiumPage(opts)
page.get(base)
print(" page.address:", page.address)
print(" 第一个标签页 tabs_count:", page.tabs_count)
page.new_tab(base + "?tab=second")
print(" 新标签后 tabs_count:", page.tabs_count)
browser = Chromium("127.0.0.1:8071")
print(" Chromium('127.0.0.1:8071') 接管成功 | 类型:", type(browser).__name__)
print(" browser.tabs_count:", browser.tabs_count)
print(" browser.tab_ids 数量:", len(browser.tab_ids))
titles = sorted(t.title for t in browser.get_tabs())
print(" 两个标签的标题:", titles)
page.quit()

print()
print("== 9. existing_only(True) :没有现成浏览器时 ==")
try:
ChromiumPage(ChromiumOptions().existing_only(True).set_local_port(8071)).get(base)
print(" 居然连上了(说明 8071 上还有活的浏览器)")
except Exception as exc:
print(" ->", type(exc).__name__ + ":", fmt_exc(exc))
finally:
if page is not None:
try:
page.quit()
except Exception:
pass
srv.shutdown()
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
== 1. headless(True) ==
page.user_agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) HeadlessChrome/156.0.0.0 Safari/537.36
服务端看到的 UA: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) HeadlessChrome/156.0.0.0 Safari/537.36
JS 读到的 navigator.userAgent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) HeadlessChrome/156.0.0.0 Safari/537.36
JS 是否执行(#js 文本): 'JS已执行'
UA 里含 Headless 字样: True
navigator.webdriver -> False
page.browser_version -> 'Chrome/156.0.8078.80'
page.title -> 'ua-probe'
== 2. headless(False)(有桌面会话才能起) ==
启动成功 | UA: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/156.0.0.0 Safari/537.36
UA 里含 Headless 字样: False
JS 读到的 navigator.userAgent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/156.0.0.0 Safari/537.36
== 3. set_user_agent 改写 ==
page.user_agent: DP-Probe/9.9 (custom)
服务端看到的 UA: DP-Probe/9.9 (custom)
JS 读到的 navigator.userAgent: DP-Probe/9.9 (custom)
== 4. no_js(True) 到底管不管用 ==
对照(普通模式) #js = 'JS已执行'
no_js(True) 之后 #js = 'JS已执行'
no_js 实际写入的启动参数: ['--disable-javascript']
改用 CDP setScriptExecutionDisabled 后 #js = '' (空=JS 被关)
== 5. no_imgs(True) 到底管不管用 ==
普通模式 naturalWidth -> 1
no_imgs(True) naturalWidth -> 1
改用 set.blocked_urls 后再载入 naturalWidth -> 0
== 6. incognito(True) ==
启动成功,标题: ua-probe
page.set.local_storage 调用 -> 'NoneType'
无痕下 local_storage 读回 -> 'v1'
== 7. set_load_mode 三种取值 ==
normal -> page.load_mode: normal | get() 立刻数 p 元素: 3 | 允许 timeout=5 再数: 3
eager -> page.load_mode: eager | get() 立刻数 p 元素: 3 | 允许 timeout=5 再数: 3
none -> page.load_mode: none | get() 立刻数 p 元素: 3 | 允许 timeout=5 再数: 3
非法取值 -> ValueError: value参数值错误。 | 允许值: 'normal', 'eager', 'none' | 当前值: wrong_mode | 版本: 4.1.1.4
== 8. set_local_port 固定端口 + Chromium 接管 ==
page.address: 127.0.0.1:8071
第一个标签页 tabs_count: 1
新标签后 tabs_count: 2
Chromium('127.0.0.1:8071') 接管成功 | 类型: Chromium
browser.tabs_count: 2
browser.tab_ids 数量: 2
两个标签的标题: ['ua-probe', 'ua-probe']
== 9. existing_only(True) :没有现成浏览器时 ==
-> BrowserConnectError: 浏览器连接失败,请确认浏览器已启动。 | 地址: 127.0.0.1:8071 | 版本: 4.1.1.4
  • 无头模式下,页面属性、服务端请求头、页面 JS 三处 UA 记录一致,包含 HeadlessChrome/156.0.0.0;有头模式记录了页面属性和 JS 两处,包含 Chrome/156.0.0.0,没有打印有头模式的服务端 UA 对照。这只是一项可观察差异,不是两种模式的全部差异。
  • set_user_agent('DP-Probe/9.9 (custom)') 后,上述三处记录同为该字符串;结论限于本机靶页的这三处观测,不声称所有客户端特征均已改变。
  • navigator.webdriver 只在无头模式那组被读取,记录为 False(JS 的 false);脚本没有读取有头模式的此属性。单个本机结果既不是跨环境保证,也不表示站点无法检测自动化。
  • 在本机 Chrome 156 的记录中,no_js(True) 写入了 --disable-javascript,但 #js 仍为 'JS已执行'。07 实际使用的是 page.run_cdp('Emulation.setScriptExecutionDisabled', value=True),随后 page.refresh(),记录 #js 变空,再用 value=False 恢复。不要把这个命令写成其他 CDP 域的方法。
  • 同样只限本机 Chrome 156:no_imgs(True) 后图片 naturalWidth 仍为 1;脚本调用 page.set.blocked_urls('/px.png') 再加载,记录变为 0。这是本地单张图片的对照,不证明跨浏览器版本效果或总体流量节省。
  • set_load_mode() 的 normal、eager、none 三组在两次取值时均数到 3 个 <p>,非法值抛 ValueError。第一次是 timeout=0,第二次是 eles(..., timeout=5),表示最多等待 5 秒,不是固定暂停。这些 <p> 本来就在 HTML 中,不能用这组数量判断三种加载模式的耗时差异。
  • incognito(True) 那组启动后,page.set.local_storage('dp', 'v1') 写入,再读回 'v1'。这只验证当前无痕会话内可读写,不代表数据持久化;没有关闭并重开后的存储对照,不据此推断重启、落盘或清除行为。
  • 固定端口那组先用 ChromiumOptions().headless(True).set_local_port(8071) 创建页面,再在同一 Python 进程内调用 Chromium('127.0.0.1:8071');读到 2 个本机标签,标题均为 ua-probe。该调用可能复用进程内已有对象,不能作为另一个进程独立接管成功的证据。
  • existing_only(True) 那组在关闭浏览器后尝试连接 8071,得到 BrowserConnectError。这个选项表示「只连接已有浏览器」;示例展示无可连接实例时的失败路径,不涉及已登录浏览器或跨进程接管。

标签页与多实例

先分清浏览器对象、绑定某个标签的页面对象,以及单独的标签对象。本段从 ChromiumPage 开始,打开多个页面,演示查找、激活、新建和关闭标签;多标签不等于多浏览器实例,这里也没有多进程并发实验。涉及在线页面的标题或 URL 过滤时,先确认导航成功。

08_tabs.py
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
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
# -*- coding: utf-8 -*-
"""多标签页 / 多窗口:new_tab、get_tab、get_tabs、tab_ids、activate_tab、close_tabs、latest_tab 。"""

import sys

from DrissionPage import ChromiumOptions, ChromiumPage
from DrissionPage.items import ChromiumTab, MixTab

URL_A = "https://spa1.scrape.center/"
URL_B = "https://quotes.toscrape.com/js/"
URL_C = "https://quotes.toscrape.com/"

page = None
exit_code = 0


def show(expr, fn):
try:
value = fn()
print("%-44s -> %r" % (expr, value))
return value
except Exception as exc:
print("%-44s %s: %s" % (
expr, type(exc).__name__,
" | ".join(l.strip() for l in str(exc).strip().splitlines())))
return None


def get(tab, url):
global exit_code
ok = tab.get(url, retry=0, timeout=30)
print("导航 get:", ok, "| url:", tab.url, "| title:", tab.title)
if not ok:
exit_code = 1
print("导航失败:保留真实标签,标题过滤可能无匹配,不视为标签 API 错误。")
return ok


def visibility(label):
print(label, "| B hidden:", tab_b.run_js("return document.hidden"),
"| C hidden:", tab_c.run_js("return document.hidden"))
print("tab_ids 实际顺序(不排序):",
[page.get_tab(tid).url for tid in page.tab_ids])


try:
page = ChromiumPage(ChromiumOptions().headless(True).auto_port())
print("站点 A=spa1,B=quotes/js,C=quotes 根;不用 ssr1 做标题靶站。")
print("== 两种标签类型 ==")
_a = {n for n in dir(ChromiumTab) if not n.startswith("_")}
_b = {n for n in dir(MixTab) if not n.startswith("_")}
print("ChromiumTab MRO:", " -> ".join(c.__name__ for c in ChromiumTab.__mro__[:4]))
print("MixTab MRO:", " -> ".join(c.__name__ for c in MixTab.__mro__[:5]))
print("MixTab 比 ChromiumTab 多:", sorted(_b - _a))
print()
get(page, URL_A)
print("起始: tabs_count =", page.tabs_count, "| page.url =", page.url)
print("type(page.latest_tab) =", type(page.latest_tab).__name__,
"| latest_tab.url =", page.latest_tab.url)

print()
print("== 新开两个标签 ==")
tab_b = page.new_tab(URL_B)
print("new_tab() 返回类型:", type(tab_b).__name__)
print(" tab_b.url:", tab_b.url)
print(" tab_b.title:", tab_b.title)
print(" tab_b.tab_id 长度:", len(tab_b.tab_id))
print(" tab_b.browser 类型:", type(tab_b.browser).__name__)
print(" tab_b 是否有 quit/close:", hasattr(tab_b, "quit"), hasattr(tab_b, "close"))
print(" tab_b 是否有 change_mode(切换 d/s 模式):", hasattr(tab_b, "change_mode"))
tab_c = page.new_tab()
tab_c.get(URL_C)
print(" 第二个新标签 tab_c.url:", tab_c.url, "| title:", tab_c.title)
print("tabs_count =", page.tabs_count, "| len(tab_ids) =", len(page.tab_ids))
print("page.url 现在指向:", page.url)
print("page.title 现在指向:", page.title)
print("page.latest_tab.url:", page.latest_tab.url)

print()
print("== get_tab / get_tabs ==")
print("get_tabs() 数量:", len(page.get_tabs()))
print("get_tabs() 类型:", sorted({type(t).__name__ for t in page.get_tabs()}))
print("get_tabs() 标题:", sorted(t.title for t in page.get_tabs()))
print("get_tabs() 每项类型:", [type(t).__name__ for t in page.get_tabs()])
for index in (0, 1, -1):
show("get_tab(%d).url" % index, lambda index=index: page.get_tab(index).url)
show("get_tab(title='Scrape | Movie').url",
lambda: page.get_tab(title="Scrape | Movie").url)
show("get_tab(url='quotes').url", lambda: page.get_tab(url="quotes").url)
show("get_tab(9)", lambda: page.get_tab(9))
show("get_tab(title='zzz')", lambda: page.get_tab(title="zzz"))
print("get_tabs(url='scrape.center') 数量:", len(page.get_tabs(url="scrape.center")))
print("get_tabs(url='quotes') 数量:", len(page.get_tabs(url="quotes")))
show("get_tab(tab_c.tab_id) is tab_c", lambda: page.get_tab(tab_c.tab_id) is tab_c)
show("get_tab(1) 两次是否同一对象", lambda: page.get_tab(1) is page.get_tab(1))

print()
print("== 各序号对应的标签(保留实际 tab_ids 顺序) ==")
print("按创建顺序的 url:", [URL_A, URL_B, URL_C])
for index in (0, 1, 2):
show("get_tab(%d).url" % index, lambda index=index: page.get_tab(index).url)
visibility("激活前")

print()
print("== activate_tab 只切换前台标签,不改变 page 绑定的标签 ==")
before_tab_id = page.tab_id
page.activate_tab(tab_b.tab_id)
print("activate_tab(tab_b.tab_id) 后 page.url:", page.url)
print(" page.tab_id 是否仍是绑定时的标签:", page.tab_id == before_tab_id)
print(" page.tab_id 是否等于 tab_b.tab_id:", page.tab_id == tab_b.tab_id)
visibility("激活 B 后")
page.activate_tab(tab_c)
print("activate_tab(tab_c 对象) 后 page.url:", page.url)
print(" page.tab_id 仍等于最初绑定的标签:", page.tab_id == before_tab_id)
visibility("激活 C 后")
get(page, URL_A)
print(" page.get() 作用于绑定标签: page.url =", page.url, "| tab_c.url =", tab_c.url)
print(" 未激活的 tab_b.url 保持不变 =", tab_b.url)

print()
print("== 从浏览器对象拿标签页(page.browser) ==")
browser = page.browser
print("type(page.browser):", type(browser).__name__, "| MRO:",
" -> ".join(c.__name__ for c in type(browser).__mro__))
print("browser.tab_ids 数量:", len(browser.tab_ids))
show("browser.get_tab(url='quotes')", lambda: browser.get_tab(url="quotes"))
tb_url = show("page.get_tab(url='quotes').url", lambda: page.get_tab(url="quotes").url)
if tb_url is not None:
tb = page.get_tab(url="quotes")
print("page.get_tab(url='quotes') 类型:", type(tb).__name__, "| url:", tb.url)
get(tb, URL_A)
print("tb.get(spa1) 后 tb.url:", tb.url, "| 主 page.url:", page.url)
else:
print("标签查找失败:跳过 tb 导航,继续其他实验。")
print("page.get_tabs(url='spa1') 数量:", len(page.get_tabs(url="spa1")))

print()
print("== 新窗口 / 后台 / 独立上下文 ==")
t_win = page.new_tab(URL_B, new_window=True)
print("new_window=True 返回类型:", type(t_win).__name__, "| url:", t_win.url)
print("标签对象有 window_id 吗:", hasattr(t_win, "window_id"))
t_bg = page.new_tab(URL_B, background=True)
print("background=True 后 page.url 仍是:", page.url)
t_ctx = page.new_tab(URL_B, new_context=True)
print("new_context=True 返回类型:", type(t_ctx).__name__, "| url:", t_ctx.url)
print("当前标签总数:", page.tabs_count, "| 标题(去重排序):",
sorted({t.title for t in page.get_tabs()}))

print()
print("== latest_tab 的语义 ==")
t_new = page.new_tab(URL_B)
print(" 新建后 latest_tab.url:", page.latest_tab.url)
print(" 新建后 page.url(绑定标签,不称为前台):", page.url)
print(" latest_tab 就是刚建的那个吗:", page.latest_tab.tab_id == t_new.tab_id)
page.activate_tab(tab_b)
print(" 激活 tab_b 之后 latest_tab.url:", page.latest_tab.url)
print(" latest_tab.tab_id == tab_ids[-1]:",
page.latest_tab.tab_id == page.tab_ids[-1])

print()
print("== close_tabs:传 ChromiumTab 对象会撞上 MixTab ==")
print("关闭前 tabs_count:", page.tabs_count)
show("close_tabs(t_new) 直接传对象", lambda: page.close_tabs(t_new))
show("close_tabs(t_new.tab_id) 传 id", lambda: page.close_tabs(t_new.tab_id))
t_new.close()
print("改用 t_new.close() 后 tabs_count:", page.tabs_count)
for t in (t_win, t_bg, t_ctx):
t.close()
print("逐个 close() 后 tabs_count:", page.tabs_count)
page.new_tab(URL_A)
page.new_tab(URL_A)
print("补两个标签后 tabs_count:", page.tabs_count)
for t in page.get_tabs():
if t.tab_id != page.tab_id:
t.close()
print("逐个 close 掉非绑定标签后 tabs_count:", page.tabs_count)
show("close_tabs([], others=True)", lambda: page.close_tabs([], others=True))
print("(上面这行会把所有标签都关掉,等价于关掉浏览器)")
except Exception as exc:
exit_code = 1
print("块执行失败:", type(exc).__name__, " | ".join(str(exc).splitlines()))
finally:
if page is not None:
try:
page.quit(del_data=True)
print("page.quit() 执行完毕,清理本次用户数据。")
except Exception as exc:
exit_code = 1
print("page.quit():", type(exc).__name__,
"|", " | ".join(l.strip() for l in str(exc).strip().splitlines()))
sys.exit(exit_code)
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
站点 A=spa1,B=quotes/js,C=quotes 根;不用 ssr1 做标题靶站。
== 两种标签类型 ==
ChromiumTab MRO: ChromiumTab -> ChromiumBase -> BasePage -> BaseParser
MixTab MRO: MixTab -> SessionPage -> ChromiumTab -> ChromiumBase -> BasePage
MixTab 比 ChromiumTab 多: ['change_mode', 'cookies_to_browser', 'cookies_to_session', 'encoding', 'mode', 'post', 'raw_data', 'response']
导航 get: True | url: https://spa1.scrape.center/ | title: Scrape | Movie
起始: tabs_count = 1 | page.url = https://spa1.scrape.center/
type(page.latest_tab) = MixTab | latest_tab.url = https://spa1.scrape.center/
== 新开两个标签 ==
new_tab() 返回类型: ChromiumTab
tab_b.url: https://quotes.toscrape.com/js/
tab_b.title: Quotes to Scrape
tab_b.tab_id 长度: 32
tab_b.browser 类型: Chromium
tab_b 是否有 quit/close: False True
tab_b 是否有 change_mode(切换 d/s 模式): False
第二个新标签 tab_c.url: https://quotes.toscrape.com/ | title: Quotes to Scrape
tabs_count = 3 | len(tab_ids) = 3
page.url 现在指向: https://spa1.scrape.center/
page.title 现在指向: Scrape | Movie
page.latest_tab.url: https://quotes.toscrape.com/
== get_tab / get_tabs ==
get_tabs() 数量: 3
get_tabs() 类型: ['ChromiumTab', 'MixTab']
get_tabs() 标题: ['Quotes to Scrape', 'Quotes to Scrape', 'Scrape | Movie']
get_tabs() 每项类型: ['ChromiumTab', 'ChromiumTab', 'MixTab']
get_tab(0).url -> 'https://quotes.toscrape.com/'
get_tab(1).url -> 'https://quotes.toscrape.com/'
get_tab(-1).url -> 'https://spa1.scrape.center/'
get_tab(title='Scrape | Movie').url -> 'https://spa1.scrape.center/'
get_tab(url='quotes').url -> 'https://quotes.toscrape.com/'
get_tab(9) IndexError: list index out of range
get_tab(title='zzz') RuntimeError: 没有找到指定标签页。 | 参数: {'id_or_num': None, 'title': 'zzz', 'url': None, 'tab_type': 'page'} | 版本: 4.1.1.4
get_tabs(url='scrape.center') 数量: 1
get_tabs(url='quotes') 数量: 2
get_tab(tab_c.tab_id) is tab_c -> True
get_tab(1) 两次是否同一对象 -> True
== 各序号对应的标签(保留实际 tab_ids 顺序) ==
按创建顺序的 url: ['https://spa1.scrape.center/', 'https://quotes.toscrape.com/js/', 'https://quotes.toscrape.com/']
get_tab(0).url -> 'https://quotes.toscrape.com/'
get_tab(1).url -> 'https://quotes.toscrape.com/'
get_tab(2).url -> 'https://quotes.toscrape.com/js/'
激活前 | B hidden: False | C hidden: False
tab_ids 实际顺序(不排序): ['https://quotes.toscrape.com/', 'https://quotes.toscrape.com/js/', 'https://spa1.scrape.center/']
== activate_tab 只切换前台标签,不改变 page 绑定的标签 ==
activate_tab(tab_b.tab_id) 后 page.url: https://spa1.scrape.center/
page.tab_id 是否仍是绑定时的标签: True
page.tab_id 是否等于 tab_b.tab_id: False
激活 B 后 | B hidden: False | C hidden: False
tab_ids 实际顺序(不排序): ['https://quotes.toscrape.com/js/', 'https://quotes.toscrape.com/', 'https://spa1.scrape.center/']
activate_tab(tab_c 对象) 后 page.url: https://spa1.scrape.center/
page.tab_id 仍等于最初绑定的标签: True
激活 C 后 | B hidden: False | C hidden: False
tab_ids 实际顺序(不排序): ['https://quotes.toscrape.com/', 'https://quotes.toscrape.com/js/', 'https://spa1.scrape.center/']
导航 get: True | url: https://spa1.scrape.center/ | title: Scrape | Movie
page.get() 作用于绑定标签: page.url = https://spa1.scrape.center/ | tab_c.url = https://quotes.toscrape.com/
未激活的 tab_b.url 保持不变 = https://quotes.toscrape.com/js/
== 从浏览器对象拿标签页(page.browser) ==
type(page.browser): Chromium | MRO: Chromium -> object
browser.tab_ids 数量: 3
browser.get_tab(url='quotes') RuntimeError: 该标签页已有非MixTab版本,如需多对象共用标签页请设置Settings.set_singleton_tab_obj(False)。 | 版本: 4.1.1.4
page.get_tab(url='quotes').url -> 'https://quotes.toscrape.com/'
page.get_tab(url='quotes') 类型: ChromiumTab | url: https://quotes.toscrape.com/
导航 get: True | url: https://spa1.scrape.center/ | title: Scrape | Movie
tb.get(spa1) 后 tb.url: https://spa1.scrape.center/ | 主 page.url: https://spa1.scrape.center/
page.get_tabs(url='spa1') 数量: 2
== 新窗口 / 后台 / 独立上下文 ==
new_window=True 返回类型: ChromiumTab | url: https://quotes.toscrape.com/js/
标签对象有 window_id 吗: False
background=True 后 page.url 仍是: https://spa1.scrape.center/
new_context=True 返回类型: ChromiumTab | url: https://quotes.toscrape.com/js/
当前标签总数: 6 | 标题(去重排序): ['Quotes to Scrape', 'Scrape | Movie']
== latest_tab 的语义 ==
新建后 latest_tab.url: https://quotes.toscrape.com/js/
新建后 page.url(绑定标签,不称为前台): https://spa1.scrape.center/
latest_tab 就是刚建的那个吗: True
激活 tab_b 之后 latest_tab.url: https://quotes.toscrape.com/js/
latest_tab.tab_id == tab_ids[-1]: False
== close_tabs:传 ChromiumTab 对象会撞上 MixTab ==
关闭前 tabs_count: 7
close_tabs(t_new) 直接传对象 RuntimeError: 该标签页已有非MixTab版本,如需多对象共用标签页请设置Settings.set_singleton_tab_obj(False)。 | 版本: 4.1.1.4
close_tabs(t_new.tab_id) 传 id RuntimeError: 该标签页已有非MixTab版本,如需多对象共用标签页请设置Settings.set_singleton_tab_obj(False)。 | 版本: 4.1.1.4
改用 t_new.close() 后 tabs_count: 6
逐个 close() 后 tabs_count: 3
补两个标签后 tabs_count: 5
逐个 close 掉非绑定标签后 tabs_count: 1
close_tabs([], others=True) -> None
(上面这行会把所有标签都关掉,等价于关掉浏览器)
page.quit() 执行完毕,清理本次用户数据。
  • MixTab 继承 SessionPage 和 ChromiumTab,比 ChromiumTab 多出 change_mode、mode、post、raw_data、response、cookies_to_browser、cookies_to_session、encoding 这 8 个公开名称。操作前要分清取得的标签类型,不因一次 latest_tab 的类型就推广到所有获取方式。
  • ChromiumPage.new_tab() 返回 ChromiumTab,它有 close(),本版本未提供 quit();标签由 tab_id 标识。不要把这个返回类型推广到所有对象的同名方法:本版 Chromium.new_tab() 默认请求的是 MixTab。
  • get_tab(0) 与 get_tab(1) 都取当前 tab_ids 的首项,正数从 1 起,负数从末尾倒数。源码用 id_or_num - 1 if id_or_num > 0 else id_or_num 索引列表。索引指向当前列表位置,不是创建时的固定编号,也不绑定某个固定 URL。
  • tab_id 和对象引用适合保留对指定标签的引用;url=/title= 可用于过滤。多个标签可能有相同标题或 URL,不能把过滤条件当作唯一身份标识;导航失败导致标题不匹配,也不等于标签 API 出错。
  • 标签数量不足时 get_tab(9) 会抛 IndexError;没有匹配标题时,get_tab(title='zzz') 抛 RuntimeError: 没有找到指定标签页。按位置查找与按标题过滤是两种不同的查找方式。
  • 启用单例的示例中,page.get_tab(1) 两次 is 比较为 True,按 tab_c.tab_id 取回的对象也与 tab_c 相同。这说明同一标签对象可以复用,不是每次获取都会新建对象。
  • activate_tab() 只激活标签,不改变 page 绑定的标签。比较激活前后的 page.tab_id 是否相等,可以确认绑定不变;之后 page.get() 仍作用于绑定标签。要导航另一个标签,应取得对应的 tab 对象再调用 tab.get(...)。
  • MixTab 冲突的方向是:先由 page.new_tab() 建立 ChromiumTab 对象,再由浏览器对象请求同标签的 MixTab,于是出现「已有非MixTab版本」错误。示例的 page.close_tabs(t_new) 和传其 ID 的调用也遇到该错误,改用 t_new.close() 可以关闭。错误提示提到的 Settings.set_singleton_tab_obj(False) 不属于这里已验证的解决步骤。
  • new_window=True、background=True、new_context=True 分别用于请求新窗口、后台标签和新上下文。标签对象未提供 window_id 不等于库没有窗口概念;绑定的 page.url 不变也不能证明前台焦点。判断可见性可结合页面状态,判断上下文隔离则应比较实际存储数据。
  • latest_tab 不是前台标签查询接口。本版取当前 tab_ids[0];完整列表的排序不能仅靠几次 URL 回显推断。多个标签可访问同一 URL,要确认是否为同一标签,应比较 tab_id。
  • page.url 始终描述绑定标签,不代表前台。需要继续操作已取得的标签时,使用对应对象或 ID,不把 latest_tab 当成前台状态。
  • 只关闭非绑定标签,可遍历标签并比较 tab_id,对其他标签调用 close()。close_tabs([], others=True) 不保留任何标签,不是「保留当前,只关多余标签」。关闭调用返回 None 也不是浏览器进程已退出的证明,应根据需要确认关闭后的状态。

网络监听抓接口

page.listen 可以获取浏览器发出的请求和收到的响应。监听时仍启动浏览器、页面仍会渲染;直接读接口数据省掉的是 DOM 解析,不是浏览器渲染。下面用 spa1 演示。

09_listen_packets.py
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
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
# -*- coding: utf-8 -*-
"""网络监听 listen:抓 SPA 的接口数据包,并与渲染取数对照。目标站 spa1 / ssr1 。"""

import sys

from DrissionPage import ChromiumOptions, ChromiumPage

SPA = "https://spa1.scrape.center/page/2"
SSR = "https://ssr1.scrape.center/"

page = None
first = None
failures = 0


def fail(message):
global failures
failures += 1
print("失败:", message)


def movie_body(packet, label):
if not packet:
fail(label + ":20 秒内未捕获 api/movie 数据包")
return None
print(label, "| url:", packet.url, "| is_failed:", packet.is_failed)
if packet.is_failed:
fail(label + ":网络请求失败,不能使用 response.body")
return None
response = packet.response
body = response.body
print(" HTTP status:", response.status, "| body 类型:", type(body).__name__)
if not isinstance(response.status, int) or not 200 <= response.status < 300:
fail(label + ":接口 HTTP 状态异常")
return None
if not isinstance(body, dict):
fail(label + ":response.body 不是 dict")
return None
results = body.get("results")
if not isinstance(results, list) or not results:
fail(label + ":results 必须是非空 list,实际为 %r" % results)
return None
fields = {"name", "score", "minute", "regions"}
if not isinstance(results[0], dict) or not fields.issubset(results[0]):
fail(label + ":results[0] 缺少预期字段")
return None
print(" body 顶层键:", sorted(body))
print(" count 字段存在:", "count" in body,
"| count 类型:", type(body.get("count")).__name__)
print(" len(body['results']):", len(results))
return body


def show(expr, fn):
try:
print("%-48s -> %r" % (expr, fn()))
except Exception as exc:
print("%-48s %s: %s" % (
expr, type(exc).__name__,
" | ".join(l.strip() for l in str(exc).strip().splitlines())))


try:
page = ChromiumPage(ChromiumOptions().headless(True).auto_port())
print("== Listener 对象公开成员 ==")
lis = page.listen
print("type:", type(lis).__name__)
print(sorted(n for n in dir(lis) if not n.startswith("_")))
show("未 start 前 wait(timeout=1)", lambda: page.listen.wait(timeout=1))
print("listen.once 是否存在:", hasattr(page.listen, "once"))
print("listen.targets 初始:", page.listen.targets)

print()
print("== start('api/movie') 后访问 SPA ==")
ret = page.listen.start("api/movie")
print("start() 返回值:", ret, "| listen.targets:", page.listen.targets)
print("listening:", page.listen.listening)
ok = page.get(SPA, retry=0, timeout=30)
print("首轮 get:", ok, "| url:", page.url, "| title:", page.title)
item = page.listen.wait(timeout=20, raise_err=False)
print("item 类型:", type(item).__name__)
body = movie_body(item, "首轮抓包")
if not ok:
fail("首轮导航失败,不使用接口首条结果")
elif body is not None:
first = body["results"][0]
print("item.target:", item.target)
print("item.method:", item.method)
print("item.resourceType:", item.resourceType)
print("type(item.request):", type(item.request).__name__,
"| type(item.response):", type(item.response).__name__)
print("item.request.params:", item.request.params)
print("item.request.postData:", item.request.postData)
print("item.request.headers['accept']:", item.request.headers.get("accept"))
print("item.response.statusText:", item.response.statusText)
print("item.response.status_text(下划线写法):", item.response.status_text)
print("item.response.mimeType:", item.response.mimeType)
print("item.response.headers['content-type']:",
item.response.headers.get("content-type"))
print("results[0] 字段:", sorted(first))
print("results[0]['name']:", first["name"])
show("item.response.json 属性实测", lambda: item.response.json)
print("item.response.raw_body 类型:", type(item.response.raw_body).__name__)
show("item.frames()", lambda: item.frames())
else:
print("首轮接口校验失败,跳过 first 字段展示。")

print()
print("== 再次 start() 是替换目标,不是追加 ==")
page.listen.start("api/movie")
print("第一次 start 后 targets:", sorted(page.listen.targets))
page.listen.start("static")
print("第二次 start 后 targets:", sorted(page.listen.targets))
page.listen.set_targets(["api", "scrape.center"])
print("set_targets(列表) 后 targets:", sorted(page.listen.targets))
page.listen.stop()

print()
print("== 一次访问抓多个包(query 不保证子资源绕过缓存) ==")
page.listen.start("scrape.center", method="GET")
ok = page.get("https://spa1.scrape.center/page/2?p=2", retry=0, timeout=30)
print("多包导航 get:", ok)
if not ok:
fail("多包实验导航失败")
packets = page.listen.wait(count=2, timeout=25, fit_count=False, raise_err=False)
print("wait(count=2, fit_count=False) 类型:", type(packets).__name__)
if isinstance(packets, list):
print("条数:", len(packets))
for packet in packets:
print(" resourceType:", packet.resourceType, "| url:", packet.url,
"| is_failed:", packet.is_failed,
"| status:", None if packet.is_failed else packet.response.status)
if len(packets) < 2:
fail("多包等待超时,只返回部分数据包")
else:
fail("多包等待没有收到数据,实际返回 %r" % packets)
page.listen.stop()

print()
print("== steps() 迭代 ==")
page.listen.start("api/movie", method="GET")
print("steps 导航 get:", page.get(SPA, retry=0, timeout=30))
step_packet = None
for packet in page.listen.steps(count=1, timeout=20):
step_packet = packet
print("steps() 拿到:", repr(packet))
break
movie_body(step_packet, "steps 抓包")
page.listen.stop()

print()
print("== 正则目标 / method 过滤 / pause-resume-clear ==")
page.listen.start(r"api/\w+", is_regex=True, method="GET")
print("is_regex=True 时 targets:", page.listen.targets)
print("正则导航 get:", page.get(SPA, retry=0, timeout=30))
it2 = page.listen.wait(timeout=20, raise_err=False)
movie_body(it2, "正则抓包")
page.listen.pause()
print("pause() 后 listening:", page.listen.listening)
page.listen.resume()
print("resume() 后 listening:", page.listen.listening)
page.listen.clear()
print("clear() 后公开 cached_count 属性存在:", hasattr(page.listen, "cached_count"))
page.listen.stop()
page.listen.start("api/movie", method="POST")
print("method='POST' 时 targets:", page.listen.targets)
ok = page.get(SPA, retry=0, timeout=30)
loaded = page.wait.eles_loaded("css:div.item", timeout=20, raise_err=False) if ok else False
print("POST 过滤实验 get:", ok, "| DOM 就绪:", loaded)
print("POST 过滤后 wait(timeout=3):", page.listen.wait(timeout=3, raise_err=False))
if not loaded:
fail("页面未渲染,POST 无包不能单独证明过滤效果")
page.listen.stop()

print()
print("== ssr1:文档失败时不能以无包证明没有接口 ==")
# method 不传会沿用 POST,故显式重置为 GET,同时捕获文档状态。
page.listen.start(True, method="GET")
ok = page.get(SSR, retry=0, timeout=30)
packets = page.listen.wait(count=20, timeout=8, fit_count=False, raise_err=False)
packets = packets if isinstance(packets, list) else []
statuses = [None if p.is_failed else p.response.status
for p in packets if p.resourceType == "Document"]
print("ssr1 get:", ok, "| 文档 HTTP 状态:", statuses,
"| api/movie 包数:", sum("api/movie" in p.url for p in packets))
if not ok or not statuses or not any(s and 200 <= s < 300 for s in statuses):
print("跳过 ssr1 接口边界结论:文档请求未成功。")
else:
print("以上只代表本次有限等待窗口的观察,不证明站点永远没有接口。")
page.listen.stop()

print()
print("== 监听 vs 渲染:同次 spa1 导航 ==")
first = None # 不使用先前导航的过期数据。
page.listen.start("api/movie", method="GET")
ok = page.get(SPA, retry=0, timeout=30)
print("对照 get:", ok, "| url:", page.url, "| title:", page.title)
current = page.listen.wait(timeout=20, raise_err=False)
body = movie_body(current, "对照抓包")
if ok and body is not None:
first = body["results"][0]
loaded = page.wait.eles_loaded("css:div.item", timeout=20, raise_err=False) if ok else False
print("wait.eles_loaded('css:div.item', timeout=20):", loaded)
cards = page.eles("css:div.item", timeout=0) if loaded else []
headings = page.eles("t:h2", timeout=0) if loaded else []
print("渲染路线卡片数量:", len(cards), "| h2 数量:", len(headings))
if not ok or not loaded or not cards or not headings:
fail("DOM 对照失败:get=%r, loaded=%r, cards=%d, h2=%d, url=%s, title=%s"
% (ok, loaded, len(cards), len(headings), page.url, page.title))
elif first is None:
print("DOM 非空但接口校验失败:跳过接口字段及一致性比较。")
else:
text = " ".join(cards[0].text.split())
print("渲染路线第一条卡片文本:", text[:80])
print("接口路线第一条的同一批字段:", sorted(first))
print("接口里 score=%r、minute=%r、regions=%r"
% (first["score"], first["minute"], first["regions"]))
matched = bool(first["name"]) and first["name"] in text
print("非空首条片名出现在同次 DOM 中:", matched)
if not matched:
fail("接口与 DOM 非空首条片名不匹配")
page.listen.stop()
except Exception as exc:
fail("块执行异常 %s: %s" % (type(exc).__name__, " | ".join(str(exc).splitlines())))
finally:
if page is not None:
try:
page.quit(del_data=True)
print("浏览器已退出并清理本次用户数据。")
except Exception as exc:
fail("浏览器清理失败 %s: %s" % (type(exc).__name__, " | ".join(str(exc).splitlines())))
print("非预期失败次数:", failures)
sys.exit(1 if failures else 0)
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
== Listener 对象公开成员 ==
type: Listener
['clear', 'listening', 'pause', 'resume', 'set_targets', 'start', 'steps', 'stop', 'tab_id', 'targets', 'wait', 'wait_silent']
未 start 前 wait(timeout=1) RuntimeError: 监听未启动或已停止。 | 版本: 4.1.1.4
listen.once 是否存在: False
listen.targets 初始: True
== start('api/movie') 后访问 SPA ==
start() 返回值: None | listen.targets: {'api/movie'}
listening: True
首轮 get: True | url: https://spa1.scrape.center/page/2 | title: Scrape | Movie
item 类型: DataPacket
首轮抓包 | url: https://spa1.scrape.center/api/movie/?limit=10&offset=10 | is_failed: False
HTTP status: 200 | body 类型: dict
body 顶层键: ['count', 'results']
count 字段存在: True | count 类型: int
len(body['results']): 10
item.target: api/movie
item.method: GET
item.resourceType: XHR
type(item.request): Request | type(item.response): Response
item.request.params: {'limit': '10', 'offset': '10'}
item.request.postData: False
item.request.headers['accept']: application/json, text/plain, */*
item.response.statusText: OK
item.response.status_text(下划线写法): None
item.response.mimeType: application/json
item.response.headers['content-type']: application/json
results[0] 字段: ['alias', 'categories', 'cover', 'id', 'minute', 'name', 'published_at', 'regions', 'score']
results[0]['name']: 狮子王
item.response.json 属性实测 -> None
item.response.raw_body 类型: str
item.frames() AttributeError: 'DataPacket' object has no attribute 'frames'
== 再次 start() 是替换目标,不是追加 ==
第一次 start 后 targets: ['api/movie']
第二次 start 后 targets: ['static']
set_targets(列表) 后 targets: ['api', 'scrape.center']
== 一次访问抓多个包(query 不保证子资源绕过缓存) ==
多包导航 get: True
wait(count=2, fit_count=False) 类型: list
条数: 2
resourceType: Stylesheet | url: https://spa1.scrape.center/css/app.ea9d802a.css | is_failed: False | status: 200
resourceType: Script | url: https://spa1.scrape.center/js/app.17b3aaa5.js | is_failed: False | status: 200
== steps() 迭代 ==
steps 导航 get: True
steps() 拿到: <DataPacket target="api/movie" url="https://spa1.scrape.center/api/movie/?limit=10&offset=10">
steps 抓包 | url: https://spa1.scrape.center/api/movie/?limit=10&offset=10 | is_failed: False
HTTP status: 200 | body 类型: dict
body 顶层键: ['count', 'results']
count 字段存在: True | count 类型: int
len(body['results']): 10
== 正则目标 / method 过滤 / pause-resume-clear ==
is_regex=True 时 targets: {'api/\\w+'}
正则导航 get: True
正则抓包 | url: https://spa1.scrape.center/api/movie/?limit=10&offset=10 | is_failed: False
HTTP status: 200 | body 类型: dict
body 顶层键: ['count', 'results']
count 字段存在: True | count 类型: int
len(body['results']): 10
pause() 后 listening: False
resume() 后 listening: True
clear() 后公开 cached_count 属性存在: False
method='POST' 时 targets: {'api/movie'}
POST 过滤实验 get: True | DOM 就绪: True
POST 过滤后 wait(timeout=3): False
== ssr1:文档失败时不能以无包证明没有接口 ==
ssr1 get: True | 文档 HTTP 状态: [502] | api/movie 包数: 0
跳过 ssr1 接口边界结论:文档请求未成功。
== 监听 vs 渲染:同次 spa1 导航 ==
对照 get: True | url: https://spa1.scrape.center/page/2 | title: Scrape | Movie
对照抓包 | url: https://spa1.scrape.center/api/movie/?limit=10&offset=10 | is_failed: False
HTTP status: 200 | body 类型: dict
body 顶层键: ['count', 'results']
count 字段存在: True | count 类型: int
len(body['results']): 10
wait.eles_loaded('css:div.item', timeout=20): True
渲染路线卡片数量: 10 | h2 数量: 10
渲染路线第一条卡片文本: 狮子王 - The Lion King 动画 歌舞 冒险 美国 / 89 分钟 1995-07-15 上映 9.0
接口路线第一条的同一批字段: ['alias', 'categories', 'cover', 'id', 'minute', 'name', 'published_at', 'regions', 'score']
接口里 score=9.0、minute=89、regions=['美国']
非空首条片名出现在同次 DOM 中: True
浏览器已退出并清理本次用户数据。
非预期失败次数: 0
  • 这次 dir() 列出 Listener 的 12 个公开成员:clear、listening、pause、resume、set_targets、start、steps、stop、tab_id、targets、wait、wait_silent。没有 listen.once;只取一个包可用 wait(count=1)。
  • 先 start(目标串) 再导航:page.listen.start("api/movie") 返回 None,targets 变成 {'api/movie'},然后访问页面并 wait()。记录中未启动监听就调用 wait() 抛 RuntimeError。
  • 成功取得的 item 是 DataPacket,可读 target、url、method、resourceType、is_failed。本例为 GET/XHR;item.request.params 中 limit=10、offset=10 对应第二页,值是字符串。请求另有 postData、headers;响应可读 status、statusText、mimeType、headers、body、raw_body。
  • 本次 JSON 响应的 item.response.body 是 dict,确认 results 非空后可取 body["results"][0]["name"];raw_body 是原始字符串。本例访问 item.response.json 得到 None,调用 item.frames() 抛 AttributeError,不要据此把所有响应体都当 dict。
  • 本次 statusText 为 'OK',status_text 为 None,应使用实际字段名 statusText。
  • 连调 start("api/movie")、start("static"),目标从前者变成后者,并非累加;多个目标可一次传给 set_targets(["api", "scrape.center"])。
  • 等到指定数量时,wait(count=2) 返回列表,wait(count=1) 返回单个包。已有输出中的两个包为 CSS 和 JS,类型是 Stylesheet/Script;请求顺序与缓存状态可能影响实际结果。
  • steps() 提供迭代方式,例如代码中的 steps(count=1, timeout=20) 循环。
  • 目标串默认按包含关系匹配;本例 start(r"api/\w+", is_regex=True) 使用正则并命中接口。
  • method='POST' 按请求方法过滤;本例 GET 接口未命中,wait(timeout=3) 返回 False。这不代表等待一定不抛异常:源码中 raise_err=True 或相应全局等待设置可令超时抛 WaitTimeoutError。
  • pause() 后 listening 为 False,resume() 后为 True;clear() 清空已缓存包。当前版本没有公开的 cached_count 属性,存在性检查为 False,不要把它当缓存数量。
  • 同次导航中,接口与 DOM 都先检查非空,再比较首条片名。接口直接提供评分、时长、地区等字段,DOM 路线需要定位和提取。ssr1 的文档请求实测出现 502/503,因而跳过接口边界结论;错误页没有匹配包不能证明正常页面永远没有接口。

浏览器里的 cookie 可以交给 SessionPage 复用。下面的本机靶站会设置 cookie,并按收到的 cookie 返回 COOKIE-HIT 或 COOKIE-MISS,便于观察搬运结果;这只是 cookie 检查示例,不是完整登录流程。

10_cookies_storage.py
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
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
# -*- coding: utf-8 -*-
"""cookie 与 storage:浏览器读/写 cookie、local/session storage,以及把浏览器 cookie 喂给 SessionPage 。

本机靶站 127.0.0.1:8069 : / 会 Set-Cookie ,/check 依据 cookie 返回不同文本。
"""

import socket
import sys
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

from DrissionPage import ChromiumOptions, ChromiumPage, SessionPage

PORT = 8069
HIT = "COOKIE-HIT"
MISS = "COOKIE-MISS"


class CookieHandler(BaseHTTPRequestHandler):
def log_message(self, *args):
pass

def do_GET(self):
if self.path.startswith("/check"):
hit = "dp_probe=browser-set" in self.headers.get("Cookie", "")
body = (HIT if hit else MISS).encode("utf-8")
ctype = "text/plain; charset=utf-8"
else:
body = (b"<html><head><title>cookie-probe</title><meta charset='utf-8'></head>"
b"<body><p id='a'>hello</p></body></html>")
ctype = "text/html; charset=utf-8"
self.send_response(200)
self.send_header("Content-Type", ctype)
self.send_header("Content-Length", str(len(body)))
if self.path.startswith("/") and not self.path.startswith("/check"):
self.send_header("Set-Cookie", "dp_probe=browser-set; Path=/")
self.end_headers()
self.wfile.write(body)


def start_target():
try:
with socket.create_connection(("127.0.0.1", PORT), 0.2):
print("端口 %d 已被占用,脚本终止" % PORT)
sys.exit(1)
except OSError:
pass
srv = ThreadingHTTPServer(("127.0.0.1", PORT), CookieHandler)
threading.Thread(target=srv.serve_forever, daemon=True).start()
return srv


def show(expr, fn):
try:
print("%-46s -> %r" % (expr, fn()))
except Exception as exc:
print("%-46s %s: %s" % (
expr, type(exc).__name__,
" | ".join(l.strip() for l in str(exc).strip().splitlines())))


srv = start_target()
base = "http://127.0.0.1:%d/" % PORT
page = ChromiumPage(ChromiumOptions().headless(True).auto_port())
try:
page.get(base)
print("页面标题:", page.title)

print()
print("== 浏览器侧 cookie ==")
print("type(page.cookies):", type(page.cookies).__name__, "(方法还是属性: 见 dir)")
cks = page.cookies()
print("cookies() 条数:", len(cks))
print("本条 cookie 的键:", sorted(cks[0]) if cks else "-")
print("name=value:", [(c["name"], c["value"]) for c in sorted(cks, key=lambda x: x["name"])])
print("all_info=True 的键:", sorted(page.cookies(all_info=True)[0]))
show("page.set.cookies 签名", lambda: __import__("inspect").signature(page.set.cookies))
show("page.cookies_to_session 是否存在", lambda: hasattr(page, "cookies_to_session"))

print()
print("== 写入 / 删除 cookie ==")
page.set.cookies({"name": "dp_add", "value": "1", "domain": "127.0.0.1", "path": "/"})
print("set.cookies(dict) 之后:", sorted((c["name"], c["value"]) for c in page.cookies()))
page.set.cookies("dp_add2=2")
print("set.cookies('k=v') 之后:", sorted((c["name"], c["value"]) for c in page.cookies()))
show("page.set.cookies(非法值)", lambda: page.set.cookies(123))
page.run_cdp("Network.clearBrowserCookies")
print("run_cdp('Network.clearBrowserCookies') 之后:",
sorted((c["name"], c["value"]) for c in page.cookies()))
page.get(base)
print("重新载入后(服务端再种一次):",
sorted((c["name"], c["value"]) for c in page.cookies()))

print()
print("== 把浏览器 cookie 交给 SessionPage ==")
sess = SessionPage()
r0 = sess.get(base + "check")
print("SessionPage 无 cookie -> get() 返回:", r0,
"| requests 的 status_code:", sess.response.status_code, "| 文本:", sess.html.strip())
sess.set.cookies(page.cookies())
r1 = sess.get(base + "check")
print("SessionPage 带浏览器 cookie -> get() 返回:", r1,
"| requests 的 status_code:", sess.response.status_code, "| 文本:", sess.html.strip())
sess2 = SessionPage()
sess2.set.cookies([{"name": "dp_probe", "value": "browser-set",
"domain": "127.0.0.1", "path": "/"}])
sess2.get(base + "check")
print("手工构造 cookie 也认:", sess2.html.strip())

print()
print("== WebPage 的 cookies_to_session / cookies_to_browser ==")
from DrissionPage import WebPage
web = WebPage(chromium_options=ChromiumOptions().headless(True).auto_port())
show("type(web)", lambda: type(web).__name__)
print(" web.mode:", web.mode)
web.get(base)
print(" 浏览器模式 cookies 数:", len(web.cookies()))
show("web.cookies_to_session()", lambda: web.cookies_to_session())
web.change_mode("s")
print(" 切到 s 模式后 mode:", web.mode, "| url:", web.url)
print(" s 模式 cookies:", sorted((c["name"], c["value"]) for c in web.cookies()))
show("web.cookies_to_browser()", lambda: web.cookies_to_browser())
web.quit()

print()
print("== local_storage / session_storage ==")
print("type(page.local_storage):", type(page.local_storage).__name__)
print("page.local_storage() 初始:", page.local_storage())
page.set.local_storage("dp_key", "dp_val")
print("set 之后读单个:", repr(page.local_storage("dp_key")))
print("读全部:", page.local_storage())
page.set.local_storage("dp_key", False)
print("setter 删除后:", page.local_storage(), "| 读单个:", repr(page.local_storage("dp_key")))
page.set.session_storage("sk", "sv")
print("session_storage 读单个:", repr(page.session_storage("sk")), "| 全部:", page.session_storage())
page.run_js("sessionStorage.clear(); localStorage.clear();")
print("清空后 local:", page.local_storage(), "| session:", page.session_storage())
show("不存在的用法 page.local_storage.remove", lambda: page.local_storage.remove)

sess.close()
sess2.close()
finally:
page.quit()
srv.shutdown()
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
页面标题: cookie-probe
== 浏览器侧 cookie ==
type(page.cookies): method (方法还是属性: 见 dir)
cookies() 条数: 1
本条 cookie 的键: ['domain', 'name', 'value']
name=value: [('dp_probe', 'browser-set')]
all_info=True 的键: ['domain', 'expires', 'httpOnly', 'name', 'path', 'priority', 'secure', 'session', 'size', 'sourcePort', 'sourceScheme', 'value']
page.set.cookies 签名 -> <Signature (cookies)>
page.cookies_to_session 是否存在 -> False
== 写入 / 删除 cookie ==
set.cookies(dict) 之后: [('dp_add', '1'), ('dp_probe', 'browser-set')]
set.cookies('k=v') 之后: [('dp_add', '1'), ('dp_add2', '2'), ('dp_probe', 'browser-set')]
page.set.cookies(非法值) ValueError: cookies参数类型错误。 | 允许类型: Cookie, CookieJar, list, tuple, str, dict | 当前类型: <class 'int'> | 版本: 4.1.1.4
run_cdp('Network.clearBrowserCookies') 之后: []
重新载入后(服务端再种一次): [('dp_probe', 'browser-set')]
== 把浏览器 cookie 交给 SessionPage ==
SessionPage 无 cookie -> get() 返回: True | requests 的 status_code: 200 | 文本: COOKIE-MISS
SessionPage 带浏览器 cookie -> get() 返回: True | requests 的 status_code: 200 | 文本: COOKIE-HIT
手工构造 cookie 也认: COOKIE-HIT
== WebPage 的 cookies_to_session / cookies_to_browser ==
type(web) -> 'WebPage'
web.mode: d
浏览器模式 cookies 数: 1
web.cookies_to_session() -> None
切到 s 模式后 mode: s | url: http://127.0.0.1:8069/
s 模式 cookies: [('dp_probe', 'browser-set')]
web.cookies_to_browser() -> None
== local_storage / session_storage ==
type(page.local_storage): method
page.local_storage() 初始: {}
set 之后读单个: 'dp_val'
读全部: {'dp_key': 'dp_val'}
setter 删除后: {} | 读单个: None
session_storage 读单个: 'sv' | 全部: {'sk': 'sv'}
清空后 local: {} | session: {}
不存在的用法 page.local_storage.remove AttributeError: 'function' object has no attribute 'remove'
  • cookies 是方法,用 page.cookies() 或 sess.cookies() 读取。本次浏览器侧返回的条目默认有 domain、name、value;page.cookies(all_info=True) 还显示过期时间、路径等信息,按需要选择。
  • 浏览器写 cookie 用 page.set.cookies(...),本例试了 dict 和 'k=v' 字符串;SessionPage 也试了字典列表。传整数的错误信息列出允许类型为 Cookie, CookieJar, list, tuple, str, dict,不表示本例逐一验证了所有类型。
  • 本例用 page.run_cdp('Network.clearBrowserCookies') 清空浏览器 cookie,随后读到空列表;重新访问靶站后,服务端又设置了 cookie。这是清空示例,不是删除单条 cookie 的测试。
  • 不给 cookie 时靶站回 COOKIE-MISS,但 sess.get() 仍返回 True:请求成功不等于登录态满足。执行 sess.set.cookies(page.cookies()) 后,同一地址回 COOKIE-HIT;手工构造匹配的 cookie 也得到 COOKIE-HIT。本例的 ChromiumPage 没有 cookies_to_session 成员,采用上述显式搬运即可。
  • WebPage 提供 cookies_to_session() 和 cookies_to_browser() 两个搬运入口。已有输出中,前者调用后切到 s 模式能读到 dp_probe=browser-set,后者返回 None;脚本未另造差异 cookie 核验回传后的浏览器状态。这是两侧 cookie 的复制,不是持续共享同一份 cookie。
  • storage 的读写入口不同:page.local_storage() 读全部,page.local_storage('dp_key') 读单项;写入用 page.set.local_storage('dp_key', 'dp_val')。session_storage 对应使用 page.session_storage(...) 与 page.set.session_storage(key, value)。
  • 删除单项用 page.set.local_storage('dp_key', False):示例实际执行后,page.local_storage() 得到空字典 {},page.local_storage('dp_key') 得到 None。本版两个 storage setter 都以 value is False 作为删除条件;读取缺失项返回 None,不表示写入 None 就是删除。
  • 清空全部 storage 可执行 page.run_js('localStorage.clear(); sessionStorage.clear();'),示例中两者随后均为空字典。page.local_storage.remove 会抛 AttributeError,不要把读取方法当成带 remove 的对象。

SessionPage 与边界

SessionPage 用 requests 发请求,不启动浏览器,也不执行页面 JavaScript。下面对照静态 HTML 与前端渲染页面,再演示请求头、POST、本机代理和 WebPage 模式切换。

11_sessionpage_and_boundary.py
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
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
# -*- coding: utf-8 -*-
"""SessionPage / WebPage :不启动浏览器的路线,代理、请求头、verify ,以及与浏览器路线的能力边界。

本机靶站 127.0.0.1:8052 ;本机代理 127.0.0.1:8053(只回一小段文本,用来证明请求真的走了代理)。
"""

import json
import socket
import sys
import threading
import warnings
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

warnings.filterwarnings("ignore") # verify=False 会打到 stderr,取证不要它

from DrissionPage import ChromiumOptions, ChromiumPage, SessionOptions, SessionPage, WebPage

PORT = 8052
PROXY_PORT = 8053
SSR = "https://ssr1.scrape.center/"
SPA = "https://spa1.scrape.center/page/2"
QUOTES_JS = "https://quotes.toscrape.com/js/"


class EchoHandler(BaseHTTPRequestHandler):
def log_message(self, *args):
pass

def _send(self, body, ctype="application/json; charset=utf-8"):
data = body.encode("utf-8") if isinstance(body, str) else body
self.send_response(200)
self.send_header("Content-Type", ctype)
self.send_header("Content-Length", str(len(data)))
self.end_headers()
self.wfile.write(data)

def do_GET(self):
if self.path.startswith("/header"):
payload = {"method": self.command,
"agent": self.headers.get("User-Agent", ""),
"x_probe": self.headers.get("X-Probe", "<无>"),
"cookie": self.headers.get("Cookie", "<无>")}
self._send(json.dumps(payload, sort_keys=True))
else:
self._send("<html><head><title>echo</title></head><body>"
"<p class='name'>静态第一行</p><p class='name'>静态第二行</p>"
"</body></html>", "text/html; charset=utf-8")

def do_POST(self):
length = int(self.headers.get("Content-Length", 0))
raw = self.rfile.read(length)
self._send(json.dumps({"method": "POST", "body_len": len(raw)}))


class ProxyHandler(BaseHTTPRequestHandler):
def log_message(self, *args):
pass

def do_GET(self):
self._send(b"PROXY-HIT")

def CONNECT(self, *args):
pass

def _send(self, data):
self.send_response(200)
self.send_header("Content-Type", "text/plain; charset=utf-8")
self.send_header("Content-Length", str(len(data)))
self.end_headers()
self.wfile.write(data)


class QuietServer(ThreadingHTTPServer):
def handle_error(self, request, client_address):
pass # 探针流量/TLS 握手打到明文端口时不往 stderr 刷栈


def bind(port, handler, name):
try:
with socket.create_connection(("127.0.0.1", port), 0.2):
print("端口 %d(%s) 已被占用,脚本终止" % (port, name))
sys.exit(1)
except OSError:
pass
srv = QuietServer(("127.0.0.1", port), handler)
threading.Thread(target=srv.serve_forever, daemon=True).start()
return srv


def show(expr, fn):
try:
print("%-48s -> %r" % (expr, fn()))
except Exception as exc:
print("%-48s %s: %s" % (
expr, type(exc).__name__,
" | ".join(l.strip() for l in str(exc).strip().splitlines())))


def session_get(target, url, label, **kwargs):
try:
ok = target.get(url, retry=0, timeout=25, **kwargs)
response = target.response
status = response.status_code if response is not None else None
reason = response.reason if response is not None else "无响应"
print(label, "| get:", ok, "| HTTP:", status, "| reason:", reason, "| url:", url)
return bool(ok and status is not None and 200 <= status < 300)
except Exception as exc:
print(label, "请求失败:", type(exc).__name__, " | ".join(str(exc).splitlines()))
return False


def counts(target):
return {loc: len(target.eles(loc, timeout=0))
for loc in ("t:blockquote", ".quote", "css:div.item")}


def browser_get(target, url, locator):
ok = target.get(url, retry=0, timeout=30)
loaded = target.wait.eles_loaded(locator, timeout=20, raise_err=False) if ok else False
elements = target.eles(locator, timeout=0) if loaded else []
print("浏览器 get:", ok, "| wait:", loaded, "| locator:", locator,
"| 数量:", len(elements), "| url:", target.url, "| title:", target.title)
if not ok or not loaded or not elements:
print("失败:页面/渲染未就绪,不能将空数组作为边界或一致性证明。")
return None
return elements


srv = proxy = sess = browser = web = sp = None
exit_code = 0
try:
srv = bind(PORT, EchoHandler, "靶站")
proxy = bind(PROXY_PORT, ProxyHandler, "代理")
sess = SessionPage()
print("== SessionPage 成员 ==")
print("type:", type(sess).__name__, "| MRO:",
" -> ".join(c.__name__ for c in type(sess).__mro__))
_sm = sorted(n for n in dir(sess) if not n.startswith("_"))
print("公开成员 %d 个:" % len(_sm))
for _i in range(0, len(_sm), 8):
print(" " + " ".join("%-16s" % n for n in _sm[_i:_i + 8]))
show("inspect.signature(SessionPage.get)",
lambda: __import__("inspect").signature(SessionPage.get))
show("inspect.signature(SessionPage.post)",
lambda: __import__("inspect").signature(SessionPage.post))
print("type(sess.session):", type(sess.session).__name__)
print("type(sess.response):", type(sess.response).__name__)

print()
print("== 抓服务端渲染页(ssr1) ==")
if session_get(sess, SSR, "ssr1"):
try:
names = sess.eles(".name")
print("sess.eles('.name') 数量:", len(names))
if names:
print("sess.title:", sess.title)
print("第一个 .name:", names[0].text)
print("type(sess.ele('.name')):", type(names[0]).__name__)
prefix = names[0].text[:4]
print("非空片名前缀存在于 HTML:", bool(prefix) and prefix in sess.html)
show("sess.json(非 JSON 响应)", lambda: sess.json)
else:
print("跳过 ssr1 取数小节:HTTP 成功但 .name 数量为 0。")
except Exception as exc:
print("跳过 ssr1 取数小节:解析失败", type(exc).__name__,
" | ".join(str(exc).splitlines()))
else:
print("跳过 ssr1 取数小节:get/HTTP 检查未通过;元素数量未解析,不当成 0。")
show("sess.status 属性探测", lambda: sess.status)
print("继续本机 POST/代理与 spa1、quotes/js 边界,不以空数组相等证明成功。")

print()
print("== 请求头 / session 对象 ==")
print("SessionPage 有 headers 吗:", hasattr(sess, "headers"))
print("默认 UA 前 24 字:", sess.session.headers.get("User-Agent", "")[:24])
sess.set.headers({"X-Probe": "dp-session", "User-Agent": "DP-Session/1.0"})
sess.get("http://127.0.0.1:%d/header" % PORT)
print("靶站回显:", sess.html)
print("set.headers 后 sess.headers 有吗:", hasattr(sess, "headers"))
print("sess.session.headers['X-Probe']:", sess.session.headers.get("X-Probe"))
show("set.a_header", lambda: sess.set.a_header("X-Probe", "dp-single"))
sess.get("http://127.0.0.1:%d/header" % PORT)
print("set.a_header 之后:", json.loads(sess.html)["x_probe"])
show("clear_headers", lambda: sess.clear_headers())
sess.get("http://127.0.0.1:%d/header" % PORT)
print("clear_headers 之后:", json.loads(sess.html)["x_probe"])

print()
print("== POST / 参数 ==")
r = sess.post("http://127.0.0.1:%d/post" % PORT, data={"a": "1" * 100})
print("post() 返回:", r, "| 靶站收到:", sess.html)

print()
print("== 代理(本机 8053 假代理) ==")
print("直连靶站:", sess.get("http://127.0.0.1:%d/header" % PORT),
"| method:", json.loads(sess.html)["method"])
sess.set.proxies("http://127.0.0.1:%d" % PROXY_PORT)
got = sess.get("http://example.invalid/anything")
print("set.proxies 后 get() 返回:", got, "| 响应体:", sess.html,
"| sess.url:", sess.url)
show("proxies 传 int", lambda: sess.set.proxies(12345))
print(" 传 int 后 proxies 值:", sess.session.proxies if hasattr(sess.session, "proxies") else "-")
print("浏览器侧代理参数:",
[a for a in ChromiumOptions().set_proxy("http://127.0.0.1:%d" % PROXY_PORT)
.arguments if "proxy" in a])

print()
print("== verify 参数 ==")
show("set.verify 签名", lambda: __import__("inspect").signature(sess.set.verify))
print("SessionPage 有 verify 属性吗:", hasattr(sess, "verify"))
sess.set.verify(False)
print("set.verify(False) 后 requests 侧:", sess.session.verify)
sess.set.verify(True)
print("set.verify(True) 后 requests 侧:", sess.session.verify)
# 错误代理值只是独立 API 探测,不得污染 verify 与在线边界实验。
sess.set.proxies(None, None)
print("重置代理后的 requests.proxies:", sess.session.proxies)
session_get(sess, SPA, "get(verify=123) 实测", verify=123)
session_get(sess, SPA, "get(verify=False) 实测", verify=False)
print("未搭建自签证书靶站;只记录参数设置和本次请求结果,不推断验证机制。")

print()
print("== 能力边界:成功响应与非空渲染才构成证据 ==")
browser = ChromiumPage(ChromiumOptions().headless(True).auto_port())
for name, url, locator in (("spa1", SPA, "css:div.item"),
("quotes/js", QUOTES_JS, ".quote")):
try:
session_counts = None
if session_get(sess, url, name + " SessionPage"):
session_counts = counts(sess)
print(name, "SessionPage 元素数量:", session_counts)
else:
print(name, "SessionPage 请求失败,不能宣称静态页没有元素。")
exit_code = 1
elements = browser_get(browser, url, locator)
if elements:
rendered_counts = counts(browser)
print(name, "ChromiumPage 元素数量:", rendered_counts)
if session_counts is not None:
print(name, "成功响应后的静态/渲染目标数量:",
session_counts[locator], "/", rendered_counts[locator])
else:
exit_code = 1
except Exception as exc:
exit_code = 1
print(name, "边界实验失败:", type(exc).__name__, " | ".join(str(exc).splitlines()))

api = "https://spa1.scrape.center/api/movie/?limit=10&offset=10"
if session_get(sess, api, "spa1 API"):
try:
data = sess.json
results = data.get("results") if isinstance(data, dict) else None
print("接口 JSON 类型:", type(data).__name__)
if isinstance(data, dict) and isinstance(results, list) and results:
print("count 类型:", type(data.get("count")).__name__, "| results 数量:", len(results))
else:
exit_code = 1
print("失败:接口不是 dict 或 results 不是非空 list。")
except Exception as exc:
exit_code = 1
print("接口解析失败:", type(exc).__name__, " | ".join(str(exc).splitlines()))
else:
exit_code = 1
print("接口请求失败:不访问 JSON 字段。")

print("\n== WebPage(d/s 双模式) ==")
try:
web = WebPage(chromium_options=ChromiumOptions().headless(True).auto_port())
print("初始 mode:", web.mode, "| type:", type(web).__name__)
d_elements = browser_get(web, SPA, "css:div.item")
if d_elements:
show("d 模式下 web.s_ele('t:body')", lambda: type(web.s_ele("t:body")).__name__)
else:
exit_code = 1
# go=False 避免隐式请求失败时中断;随后显式 get 并校验 HTTP 状态。
web.change_mode("s", go=False)
print("切 s 模式后 mode:", web.mode, "| type(web):", type(web).__name__)
if session_get(web, SPA, "WebPage s/spa1"):
print("s 模式电影卡片数量:", len(web.eles("css:div.item")))
show("s 模式下 web.s_ele('.name')", lambda: type(web.s_ele(".name")).__name__)
show("s 模式下 web.ele('.name')", lambda: type(web.ele(".name")).__name__)
show("s 模式下 web.s_eles('.name')", lambda: type(web.s_eles(".name")).__name__)
else:
exit_code = 1
print("s 模式请求失败,跳过元素探测。")
print("SessionPage.s_ele 签名:", __import__("inspect").signature(SessionPage.s_ele))
print("SessionPage.s_eles 签名:", __import__("inspect").signature(SessionPage.s_eles))

# SSR 在线失败不阻断模式切换;用本机静态页做非空的同内容对照。
local = "http://127.0.0.1:%d/" % PORT
s_names = None
if session_get(web, local, "WebPage s/本机静态页"):
s_names = [e.text for e in web.eles(".name")]
print("s 模式本机 .name 数量:", len(s_names))
web.change_mode("d", go=False)
print("切回 d 模式:", web.mode)
d_names = browser_get(web, local, ".name")
if s_names and d_names:
same = s_names == [e.text for e in d_names]
print("两侧非空后比较本机 .name 文本:", same)
if not same:
exit_code = 1
else:
exit_code = 1
print("失败:模式对照一侧为空或请求失败,不执行空数组相等比较。")
print("web.user_agent 前 20 字:", web.user_agent[:20])
except Exception as exc:
exit_code = 1
print("WebPage 实验失败:", type(exc).__name__, " | ".join(str(exc).splitlines()))

print()
print("== SessionOptions ==")
so = SessionOptions(read_file=False)
so.set_headers({"X-Probe": "from-options"})
so.set_timeout(7)
so.set_retry(times=2, interval=0.5)
sp = SessionPage(so)
print("用 SessionOptions 建页面: headers:", sp.session.headers.get("X-Probe"),
"| timeout:", sp.timeout, "| retry_times:", sp.retry_times)
sp.get("http://127.0.0.1:%d/header" % PORT)
print(" 靶站看到的 X-Probe:", json.loads(sp.html)["x_probe"])
except Exception as exc:
exit_code = 1
print("块执行失败:", type(exc).__name__, " | ".join(str(exc).splitlines()))
finally:
for label, target in (("WebPage", web), ("ChromiumPage", browser)):
if target is not None:
try:
target.quit(del_data=True)
print(label, "浏览器已退出并清理本次用户数据。")
except Exception as exc:
exit_code = 1
print(label, "清理失败:", type(exc).__name__, " | ".join(str(exc).splitlines()))
for target in (sess, sp):
if target is not None:
target.session.close()
for server in (srv, proxy):
if server is not None:
server.shutdown()
server.server_close()
sys.exit(exit_code)
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
== SessionPage 成员 ==
type: SessionPage | MRO: SessionPage -> BasePage -> BaseParser -> object
公开成员 25 个:
close cookies download download_path ele eles encoding find
get html json post raw_data response retry_interval retry_times
s_ele s_eles session set timeout title url url_available
user_agent
inspect.signature(SessionPage.get) -> <Signature (self, url, show_errmsg=False, retry=None, interval=None, timeout=None, **kwargs)>
inspect.signature(SessionPage.post) -> <Signature (self, url, show_errmsg=False, retry=None, interval=None, **kwargs)>
type(sess.session): Session
type(sess.response): NoneType
== 抓服务端渲染页(ssr1) ==
ssr1 | get: False | HTTP: 502 | reason: Bad Gateway | url: https://ssr1.scrape.center/
跳过 ssr1 取数小节:get/HTTP 检查未通过;元素数量未解析,不当成 0。
sess.status 属性探测 AttributeError: 'SessionPage' object has no attribute 'status'
继续本机 POST/代理与 spa1、quotes/js 边界,不以空数组相等证明成功。
== 请求头 / session 对象 ==
SessionPage 有 headers 吗: False
默认 UA 前 24 字: python-requests/2.34.2
靶站回显: {"agent": "DP-Session/1.0", "cookie": "<\u65e0>", "method": "GET", "x_probe": "dp-session"}
set.headers 后 sess.headers 有吗: False
sess.session.headers['X-Probe']: None
set.a_header AttributeError: 'SessionPageSetter' object has no attribute 'a_header'
set.a_header 之后: dp-session
clear_headers AttributeError: 'SessionPage' object has no attribute 'clear_headers'
clear_headers 之后: dp-session
== POST / 参数 ==
post() 返回: True | 靶站收到: {"method": "POST", "body_len": 102}
== 代理(本机 8053 假代理) ==
直连靶站: True | method: GET
set.proxies 后 get() 返回: True | 响应体: PROXY-HIT | sess.url: http://example.invalid/anything
proxies 传 int -> None
传 int 后 proxies 值: {'http': 12345, 'https': None}
浏览器侧代理参数: ['--proxy-server=http://127.0.0.1:8053']
== verify 参数 ==
set.verify 签名 -> <Signature (on_off)>
SessionPage 有 verify 属性吗: False
set.verify(False) 后 requests 侧: False
set.verify(True) 后 requests 侧: True
重置代理后的 requests.proxies: {'http': None, 'https': None}
get(verify=123) 实测 | get: False | HTTP: None | reason: 无响应 | url: https://spa1.scrape.center/page/2
get(verify=False) 实测 | get: True | HTTP: 200 | reason: OK | url: https://spa1.scrape.center/page/2
未搭建自签证书靶站;只记录参数设置和本次请求结果,不推断验证机制。
== 能力边界:成功响应与非空渲染才构成证据 ==
spa1 SessionPage | get: True | HTTP: 200 | reason: OK | url: https://spa1.scrape.center/page/2
spa1 SessionPage 元素数量: {'t:blockquote': 0, '.quote': 0, 'css:div.item': 0}
浏览器 get: True | wait: True | locator: css:div.item | 数量: 10 | url: https://spa1.scrape.center/page/2 | title: Scrape | Movie
spa1 ChromiumPage 元素数量: {'t:blockquote': 0, '.quote': 0, 'css:div.item': 10}
spa1 成功响应后的静态/渲染目标数量: 0 / 10
quotes/js SessionPage | get: True | HTTP: 200 | reason: OK | url: https://quotes.toscrape.com/js/
quotes/js SessionPage 元素数量: {'t:blockquote': 0, '.quote': 0, 'css:div.item': 0}
浏览器 get: True | wait: True | locator: .quote | 数量: 10 | url: https://quotes.toscrape.com/js/ | title: Quotes to Scrape
quotes/js ChromiumPage 元素数量: {'t:blockquote': 0, '.quote': 10, 'css:div.item': 0}
quotes/js 成功响应后的静态/渲染目标数量: 0 / 10
spa1 API | get: True | HTTP: 200 | reason: OK | url: https://spa1.scrape.center/api/movie/?limit=10&offset=10
接口 JSON 类型: dict
count 类型: int | results 数量: 10
== WebPage(d/s 双模式) ==
初始 mode: d | type: WebPage
浏览器 get: True | wait: True | locator: css:div.item | 数量: 10 | url: https://spa1.scrape.center/page/2 | title: Scrape | Movie
d 模式下 web.s_ele('t:body') -> 'SessionElement'
切 s 模式后 mode: s | type(web): WebPage
WebPage s/spa1 | get: True | HTTP: 200 | reason: OK | url: https://spa1.scrape.center/page/2
s 模式电影卡片数量: 0
s 模式下 web.s_ele('.name') TypeError: SessionPage.s_ele() got an unexpected keyword argument 'timeout'
s 模式下 web.ele('.name') -> 'NoneElement'
s 模式下 web.s_eles('.name') TypeError: SessionPage.s_eles() got an unexpected keyword argument 'timeout'
SessionPage.s_ele 签名: (self, locator=None, index=1)
SessionPage.s_eles 签名: (self, locator)
WebPage s/本机静态页 | get: True | HTTP: 200 | reason: OK | url: http://127.0.0.1:8052/
s 模式本机 .name 数量: 2
切回 d 模式: d
浏览器 get: True | wait: True | locator: .name | 数量: 2 | url: http://127.0.0.1:8052/ | title: echo
两侧非空后比较本机 .name 文本: True
web.user_agent 前 20 字: Mozilla/5.0 (Windows
== SessionOptions ==
用 SessionOptions 建页面: headers: None | timeout: 7 | retry_times: 2
靶站看到的 X-Probe: from-options
WebPage 浏览器已退出并清理本次用户数据。
ChromiumPage 浏览器已退出并清理本次用户数据。
  • 本次实例的 dir() 列出 25 个公开成员:close、cookies、download、download_path、ele、eles、encoding、find、get、html、json、post、raw_data、response、retry_interval、retry_times、s_ele、s_eles、session、set、timeout、title、url、url_available、user_agent。
  • 本例 hasattr(sess, 'headers') 为 False;sess.session 是内部持有的 requests.Session。构造签名为 SessionPage(session_or_options=None, timeout=None)。
  • sess.status 抛 AttributeError,HTTP 状态码要看 sess.response.status_code,原因短语看 reason。初始 sess.response 为 None;请求后也应先确认响应存在,再访问这些字段。
  • get() 返回布尔值;HTTP 状态和业务数据还需单独核对。sess.url、sess.title、sess.html 用属性读取;元素可用 ele() / eles() 定位。ssr1 本轮返回 502/503,示例跳过该页取数,继续其他实验,不把失败页面上的空结果当有效数据。
  • sess.json 按 JSON 解析响应,解析失败可返回 None。本例接口为 dict,仍检查 results 是非空列表;不要从 HTTP 200 直接推断字段齐全。
  • 同一定位串 css:div.item,spa1 第二页的静态源码与浏览器 DOM 分别为 0 和 10;quotes/js 的 .quote 同样为 0 和 10。t:blockquote 在该浏览器页也为 0,因此零结果还可能是选择器不符合页面结构,不能全归因于缺少 JS 执行。
  • 直接请求 spa1 的接口时,已有输出中的 sess.json 是 dict,results 有 10 条。知道可用接口及所需请求条件时,可考虑这条不启动浏览器的路线。
  • sess.set.headers 已生效,但读取底层会话头中的 X-Probe 仍为 None,靶站收到的却是 dp-session。源码中 setter 修改的是 SessionPage 自持的 _headers,发请求时再合并为请求头;它与 requests 底层 Session.headers 分开,不能只读后者判断设置是否生效,也不建议直接操作私有字段。
  • 本例 sess.set.a_header(...) 和 sess.clear_headers() 均抛 AttributeError。已验证的整组配置方式是 sess.set.headers({...}),或先 SessionOptions.set_headers({...}) 再构造页面;后者的靶站回显为 from-options。
  • post() 签名为 (self, url, show_errmsg=False, retry=None, interval=None, **kwargs)。本例实际测试了 data={'a': '1' * 100},靶站收到 102 字节;不能仅因有 **kwargs 就宣称 requests 的所有参数组合均已验证。
  • sess.set.proxies('http://127.0.0.1:8053') 后,本例请求得到假代理的 PROXY-HIT,sess.url 仍为给定地址。set.proxies(12345) 调用未报错,底层代理配置中出现整数,但这不说明该配置有效。浏览器代理部分只检查了 ChromiumOptions.set_proxy() 生成的启动参数,未验证浏览器流量。
  • set.verify(False) / set.verify(True) 改变底层会话设置。示例先清除非法代理,再独立测试请求级 verify=123,本次得到 False 和无响应;verify=False 请求得到 200。这不等于验证了自签证书场景,也不建议为了通网长期关闭证书校验;show_errmsg=True 可让请求错误抛出。
  • 模式切换用 web.change_mode('s') / web.change_mode('d'),不是给 web.mode 赋值。 本例先展示 SPA 在两模式下的区别,再用本机静态页确认两侧均取到 2 个元素且文本一致。go=False 暂不自动访问,随后显式 get() 便于检查结果;copy_cookies=True 控制切换时的复制,不是永久共享。
  • SessionPage.s_ele(locator=None, index=1)、s_eles(locator) 是可用解析入口,但没有 timeout 参数。本版 WebPage 在 s 模式的同名转发仍传入 timeout,即使调用者没写也会出现所示 TypeError;不能因此说 SessionPage.s_ele() 本身不存在。示例分别标明当前模式。

弹窗上传下载与截图

这一组演示浏览器侧的 JavaScript 弹窗、文本输入、文件上传、下载、iframe 和截图。全部使用同一个本机靶站,不把这些交互与单纯发送 HTTP 请求混为一谈。

13_upload_download_dialog_iframe.py
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
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
# -*- coding: utf-8 -*-
"""协议级四件套:弹窗 / 上传 / 下载 / iframe ,外加截图。全部打在 127.0.0.1:8067 的本机靶站上。"""

import hashlib
import os
import shutil
import socket
import sys
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs, urlparse

from DrissionPage import ChromiumOptions, ChromiumPage

PORT = 8067
HERE = os.path.dirname(os.path.abspath(__file__))
TMP = os.path.join(os.path.dirname(HERE), "_tmp")
DL_DIR = os.path.join(TMP, "dl")
UP_MARK = "DP-UPLOAD-MARKER-0123456789"

MAIN = """<!DOCTYPE html>
<html><head><title>proto-probe</title><meta charset="utf-8"></head>
<body>
<h1 id="hd">协议级靶站</h1>
<button id="alert-btn" onclick="alert('DP-ALERT-TEXT')">弹窗</button>
<button id="confirm-btn" onclick="document.getElementById('cb').textContent = confirm('q?') ? '已接受' : '已取消'">
确认框</button><p id="cb"></p>
<div id="hover-box" onmouseover="this.className='hovered'" onmouseout="this.className=''">悬停我</div>
<input id="txt" type="text" value="原始">
<button id="show-text" onclick="document.getElementById('out').textContent=document.getElementById('txt').value">
读文本</button><p id="out"></p>
<form id="fm" method="post" action="/upload">
<input id="file" type="file" name="up"><button id="send" type="submit">默认 enctype 上传</button>
</form>
<form id="fm2" method="post" action="/upload" enctype="multipart/form-data">
<input id="file2" type="file" name="up2"><button id="send2" type="submit">multipart 上传</button>
</form>
<a id="dl" href="/dl.bin?name=dp-probe.txt">下载</a>
<iframe id="fr" src="/frame.html" width="400" height="120"></iframe>
</body></html>
"""

FRAME = """<!DOCTYPE html>
<html><head><title>inner-frame</title><meta charset="utf-8"></head>
<body><p id="fp">iframe 内的段落</p><button id="fbtn" onclick="document.getElementById('fs').textContent='iframe 已点击'">
点我</button><p id="fs"></p></body></html>
"""

DL_BODY = b"drissionpage-download-probe\n" * 8


class ProtoHandler(BaseHTTPRequestHandler):
def log_message(self, *args):
pass

def _send(self, body, ctype="text/html; charset=utf-8", extra=None):
data = body.encode("utf-8") if isinstance(body, str) else body
self.send_response(200)
self.send_header("Content-Type", ctype)
self.send_header("Content-Length", str(len(data)))
for k, v in (extra or {}).items():
self.send_header(k, v)
self.end_headers()
self.wfile.write(data)

def do_GET(self):
parsed = urlparse(self.path)
if parsed.path == "/frame.html":
self._send(FRAME)
elif parsed.path == "/dl.bin":
name = parse_qs(parsed.query).get("name", ["file.bin"])[0]
self._send(DL_BODY, "application/octet-stream",
{"Content-Disposition": 'attachment; filename="%s"' % name})
else:
self._send(MAIN)

def do_POST(self):
length = int(self.headers.get("Content-Length", 0))
raw = self.rfile.read(length)
has_name = b'filename="upload_sample.txt"' in raw
has_mark = UP_MARK.encode() in raw
text = "收到文件: %s | 内容标记: %s | 请求体字节: %s" % (
"是" if has_name else "否", "是" if has_mark else "否", "非零" if length else "零")
self._send("<html><body><p id='r'>%s</p></body></html>" % text)


class QuietServer(ThreadingHTTPServer):
def handle_error(self, request, client_address):
pass


def start_target():
try:
with socket.create_connection(("127.0.0.1", PORT), 0.2):
print("端口 %d 已被占用,脚本终止" % PORT)
sys.exit(1)
except OSError:
pass
srv = QuietServer(("127.0.0.1", PORT), ProtoHandler)
threading.Thread(target=srv.serve_forever, daemon=True).start()
return srv


def show(expr, fn):
try:
print("%-44s -> %r" % (expr, fn()))
except Exception as exc:
print("%-44s %s: %s" % (
expr, type(exc).__name__,
" | ".join(l.strip() for l in str(exc).strip().splitlines())))


def clean_dirs():
shutil.rmtree(TMP, ignore_errors=True)
os.makedirs(DL_DIR, exist_ok=True)
sample = os.path.join(TMP, "upload_sample.txt")
with open(sample, "w", encoding="utf-8") as f:
f.write(UP_MARK)
return sample


srv = start_target()
base = "http://127.0.0.1:%d/" % PORT
sample = clean_dirs()
page = ChromiumPage(ChromiumOptions().headless(True).auto_port())
try:
print("== 1. 弹窗 handle_alert ==")
page.get(base)
page.ele("#alert-btn").click()
show("page.get_alert()", lambda: page.get_alert())
text = page.handle_alert()
print("handle_alert() 返回(接受):", repr(text))
show("弹窗处理后页面仍在:", lambda: page.title)
page.ele("#confirm-btn").click()
show("handle_alert(accept=False) 取消确认框", lambda: page.handle_alert(accept=False))
show("取消后页面文本", lambda: page.ele("#cb").text)
show("没有弹窗时 handle_alert(timeout=0.5)",
lambda: page.handle_alert(timeout=0.5))
from DrissionPage.common import Settings
print("Settings.auto_handle_alert:", Settings.auto_handle_alert)
show("Settings.set_auto_handle_alert(True)", lambda: Settings.set_auto_handle_alert(True))
page.ele("#alert-btn").click()
page.wait(0.3)
show("自动处理后是否还有弹窗", lambda: page.handle_alert(timeout=0.5))
Settings.set_auto_handle_alert(None)
print("恢复手动处理:", Settings.auto_handle_alert is None)

print()
print("== 2. input 与点击 / 悬停 ==")
page.get(base)
box = page.ele("#txt")
print("input 前 value:", box.value)
box.input("第一段")
print("input(文本) 后 value:", box.value)
box.input("追加")
print("再次 input(不清空) 后 value:", box.value)
box.input("清空后重写", clear=True)
print("input(clear=True) 后 value:", box.value)
box.input("by_js 写入", by_js=True)
print("input(by_js=True) 后 value:", box.value)
page.ele("#show-text").click()
print("点按钮读到的文本框值:", page.ele("#out").text)
page.get(base)
hb = page.ele("#hover-box")
print("hover 前 class:", repr(hb.attr("class")))
show("ele.hover()", lambda: type(hb.hover()).__name__)
page.wait(0.2)
print("hover 后 class:", repr(page.ele("#hover-box").attr("class")))
print("ele.click 是 Clicker 对象:", type(page.ele("#alert-btn").click).__name__)
print("inspect.signature(Clicker.__call__):",
__import__("inspect").signature(type(page.ele("#alert-btn").click).__call__))
print("元素有 actions 吗:", hasattr(page.ele("#hd"), "actions"),
"| 页面有 actions 吗:", hasattr(page, "actions"),
type(page.actions).__name__)
page.get(base)
c = page.ele("#confirm-btn").click(by_js=True)
print("处理 JS 点击弹窗:", page.handle_alert())
print("click(by_js=True) 返回类型:", type(c).__name__,
"| 是否就是被点的元素:", c.attr("id"))
page.get(base)
c = page.ele("#confirm-btn").click()
print("处理普通点击弹窗:", page.handle_alert())
print("click() 返回类型:", type(c).__name__,
"| 是否就是被点的元素:", c.attr("id"))
show("ele.click(actions=True)", lambda: page.ele("#hd").click(actions=True))
show("page.actions 成员", lambda: sorted(n for n in dir(page.actions) if not n.startswith("_")))

print()
print("== 3. 上传文件(不碰系统对话框) ==")
page.get(base)
up = page.ele("#file")
print("type(#file):", type(up).__name__, "| tag:", up.tag)
show("ele.input(文件路径)", lambda: type(up.input(sample)).__name__)
print("page.upload_list:", page.upload_list)
page.set.upload_files(sample)
print("set.upload_files 之后 page.upload_list 条数:", len(page.upload_list))
page.get(base)
f_el = page.ele("#file")
f_el.input(sample)
print("input(路径) 后 files.length:",
page.run_js("return document.getElementById('file').files.length"))
print("files[0].name:", page.run_js("return document.getElementById('file').files[0].name"))
page.ele("#send").click()
page.wait.eles_loaded("#r", timeout=8)
print("默认 enctype 的表单,靶站回显:", page.ele("#r").text)

page.get(base)
page.ele("#file2").input(sample)
print("第二个表单(enctype=multipart/form-data)files.length:",
page.run_js("return document.getElementById('file2').files.length"))
page.ele("#send2").click()
page.wait.eles_loaded("#r", timeout=8)
print("multipart 表单,靶站回显:", page.ele("#r").text)

print()
print("== 4. 下载 ==")
page.get(base)
page.set.download_path(DL_DIR)
print("page.download_path:", page.download_path == DL_DIR)
show("page.download 类型", lambda: type(page.download).__name__)
print("下载前目录文件:", sorted(os.listdir(DL_DIR)))
page.ele("#dl").click()
got = page.wait.download_begin(timeout=8)
print("wait.download_begin() 返回类型:", type(got).__name__)
show("wait.downloads_done() 返回类型", lambda: type(page.wait.downloads_done(timeout=15)).__name__)
show("page.download.wait() 返回类型", lambda: type(page.download.wait(timeout=15)).__name__)
files = sorted(os.listdir(DL_DIR))
print("下载后目录文件:", files)
if files:
with open(os.path.join(DL_DIR, files[0]), "rb") as f:
data = f.read()
print("文件名:", files[0], "| 字节数:", len(data),
"| sha256 前 12 位:", hashlib.sha256(data).hexdigest()[:12])
show("page.download_dir 是否存在", lambda: hasattr(page, "download_dir"))
print("type(page.download):", type(page.download).__name__,
"| 公开成员:", sorted(n for n in dir(page.download) if not n.startswith("_")))

print()
print("== 5. iframe ==")
page.get(base)
print("page.get_frames() 数量:", len(page.get_frames()))
frame = page.get_frame("#fr")
print("type(get_frame('#fr')):", type(frame).__name__)
print("frame.url:", frame.url)
print("frame.title:", frame.title)
print("frame.ele('#fp').text:", frame.ele("#fp").text)
frame.ele("#fbtn").click()
print("iframe 内点击后:", frame.ele("#fs").text)
show("page.get_frame(0) 索引方式", lambda: page.get_frame(0).tag)
show("page.get_frame(1) 第一个 iframe", lambda: page.get_frame(1).tag)
show("iframe 元素自身 doc()", lambda: page.ele("#fr").doc().ele("#fp").text)
inner = page.ele("#fp", timeout=0.3)
print("主页面直接找 iframe 内 #fp:", type(inner).__name__, "| tag:", inner.tag,
"| text:", inner.text)
show("把 iframe 当元素:page.ele('#fr').eles('t:p') 数量",
lambda: len(page.ele("#fr").eles("t:p")))

print()
print("== 6. 截图 ==")
shot_dir = os.path.join(TMP, "shot")
os.makedirs(shot_dir, exist_ok=True)
page.get(base)
r1 = page.get_screenshot(path=shot_dir, name="dp-page.png")
print("page.get_screenshot 返回类型:", type(r1).__name__)
ele = page.ele("#hd")
r2 = ele.get_screenshot(path=shot_dir, name="dp-ele.png")
print("ele.get_screenshot 返回类型:", type(r2).__name__)
files = sorted(os.listdir(shot_dir))
print("生成文件:", files)
for name in files:
with open(os.path.join(shot_dir, name), "rb") as f:
head = f.read(8)
size = os.path.getsize(os.path.join(shot_dir, name))
print(" %s: 字节数>0 = %s | PNG 魔数 = %s" % (name, size > 0, head == b"\x89PNG\r\n\x1a\n"))
b = page.get_screenshot(as_bytes="png")
print("as_bytes='png' 类型:", type(b).__name__, "| 长度>0:", len(b) > 0)
show("full_page=True 长度是否更大",
lambda: len(page.get_screenshot(as_bytes="png", full_page=True)) > len(b))
finally:
page.quit()
srv.shutdown()
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
== 1. 弹窗 handle_alert ==
page.get_alert() AttributeError: 'ChromiumPage' object has no attribute 'get_alert'
handle_alert() 返回(接受): 'DP-ALERT-TEXT'
弹窗处理后页面仍在: -> 'proto-probe'
handle_alert(accept=False) 取消确认框 -> 'q?'
取消后页面文本 -> '已取消'
没有弹窗时 handle_alert(timeout=0.5) -> False
Settings.auto_handle_alert: None
Settings.set_auto_handle_alert(True) -> <class 'DrissionPage._functions.settings.Settings'>
自动处理后是否还有弹窗 -> False
恢复手动处理: True
== 2. input 与点击 / 悬停 ==
input 前 value: 原始
input(文本) 后 value: 第一段原始
再次 input(不清空) 后 value: 第一段追加原始
input(clear=True) 后 value: 清空后重写
input(by_js=True) 后 value: by_js 写入
点按钮读到的文本框值: by_js 写入
hover 前 class: None
ele.hover() -> 'ChromiumElement'
hover 后 class: 'hovered'
ele.click 是 Clicker 对象: Clicker
inspect.signature(Clicker.__call__): (self, by_js=False, timeout=1.5, wait_stop=True)
元素有 actions 吗: False | 页面有 actions 吗: True Actions
处理 JS 点击弹窗: q?
click(by_js=True) 返回类型: ChromiumElement | 是否就是被点的元素: confirm-btn
处理普通点击弹窗: q?
click() 返回类型: ChromiumElement | 是否就是被点的元素: confirm-btn
ele.click(actions=True) TypeError: Clicker.__call__() got an unexpected keyword argument 'actions'
page.actions 成员 -> ['click', 'curr_x', 'curr_y', 'down', 'drag_in', 'hold', 'input', 'key_down', 'key_up', 'left', 'm_click', 'm_hold', 'm_release', 'modifier', 'move', 'move_to', 'owner', 'r_click', 'r_hold', 'r_release', 'release', 'right', 'scroll', 'type', 'up', 'wait']
== 3. 上传文件(不碰系统对话框) ==
type(#file): ChromiumElement | tag: input
ele.input(文件路径) -> 'ChromiumElement'
page.upload_list: None
set.upload_files 之后 page.upload_list 条数: 1
input(路径) 后 files.length: 1
files[0].name: upload_sample.txt
默认 enctype 的表单,靶站回显: 收到文件: 否 | 内容标记: 否 | 请求体字节: 非零
第二个表单(enctype=multipart/form-data)files.length: 1
multipart 表单,靶站回显: 收到文件: 是 | 内容标记: 是 | 请求体字节: 非零
== 4. 下载 ==
page.download_path: True
page.download 类型 -> 'DrissionGet'
下载前目录文件: []
wait.download_begin() 返回类型: DownloadMission
wait.downloads_done() 返回类型 -> 'ChromiumPage'
page.download.wait() 返回类型 -> 'NoneType'
下载后目录文件: ['dp-probe.txt']
文件名: dp-probe.txt | 字节数: 224 | sha256 前 12 位: 8e7913f2ff10
page.download_dir 是否存在 -> False
type(page.download): DrissionGet | 公开成员: ['add', 'block_size', 'by_browser', 'cancel', 'download', 'encoding', 'file_exists', 'get_failed_missions', 'get_mission', 'interval', 'is_running', 'missions', 'page', 'retry', 'roads', 'save_path', 'session', 'set', 'show', 'split', 'timeout', 'wait', 'waiting_list']
== 5. iframe ==
page.get_frames() 数量: 1
type(get_frame('#fr')): ChromiumFrame
frame.url: http://127.0.0.1:8067/frame.html
frame.title: inner-frame
frame.ele('#fp').text: iframe 内的段落
iframe 内点击后: iframe 已点击
page.get_frame(0) 索引方式 ElementNotFoundError: 没有找到元素。 | 方法: get_frame() | 参数: {'loc_ind_ele': 0} | 版本: 4.1.1.4
page.get_frame(1) 第一个 iframe -> 'iframe'
iframe 元素自身 doc() AttributeError: 'ChromiumFrame' object has no attribute 'doc'
主页面直接找 iframe 内 #fp: ChromiumElement | tag: p | text: iframe 内的段落
把 iframe 当元素:page.ele('#fr').eles('t:p') 数量 -> 2
== 6. 截图 ==
page.get_screenshot 返回类型: str
ele.get_screenshot 返回类型: str
生成文件: ['dp-ele.png', 'dp-page.png']
dp-ele.png: 字节数>0 = True | PNG 魔数 = True
dp-page.png: 字节数>0 = True | PNG 魔数 = True
as_bytes='png' 类型: bytes | 长度>0: True
full_page=True 长度是否更大 -> False
  • 本例 page.handle_alert() 返回弹窗文本 'DP-ALERT-TEXT';handle_alert(accept=False) 取消确认框,页面显示 '已取消'。没有弹窗时这次 handle_alert(timeout=0.5) 返回 False,不据此推断所有调用都不会抛异常。page.get_alert() 在本版抛 AttributeError。
  • Settings.set_auto_handle_alert(True) 设置全局自动接受;False 是自动取消,None 才是不启用全局自动处理。示例恢复 None 后手动处理后续确认框,再读取元素属性,避免未处理弹窗阻塞后面的操作。
  • 默认 input() 不清空,在当前光标位置插入,不是保证在末尾追加:本例依次得到 第一段原始、第一段追加原始。clear=True 先清空再输入;by_js=True 的值写入也由按钮读回验证。脚本没有记录键盘事件或 input/change 事件,不能据此判断哪些事件触发、哪些不会触发。
  • ele.click 是可调用的 Clicker 对象,其 __call__ 签名为 (self, by_js=False, timeout=1.5, wait_stop=True)。本例 click() 与 click(by_js=True) 均返回 ChromiumElement,读回 id 为 confirm-btn;这是本次成功点击的结果,不保证所有点击都返回元素。
  • ele.click(actions=True) 抛 TypeError,本例元素没有 actions,页面有 page.actions。按已有输出列表实际计数为 26 个公开成员:click、curr_x、curr_y、down、drag_in、hold、input、key_down、key_up、left、m_click、m_hold、m_release、modifier、move、move_to、owner、r_click、r_hold、r_release、release、right、scroll、type、up、wait;列出成员不等于逐一测试。
  • 本例 ele.hover() 后 class 从 None 变为 hovered,对应靶站的鼠标移入处理。
  • 对 <input type=file> 调用 ele.input(文件路径),本例返回元素,files.length 为 1、文件名正确,不需要操作系统文件对话框。另一段只验证了 page.set.upload_files(sample) 后 page.upload_list 的条数为 1,并未完整验证该入口触发选择框后的上传流程。
  • 两张表单使用同一文件:默认编码的提交未收到文件内容标记,enctype="multipart/form-data" 的提交收到文件和内容标记。文件选中与文件内容被提交是两件事;这里是浏览器原生表单提交,不能据此概括所有 JavaScript 上传方案。
  • 设置 page.set.download_path(目录) 再点下载链接;已有记录中 wait.download_begin() 返回 DownloadMission,wait.downloads_done() 返回页面对象,page.download.wait() 返回 None。仍需确认等待结果与落盘文件,不能把本次返回类型当作失败时的保证。
  • 本次落盘文件为 dp-probe.txt,224 字节,打印的 SHA-256 前 12 位为 8e7913f2ff10。脚本打印了大小与摘要,并未做完整内容相等断言。
  • 本例 page.download 类型为 DrissionGet,列出了任务管理成员;page.download_dir 不存在。未显式设下载路径的错误在 14 脚本中触发,为 RuntimeError,提示可以通过 setter、配置对象或 ini 配置。
  • 本例只有一个 iframe,page.get_frame('#fr') 返回 ChromiumFrame;可读 frame.url、frame.title,用 frame.ele('#fp').text 取文本、frame.ele('#fbtn').click() 点击。数字索引 get_frame(0) 抛 ElementNotFoundError,get_frame(1) 成功;不能说所有数字索引都不可用。调用 doc() 抛 AttributeError。
  • page.ele('#fp', timeout=0.3) 在本机同源 iframe 示例中直接取得内部元素,返回 ChromiumElement。此结果不证明任意跨域、嵌套 iframe 都能用同样方式定位。
  • 页面和元素截图在本例返回路径字符串,as_bytes='png' 返回非空 bytes,两个文件的 PNG 魔数检查通过。full_page=True 那行只比较「长度是否更大」,结果为 False,连长度相等都不能由此确定。字节长度不能证明全页截图是否有效;需另用长页面检查图像尺寸及底部内容覆盖。

异常体系

下面列出 errors 中的异常类,再记录若干错误调用的实际表现。部分探针只打印异常类名;「没有抛异常」也不等于操作成功,需要结合返回值和页面状态判断。

14_errors.py
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
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
# -*- coding: utf-8 -*-
"""异常体系:errors 里的每个类是干什么的,本机怎么真实触发,报错原文。靶站 127.0.0.1:8072 。"""

import socket
import sys
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

from DrissionPage import ChromiumOptions, ChromiumPage
from DrissionPage import errors as E
from DrissionPage.common import Settings

PORT = 8072
PAGE = """<!DOCTYPE html>
<html><head><title>err-probe</title><meta charset="utf-8"></head>
<body>
<button id="hid" style="display:none">隐藏的按钮</button>
<input id="dis" type="text" disabled value="禁用">
<button id="pop" onclick="alert('boom')">弹窗</button>
<p id="one">一</p>
</body></html>
"""


class ErrHandler(BaseHTTPRequestHandler):
def log_message(self, *args):
pass

def do_GET(self):
data = PAGE.encode("utf-8")
self.send_response(200)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.send_header("Content-Length", str(len(data)))
self.end_headers()
self.wfile.write(data)


class QuietServer(ThreadingHTTPServer):
def handle_error(self, request, client_address):
pass


def start_target():
try:
with socket.create_connection(("127.0.0.1", PORT), 0.2):
print("端口 %d 已被占用,脚本终止" % PORT)
sys.exit(1)
except OSError:
pass
srv = QuietServer(("127.0.0.1", PORT), ErrHandler)
threading.Thread(target=srv.serve_forever, daemon=True).start()
return srv


def catch(label, fn, full=True):
try:
fn()
print("%-34s 没有抛异常" % label)
except Exception as exc:
msg = " | ".join(l.strip() for l in str(exc).strip().splitlines())
if not full:
msg = "(详情含随机 id ,只类名)"
print("%-34s %s: %s" % (label, type(exc).__name__, msg))


srv = start_target()
base = "http://127.0.0.1:%d/" % PORT
names = sorted(n for n in dir(E) if not n.startswith("_") and n != "annotations")

page = ChromiumPage(ChromiumOptions().headless(True).auto_port())
try:
print("== errors 模块的异常类与继承 ==")
for n in names:
obj = getattr(E, n)
if not isinstance(obj, type):
print(" %-24s 非类: %s" % (n, type(obj).__name__))
continue
chain = " <- ".join(c.__name__ for c in obj.__mro__[1:-1])
print(" %-24s %s" % (n, chain))
print("BaseError 的 MRO:", " <- ".join(c.__name__ for c in E.BaseError.__mro__))

print()
print("== 1. ElementNotFoundError ==")
page.get(base)
miss = page.ele("t:zzz", timeout=0.3)
catch("访问 NoneElement 的属性", lambda: miss.text)
Settings.set_raise_when_ele_not_found(True)
catch("Settings 开启后直接 ele()", lambda: page.ele("t:zzz", timeout=0.3))
Settings.set_raise_when_ele_not_found(False)

print()
print("== 2. WaitTimeoutError ==")
catch("wait.eles_loaded(raise_err=True)",
lambda: page.wait.eles_loaded("t:zzz", timeout=0.3, raise_err=True))
Settings.set_raise_when_wait_failed(True)
catch("Settings 开启后 wait.download_begin",
lambda: page.wait.download_begin(timeout=0.3))
Settings.set_raise_when_wait_failed(False)

print()
print("== 3. LocatorError ==")
catch("ele(123)", lambda: page.ele(123))
catch("ele('@@a=1@@b=2@|c=3')", lambda: page.ele("@@a=1@@b=2@|c=3"))
catch("ele((错的by, 'x'))", lambda: page.ele(("wrong_by", "x")))
catch("ele((By.ID,))", lambda: page.ele(("id",)))

print()
print("== 4. 元素状态类错误 ==")
catch("点击 display:none 的元素", lambda: page.ele("#hid").click(timeout=0.5))
catch("读隐藏元素的 rect", lambda: page.ele("#hid").rect.location)
catch("向 disabled 输入框 input", lambda: page.ele("#dis").input("x", timeout=0.5))

print()
print("== 5. 弹窗未处理 ==")
page.ele("#pop").click()
page.wait(0.3)
catch("有 alert 时再点击别的元素", lambda: page.ele("#one").click(timeout=1), full=False)
catch("alert 存在时取 html", lambda: len(page.html), full=False)
page.handle_alert()
print("处理弹窗后恢复,标题:", page.title)

print()
print("== 6. URL / CDP / JS ==")
catch("get('不是url')", lambda: page.get("不是url"))
catch("get('ftp://127.0.0.1/y')", lambda: page.get("ftp://127.0.0.1/y"))
catch("run_cdp('不存在的命令')", lambda: page.run_cdp("Not_a_Command"))
catch("run_js 抛异常", lambda: page.run_js('throw new Error("boom")'), full=False)
catch("run_js 语法错误", lambda: page.run_js("return ((( "), full=False)

print()
print("== 7. cookie 格式 ==")
catch("set.cookies 缺字段", lambda: page.set.cookies([{"name": "x"}]))
catch("set.cookies 非法类型", lambda: page.set.cookies(3.14))

print()
print("== 8. 元素丢失 / 页面失联 ==")
page.get(base)
ele = page.ele("#one")
ele.run_js("this.remove()")
catch("使用已从 DOM 删除的元素", lambda: ele.text, full=False)
tab = page.new_tab(base)
tab_id = tab.tab_id
tab.close()
catch("用已关闭标签的对象", lambda: tab.url, full=False)
show_page = ChromiumPage(ChromiumOptions().headless(True).auto_port())
show_page.get(base)
print("另开一个页面正常:", show_page.title)
show_page.quit()
catch("quit 之后再用该 page", lambda: show_page.get(base), full=False)

print()
print("== 9. 连不上浏览器 ==")
catch("existing_only 且没有浏览器",
lambda: ChromiumPage(ChromiumOptions().existing_only(True).set_local_port(8074)))
catch("set_address 到一个非 CDP 端口(本机靶站)",
lambda: ChromiumPage(ChromiumOptions().set_address("127.0.0.1:%d" % PORT)))

print()
print("== 10. 参数错 ==")
catch("set_load_mode('bad')", lambda: ChromiumOptions().set_load_mode("bad"))
catch("headless('yes')", lambda: ChromiumOptions().headless("yes"))
leak = None
try:
leak = ChromiumPage(123)
print("%-34s 不抛异常,返回 %s | address=%s(int 被当成端口号)"
% ("ChromiumPage(123)", type(leak).__name__, leak.address))
except Exception as exc:
print("%-34s %s: %s" % ("ChromiumPage(123)", type(exc).__name__,
" | ".join(l.strip() for l in str(exc).strip().splitlines())))
finally:
if leak is not None:
leak.quit()
print(" 已 quit 掉这个用端口 123 起的实例")
finally:
try:
page.quit()
except Exception:
pass
srv.shutdown()
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
== errors 模块的异常类与继承 ==
AlertExistsError BaseError <- Exception <- BaseException
BaseError Exception <- BaseException
BrowserConnectError BaseError <- Exception <- BaseException
CDPError BaseError <- Exception <- BaseException
CanNotClickError BaseError <- Exception <- BaseException
ContextLostError BaseError <- Exception <- BaseException
CookieFormatError BaseError <- Exception <- BaseException
ElementLostError BaseError <- Exception <- BaseException
ElementNotFoundError BaseError <- Exception <- BaseException
GetDocumentError BaseError <- Exception <- BaseException
IncorrectURLError BaseError <- Exception <- BaseException
JavaScriptError BaseError <- Exception <- BaseException
LocatorError BaseError <- Exception <- BaseException
NoRectError BaseError <- Exception <- BaseException
NoResourceError BaseError <- Exception <- BaseException
PageDisconnectedError BaseError <- Exception <- BaseException
StorageError BaseError <- Exception <- BaseException
TargetNotFoundError BaseError <- Exception <- BaseException
UnknownError BaseError <- Exception <- BaseException
WaitTimeoutError BaseError <- Exception <- BaseException
BaseError 的 MRO: BaseError <- Exception <- BaseException <- object
== 1. ElementNotFoundError ==
访问 NoneElement 的属性 ElementNotFoundError: 没有找到元素。 | 方法: ele() | 参数: {'locator': 't:zzz', 'index': 1, 'timeout': 0.3} | 版本: 4.1.1.4
Settings 开启后直接 ele() ElementNotFoundError: 没有找到元素。 | 方法: ele() | 参数: {'locator': 't:zzz', 'index': 1, 'timeout': 0.3} | 版本: 4.1.1.4
== 2. WaitTimeoutError ==
wait.eles_loaded(raise_err=True) WaitTimeoutError: 等待元素加载失败(等待0.3秒)。 | 定位符: ('//*[name()="zzz"]',) | 版本: 4.1.1.4
Settings 开启后 wait.download_begin RuntimeError: 此功能需显式设置下载路径。 | 提示: 使用set.download_path()方法、配置对象或ini文件均可。 | 版本: 4.1.1.4
== 3. LocatorError ==
ele(123) AttributeError: 'int' object has no attribute '_type'
ele('@@a=1@@b=2@|c=3') LocatorError: @@和@|不能同时出现在一个定位语句中。 | 值: @@a=1@@b=2@|c=3 | 版本: 4.1.1.4
ele((错的by, 'x')) LocatorError: 无法识别的定位符。 | 定位符: ('wrong_by', 'x') | 版本: 4.1.1.4
ele((By.ID,)) LocatorError: 定位符长度必须为2。 | 定位符: ('id',) | 版本: 4.1.1.4
== 4. 元素状态类错误 ==
点击 display:none 的元素 NoRectError: 该元素没有位置及大小。 | 版本: 4.1.1.4
读隐藏元素的 rect NoRectError: 该元素没有位置及大小。 | 版本: 4.1.1.4
向 disabled 输入框 input TypeError: ChromiumElement.input() got an unexpected keyword argument 'timeout'
== 5. 弹窗未处理 ==
有 alert 时再点击别的元素 ElementNotFoundError: (详情含随机 id ,只类名)
alert 存在时取 html AlertExistsError: (详情含随机 id ,只类名)
处理弹窗后恢复,标题: err-probe
== 6. URL / CDP / JS ==
get('不是url') IncorrectURLError: 无效的url,也许要加上"http://"? | url: %E4%B8%8D%E6%98%AFurl | 版本: 4.1.1.4
get('ftp://127.0.0.1/y') 没有抛异常
run_cdp('不存在的命令') RuntimeError: 没有找到对应功能,方法错误或你的浏览器太旧。 | 浏览器版本: Chrome/156.0.8078.80 | 方法: Not_a_Command | 版本: 4.1.1.4
run_js 抛异常 JavaScriptError: (详情含随机 id ,只类名)
run_js 语法错误 JavaScriptError: (详情含随机 id ,只类名)
== 7. cookie 格式 ==
set.cookies 缺字段 KeyError: 'value'
set.cookies 非法类型 ValueError: cookies参数类型错误。 | 允许类型: Cookie, CookieJar, list, tuple, str, dict | 当前类型: <class 'float'> | 版本: 4.1.1.4
== 8. 元素丢失 / 页面失联 ==
使用已从 DOM 删除的元素 没有抛异常
用已关闭标签的对象 PageDisconnectedError: (详情含随机 id ,只类名)
另开一个页面正常: err-probe
quit 之后再用该 page PageDisconnectedError: (详情含随机 id ,只类名)
== 9. 连不上浏览器 ==
existing_only 且没有浏览器 BrowserConnectError: 浏览器连接失败,请确认浏览器已启动。 | 地址: 127.0.0.1:8074 | 版本: 4.1.1.4
set_address 到一个非 CDP 端口(本机靶站) BrowserConnectError: 浏览器连接失败,请检查8072端口是否浏览器,且已添加"--remote-debugging-port=8072"启动项。 | 地址: 127.0.0.1:8072 | 版本: 4.1.1.4
== 10. 参数错 ==
set_load_mode('bad') ValueError: value参数值错误。 | 允许值: 'normal', 'eager', 'none' | 当前值: bad | 版本: 4.1.1.4
headless('yes') 没有抛异常
ChromiumPage(123) 不抛异常,返回 ChromiumPage | address=127.0.0.1:123(int 被当成端口号)
已 quit 掉这个用端口 123 起的实例
  • 已有输出列出 20 个异常类,其中包含 BaseError 自身;其余 19 个继承 BaseError,而 BaseError 继承 Exception。可按需要分别捕获 ElementNotFoundError、WaitTimeoutError,或用 BaseError 捕获这一组库异常。
  • 本例默认设置下,page.ele('t:zzz', timeout=0.3) 返回 NoneElement,访问 .text 才抛 ElementNotFoundError;开启 Settings.set_raise_when_ele_not_found(True) 后,查找调用本身就抛。错误信息包含方法、定位参数和版本,有助于复现。
  • 本例 wait.eles_loaded(..., raise_err=True) 超时抛 WaitTimeoutError;没有配置下载路径时,wait.download_begin() 抛的却是 Python 的 RuntimeError。except (BaseError, RuntimeError) 只覆盖这两类及其子类,不是全兜底,也接不住这里的 ValueError、KeyError、TypeError、AttributeError;应按业务恢复方式选择捕获范围。
  • LocatorError 的三个已测输入是混用 @@ 与 @| 的字符串、未知方式元组 ('wrong_by', 'x')、长度错误元组 ('id',);page.ele(123) 则抛 AttributeError。即使都属于“定位写错”,异常类型也不同。
  • 点击 display:none 元素和读取其 rect.location,本例均抛 NoRectError。禁用输入框那行给 input 传了 timeout=0.5,先因不支持 timeout 抛 TypeError。它只测到了错误参数,没有测到使用正确签名向 disabled 元素输入的结果。
  • 在本例未处理 alert 时,再定位并点击另一元素抛 ElementNotFoundError,取 page.html 抛 AlertExistsError;处理弹窗后能正常读标题。弹窗是排查方向之一,不是这两类错误的唯一原因。
  • get('不是url') 抛 IncorrectURLError。get('ftp://127.0.0.1/y') 那行只记录了没有抛异常,没有打印返回值,不能推出访问成功或库的协议校验策略。
  • run_cdp('Not_a_Command') 抛 RuntimeError,信息包含浏览器版本和命令名;run_js 主动抛 JS 异常或包含语法错误时,记录中均为 JavaScriptError。
  • cookie 缺少 value 字段时,本例抛 KeyError: 'value';传不支持的类型则抛 ValueError,并非所有 cookie 错误都属于 BaseError。
  • 本机元素执行 this.remove() 后再读取旧引用的 .text,本次没有抛异常。从 DOM 移除不等于所有远端引用立即失效,不能写成已经触发 ElementLostError。关闭标签后读取对象、quit() 后再访问页面则都触发 PageDisconnectedError。
  • existing_only(True) 连接无浏览器的端口,以及连接本机非 CDP 靶站端口,本例都抛 BrowserConnectError。可先核对浏览器是否启动、地址端口是否正确以及远程调试是否开启。
  • 配置参数错误的补充检查如下;headless 接收字符串未报错,不等于建议传字符串。
19_api_boundaries.py
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
# -*- coding: utf-8 -*-
"""仅检查接口与错误调用,不把错误参数误称为功能验证。"""
import inspect
from DrissionPage import ChromiumOptions, SessionOptions, SessionPage

print('SessionOptions.add_adapter:', inspect.signature(SessionOptions.add_adapter))
print('SessionOptions.set_timeout:', inspect.signature(SessionOptions.set_timeout))
print('SessionOptions.from_session:', inspect.signature(SessionOptions.from_session))

page = SessionPage(SessionOptions(read_file=False), timeout=1)
print('show_errmsg=True 的不可达请求:')
try:
page.get('http://127.0.0.1:8075/', retry=0, show_errmsg=True)
except Exception as exc:
print(type(exc).__name__, '|', type(exc).__module__)
finally:
page.close()

for name, fn in (("set_load_mode('bad')", lambda: ChromiumOptions().set_load_mode('bad')),
("headless('yes')", lambda: ChromiumOptions().headless('yes'))):
try:
result = fn()
print(name, '返回:', type(result).__name__)
except Exception as exc:
print(name, type(exc).__name__, '|', ' '.join(str(exc).split()))
1
2
3
4
5
6
7
SessionOptions.add_adapter: (self, url, adapter)
SessionOptions.set_timeout: (self, second)
SessionOptions.from_session: (self, session, headers=None)
show_errmsg=True 的不可达请求:
ConnectTimeout | requests.exceptions
set_load_mode('bad') ValueError | value参数值错误。 允许值: 'normal', 'eager', 'none' 当前值: bad 版本: 4.1.1.4
headless('yes') 返回: ChromiumOptions
  • show_errmsg=True 时不可达请求抛出 requests 的 ConnectTimeout;默认返回 False 的路径见前面的会话示例。ChromiumPage(123) 的已有输出显示整数被当作端口号,地址为 127.0.0.1:123,随后执行了 quit();这是本次环境的结果,不保证该端口在其他环境也能使用。

指纹与脚本注入

下面在自控页面验证浏览器属性、文档初始化脚本和运行时 UA,并检查 Selenium 迁移时的成员名。这里只记录本机观察,不承诺能通过任何站点的检测。

15_fingerprint_and_selenium.py
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
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
# -*- coding: utf-8 -*-
"""反爬/指纹相关能力 + 与 selenium 的迁移对照。本机靶站 127.0.0.1:8073 回显 UA 与 JS 变量。"""

import inspect
import re
import socket
import sys
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

from DrissionPage import Chromium, ChromiumOptions, ChromiumPage
from DrissionPage.common import By, Keys, Settings

PORT = 8073
PAGE = """<!DOCTYPE html>
<html><head><title>fp-probe</title><meta charset="utf-8"></head>
<body><p id="ua"></p><p id="mark"></p><p id="wd"></p>
<script>
document.getElementById('ua').textContent = navigator.userAgent;
document.getElementById('mark').textContent = String(navigator.dpMark);
document.getElementById('wd').textContent = String(navigator.webdriver);
</script></body></html>
"""


class FpHandler(BaseHTTPRequestHandler):
def log_message(self, *args):
pass

def do_GET(self):
ua = self.headers.get("User-Agent", "")
body = PAGE.replace("</body>", "<p id='srv'>%s</p></body>" % re.sub(r"[<>&]", "", ua))
data = body.encode("utf-8")
self.send_response(200)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.send_header("Content-Length", str(len(data)))
self.end_headers()
self.wfile.write(data)


class QuietServer(ThreadingHTTPServer):
def handle_error(self, request, client_address):
pass


def start_target():
try:
with socket.create_connection(("127.0.0.1", PORT), 0.2):
print("端口 %d 已被占用,脚本终止" % PORT)
sys.exit(1)
except OSError:
pass
srv = QuietServer(("127.0.0.1", PORT), FpHandler)
threading.Thread(target=srv.serve_forever, daemon=True).start()
return srv


def show(expr, fn):
try:
print("%-50s -> %r" % (expr, fn()))
except Exception as exc:
print("%-50s %s: %s" % (
expr, type(exc).__name__,
" | ".join(l.strip() for l in str(exc).strip().splitlines())))


srv = start_target()
base = "http://127.0.0.1:%d/" % PORT

print("== 1. navigator.webdriver 的默认值(不加任何处理) ==")
page = ChromiumPage(ChromiumOptions().headless(True).auto_port())
page.get(base)
print(" headless 模式 webdriver:", page.ele("#wd").text)
print(" dpMark(未注入):", page.ele("#mark").text)
page.quit()
page = ChromiumPage(ChromiumOptions().headless(False).auto_port())
page.get(base)
print(" 有头模式 webdriver:", page.ele("#wd").text)
print(" 有头模式 UA 是否含 Headless:", "Headless" in page.user_agent)
page.quit()

print()
print("== 2. add_init_js :每个新文档都执行的注入 ==")
page = ChromiumPage(ChromiumOptions().headless(True).auto_port())
js = ("Object.defineProperty(navigator, 'dpMark', {get: () => '注入成功'});"
"Object.defineProperty(navigator, 'webdriver', {get: () => undefined});")
ret = page.add_init_js(js)
print("add_init_js 返回类型:", type(ret).__name__)
page.get(base)
print(" 注入后 dpMark:", page.ele("#mark").text)
print(" 注入后 webdriver 字符串:", page.ele("#wd").text)
print(" page.run_js 直接读:", page.run_js("return String(navigator.webdriver)"))
page.get(base)
print(" 第二次导航后 dpMark 还在吗:", page.ele("#mark").text)
show("remove_init_js(那个 id)", lambda: page.remove_init_js(ret))
page.get(base)
print(" 移除后 dpMark:", page.ele("#mark").text, "(undefined = 注入已撤)")
show("page.remove_init_js() 无参", lambda: page.remove_init_js())

print()
print("== 3. 运行时改 UA(不用重启浏览器) ==")
print(" 改之前服务端看到:", page.ele("#srv").text[:36])
page.get(base)
show("page.set.user_agent(...)", lambda: page.set.user_agent("DP-RunTime/1.0"))
page.get(base)
print(" 改之后 page.user_agent:", page.user_agent)
print(" 改之后 JS 读到的:", page.ele("#ua").text)
print(" 改之后服务端看到的:", page.ele("#srv").text)

print()
print("== 4. auto_port 与实例隔离 ==")
p1 = ChromiumPage(ChromiumOptions().headless(True).auto_port())
p2 = ChromiumPage(ChromiumOptions().headless(True).auto_port())
p1.get(base)
p2.get(base)
print(" 两个 page 的地址是否不同:", p1.address != p2.address)
print(" 两个 browser 的 id 是否不同:", p1.browser.id != p2.browser.id)
print(" 两个浏览器的进程号是否不同:", p1.process_id != p2.process_id)
print(" p1 标签数:", p1.tabs_count, "| p2 标签数:", p2.tabs_count)
print(" 两个实例各自打开同一页面都成功:", p1.title == p2.title)
p1.quit()
p2.quit()

print()
print("== 5. 浏览器与进程信息(只打印存在性/类型) ==")
page.get(base)
print(" type(page.browser):", type(page.browser).__name__)
print(" type(page.process_id):", type(page.process_id).__name__,
"| type(page.browser_version):", type(page.browser_version).__name__)
print(" page.browser_version:", page.browser_version)
print(" browser 公开成员:", sorted(n for n in dir(page.browser) if not n.startswith("_")))
print(" type(page.session):", type(page.session).__name__)
show("page.session.get 是否可用", lambda: type(page.session.get(base)).__name__)
print(" ChromiumOptions 有 mobile_emulation 吗:", hasattr(ChromiumOptions(), "mobile_emulation"))
print(" ChromiumOptions 有 use_cooperative_mode 吗:",
hasattr(ChromiumOptions(), "use_cooperative_mode"))
print(" ChromiumPage 有 use_cooperative_mode 吗:", hasattr(page, "use_cooperative_mode"))

print()
print("== 6. 与 selenium 的迁移对照 ==")
print(" page 有 find_element 吗:", hasattr(page, "find_element"))
print(" page 有 find_elements 吗:", hasattr(page, "find_elements"))
print(" page 有 find 吗:", hasattr(page, "find"), inspect.signature(page.find))
print(" page 有 save_screenshot 吗:", hasattr(page, "save_screenshot"))
print(" page 有 switch_to / window_handles 吗:",
hasattr(page, "switch_to"), hasattr(page, "window_handles"))
print(" By 来自:", By.__module__, "| By.ID =", By.ID, "| By.XPATH =", By.XPATH)
print(" ele((By.ID,'ua')) 命中:", type(page.ele((By.ID, "ua"))).__name__)
print(" eles((By.TAG_NAME,'p')) 数量:", len(page.eles((By.TAG_NAME, "p"))))
print(" eles((By.CLASS_NAME,'nope')) 数量:", len(page.eles((By.CLASS_NAME, "nope"))))
print(" ele((By.XPATH,\"//p[@id='ua']\")):", page.ele((By.XPATH, "//p[@id='ua']")).tag)
print(" ele((By.CSS_SELECTOR,'p')) 数量:", len(page.eles((By.CSS_SELECTOR, "p"))),
"| 简化写法 eles('t:p') 数量:", len(page.eles("t:p")))
print(" 浏览器侧 querySelectorAll('p') 个数:",
page.run_js("return document.querySelectorAll('p').length"),
"| CSS_SELECTOR 命中的 id:", [e.attr("id") for e in page.eles((By.CSS_SELECTOR, "p"))])
show("By.PARTIAL_LINK_TEXT 在非链接页", lambda: len(page.eles((By.PARTIAL_LINK_TEXT, "x"))))
show("错误的 By", lambda: page.ele(("selenium", "x")))
print(" type(page.driver):", type(page.driver).__name__,
"| driver 的模块:", type(page.driver).__module__)
print(" driver 公开成员:", sorted(n for n in dir(page.driver) if not n.startswith("_")))
print(" page.run_cdp('Page.getResourceTree') 有 frameTree 吗:",
"frameTree" in page.run_cdp("Page.getResourceTree"))
show("page.tab_id 长度", lambda: len(page.tab_id))
show("page 有 execute_script 吗", lambda: hasattr(page, "execute_script"))

print()
print("== 7. common 里给迁移用的工具 ==")
import DrissionPage.common as C
names = sorted(n for n in dir(C) if not n.startswith("_"))
print(" common 公开名数量:", len(names))
for n in ("By", "Keys", "Settings", "from_selenium", "from_playwright", "make_session_ele",
"wait_until", "configs_to_here", "get_blob", "tree"):
obj = getattr(C, n, None)
if obj is None:
print(" %-18s 不存在" % n)
elif callable(obj):
try:
print(" %-18s %s" % (n, inspect.signature(obj)))
except (TypeError, ValueError):
print(" %-18s <可调用>" % n)
else:
print(" %-18s <%s>" % (n, type(obj).__name__))
print(" Keys 里几个键:", Keys.ENTER is not None, repr(Keys.SPACE)[:20])
print(" tree(page) 是否存在:", callable(getattr(C, "tree", None)))
finally_ok = True
page.quit()
srv.shutdown()
print("(本脚本结束,浏览器已 quit)" if finally_ok else "")
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
== 1. navigator.webdriver 的默认值(不加任何处理) ==
headless 模式 webdriver: false
dpMark(未注入): undefined
有头模式 webdriver: false
有头模式 UA 是否含 Headless: False
== 2. add_init_js :每个新文档都执行的注入 ==
add_init_js 返回类型: str
注入后 dpMark: 注入成功
注入后 webdriver 字符串: undefined
page.run_js 直接读: undefined
第二次导航后 dpMark 还在吗: 注入成功
remove_init_js(那个 id) -> None
移除后 dpMark: undefined (undefined = 注入已撤)
page.remove_init_js() 无参 -> None
== 3. 运行时改 UA(不用重启浏览器) ==
改之前服务端看到: Mozilla/5.0 (Windows NT 10.0; Win64;
page.set.user_agent(...) -> None
改之后 page.user_agent: DP-RunTime/1.0
改之后 JS 读到的: DP-RunTime/1.0
改之后服务端看到的: DP-RunTime/1.0
== 4. auto_port 与实例隔离 ==
两个 page 的地址是否不同: True
两个 browser 的 id 是否不同: True
两个浏览器的进程号是否不同: True
p1 标签数: 1 | p2 标签数: 1
两个实例各自打开同一页面都成功: True
== 5. 浏览器与进程信息(只打印存在性/类型) ==
type(page.browser): Chromium
type(page.process_id): int | type(page.browser_version): str
page.browser_version: Chrome/156.0.8078.80
browser 公开成员: ['activate_tab', 'address', 'clear_cache', 'close_tabs', 'cookies', 'download_path', 'get_tab', 'get_tabs', 'id', 'latest_tab', 'load_mode', 'new_tab', 'process_id', 'quit', 'reconnect', 'retry_interval', 'retry_times', 'set', 'states', 'tab_ids', 'tabs_count', 'timeout', 'timeouts', 'user_data_path', 'version', 'wait']
type(page.session): Session
page.session.get 是否可用 -> 'Response'
ChromiumOptions 有 mobile_emulation 吗: False
ChromiumOptions 有 use_cooperative_mode 吗: False
ChromiumPage 有 use_cooperative_mode 吗: False
== 6. 与 selenium 的迁移对照 ==
page 有 find_element 吗: False
page 有 find_elements 吗: False
page 有 find 吗: True (locators, any_one=True, first_ele=True, timeout=None)
page 有 save_screenshot 吗: False
page 有 switch_to / window_handles 吗: False False
By 来自: DrissionPage._functions.by | By.ID = id | By.XPATH = xpath
ele((By.ID,'ua')) 命中: ChromiumElement
eles((By.TAG_NAME,'p')) 数量: 4
eles((By.CLASS_NAME,'nope')) 数量: 0
ele((By.XPATH,"//p[@id='ua']")): p
ele((By.CSS_SELECTOR,'p')) 数量: 5 | 简化写法 eles('t:p') 数量: 4
浏览器侧 querySelectorAll('p') 个数: 4 | CSS_SELECTOR 命中的 id: ['ua', 'mark', 'wd', 'srv', None]
By.PARTIAL_LINK_TEXT 在非链接页 -> 0
错误的 By LocatorError: 无法识别的定位符。 | 定位符: ('selenium', 'x') | 版本: 4.1.1.4
type(page.driver): Driver | driver 的模块: DrissionPage._base.driver
driver 公开成员: ['address', 'alert_flag', 'event_handlers', 'event_queue', 'id', 'immediate_event_handlers', 'immediate_event_queue', 'is_running', 'method_results', 'owner', 'run', 'session_id', 'set_callback', 'start', 'stop']
page.run_cdp('Page.getResourceTree') 有 frameTree 吗: True
page.tab_id 长度 -> 32
page 有 execute_script 吗 -> False
== 7. common 里给迁移用的工具 ==
common 公开名数量: 13
By ()
Keys ()
Settings ()
from_selenium (driver)
from_playwright (page_or_browser)
make_session_ele (html_or_ele, loc=None, index=1, method=None)
wait_until (function, kwargs=None, timeout=10)
configs_to_here (save_name=None)
get_blob (page, url, as_bytes=True)
tree (ele_or_page, text=False, show_js=False, show_css=False)
Keys 里几个键: True '\ue00d'
tree(page) 是否存在: True
(本脚本结束,浏览器已 quit)
  • 本机无头与有头的 navigator.webdriver 都是 false,但单个属性不足以判断浏览器是否可被识别。CDP 与 WebDriver 是不同控制接口,不代表“使用 CDP 就不会被检测”。
  • add_init_js(js) 返回脚本 id,后续新文档自动执行;例中第二次导航仍读到标记。传入脚本 id 调用 remove_init_js 撤销一条,不传 id 撤销全部。把 webdriver 改成 undefined 这里只演示注入,不判断它比 false 更好或更差。
  • page.set.user_agent() 后,页面属性、JS 读取、服务端请求头三处变成新 UA。但 UA 不等于完整浏览器指纹,也不等于 TLS 指纹。
  • 两个 auto_port() 实例的地址、browser.id、process_id 均不同,说明浏览器进程独立;不等于已经测过多进程调度与性能。示例没有操作用户日常浏览器。
  • page.session 可取得 requests.Session,直接 .get() 返回 Response;不要据此假设 cookie 已自动同步,浏览器与会话的搬运见前面的 cookie 节。
  • mobile_emulation、use_cooperative_mode 在本例检查的类上未提供。不要依据名字相似就套用其他库的参数。

迁移时的对应关系:

  • find_element → ele,find_elements → eles,save_screenshot → get_screenshot,execute_script → run_js;窗口切换用标签对象,不存在 Selenium 的 switch_to / window_handles 接口。
  • By 元组可用于定位,例如把 By.ID 和元素 id 组成元组。不认识的 by 抛 LocatorError;没有命中的 eles() 返回空列表。CSS 裸标签多命中的原因见「CSS 搜索边界」,不是 Selenium 的选择器语义保证。
  • page.driver 是 DrissionPage 自己的 Driver,不是 Selenium WebDriver。日常用公开接口;需底层能力时可以 run_cdp(),本例取得 Page.getResourceTree 的 frameTree。
  • common 里除了 By / Keys / Settings,还有 make_session_ele、wait_until、tree、from_selenium、from_playwright 等工具。本节对转换工具仅检查存在性与签名,未执行跨库对象转换;不要把清单当作迁移成功的测试。

怎么选

根据需要选最轻的一条路线:

需求 建议入口 边界
响应源码已有数据 SessionPage 不执行页面 JS,也能用定位语法解析
已知 JSON 接口 SessionPage.get() + .json 核对 HTTP 状态、返回类型与业务字段
必须点击或等 JS Chromium 的标签页 / ChromiumPage 要浏览器及显式的业务等待条件
观察页面接口响应 page.listen 仍启动浏览器,只免去 DOM 二次拆字段
HTTP 与浏览器交替 WebPage / MixTab 用 change_mode(),理解 cookie 搬运时点
队列与持久化调度 Scrapy / Feapder / Crawlee 等框架 DrissionPage 不提供这些框架式组件
正文 Markdown Crawl4AI DrissionPage 不以内置 Markdown 管线为主
  • DrissionPage 适合单机交互脚本、浏览器与 HTTP 混合采集;它有请求重试参数,不能笼统说“没有重试”。持久化队列、任务去重、失败入库等调度策略则需要另做。
  • 接管已有浏览器之前,确认目标确实是自己的调试实例;演示的 quit() 会关闭所连接浏览器,不应照搬到有未保存工作的日常浏览器。无头模式不需要可见窗口,有头模式需要桌面会话。
  • 本篇异常包括返回 False、NoneElement、库异常和 Python 内置异常,不能归纳成一种固定模式。对关键配置读回确认,对响应和业务元素分别校验。
  • 输出中的版本、签名、API 成员属于本次安装;布局、数量、UA 和部分字段属于本次页面。升级后复核相关小节,不因一次失败就把网络错误解释成 API 删除。

参考