☰
Playwright+Asyncio实战:批量采集LinkedIn公司信息的异步方案
2026/10/21 19:31:22 网站建设 项目流程

前阵子有个做企业数据服务的朋友找上我,说他们需要批量整理LinkedIn上公司主页的信息。单日大约2000家左右,量不算夸张,但要求很明确:字段要全、采集要稳、尽量不要触发对方的风控机制。我第一反应是直接用requests加正则去解析页面,给人家省点机器资源,结果连登录这一关都过不了——LinkedIn早就不是静态HTML的时代了,公司资料大部分是异步渲染,加上动态Token和请求头校验,传统方法根本轮不到发挥就已经被挡在门外。最后我把方案定在“Playwright + Asyncio”这个组合上,实测下来比预期好不少。

这方案一句话概括就是:Playwright负责模拟真实浏览器、渲染完整页面并提取数据,Asyncio负责把一个一个的页面访问变成可控的并发任务,两者配合起来,一台普通机器十分钟可以跑完几百家公司的公开资料采集。本文会把方案选型、环境搭建、核心代码、并发控制、踩坑记录全部整理出来,给想做类似企业数据采集的朋友做个直接可参考的实战模板。

1. 项目背景与方案选型

1.1 这个任务真正要解决的问题

很多人误以为“抓个公司信息”是一件很简单的事,真正上手才发现难点集中在四个方向:数据动态渲染、登录态维护、并发效率、风控规避。这四个问题不解决,代码跑起来就是要么抓不到数据,要么抓了几条就被网站限制访问。

LinkedIn的公司主页和普通资讯页还不一样,它内部的DOM结构非常复杂,公司简介、规模、行业、总部地址、成立时间、员工数这些字段分布在不同的区块里。直接在requests拿到HTML源码,看到的只是一堆框架代码,真正的数据要等浏览器执行完JavaScript才会出现在DOM树上。所以第一个前置条件就是选择一个能驱动真实浏览器的自动化框架。

第二个痛点是登录态。LinkedIn对未登录用户的限制极严,公司主页的字段会大量缺省,甚至整个页面被引导到登录墙。用爬虫去自动注册账号不现实,最稳妥的方式就是人工登录一次,把Token和Cookie保存下来,后续运行全部复用。这个操作放在Playwright里就是storage_state的保存与注入。

第三个是效率。单线程一次开一个页面,2000家要跑三四个小时,而且全程都在机械等待。Asyncio的异步非阻塞特性可以把等待时间叠加利用,一个浏览器上下文里同时挂着多个页面访问任务,效率能提升3到5倍。

1.2 为什么不选Selenium或者Puppeteer

同类方案里Selenium是出道最早的,很多人还在用它写自动化脚本。Selenium的问题在于API设计偏老、依赖WebDriver中间层,启动浏览器慢,并发场景下稳定性一般。Puppeteer本身很优秀,但它绑定Node.js,和Python生态集成要额外套一层桥接,维护成本高。

Playwright的优势恰恰是补齐了这两块的短板:官方支持Python异步API,browser.new_context()可以轻松创建独立浏览器上下文,隔离性强且无需额外驱动文件。它自带的auto-wait机制能自动等待元素可交互,这种WebDriver协议时代的痛点在这里被直接消化掉了。

我贴一下我当时用下来的对比感受:

维度PlaywrightSeleniumPuppeteer
启动速度快,秒级慢,依赖驱动快,但需Node环境
异步支持原生Asyncio API需要自己套线程池原生异步
浏览器上下文隔离强,天然支持多实例弱中等
元素智能等待强弱强
社区生态快速增长最成熟中等

从“2025年还在新开工的爬虫项目”角度来看,选Playwright几乎是共识。它不是没有缺点,新版本API变动略频繁,升级有大版本要注意,但比起它带来的开发效率提升,值得接受。

1.3 环境准备与初始化

我用的环境是Python 3.10以上的版本,Windows和Linux都跑过,推荐Linux服务器部署,内存占用更可控。安装命令很简单:

pip install playwright playwright install chromium

第一行安装库本体,第二行下载Chromium浏览器内核。这里提醒一点:Chromium体积比较大,如果服务器网络不好,可以先在本地执行一次,然后把浏览器缓存目录直接打包传到服务器。

Playwright提供两套API:同步版sync_api和异步版async_api。既然标题都写了Asyncio,核心代码一定用异步版,但搭建脚手架时同步版更容易验证环境:

from playwright.sync_api import sync_playwright with sync_playwright() as p: browser = p.chromium.launch(headless=False) page = browser.new_page() page.goto("https://example.com") browser.close()

跑通这段后,环境就算准备好。后续所有代码我会切换到异步版本,这是并发的基础。

2. 数据字段与页面结构拆解

2.1 公司主页的数据块划分

在写代码之前,花二十分钟把目标页面摸透是非常值的。LinkedIn公司主页的信息大概可以分成四块:

  • 顶部基础区:公司名称、一句话简介、行业、公司规模、公司类型、总部地址、成立时间
  • 右侧信息卡:公司主页链接、所属行业标签、电话邮箱(部分可见)
  • About区块:详细的公司介绍文本,字数几百到上千不等
  • 数据统计区:员工总数、分布地区、同类公司推荐等

这四块内容并不都在同一个可滚动区域内一次性渲染。关于员工数、分布地区这类数据,经常是页面滚动到对应区块时才触发异步请求补全。所以字段提取和页面滚动策略必须连在一起设计。

我建议用一段脚本先把页面所有可见文本导出,做一次“字段摸底”:

async def dump_text(page): text = await page.locator("body").inner_text() lines = [l.strip() for l in text.split("\n") if l.strip()] for line in lines: print(line)

从输出里基本就能看到哪些数据在DOM里,哪些还在“持续加载”状态。这个摸底动作一定不能省,它直接决定后续定位器的写法。

2.2 定位器的选择与验证

定位元素有CSS选择器、XPath、文本定位等多种方式,刚上手时最容易犯的错是照抄浏览器里复制出来的XPath。那种XPath通常长且脆弱,只要网站结构有一次小改版,整个脚本就废了。

我定位LinkedIn公司页面字段时,核心原则是“别太精确,但要稳定”。比如公司名称,没有稳定的id属性,但它在DOM里有着相对统一的层级位置。我用的是:

company_name = await page.locator("h1[dir='auto']").first.inner_text() overview = await page.locator("div.company-overview").inner_text()

h1[dir='auto']这个写法利用了LinkedIn里多语言文本排版方向的通用属性,比直接用h1之类宽泛选择器可靠得多,也比一长串div > div > section > div可靠得多。

行业、规模这些字段在页面里经常是以“标签键值对”形式出现的,结构近似:

<dt>行业</dt> <dd>软件开发</dd> <dt>公司规模</dt> <dd>1000-5000人</dd>

这类键值对用XPath根据左侧文本找右侧值非常舒服:

async def get_kv_value(page, key): locator = page.locator(f"dt:has-text('{key}') + dd") if await locator.count() > 0: return await locator.first.inner_text() return None

2.3 滚动加载与等待策略

公司页面的员工数据、更多介绍都是在滚动后才出现,而Playwright最优雅的地方就是它的智能等待。我直接用scroll_into_view_if_needed配合wait_for_selector,让关键元素先真实进入屏幕,再提取文本:

await page.locator("section.about").scroll_into_view_if_needed() await page.wait_for_selector("text=员工总数")

注意这里的选择器text=员工总数是Playwright特有语法,不能和普通page.query_selector混用。等待时间不用写死,Playwright默认会轮询30秒,只要元素出现就立即继续执行,这种“按需等待”机制比固定sleep(3)高端太多,也稳定太多。

3. 核心代码实战:登录态与主流程

3.1 Cookie的获取与注入

直接免登录跑触发登录墙的频次太高,所以我选择“人工登录一次、长期复用Cookie”的方案。整个流程三步走:第一步,用非无头模式手动打开页面登录;第二步,把登录后的浏览器上下文状态导出成文件;第三步,后续每次启动直接注入状态文件。

登录并保存状态的代码:

import asyncio from playwright.async_api import async_playwright async def save_state(): async with async_playwright() as p: browser = await p.chromium.launch(headless=False) context = await browser.new_context() page = await context.new_page() await page.goto("https://www.linkedin.com/login") print("请在弹出的浏览器中手动登录,登录完成后此窗口将自动关闭") await page.wait_for_selector("text=首页", timeout=180000) await context.storage_state(path="linkedin_state.json") await browser.close() asyncio.run(save_state())

这里有个细节:导出的是全套storage_state,包括Cookie、localStorage、sessionStorage。注入的时候用browser.new_context(storage_state="linkedin_state.json"),等于完整还原登录现场。

3.2 公司信息提取主流程

公司信息提取是整个爬虫的核心,我把单个公司的采集逻辑封装成一个异步函数:

async def fetch_company(browser, company_url): context = await browser.new_context( storage_state="linkedin_state.json", viewport={"width": 1366, "height": 900}, user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64)" ) page = await context.new_page() data = {"url": company_url} try: await page.goto(company_url, wait_until="domcontentloaded", timeout=30000) await page.wait_for_timeout(1500) company_name = await page.locator("h1[dir='auto']").first.inner_text() data["name"] = company_name.strip() await page.locator("section.about").scroll_into_view_if_needed() await page.wait_for_load_state("networkidle", timeout=15000) body_text = await page.locator("body").inner_text() data["overview_text"] = extract_overview(body_text) data["industry"] = await get_kv_value(page, "行业") data["size"] = await get_kv_value(page, "公司规模") except Exception as exc: data["error"] = str(exc) finally: await context.close() return data

主流程有几个关键点可以展开讲。

每个公司独立一个context。这是Playwright最有价值的能力之一。多页面之间如果共享同一个context,Cookie、User-Agent、接口请求头完全一致,反而容易暴露批量访问的特征。每次new_context造一个全新环境,指纹隔离更彻底。

先回首页再进入目标页?有的实践者会在访问每个公司页面前先跳到LinkedIn首页,模拟真实用户从推荐流点进去的行为,这确实能降低异常访问的概率,代价是每家公司要多花两三秒。我用的是直接访问,效率优先,如果发现风控敏感再调整。

提取overview_text时不要用脏正则。公司的About区块里如果包含正则里的特殊字符,处理起来会很烦。我的做法是拿到body全文后,用标记文本切出段落,再清洗换行和多余空格。这个清洗函数不难写,但值得花时间打磨,因为后续做数据分析或喂给大模型,字段清洁度直接影响结果质量。

3.3 访问节奏与低特征策略

并发高不代表可以乱来。真实用户不会在几秒内连续打开几十个公司页面,单浏览器上下文的行为如果太规律,很容易被判定为非人工操作。我在代码里加了随机等待和时间扰动:

import random random_delay = random.uniform(1.5, 4.0) await page.wait_for_timeout(int(random_delay * 1000))

这个延迟放在每个公司页面加载完成之后、数据提取之前。延迟时间我做过测试:完全不延迟跑300家,中途被要求验证的频率明显上升;加上1.5到4秒的随机等待之后,300家全程无干扰。

另外User-Agent我强调一下:不要只用Playwright默认的UA串,它的标识性太强。自己构造一段真实Chrome浏览器的UA字符串,版本号要和Chromium版本接近,否则会出现版本不匹配的怪异特征。稳定的低特征访问会在后面第5部分再展开。

4. Asyncio并发与性能调优

4.1 并发模型的设计思路

Asyncio的玩法不是简单地把N个协程丢进asyncio.gather里就完了,核心要解决的是资源上限和异常隔离。

我的实现思路是:维护一个公司URL列表,用信号量控制最大并发数,每个协程负责一个公司的所有操作。信号量的设计非常关键,它防止同时创建的浏览器Context数量超过系统资源承载上限。

import asyncio from playwright.async_api import async_playwright semaphore = asyncio.Semaphore(5) async def worker(company_url): async with semaphore: async with async_playwright() as p: browser = await p.chromium.launch(headless=True) data = await fetch_company(browser, company_url) await browser.close() return data async def main(): urls = load_urls_from_file("company_urls.txt") results = await asyncio.gather(*(worker(url) for url in urls)) save_results(results) if __name__ == "__main__": asyncio.run(main())

注意到每个worker内部都重新启动了一个Playwright实例吗?这里很多初学者会觉得“重复启动浏览器太浪费”,但真实测试下来,启动浏览器一次平均只需要800毫秒左右,相比单家公司页面加载的3到5秒,开销占比不算高。而换来的好处是浏览器实例级别的完全隔离,某个worker崩溃不会影响其他worker。

如果你觉得重复启动浏览器太浪费,也可以改为在main里启动一个浏览器,多个worker共用。但共用一个浏览器实例时,每个page共享进程资源,并发稍高就容易出现内存暴涨。我实测5个并发用独立浏览器方式跑,平均每个worker占用内存在200MB上下,整机16G内存可以舒服地跑满。

4.2 超时与重试机制

单次网页访问不可能永远成功,网络抖动、服务端限制、页面渲染失败都是家常便饭。所以重试机制必须有,而且要区分错误类型,属于临时网络问题的重试,属于数据不存在的直接跳过。

我定义一个重试封装:

async def fetch_with_retry(browser, url, retries=3): for attempt in range(1, retries + 1): try: data = await fetch_company(browser, url) if data.get("error") is None: return data except Exception as exc: print(f"第{attempt}次尝试失败: {exc}") await asyncio.sleep(2 ** attempt) return {"url": url, "error": "重试多次仍失败"}

这里用了指数退避策略:第一次失败后等2秒,第二次等4秒,第三次等8秒。指数退避比固定间隔更符合服务端对异常访问的判断预期,既给了服务端喘息空间,也让重试行为看起来更接近用户手动刷新页面。

超时设置也要分级。页面跳转超时我设30秒,元素等待超时设15秒,网络空闲等待超时设15秒。三个超时埋在fetch_company函数里的不同位置,对应的失败原因各不相同,排查时可以精准定位。如果全局只设一个大超时,网络卡顿和数据缺失会相互掩盖,调试效率会很低。

4.3 性能实测与数据落库

我自己在本地环境跑过一组数据:500个公司URL,5并发,每家公司平均延迟3.2秒,整体耗时约9分钟,成功率96.4%,剩余3.6%的失败主要来自页面加载超时和字段缺失。如果换成10并发,时间能压到6分钟以内,但失败率会小幅上升。所以并发数并不是越大越好,5是稳妥线,10是激进线。

数据落库我直接用SQLite。轻量、单文件、不用额外起服务,对2000条级别的数据量完全够用。写库用标准sqlite3即可,关键字段按照采集前规划好的表结构:

import sqlite3 def init_db(): conn = sqlite3.connect("linkedin_companies.db") conn.execute(""" CREATE TABLE IF NOT EXISTS companies ( id INTEGER PRIMARY KEY AUTOINCREMENT, url TEXT, name TEXT, industry TEXT, size TEXT, overview TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) conn.commit() return conn

注意把所有数据攒到内存里,最后一次性批量写入,避免每抓一条就开一次数据库连接。异步代码里如果同时处理数据库写入,千万注意不要让同步数据库操作阻塞事件循环,否则会拖垮整体并发性能。我在实际项目里把写库的IO也封装成异步线程执行,效果稳定。

5. 常见问题排查与实操心得

5.1 被识别为自动化工具的典型特征

在爬取过程中,网站发现异常访问时常见表现有三种:跳转到验证页面、响应内容被替换为空壳、接口开始随机返回403。这些情况多数不是某一个单一因素造成的,而是“无头浏览器特征 + 请求指纹规律 + 访问频率异常”叠加导致。

我的调试做法是把headless=False先跑一次,窗口开着观察行为是否正常。发现被验证后,逐个检查三个点:

  • User-Agent是否和真实浏览器一致
  • 是否缺失了某些常见的浏览器指纹信息
  • 并发量是不是太大了

结合标题里“高效获取”这个诉求,我额外给一个经验:不要用过高的并发去硬刚,要相信限速的价值。真实用户的阅读速度平均一分钟两三个页面,稍微高于这个值是可以接受的,但如果是每秒钟几十个页面,任何人都看得出来是爬虫。最好的状态是:植入合理的延迟,模拟人类操作的不规律性,同时保证整体耗时在业务可接受范围内。

5.2 动态加载元素提取不到怎么办

公司页面里某些区块是滚动到视野内才触发加载的,直接inner_text有可能拿到空字符串。我前面提过用scroll_into_view_if_needed(),但这个方法偶尔也会失效,因为它依赖元素已经存在于DOM里,只不过未在视野内。

真正还没出现在DOM树里的元素,要用Playwright的wait_for_selector配合轮询:

try: await page.wait_for_selector("text=员工总数", timeout=10000) except: pass

这个写法的好处是即使元素最终未出现,也不会直接抛异常,而是走一个静默跳过。主流程的稳定性就是这样一点一点磨出来的。

5.3 日志模块就是爬虫的“行车记录仪”

跑批一旦出现异常,没有日志基本等于盲人摸象。很多人的爬虫脚本从头到尾只有print,跑完看控制台一片乱麻。我自己维护的这套爬虫,从最开始就引入标准logging模块,分层记录信息、警告、错误。

在协程进入和退出、每家公司提取完成、每次重试触发时都记录一条带时间的日志。事后排查时,哪家公司出了什么问题,为什么重试,整个链路清清楚楚。日志内容里同时记录用了多少秒,这个数据积累久了就是一份性能报告,能直观看到哪个环节最耗时。

5.4 数据完整性的校验

2000家公司采集下来,最终要能经得起业务方验收。我的做法是建立“必填字段完整率”指标。公司名称、行业、规模这三个字段属于核心字段,任何一项缺失都视为不合格数据,单独输出到一个问题清单文件里,方便人工补采。

跑批结束自动统计:

def check_completeness(row): required = ["name", "industry", "size"] missing = [k for k in required if not row.get(k)] return len(missing) == 0

数据完整性校验既是自检也是交付质量证明,哪怕不是每一行都完美,但至少能明确告诉业务方哪些数据可信、哪些需要人工关注。

结尾

这套Playwright + Asyncio的采集方案,我从最初“能跑就行”,磨到现在的“又稳又快”,中间踩的坑基本都在上面了。要说个人最深的一点体会,那就是自动化采集拼的不是花哨的框架技巧,而是细节的前置思考:字段规划越细,定位器写得越稳,并发设置越克制,整个任务就越像一个有节奏的流水线,而不是一个随时会崩的实验脚本。

最后再分享一个小习惯:每次批量跑完,我都会把采集到的数据抽样几条,人工打开对应网址比对一遍。别小看这个动作,它能让你在问题变大之前就发现隐患。毕竟数据结构一变,代码里的定位器可能全部作废,如果没有抽检机制,你可能在错误的数据上跑了一个月还毫无察觉。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询