1 前提概要使用requests爬虫时经常碰到cloudflare防护的网站即使从浏览器中复制包含cookie在内的所有请求参数扔到requests的headers中返回的仍然是Just a moment...对你发出无情的嘲讽这是因为CF还要验证浏览器指纹这不是简单地加个UA参数就能解决的。2 思路既然CF必须真实地浏览器才能验证通过那我们就用真实的浏览器去爬取就行了。与正常访问网站不同的是我们不需要浏览器渲染网页不需要访问dom中的资源文件只需要借助浏览器的网络访问能力可以把浏览器当成我们爬虫的代理网关。控制浏览器访问指定网页需要用到浏览器的CDP(Chrome DevTools Protocol)功能可以通过指定端口控制浏览器的行为。python中可以通过playwright去操控浏览器。正常通过playwright操控时经常用到的是resp page.goto(https://www.baidu.com) print(resp.text()) # 原始html print(page.content()) # 选然后的html上述方式相当于在浏览器地址栏内输入了指定网址然后回车在代码执行期间我们能看到浏览器加载网页的情况。浏览器加载网页耗费了大量的性能与时间但我们不需要渲染网页不导航 Page、不执行网页、不加载图片/CSS/JS而是让已经连接的 Chrome 通过 CDP 网络层直接 GET URL再从 IO.read 读取响应体。CDP 本身提供了 Network.loadNetworkResource并支持 includeCredentialsPlaywright 也支持通过 new_cdp_session() 直接发送原始 CDP 命令。相当于劫持了浏览器的网络层。参考CDP Networkcdpsession3 部署CDP不建议使用真实浏览器去开启CDP。我的做法是启用Windows沙盒使用沙盒内的浏览器启用CDP。或者可以使用虚拟机也行。Windows沙盒其实就是Windows自带的能够快速部署的虚拟机。3.1 启用Windows沙盒使用真实浏览器或了解Windows沙盒的可以跳过这一节。在Windows搜索栏内搜索启用或关闭Windows功能勾选Windows沙盒确认后等待重启重启后能够搜到windows sandbox的程序直接使用就行。3.2 启用CDP在沙盒内的edge浏览器所在目录执行以下命令.\msedge.exe --remote-debugging-port9222 --user-data-dirC:\Users\WDAGUtilityAccount\Documents\edgedata这两个参数是必须的端口与数据目录都能自己指定。测试一下CDP是否正常在浏览器内输入以下urlhttp://127.0.0.1:9222/json有内容输出就说明CDP正常启用3.3 端口转发CDP仅绑定在127.0.0.1这个IP上外部无法访问。比如物理机启用的CDP能在本机通过127.0.0.1访问CDP但是无法通过物理机的IP(比如192.168.1.2)访问。需要利用另外一个端口(例如9223)将这个端口的流量转发到9222上。netsh interface portproxy add v4tov4 listenaddress0.0.0.0 listenport9223 connectaddress127.0.0.1 connectport9222此时仍然无法远程访问CDP因为中间还拦着一道Windows防火墙。需要添加防火墙规则放行9223端口的流量。netsh advfirewall firewall add rule nameCDP 9223 dirin actionallow protocolTCP localport9223此时可以在宿主机上访问沙盒内的CDP。参考端口转发4 代码平时爬虫时经常使用到多线程加速CDP中我们可以启用多个异步IO的任务达到同样加速的效果架构如下一个 Python 进程 │ ├── 一个 async_playwright │ ├── 一个 CDP Chrome connection │ ├── 一个 CDP Session │ └── 8 个 asyncio Task ├── Task 0 → Network.loadNetworkResource ├── Task 1 → Network.loadNetworkResource ├── Task 2 → Network.loadNetworkResource ... └── Task 7 → Network.loadNetworkResource async_playwright ↓ async CDP session ↓ asyncio Task 0 ─┐ asyncio Task 1 ─┤ asyncio Task 2 ─┤ ... ├── await cdp.send() asyncio Task 7 ─┘import asyncio import base64 import os from playwright.async_api import async_playwright CDP_URL http://172.26.70.255:9223 INPUT_FILE detail_page_urls.txt OUTPUT_DIR rY:\detail_pages WORKER_COUNT 8 MIN_SIZE 10 * 1024 # 读取任务 def load_tasks(): tasks [] with open(INPUT_FILE, r, encodingutf-8) as f: for line_no, line in enumerate(f, 1): line line.strip() if not line: continue parts line.split(None, 1) if len(parts) ! 2: print(f[跳过] 第 {line_no} 行格式错误{line}) continue file_name, url parts tasks.append((file_name, url)) return tasks # 均匀分成 count 份 def split_tasks(tasks, count): result [] total len(tasks) base total // count remainder total % count start 0 for i in range(count): size base if i remainder: size 1 end start size result.append(tasks[start:end]) start end return result # 读取 CDP Stream async def read_stream(cdp, handle): chunks [] try: while True: result await cdp.send( IO.read, { handle: handle, size: 1024 * 1024, } ) data result.get(data, ) if data: if result.get(base64Encoded, False): chunks.append(base64.b64decode(data)) else: chunks.append(data.encode(utf-8)) if result.get(eof, False): break finally: try: await cdp.send( IO.close, { handle: handle } ) except Exception: pass return b.join(chunks) # 通过 CDP Network 下载一个 URL async def download_one(cdp, frame_id, file_name, url, worker_id): html_file os.path.join(OUTPUT_DIR, file_name) # 已经存在 if os.path.exists(html_file): print(f[Worker {worker_id}] {file_name} 已存在跳过) return exists try: # ---------------------------------------------------- # 核心 # # 不 page.goto() # 不 view-source: # # 直接使用 Chrome Network 层请求 # ---------------------------------------------------- result await cdp.send( Network.loadNetworkResource, { frameId: frame_id, url: url, options: { disableCache: False, # 使用浏览器 credentials includeCredentials: True, }, } ) resource result.get( resource, {} ) # 网络请求失败 if not resource.get(success, False): error_name resource.get(netErrorName, ) error_code resource.get(netError, ) print( f[Worker {worker_id}] f{file_name} 下载异常 fNetwork error f{error_name} f({error_code}) ) return stop status_code int(resource.get(httpStatusCode, 0)) # 获取 stream stream resource.get(stream) if not stream: print( f[Worker {worker_id}] f{file_name} 下载异常 fChrome 没有返回 stream ) return error # 读取 response body body await read_stream(cdp, stream) size len(body) # 异常文件处理 # 小于 10KB if size MIN_SIZE: html body.decode(utf-8, errorsignore) # ----------------------------------------------- # Server Error # 虽然小于 10KB但是保存 # ----------------------------------------------- if OK in html: with open(html_file, wb) as f: f.write(body) print( f[Worker {worker_id}] f{file_name} size Error f按规则保存 f{size / 1024:.2f} KB ) return ok # ----------------------------------------------- # 其他小文件 # 不保存停止当前 worker # ----------------------------------------------- print( f[Worker {worker_id}] f{file_name} 下载异常 f{size / 1024:.2f} KB f不是 Server Error f停止当前 Worker ) return stop # 正常保存 with open(html_file, wb) as f: f.write(body) print( f[Worker {worker_id}] f{file_name} OK f{size / 1024:.2f} KB fHTTP {status_code} ) return success except Exception as e: print( f[Worker {worker_id}] f{file_name} 下载异常 f{repr(e)} ) return error # Worker async def worker(worker_id, tasks, cdp, frame_id): print(f[Worker {worker_id}] 开始任务数{len(tasks)}) for file_name, url in tasks: result await download_one(cdp, frame_id, file_name, url, worker_id) # ---------------------------------------------------- # 遇到小于 10KB 且无OK # 停止当前 Worker # ---------------------------------------------------- if result stop: break print(f[Worker {worker_id}] 结束) # 主程序 async def main(): os.makedirs(OUTPUT_DIR, exist_okTrue) # 读取全部任务 tasks load_tasks() print(f总任务数{len(tasks)}) # 分成 WORKER_COUNT 份 batches split_tasks(tasks, WORKER_COUNT) for i, batch in enumerate(batches): print(fWorker {i}: {len(batch)} 个任务) # 连接已有 Chrome async with async_playwright() as p: print(f连接 CDP{CDP_URL}) browser await p.chromium.connect_over_cdp(CDP_URL, timeout30_000) print(Chrome connected:, browser.is_connected()) # 获取现有 BrowserContext if not browser.contexts: raise RuntimeError(CDP Chrome 没有 BrowserContext) context browser.contexts[0] # 获取当前 Page pages context.pages if not pages: raise RuntimeError(CDP Chrome 当前没有 Page) page pages[0] print(当前 Page, page.url) # ---------------------------------------------------- # 建立 CDP Session # # 注意 # 这里整个程序只创建一个 session # 所有 asyncio worker 共用 # # 不存在 Python Thread # 所以不会出现 greenlet 跨线程 # ---------------------------------------------------- cdp await context.new_cdp_session(page) # 获取当前 frame ID frame_tree await cdp.send(Page.getFrameTree) frame_id (frame_tree[frameTree][frame][id]) print(Frame ID, frame_id) # 创建 WORKER_COUNT 个 asyncio Worker workers [] for worker_id in range(WORKER_COUNT): task asyncio.create_task( worker(worker_id, batches[worker_id], cdp, frame_id) ) workers.append(task) # 等待 所有 Worker await asyncio.gather(*workers) print(\n全部 Worker 执行完成) # 只断开 CDP Session不关闭 Chrome try: await cdp.detach() except Exception: pass # Windows if __name__ __main__: asyncio.run(main())代码解释我将所有的爬虫任务收集到一个文件中即代码中的INPUT_FILE变量这个文件有多行每行2列用空格隔开第一列是爬取后的结果保存到的文件名文件保存到OUTPUT_DIR目录下第二列是url。代码中我定义了8个worker每个worker负责INPUT_FILE中任务的1/8当网络请求失败时停止当前worker其他worker不受影响。当所有worker异常停止或任务完成后整个python进程结束。代码中我还定义了一些异常处理比如下载的文件小于10KB需要检查其内容后再判断是否需要保存为文件。5 执行爬虫使用上述方法并不能完全避免CF的防护因为浏览器指纹、cookie会过期无人干预的情况下某个时间点之后所有的任务都会失败。使用之前需要我们打开目标网站手动过一遍CF的验证注意这个标签页不能关闭。6 扩展骚操作在浏览器指纹、cookie过期后爬虫脚本就空转或结束了在CF拦截导致任务停止后手动刷新目标网页能让爬虫重新工作但我不可能盯着爬虫。通过CDP去刷新网页是过不了CF验证的。我想到了自动化操作。思路其实很简单写一个脚本操纵键鼠让他能够刷新网页。再另起一个脚本调用执行爬虫脚本与自动化脚本在这两个任务中循环。自动化脚本示例from time import sleep import pyautogui if __name__ __main__: # 点击任务栏 pyautogui.moveTo(1400, 1060, duration0.5) pyautogui.click() # 点击沙盒 pyautogui.moveTo(975, 1051, duration0.5) pyautogui.click() # 新建标签页 pyautogui.moveTo(1011, 165, duration0.5) pyautogui.click() # 输入目标网址 pyautogui.write(https://target.com) pyautogui.press(enter) sleep(10) # 鼠标滑动模拟真人 pyautogui.moveTo(1000, 440, duration0.5) sleep(10) # 关闭标签页 pyautogui.hotkey(ctrl, w)循环脚本示例import subprocess while True: print(开始执行 爬虫) subprocess.run([python, tasks.py], checkTrue) print(爬虫 执行完成开始执行 自动化) subprocess.run([python, refresh.py], checkTrue) print(自动化 执行完成重新执行 爬虫)