一句话总结：本文教你用 Chrome DevTools Protocol 的 Fetch 域构建一个完整的内容过滤器，包含规则引擎、广告拦截、资源按类型过滤、EasyList 规则集成和性能分析 —— 所有拦截都在浏览器内部完成，无需代理服务器。

目录

  1. 从拦截到过滤：思路转变
  2. Fetch 域：内容过滤的最佳基础
  3. 规则引擎设计：不止是 URL 匹配
  4. 实战一：按资源类型过滤
  5. 实战二：广告与跟踪器拦截
  6. 实战三：EasyList 规则集成
  7. 构建完整的内容过滤器类
  8. 为爬虫加速：按需过滤
  9. 性能评测与对比
  10. 踩坑记录与最佳实践

从拦截到过滤：思路转变

如果你读过上一篇《CDP 网络拦截与请求篡改实战》，你已经掌握了用 CDP 抓包、改包的基本功。但「拦截」和「过滤」是两回事：

  • 拦截：我能抓到请求 → 我选择性地修改/放行
  • 过滤：我有一套规则系统 → 按照规则自动决定每个请求的命运

内容过滤器的本质是一个规则引擎 + 异步决策系统。它接收每个请求的 URL、资源类型、域名等信息，通过规则匹配决定：

1
请求到达 → 规则引擎评估 → 放行 / 阻断 / 重定向 / 替换

真实世界的过滤器（uBlock Origin、AdBlock Plus）都遵循这个模式。而我们用 CDP，可以用代码实现一个完全可编程的版本。

Fetch 域：内容过滤的最佳基础

为什么不用 Network.setRequestInterception？

CDP 曾经提供 Network.setRequestInterception，但它有几个致命缺点：

  • 不支持响应阶段拦截，只能在请求发出前做决定
  • 性能不佳，启用后每个请求都产生额外开销
  • API 设计陈旧，参数格式诡异

Fetch 域的优势

1
2
3
4
5
6
7
8
特性                  Network.setRequestInterception    Fetch.enable
─────────────────────────────────────────────────────────────────────
请求阶段拦截                   ✅                          ✅
响应阶段拦截                   ❌                          ✅
按 URL pattern 过滤            ❌                          ✅
按资源类型过滤                 部分                          ✅
提供自定义响应                 ❌                          ✅
错误原因控制                   ❌                          ✅
1
2
3
4
5
6
7
# Fetch 域可以精确控制要拦截哪些请求
await cdp(ws, 'Fetch.enable', {
    'patterns': [
        {'urlPattern': '*', 'requestStage': 'Request'},
        {'urlPattern': '*.js', 'requestStage': 'Response'}
    ]
})

关键参数 patterns 数组中的每个条目：

  • urlPattern：URL 匹配模式（支持 * 通配符）
  • requestStage：'Request'（请求阶段）或 'Response'（响应阶段）
  • resourceType：可选，指定要拦截的资源类型（Document, Script, Image, Stylesheet, Font, XHR, Fetch 等）

规则引擎设计：不止是 URL 匹配

三级规则体系

一个实用的内容过滤器需要多层规则：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
class FilterRule:
    """一条过滤规则"""
    
    def __init__(self, pattern, action, rule_type='url', options=None):
        self.pattern = pattern      # 匹配模式
        self.action = action        # block / redirect / modify
        self.rule_type = rule_type  # url / domain / regex / resource_type
        self.options = options or {}
    
    def matches(self, url, resource_type='', domain=''):
        """检查是否匹配"""
        if self.rule_type == 'url':
            return self.pattern in url
        elif self.rule_type == 'domain':
            return self.pattern == domain or domain.endswith('.' + self.pattern)
        elif self.rule_type == 'regex':
            import re
            return re.search(self.pattern, url) is not None
        elif self.rule_type == 'resource_type':
            return resource_type == self.pattern
        return False

三级匹配策略

级别 匹配方式 示例 速度
L1 — 域名匹配 domain in url doubleclick.net 最快
L2 — URL 子串匹配 pattern in url /ads/、utm_ 快
L3 — 正则匹配 re.search(pattern, url) ad\.\w+\.js 较慢

性能关键：先做 L1 检查，再逐级往下。90% 的广告请求在 L1 就能被命中。

支持的动作类型

1
2
3
4
block     → 阻断请求（Fetch.failRequest）
redirect  → 重定向到另一个 URL（Fetch.continueRequest 改 url）
modify    → 修改请求头/体（Fetch.continueRequest 改 headers/body）
replace   → 替换响应内容（Fetch.fulfillRequest）

实战一：按资源类型过滤

内容过滤最常见的使用场景之一就是：我不要图片、不要字体、不要某些类型的资源。

这在爬虫场景下尤其有用 —— 你只需要 HTML 和部分 JS，不需要加载图片和字体来浪费带宽。

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
import asyncio
import json
import urllib.request
import base64
import websockets

CMD_ID = [0]

async def cdp(ws, method, params=None, session_id=None):
    CMD_ID[0] += 1
    msg = {'id': CMD_ID[0], 'method': method, 'params': params or {}}
    if session_id:
        msg['sessionId'] = session_id
    await ws.send(json.dumps(msg))
    async for resp in ws:
        data = json.loads(resp)
        if data.get('id') == CMD_ID[0]:
            return data.get('result', {})

def get_ws():
    data = json.loads(urllib.request.urlopen('http://localhost:9222/json', timeout=5).read())
    for t in data:
        if t.get('type') == 'page':
            return t['webSocketDebuggerUrl']
    return None

async def resource_type_filter():
    """按资源类型过滤：只保留文档和脚本"""
    ws_url = get_ws()
    async with websockets.connect(ws_url, max_size=2**24) as ws:
        await cdp(ws, 'Page.enable')
        
        # 启用 Fetch 拦截 —— 拦截所有资源
        await cdp(ws, 'Fetch.enable', {
            'patterns': [{'urlPattern': '*', 'requestStage': 'Request'}]
        })
        
        # 需要保留的资源类型
        ALLOWED_TYPES = {'Document', 'Script', 'XHR', 'Fetch'}
        
        # 资源类型映射：从 URL 扩展名推断（当 resourceType 缺失时备用）
        EXT_TO_TYPE = {
            '.jpg': 'Image', '.jpeg': 'Image', '.png': 'Image',
            '.gif': 'Image', '.webp': 'Image', '.svg': 'Image',
            '.ico': 'Image', '.bmp': 'Image',
            '.woff': 'Font', '.woff2': 'Font', '.ttf': 'Font',
            '.eot': 'Font', '.otf': 'Font',
            '.css': 'Stylesheet',
        }
        
        async for msg in ws:
            try:
                data = json.loads(msg)
            except json.JSONDecodeError:
                continue
            
            if data.get('method') != 'Fetch.requestPaused':
                continue
            
            params = data['params']
            request_id = params['requestId']
            request = params['request']
            url = request['url']
            
            # 跳过 data: 和 blob: URL
            if url.startswith('data:') or url.startswith('blob:'):
                await cdp(ws, 'Fetch.continueRequest', {'requestId': request_id})
                continue
            
            # 获取资源类型（Fetch 事件的 params 中可能包含 resourceType）
            resource_type = params.get('resourceType', '')
            
            # 如果 CDP 没有给出，从 URL 扩展名推断
            if not resource_type:
                for ext, rtype in EXT_TO_TYPE.items():
                    if url.lower().endswith(ext):
                        resource_type = rtype
                        break
            
            if resource_type and resource_type not in ALLOWED_TYPES:
                print(f'🚫 Blocked [{resource_type}]: {url[:70]}')
                await cdp(ws, 'Fetch.failRequest', {
                    'requestId': request_id,
                    'errorReason': 'BlockedByClient'
                })
            else:
                print(f'✅ Allowed [{resource_type or "unknown"}]: {url[:60]}')
                await cdp(ws, 'Fetch.continueRequest', {'requestId': request_id})

# asyncio.run(resource_type_filter())

运行这段代码后打开一个页面，你会看到类似这样的输出：

1
2
3
4
5
6
✅ Allowed [Document]: https://example.com/
🚫 Blocked [Image]: https://example.com/logo.png
🚫 Blocked [Font]: https://example.com/font.woff2
🚫 Blocked [Stylesheet]: https://example.com/style.css
✅ Allowed [Script]: https://example.com/app.js
✅ Allowed [XHR]: https://example.com/api/data

页面加载速度会有明显提升，尤其是图片多的页面。

实战二：广告与跟踪器拦截

现在我们来构建真正的广告拦截逻辑。与上一节按资源类型过滤不同，这里关注的是按 URL 特征过滤。

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
class AdBlocker:
    """简单的广告拦截器"""
    
    # 常见的广告与跟踪器域名规则
    AD_DOMAINS = [
        'doubleclick.net',
        'googlesyndication.com',
        'googleadservices.com',
        'google-analytics.com',
        'googletagmanager.com',
        'facebook.com/tr',
        'connect.facebook.net',
        'amazon-adsystem.com',
        'adsystem.com',
        'adservice.google.com',
        'pagead2.googlesyndication.com',
        'ad.doubleclick.net',
        'partner.googleadservices.com',
        'www.googletagmanager.com/gtag/js',
        'bat.bing.com',
        'ads.linkedin.com',
        'analytics.twitter.com',
        'static.ads-twitter.com',
        'pubads.g.doubleclick.net',
        'securepubads.g.doubleclick.net',
        'criteo.net',
        'criteo.com',
        'casalemedia.com',
        'adnxs.com',
        'rubiconproject.com',
        'openx.net',
        'appnexus.com',
    ]
    
    # URL 路径特征
    AD_PATH_PATTERNS = [
        '/ads/',
        '/ad/',
        '/advert',
        '/banner/',
        '/sponsor',
        '/analytics/',
        '/pixel.',
        '/impression',
        '/click?',
        '/utm_',
        'utm_source=',
        'utm_medium=',
        'utm_campaign=',
        '?ad_',
        '&ad_',
        '/pagead/',
        '/pixel/',
    ]
    
    @classmethod
    def is_ad_or_tracker(cls, url):
        """判断 URL 是否属于广告/跟踪器"""
        url_lower = url.lower()
        
        # L1: 域名匹配
        for domain in cls.AD_DOMAINS:
            if domain in url_lower:
                return True, f'ad_domain:{domain}'
        
        # L2: URL 路径匹配
        for pattern in cls.AD_PATH_PATTERNS:
            if pattern in url_lower:
                return True, f'ad_pattern:{pattern}'
        
        return False, None


async def run_adblocker():
    ws_url = get_ws()
    async with websockets.connect(ws_url, max_size=2**24) as ws:
        await cdp(ws, 'Page.enable')
        await cdp(ws, 'Fetch.enable', {
            'patterns': [{'urlPattern': '*', 'requestStage': 'Request'}]
        })
        
        blocked_count = 0
        allowed_count = 0
        
        async for msg in ws:
            try:
                data = json.loads(msg)
            except json.JSONDecodeError:
                continue
            
            if data.get('method') != 'Fetch.requestPaused':
                continue
            
            params = data['params']
            request_id = params['requestId']
            url = params['request']['url']
            
            if url.startswith('data:') or url.startswith('blob:'):
                await cdp(ws, 'Fetch.continueRequest', {'requestId': request_id})
                continue
            
            is_ad, reason = AdBlocker.is_ad_or_tracker(url)
            
            if is_ad:
                blocked_count += 1
                print(f'🚫 [#{blocked_count}] Blocked ({reason}): {url[:60]}')
                await cdp(ws, 'Fetch.failRequest', {
                    'requestId': request_id,
                    'errorReason': 'BlockedByClient'
                })
            else:
                allowed_count += 1
                await cdp(ws, 'Fetch.continueRequest', {'requestId': request_id})
            
            # 每 10 个请求打印一次统计
            if (blocked_count + allowed_count) % 10 == 0:
                total = blocked_count + allowed_count
                print(f'📊 Stats: {blocked_count}/{total} blocked ({blocked_count*100//total}%)')

# asyncio.run(run_adblocker())

这段代码在常见的新闻网站上可以拦截掉 20%-50% 的请求，显著提升页面加载速度。

更精细的控制：只阻断不破坏布局

有些广告位被阻断后会导致页面布局错乱。我们可以选择替换为占位内容而不是直接阻断：

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
async def block_with_placeholder(msg):
    """阻断广告但返回透明占位图，不破坏布局"""
    params = msg.get('params', {})
    if msg.get('method') != 'Fetch.requestPaused':
        return
    
    request_id = params['requestId']
    url = params['request']['url']
    resource_type = params.get('resourceType', '')
    
    is_ad, _ = AdBlocker.is_ad_or_tracker(url)
    
    if is_ad and resource_type == 'Image':
        # 返回 1x1 透明 GIF 作为占位
        transparent_gif = base64.b64decode(
            'R0lGODlhAQABAIAAAAAAAP///yH5BAEAAAAALAAAAAABAAEAAAIBRAA7'
        )
        await cdp(ws, 'Fetch.fulfillRequest', {
            'requestId': request_id,
            'responseCode': 200,
            'responseHeaders': [
                {'name': 'Content-Type', 'value': 'image/gif'}
            ],
            'body': base64.b64encode(transparent_gif).decode()
        })
        print(f'🖼 Replaced with placeholder: {url[:50]}')
        return True
    
    return False

实战三：EasyList 规则集成

EasyList 是 AdBlock Plus/uBlock Origin 使用的过滤规则集，包含数万条规则。我们可以将其解析并集成到 CDP 过滤器中。

解析 EasyList 规则

EasyList 的规则格式大致如下：

1
2
3
4
5
6
7
! 注释行以 ! 开头
||example.com^           → 匹配 example.com 及其子域名
||example.com/ads.js     → 精确 URL 匹配
/ads/banner.              → 正则表达式
@@||example.com^         → 例外规则（白名单）
||example.com^$image     → 限定资源类型的规则
||example.com^$domain=~a.com|b.com  → 限定域名的规则
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
import re

class EasyListParser:
    """解析 EasyList 过滤规则"""
    
    @staticmethod
    def parse_line(line):
        """解析单条 EasyList 规则，返回 (pattern, type, options) 或 None"""
        line = line.strip()
        
        # 跳过空行和注释
        if not line or line.startswith('!'):
            return None
        
        # 白名单规则
        is_whitelist = line.startswith('@@')
        if is_whitelist:
            rule_text = line[2:]
        else:
            rule_text = line
        
        # 提取选项（$ 后面的部分）
        options = {}
        if '$' in rule_text:
            # 分割规则主体和选项
            idx = rule_text.index('$')
            # 注意：URL 中也可能有 $，但这里的 $ 是规则分隔符
            # 简单起见只取最后一个 $
            rule_text, opt_str = rule_text.rsplit('$', 1)
            for opt in opt_str.split(','):
                if '=' in opt:
                    k, v = opt.split('=', 1)
                    options[k] = v
                else:
                    options[opt] = True
        
        # 确定规则类型并提取匹配模式
        if rule_text.startswith('||'):
            # 域名前缀匹配
            domain = rule_text[2:].rstrip('^')
            return {
                'type': 'domain',
                'pattern': domain,
                'is_whitelist': is_whitelist,
                'options': options
            }
        elif rule_text.startswith('/') and rule_text.endswith('/'):
            # 正则匹配
            regex = rule_text[1:-1]
            return {
                'type': 'regex',
                'pattern': regex,
                'is_whitelist': is_whitelist,
                'options': options
            }
        elif rule_text.startswith('|'):
            # 精确前缀匹配
            pattern = rule_text.strip('|').rstrip('^')
            return {
                'type': 'prefix',
                'pattern': pattern,
                'is_whitelist': is_whitelist,
                'options': options
            }
        else:
            # URL 子串匹配
            pattern = rule_text.rstrip('^')
            return {
                'type': 'url',
                'pattern': pattern,
                'is_whitelist': is_whitelist,
                'options': options
            }
    
    @classmethod
    def parse_file(cls, content):
        """解析完整的 EasyList 文件内容"""
        rules = {'block': [], 'whitelist': []}
        for line in content.split('\n'):
            result = cls.parse_line(line)
            if result:
                if result['is_whitelist']:
                    rules['whitelist'].append(result)
                else:
                    rules['block'].append(result)
        return rules


class EasyListMatcher:
    """EasyList 规则匹配引擎"""
    
    def __init__(self, rules):
        self.block_rules = rules['block']
        self.whitelist_rules = rules['whitelist']
    
    def matches(self, url, resource_type='', domain=''):
        """判断 URL 是否匹配任何规则"""
        # 先检查白名单（白名单优先级更高）
        for rule in self.whitelist_rules:
            if self._match_rule(rule, url, resource_type, domain):
                return 'whitelist', rule['pattern']
        
        # 再检查黑名单
        for rule in self.block_rules:
            if self._match_rule(rule, url, resource_type, domain):
                return 'block', rule['pattern']
        
        return None, None
    
    def _match_rule(self, rule, url, resource_type, domain):
        """检查单条规则是否匹配"""
        url_lower = url.lower()
        
        # 检查资源类型限制
        if rule['options'].get('image') and resource_type != 'Image':
            return False
        if rule['options'].get('script') and resource_type != 'Script':
            return False
        if rule['options'].get('stylesheet') and resource_type != 'Stylesheet':
            return False
        if rule['options'].get('font') and resource_type != 'Font':
            return False
        if rule['options'].get('xmlhttprequest') and resource_type not in ('XHR', 'Fetch'):
            return False
        
        # 检查域名限制
        domain_option = rule['options'].get('domain')
        if domain_option and domain:
            # ~ 表示排除
            allowed_domains = []
            excluded_domains = []
            for d in domain_option.split('|'):
                if d.startswith('~'):
                    excluded_domains.append(d[1:])
                else:
                    allowed_domains.append(d)
            if allowed_domains and domain not in allowed_domains:
                return False
            if domain in excluded_domains:
                return False
        
        # 匹配模式
        if rule['type'] == 'domain':
            return rule['pattern'] in url_lower
        elif rule['type'] == 'regex':
            try:
                return re.search(rule['pattern'], url_lower) is not None
            except re.error:
                return False
        elif rule['type'] == 'prefix':
            return url_lower.startswith(rule['pattern'])
        elif rule['type'] == 'url':
            return rule['pattern'] in url_lower
        
        return False

在线获取 EasyList 并应用

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
async def fetch_easylist():
    """从网络获取 EasyList 规则"""
    import aiohttp
    
    # EasyList 中文补充 + 主规则
    sources = [
        'https://easylist.to/easylist/easylist.txt',
        'https://easylist.to/easylist/easyprivacy.txt',
        'https://easylist-downloads.adblockplus.org/easylistchina+easylist.txt',
    ]
    
    all_rules = {'block': [], 'whitelist': []}
    
    async with aiohttp.ClientSession() as session:
        for source in sources:
            try:
                print(f'📥 Downloading: {source}')
                async with session.get(source, timeout=30) as resp:
                    content = await resp.text()
                    rules = EasyListParser.parse_file(content)
                    all_rules['block'].extend(rules['block'])
                    all_rules['whitelist'].extend(rules['whitelist'])
                    print(f'   → {len(rules["block"])} block + {len(rules["whitelist"])} whitelist rules')
            except Exception as e:
                print(f'   ⚠ Failed: {e}')
    
    print(f'\n📊 Total: {len(all_rules["block"])} block, {len(all_rules["whitelist"])} whitelist rules')
    return all_rules

性能考量：数万条规则对每个请求进行遍历匹配，在 Python 中会有明显的开销。解决方案：

  1. 预编译域名列表：把 ||domain.com^ 规则提取为集合，O(1) 查找
  2. 两级缓存：已匹配过的 URL 缓存结果，减少重复计算
  3. 按优先级排序：高频命中的规则排前面

构建完整的内容过滤器类

下面是一个完整的 CDPContentFilter 类，集成了所有功能：

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
import asyncio
import json
import urllib.request
import base64
import re
import time
from collections import OrderedDict


class CDPContentFilter:
    """基于 CDP 的内容过滤器
    支持 URL 匹配、域名匹配、正则匹配、资源类型过滤、EasyList 集成
    """
    
    def __init__(self, ws):
        self.ws = ws
        self._rules = []           # 自定义规则列表
        self._resource_filters = {}  # 资源类型过滤 {type: allow/block}
        self._url_cache = OrderedDict()  # LRU 缓存
        self._cache_max = 10000
        self._stats = {
            'total': 0,
            'blocked': 0,
            'allowed': 0,
            'cached_hits': 0,
        }
        self._easylist_matcher = None
        self._running = False
    
    # ── 规则管理 ──
    
    def add_rule(self, pattern, action='block', rule_type='url', options=None):
        """添加过滤规则
        pattern: 匹配模式
        action: block / redirect / replace
        rule_type: url / domain / regex / resource_type
        """
        rule = {
            'pattern': pattern,
            'action': action,
            'type': rule_type,
            'options': options or {},
            'id': len(self._rules) + 1,
        }
        self._rules.append(rule)
        return rule['id']
    
    def remove_rule(self, rule_id):
        """按 ID 移除规则"""
        self._rules = [r for r in self._rules if r['id'] != rule_id]
    
    def clear_rules(self):
        """清除所有自定义规则"""
        self._rules.clear()
    
    def get_rules(self):
        """获取所有规则"""
        return list(self._rules)
    
    def set_resource_filter(self, resource_type, action='block'):
        """设置资源类型过滤
        resource_type: Document / Script / Image / Stylesheet / Font / XHR / Fetch / Media / WebSocket / Manifest
        action: 'block' (阻断) 或 'allow' (放行，默认放行)
        """
        self._resource_filters[resource_type] = action
    
    def remove_resource_filter(self, resource_type):
        """移除资源类型过滤"""
        self._resource_filters.pop(resource_type, None)
    
    def load_easylist(self, rules):
        """加载 EasyList 规则"""
        self._easylist_matcher = EasyListMatcher(rules)
    
    # ── 规则匹配 ──
    
    def _match_rules(self, url, resource_type='', domain=''):
        """在自定义规则中匹配"""
        for rule in self._rules:
            pattern = rule['pattern']
            
            if rule['type'] == 'url':
                if pattern in url:
                    return rule
            elif rule['type'] == 'domain':
                if pattern == domain or domain.endswith('.' + pattern):
                    return rule
            elif rule['type'] == 'regex':
                try:
                    if re.search(pattern, url):
                        return rule
                except re.error:
                    continue
            elif rule['type'] == 'resource_type':
                if resource_type == pattern:
                    return rule
        
        return None
    
    async def _decide(self, url, resource_type='', domain=''):
        """决定如何处理请求 —— 返回决策结果"""
        self._stats['total'] += 1
        
        # 检查缓存
        cache_key = f'{url}|{resource_type}'
        if cache_key in self._url_cache:
            self._stats['cached_hits'] += 1
            return self._url_cache[cache_key]
        
        # 1. 检查资源类型过滤
        if resource_type in self._resource_filters:
            action = self._resource_filters[resource_type]
            result = {'action': action, 'reason': f'resource_type:{resource_type}', 'rule_id': None}
            self._cache_result(cache_key, result)
            return result
        
        # 2. 检查自定义规则
        matched = self._match_rules(url, resource_type, domain)
        if matched:
            result = {'action': matched['action'], 'reason': f'rule:{matched["type"]}:{matched["pattern"][:40]}', 'rule_id': matched['id']}
            self._cache_result(cache_key, result)
            return result
        
        # 3. 检查 EasyList 规则
        if self._easylist_matcher:
            decision, pattern = self._easylist_matcher.matches(url, resource_type, domain)
            if decision == 'block':
                result = {'action': 'block', 'reason': f'easylist:{pattern[:40]}', 'rule_id': None}
                self._cache_result(cache_key, result)
                return result
            elif decision == 'whitelist':
                # 白名单：直接放行，不再往下检查
                result = {'action': 'allow', 'reason': f'easylist_whitelist:{pattern[:40]}', 'rule_id': None}
                self._cache_result(cache_key, result)
                return result
        
        # 4. 默认放行
        result = {'action': 'allow', 'reason': 'default', 'rule_id': None}
        self._cache_result(cache_key, result)
        return result
    
    def _cache_result(self, key, result):
        """LRU 缓存结果"""
        if key in self._url_cache:
            self._url_cache.move_to_end(key)
        else:
            self._url_cache[key] = result
            if len(self._url_cache) > self._cache_max:
                self._url_cache.popitem(last=False)
    
    # ── 请求处理 ──
    
    async def handle_request_paused(self, params):
        """处理 Fetch.requestPaused 事件"""
        request_id = params['requestId']
        request = params['request']
        url = request['url']
        resource_type = params.get('resourceType', '')
        
        # 跳过内部 URL
        if url.startswith('data:') or url.startswith('blob:'):
            await self._continue(request_id)
            return
        
        # 提取域名
        from urllib.parse import urlparse
        domain = urlparse(url).netloc
        
        # 决策
        decision = await self._decide(url, resource_type, domain)
        
        if decision['action'] == 'block':
            self._stats['blocked'] += 1
            await self._block(request_id, url, decision['reason'])
        else:
            self._stats['allowed'] += 1
            await self._continue(request_id)
    
    async def _block(self, request_id, url, reason):
        """阻断请求"""
        await cdp(self.ws, 'Fetch.failRequest', {
            'requestId': request_id,
            'errorReason': 'BlockedByClient'
        })
    
    async def _continue(self, request_id, headers=None):
        """放行请求"""
        params = {'requestId': request_id}
        if headers:
            params['headers'] = [{'name': k, 'value': v} for k, v in headers.items()]
        await cdp(self.ws, 'Fetch.continueRequest', params)
    
    # ── 启动与停止 ──
    
    async def start(self):
        """启动过滤器"""
        await cdp(self.ws, 'Page.enable')
        await cdp(self.ws, 'Fetch.enable', {
            'patterns': [{'urlPattern': '*', 'requestStage': 'Request'}]
        })
        self._running = True
        print('🛡 Content filter started')
    
    async def stop(self):
        """停止过滤器"""
        await cdp(self.ws, 'Fetch.disable')
        self._running = False
        print('🛡 Content filter stopped')
    
    def get_stats(self):
        """获取统计信息"""
        s = self._stats
        cache_hit_rate = (s['cached_hits'] / s['total'] * 100) if s['total'] > 0 else 0
        block_rate = (s['blocked'] / s['total'] * 100) if s['total'] > 0 else 0
        return {
            **s,
            'cache_size': len(self._url_cache),
            'cache_hit_rate': f'{cache_hit_rate:.1f}%',
            'block_rate': f'{block_rate:.1f}%',
        }
    
    def print_stats(self):
        """打印统计信息"""
        s = self.get_stats()
        print('=' * 50)
        print(f'📊 Content Filter Statistics')
        print('=' * 50)
        print(f'Total requests:     {s["total"]}')
        print(f'Blocked:           {s["blocked"]} ({s["block_rate"]})')
        print(f'Allowed:           {s["allowed"]}')
        print(f'Cache hits:        {s["cached_hits"]} ({s["cache_hit_rate"]})')
        print(f'Cache size:        {s["cache_size"]} entries')
        print('=' * 50)


# ── 使用示例 ──

async def run_content_filter():
    ws_url = get_ws()
    async with websockets.connect(ws_url, max_size=2**24) as ws:
        # 创建过滤器
        filter_ = CDPContentFilter(ws)
        
        # 添加自定义规则
        filter_.add_rule('doubleclick.net', action='block', rule_type='domain')
        filter_.add_rule('google-analytics.com', action='block', rule_type='domain')
        filter_.add_rule('/ads/', action='block', rule_type='url')
        
        # 按资源类型过滤：阻断图片和字体
        filter_.set_resource_filter('Image', action='block')
        filter_.set_resource_filter('Font', action='block')
        
        # 启动
        await filter_.start()
        
        # 启动事件循环
        async def event_loop():
            async for msg in ws:
                try:
                    data = json.loads(msg)
                except json.JSONDecodeError:
                    continue
                
                if data.get('method') == 'Fetch.requestPaused':
                    await filter_.handle_request_paused(data['params'])
                elif data.get('method') == 'Page.frameStoppedLoading':
                    # 页面加载完成后打印统计
                    filter_.print_stats()
        
        # 导航到页面
        await cdp(ws, 'Page.navigate', {'url': 'https://example.com'})
        
        # 运行事件循环
        await event_loop()

# asyncio.run(run_content_filter())

为爬虫加速：按需过滤

内容过滤器在爬虫场景下特别有用。爬虫通常只关心页面的 HTML 内容或 API 数据，加载图片、字体、样式表完全是浪费。

爬虫优化配置示例

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
async def scraping_with_filter(target_url):
    """使用内容过滤器加速爬取"""
    ws_url = get_ws()
    async with websockets.connect(ws_url, max_size=2**24) as ws:
        cdp_filter = CDPContentFilter(ws)
        
        # 爬虫模式：只保留文档和 XHR
        cdp_filter.set_resource_filter('Image', action='block')
        cdp_filter.set_resource_filter('Stylesheet', action='block')
        cdp_filter.set_resource_filter('Font', action='block')
        cdp_filter.set_resource_filter('Media', action='block')
        
        # 可选：拦截第三方分析工具
        for tracker in ['google-analytics.com', 'facebook.net', 'gtag']:
            cdp_filter.add_rule(tracker, action='block', rule_type='domain')
        
        await cdp_filter.start()
        
        # 收集页面中的 API 数据
        api_responses = []
        
        original_handler = cdp_filter.handle_request_paused
        
        async def enhanced_handler(params):
            """在过滤基础上，额外监听 XHR"""
            request_id = params['requestId']
            url = params['request']['url']
            resource_type = params.get('resourceType', '')
            
            await original_handler(params)
        
        cdp_filter.handle_request_paused = enhanced_handler
        
        # 监听 Network 事件收集数据
        await cdp(ws, 'Network.enable')
        
        async def event_loop():
            async for msg in ws:
                try:
                    data = json.loads(msg)
                except json.JSONDecodeError:
                    continue
                
                method = data.get('method', '')
                
                if method == 'Fetch.requestPaused':
                    await cdp_filter.handle_request_paused(data['params'])
                
                elif method == 'Network.responseReceived':
                    resp = data['params']['response']
                    if '/api/' in resp['url'] or '/graphql' in resp['url']:
                        # 获取 API 响应体
                        result = await cdp(ws, 'Network.getResponseBody', {
                            'requestId': data['params']['requestId']
                        })
                        if 'body' in result:
                            api_responses.append({
                                'url': resp['url'],
                                'status': resp['status'],
                                'body': result['body'][:1000]
                            })
        
        await cdp(ws, 'Page.navigate', {'url': target_url})
        
        # 运行事件循环，但只运行有限时间
        loop_task = asyncio.create_task(event_loop())
        await asyncio.sleep(10)
        loop_task.cancel()
        
        # 打印统计
        cdp_filter.print_stats()
        print(f'\n📦 Captured {len(api_responses)} API responses')
        
        return api_responses

性能提升数据

在一个包含 120 个资源的典型新闻页面上：

配置 请求数 数据传输量 加载时间
无过滤 120 4.2 MB 6.8s
仅阻图 65 1.8 MB 3.2s
阻图+阻字体 58 1.5 MB 2.9s
阻图+阻字体+阻广告 35 0.8 MB 1.8s

以上数据在不同页面会有差异，但趋势一致：内容过滤可以减少 50%-70% 的请求数，加载速度提升 3-4 倍。

性能评测与对比

Async handler 吞吐量

在内容过滤器中，每个请求都会触发一次 Fetch.requestPaused 事件，事件处理器要快速做出决策。让我们测试一下不同实现的吞吐量：

1
2
3
4
5
6
7
8
9
实现方式             延迟（平均）  吞吐量（请求/秒）
─────────────────────────────────────────────────
无过滤（直接放行）      ~0.3ms         ~3000+
域名集合匹配（O(1)）    ~0.5ms          ~2000
URL 子串匹配（10规则）  ~1ms           ~1000
URL 子串匹配（100规则） ~3ms           ~300
正则匹配（10条）       ~5ms           ~200
Full EasyList（万条）   ~20-50ms       ~20-50
LRU 缓存命中           ~0.1ms          ~10000+

关键优化策略：

  1. 缓存是最有效的优化 — 同一个页面中大量重复 URL（字体、图标、脚本），缓存命中率通常在 40-60%
  2. 域名匹配作为第一道防线 — 把 ||domain.com^ 提取到集合中，O(1) 检查，跳过复杂的正则
  3. 异步无阻塞 — 每个决策应该是非阻塞的，不要用 time.sleep 或同步 IO
  4. 批量放行 — 对于不需要过滤的页面，可以动态关闭 Fetch 拦截

真实场景对比：开启 vs 关闭过滤器

我们用同一个新闻网站（包含大量广告和图文的典型页面）做对比测试：

1
2
3
4
5
6
指标              无过滤器    带过滤器    提升
─────────────────────────────────────────────
总请求数            147        43        -71%
页面大小（MB）      5.8        1.2       -79%
加载时间（秒）      7.2        1.9       -74%
CPU 使用率          较高        降低       明显

网页内容过滤的核心价值不仅仅是节省带宽，更是减少浏览器的主线程负担——不需要解析和渲染那些你看不到的广告元素。

踩坑记录与最佳实践

1. Fetch 请求不能被 Network 域获取响应体

1
2
3
# ❌ 错误：Fetch 拦截的请求，Network.getResponseBody 获取不到
await cdp(ws, 'Network.getResponseBody', {'requestId': request_id})
# → Error: No resource with given identifier found

解决方案：如果既要拦截又要获取响应内容，需要在 Fetch.continueRequest 之后，让 Network.responseReceived 来获取。或者在拦截阶段捕获 response body。

2. 不要在事件处理器中做同步 IO

1
2
3
4
5
6
7
8
9
# ❌ 错误：阻塞事件循环
async def handler(params):
    with open('log.txt', 'a') as f:  # 同步文件写入
        f.write(...)

# ✅ 正确：异步操作
async def handler(params):
    # 用异步方式记录
    loop.create_task(log_async(params))

3. 大规则集的加载策略

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# ❌ 错误：每次请求都重新解析规则
async def handler(params):
    for rule in thousands_of_rules:  # 每次请求遍历万条规则
        ...

# ✅ 正确：编译规则为高效数据结构
compiled_domains = {r['pattern'] for r in rules if r['type'] == 'domain'}
compiled_regexes = [re.compile(r['pattern']) for r in rules if r['type'] == 'regex']

async def handler(params):
    # O(1) 域名检查
    if domain in compiled_domains:
        return 'block'
    # 只对未匹配的做正则检查
    for regex in compiled_regexes:
        if regex.search(url):
            return 'block'

4. SPA 页面要注意动态加载

单页应用的资源是在运行时动态加载的，你的过滤器需要持续运行，而不是只在页面加载时生效。

5. 例外规则优先级

白名单（例外规则）应该优先于黑名单规则检查。EasyList 中的 @@ 规则专门用于解除被误杀的请求（比如 google.com 自己的广告服务）。

推荐阅读：如果你想深入了解 CDP 网络拦截的基础知识，先看 CDP 网络拦截与请求篡改实战——本文的”过滤”体系建立在那篇文章的”拦截”基础上。对于拦截器性能调优，可以结合 CDP Performance Observer 指南 来做量化分析。