TL;DR
最近在做一个项目时遇到了需要批量获取登录态保护API数据的问题,传统的HTTP请求方式因为复杂的认证机制屡屡碰壁。我发现用Playwright来模拟真实浏览器操作是个不错的解决方案——通过维持登录会话状态,就能稳定地获取到想要的数据。这个思路在实践中效果很好,成功率比预想的要高很多。
为什么要做这个
最近在做项目的时候发现一个很头疼的问题:很多有用的API接口和数据都被登录态保护着,直接用Python的requests库去访问基本都会被挡回来。特别是那些需要复杂认证流程的系统,比如企业内部的SaaS平台、管理后台之类的,传统的HTTP请求方式根本拿不到数据。
我尝试了几个思路:
直接数据抓取
- 需要登录才能看的页面:很多后台管理系统的数据报表、用户列表等,必须先登录才能访问
- 认证后的API接口:很多系统的API都要求Bearer Token或者Session验证,直接访问会返回401
- 批量处理需求:有时候需要从多个页面或者接口获取数据,一个个手动弄太费时间
自动化测试场景
- 接口功能验证:想测试某些接口在不同登录状态下的表现
- 会话保持测试:看看长时间不操作后登录状态是否还能保持
- 批量数据操作:通过脚本自动执行一些需要权限的数据操作
业务自动化需求
- 定期同步数据:比如每天自动从某个管理系统导出报表数据
- 监控数据收集:获取需要管理员权限才能看到的系统监控信息
- 配置批量管理:批量修改一些需要管理员权限的配置项
这些问题在开发过程中其实挺常见的,特别是在做企业内部系统开发或者数据分析的时候。我的这个实现主要就是针对这类需要”先登录,后办事”的场景。
我的实现思路
整个实现我采用了分层的设计,这样每个部分的职责都比较清晰,以后维护起来也方便。大概分成这么几层:
- 配置管理层:主要负责读取目标列表、设置输出路径这些基础配置
- 浏览器控制层:用Playwright管理浏览器的启动、关闭,还有会话状态的维护
- API交互层:专门负责发起HTTP请求、设置请求头、获取响应数据
- 数据处理层:解析API返回的JSON数据,转换成需要的格式然后保存
- 过程控制层:处理各种异常情况、记录进度、添加延迟防止被识别
这样做的好处是每层都很独立,比如以后想换个数据存储方式,只需要改数据处理层那一块就行了,其他地方都不用动。而且调试起来也方便,出了问题能很快定位到是哪一层的责任。
几个关键实现细节
浏览器会话管理
这个实现的基础就是浏览器会话的管理。我用p.chromium.launch(headless=False)
启动了一个非无头模式的浏览器,这样调试起来方便,能看到实际发生了什么。然后通过browser.new_context()
创建独立的浏览器环境,context.new_page()
获取到操作页面。
在构建的入参里面设置了User-Agent和一些HTTP头部。最重要的是,浏览器会自动处理Cookie,这样只要登录一次,后续的请求都会带上登录状态,不需要手动管理Token。
1 2 3 4 5 6 7
| browser.new_context( user_agent="Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36", extra_http_headers={ "Accept": "application/json, text/plain, */*", "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", } )
|
直接访问API
需要通过前端的页面去操作,可以直接用浏览器访问API端点。用page.goto(url, wait_until="networkidle")
直接请求API,wait_until="networkidle"
这个参数确保网络请求都完成了。然后用page.inner_text("body")
就能拿到完整的响应数据。
这样做的好处是绕过了前端的渲染过程,直接拿到后端的原始数据,效率高很多。
登录状态检测
在实际使用中,我发现登录状态可能会过期,所以需要检测机制。我的做法很简单:检查返回的页面内容是否包含”登录”、”密码登录”或”Login”这些关键词。
如果检测到需要登录,我会让用户手动完成登录(通过input("登录完成后请按Enter键继续...")
),然后再验证一下登录是否成功。这种半自动的方式在复杂认证场景下反而比全自动更可靠,另外也不建议本地存储我们的账密来让AI输入,对于账号和密码这些数据,最好是由我们自己来输入。
数据处理和保存
拿到API响应后,我用标准的json.loads()
解析JSON数据,检查返回的状态码,然后提取需要的数据字段。保存数据时,我用了CSV格式,通过追加模式写入,这样即使程序中断了,之前获取的数据也不会丢失。
实际使用中发现,很多API返回的数据结构可能不一致,所以在解析时要做好异常处理,避免因为某个字段缺失导致整个程序崩溃。
这样做的好处
会话管理很省心
最大的好处就是不用自己管理登录状态了。浏览器会自动处理Cookie,只要登录一次,后续请求都会带着登录信息。而且和真实用户操作完全一致,很多复杂的认证机制(比如多因素认证、动态Token)都能自动处理。
减少识别为爬虫或者恶意刷借口
我加了一些简单的措施:随机延迟random.uniform(1, 3)
避免请求太频繁,设置了标准的User-Agent和HTTP头部。因为用的是真实的浏览器环境,比那些模拟HTTP请求的脚本更像真实用户,被识别的概率低很多。
核心代码实现
以下是Playwright自动化浏览器技术的核心实现流程
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
| class APIDataCollector: def __init__(self): self.config = self._load_config() self.setup_directories()
def run_collection(self): """执行数据采集的核心流程""" with sync_playwright() as p: browser = p.chromium.launch(headless=False) context = browser.new_context( user_agent="Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36", extra_http_headers={ "Accept": "application/json, text/plain, */*", "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", } ) page = context.new_page()
self._ensure_login_status(page)
for target in self._get_target_list(): self._random_delay()
api_data = self._fetch_api_data(page, target)
if api_data: self._parse_and_save_data(api_data, target)
browser.close()
def _ensure_login_status(self, page): """确保登录状态,必要时引导用户手动登录""" test_endpoint = self._build_test_url() page.goto(test_endpoint, wait_until="networkidle")
page_content = page.inner_text("body") login_keywords = ["登录", "密码登录", "Login", "signin"]
if any(keyword in page_content for keyword in login_keywords): print("检测到需要登录,请手动完成登录后继续...") input("登录完成后按Enter键继续...")
page.goto(test_endpoint, wait_until="networkidle") page_content = page.inner_text("body") if any(keyword in page_content for keyword in login_keywords): raise Exception("登录验证失败")
def _fetch_api_data(self, page, target): """通过浏览器获取API数据""" try: api_url = self._build_api_url(target)
page.goto(api_url, wait_until="networkidle")
response_text = page.inner_text("body")
if self._is_login_required(response_text): self.log("检测到会话过期,跳过当前请求") return None
return self._parse_api_response(response_text)
except Exception as e: return None
def _parse_api_response(self, response_text): """解析API响应数据""" try: json_data = json.loads(response_text)
if json_data.get("status") == "success" and "data" in json_data: return json_data["data"].get("items", []) else: return None
except json.JSONDecodeError: return None
def _parse_and_save_data(self, api_data, target_info): """解析并保存数据到文件""" if not api_data: return
file_exists = os.path.exists(self.output_file)
with open(self.output_file, "a", encoding="utf-8", newline="") as f: writer = csv.writer(f)
if not file_exists: headers = self._generate_csv_headers() writer.writerow(headers)
for item in api_data: row_data = self._extract_data_fields(item, target_info) writer.writerow(row_data)
def _extract_data_fields(self, item, target_info): """从API响应中提取需要的数据字段""" return [ item.get("field1", ""), item.get("field2", ""), item.get("field3", ""), target_info, item.get("metadata", {}).get("subfield1", ""), ]
def _load_config(self): pass def setup_directories(self): pass def _get_target_list(self): pass def _build_test_url(self): pass def _build_api_url(self, target): pass def _is_login_required(self, content): pass def _random_delay(self): pass def _generate_csv_headers(self): pass
|
什么时候能用这个
比较适合的场景
这个方法特别适合那些需要登录态的数据获取任务:
- 单页应用数据抓取:很多前端框架渲染的页面,直接HTTP请求拿不到数据
- 复杂认证的API:需要多步验证、动态Token之类的接口
- 需要浏览器环境的接口:有些API会检查浏览器环境或者执行JavaScript
一些局限性
当然这个方法也有自己的限制:
- 性能开销大一些:要启动完整的浏览器,内存和CPU消耗都比直接HTTP请求高
- 速度相对慢:页面加载需要时间,不如直接API请求快
- 依赖浏览器环境:如果目标网站的页面结构变了,可能需要调整代码
- 资源消耗:同时处理大量任务时,浏览器实例会占用不少系统资源
总的来说,如果你的主要目标是稳定获取登录态保护的数据,而且对性能要求不是特别苛刻,这个方法还是很实用的。