做爬虫这行最怕遇到什么不是验证码是那种 URL 往里一怼requests 连响应头都拿不全页面内容全靠 JavaScript 现场渲染的站点。你翻遍返回的 HTML 找到的只有一堆 script 标签和空壳 div。Chrome 扩展商店就是这类页面里很典型的一个目标——列表无限滚动详情页动态渲染每个扩展还嵌套着一堆异步接口。这篇实战笔记就是记录我用 Playwright 硬啃 Chrome 扩展商店的全过程从环境搭建、页面分析、字段解析到 SQLAlchemy 落库、增量去重、并发加速最后把常见坑一次性列清楚。适合已经会用 requests 写简单爬虫、想进阶动态渲染抓取的开发者参考。在动手之前先把规矩说清楚这个项目只采集公开页面上的扩展元数据不是去破解任何接口或者搞批量滥用。请求频率我会控制在很保守的水平代码里也会强制加随机延时。你要拿去商用或者大规模采集务必先读目标站点的服务条款和 robots 协议出了合规问题别怪我没提醒。技术本身是中性的用在哪、怎么用是我们自己得负责的事。1. 为什么盯上 Chrome 扩展商店数据价值与目标拆解1.1 扩展商店里的数据到底有什么用Chrome 扩展商店可能是很多人忽略的一个高质量数据源。作为最大的浏览器扩展分发渠道之一它上面每一个扩展都携带了非常完整的结构化信息扩展名称、简介、评分、评分人数、累计用户数、开发者名称、当前版本、最近更新时间、隐私权属声明、所需权限、分类、官网链接等等。把这些字段拼起来能做的事情非常多。从产品角度看可以按分类统计扩展市场的用户规模分布看哪些细分赛道已经拥挤、哪些还空着从市场分析角度看可以追踪某个竞品扩展的用户量变化曲线判断它最近一次更新后评分是涨是跌从安全研究角度看可以通过权限列表寻找那些只要一个很小功能却申请了海量权限的可疑扩展。标题里提到的千万级数据并不夸张商店里扩展数量庞大如果再把历史版本、权限变更、评论内容这些维度加进来单个扩展就能衍生出几十条甚至上百条记录。但千万级是最终目标不是一上来就闷头跑全量先把单条数据抓深抓稳再谈扩展规模。1.2 为什么不用 requests 硬怼很多人拿到这类页面的第一反应是打开 DevTools 找 XHR 接口找到那个返回 JSON 的 URL然后用 requests 直接请求。我也这么干过但实际情况是Chrome 扩展商店的页面容器非常重列表页的滚动加载数据确实来自异步请求可这些请求携带了复杂的动态参数和令牌签名逻辑藏在压缩后的 JS 代码里调试成本极高。而且一旦请求方式跟页面实际行为不一致响应内容随时可能变成一段带验证逻辑的 HTML。换个思路看问题我要的数据最终都渲染在 DOM 上了那与其逆向接口不如直接驱动一个真实浏览器去取渲染完成后的结果。这样做的优点是逻辑直观——你眼睛看到什么代码就能抓到什么缺点是需要承担浏览器进程的开销但在批量抓取场景下这个代价完全值得。Playwright 就是冲这个场景来的。1.3 Playwright 比 Selenium 强在哪Selenium 用了这么多年生态成熟但用起来总有一种老派的感觉。首先是 driver 管理麻烦Chrome 一升级chromedriver 版本对不上就开始报错其次定位和等待的 API 设计得绕显式等待写起来啰嗦。Playwright 最大的优势是把这一切收敛了pip install playwright一条命令装完再playwright install chromium就能拉起浏览器不用单独维护 driver。页面操作用的是天然带自动等待的 Locator API而且原生支持异步写并发比 Selenium 顺手太多。另外一个我特别喜欢的功能是playwright codegen。它可以记录你在页面上的真实操作自动生成对应的选择器和代码分析目标页面结构的时候效率非常高。后面讲页面分析时我会再细说。2. 环境准备与目标页面拆解2.1 安装 Playwright 和浏览器内核老规矩先建虚拟环境再装依赖免得把系统 Python 搞乱python -m venv venv source venv/bin/activate # Windows 上执行 venv\Scripts\activate pip install playwright sqlalchemy pandas playwright install chromium这里说明几个关键点。playwright install chromium只会装 Playwright 自己管理的 Chromium 内核跟系统里的 Chrome 互不干扰这样版本可控也避免项目跑到服务器上时找不到浏览器。如果你希望 Playwright 直接驱动已经安装好的正式版 Chrome可以在启动时传channelchrome比如p.chromium.launch(channelchrome)但不是必须的默认的 Chromium 已经完全够用。如果部署到 Linux 服务器上还需要装浏览器运行依赖playwright install-deps chromium这一步是为了解决类似缺少 libnss3 这些底层库的问题。另外几个我常加的启动参数后面代码里再给。2.2 商店页面结构分析Chrome 扩展商店现在的域名已经统一到chromewebstore.google.com老域名chrome.google.com/webstore会自动跳转。列表页和搜索页都是典型的无限滚动设计页面底部有一个加载触发器滚到那里就会通过异步请求追加新的扩展卡片但卡片是渲染成 DOM 的所以我们只需要模拟滚动然后从 DOM 里取链接和卡片信息。详情页的 URL 结构通常是这样的https://chromewebstore.google.com/detail/扩展名slug/一串32位十六进制ID这串 32 位 ID 是扩展在商店内部的主键天然适合作数据库的主键。所有详情页链接都可以从列表页卡片元素的a[href*/detail/]选择器里提取。用playwright codegen可以快速确认关键元素playwright codegen https://chromewebstore.google.com/打开工具后点击任意一个扩展进入详情页codegen 会实时把对应的选择器生成在右侧面板里。比如标题元素通常是h1评分在div[aria-label*评分]里用户数在包含位用户的文本节点中。这些信息不是一成不变的建议每次项目开始前花十分钟重新确认一遍别守着一年前的选择器跑今天的页面。2.3 核心字段建模在开始写爬虫之前先把要抓的字段列清楚方便后面设计数据库表。我这次最终确定的字段如下字段名来源示例目标类型扩展 IDURL 中的 32 位串字符串主键名称页面h1字符串简介详情页描述段文本评分评分区域文本如 4.5/5浮点数评分人数评分区括号里的数字整数累计用户数1,234 位用户整数开发者开发者栏链接文本字符串当前版本版本号文本字符串更新日期详情信息里的日期日期类型分类分类标签字符串权限列表查看权限弹层中的列表JSON 文本官网链接相关链接字符串抓取时间系统当前时间时间戳这里有个小经验抓取之前先在本地跑三五个详情页把所有文本导出成纯文本看一看实际格式再做清洗逻辑。因为页面上可能把用户数显示成1.2万位用户或1,234 位用户不同地区语言环境下格式完全不同。我一律通过设置localezh-CN来保证页面返回简体中文清洗规则相对固定。3. 动态页面抓取核心实现3.1 初始化浏览器上下文Playwright 里有一个容易被新手忽略的概念Browser Context浏览器上下文。它相当于一个独立的浏览器会话cookie、缓存、localStorage 都是隔离的。多个任务并行时让每个任务使用独立 context可以避免登录状态、地理位置这些脏数据互相污染。我通常会封装一个初始化函数from playwright.sync_api import sync_playwright def init_browser(playwright): browser playwright.chromium.launch( headlessTrue, args[--disable-dev-shm-usage], ) context browser.new_context( user_agent( Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36 ), viewport{width: 1920, height: 1080}, localezh-CN, ) page context.new_page() page.set_default_timeout(15000) return browser, context, page解释几个参数headlessTrue保证在服务器上稳定运行调试时可以临时改成 False 看实际渲染过程--disable-dev-shm-usage是容器环境里非常关键的参数避免/dev/shm太小导致 Chromium 崩溃viewport 设置成 1920x1080 是因为有些懒加载逻辑会判断你看到了哪里窗口太小可能导致页面认为某些卡片不在视野内而不加载localezh-CN则让页面输出中文字段解析时统一用位用户、分这些关键字。3.2 列表页滚动加载的稳定实现无限滚动页面的通关口诀是一边滚动一边数卡片连续几次没有新卡片出现就停止。我的实现是这样的def scroll_list(page, max_scrolls60): prev_count 0 stable_rounds 0 for _ in range(max_scrolls): cards page.locator(a[href*/detail/]) count cards.count() page.mouse.wheel(0, 1200) page.wait_for_timeout(1200) if count prev_count: stable_rounds 1 if stable_rounds 3: break else: stable_rounds 0 prev_count count return page.locator(a[href*/detail/]).count()注意几个细节。第一滚动用的是page.mouse.wheel(0, 1200)而不是page.evaluate(window.scrollTo(...))。虽然两者都能触发滚动但鼠标滚轮事件更接近真实用户行为很多页面会根据事件类型决定是否加载后续内容用鼠标滚轮踩坑少。第二每次滚动后等待 1200 毫秒给异步加载留下时间等待太短会漏数据等待太长会拖慢整体进度。第三连续 3 轮没有新增就退出这个阈值可以根据网络情况调整网络差的站点可能需要连续 5 轮才算真到底。如果你抓的是某个分类页可以在 URL 上追加排序参数比如按评分最高或用户数最多排序这样同一页的扩展质量更高也更容易控制数据规模。3.3 提取详情页链接滚动结束后把所有详情页链接一次性取出来links page.locator(a[href*/detail/]).evaluate_all( els els.map(e e.href) ) unique_links list(dict.fromkeys(links))这里用evaluate_all直接在浏览器上下文里把 href 全部映射出来比 Python 端逐个get_attribute快很多。dict.fromkeys去重保序比set更能保持原始顺序调试时方便比对。从链接里提取扩展 ID 用正则import re def extract_ext_id(url): m re.search(r/detail/(?:[^/]/)?([a-z0-9]{32}), url) return m.group(1) if m else None这个正则可同时兼容新旧两种 URL 格式。新版 URL 里扩展名和 ID 之间有个斜杠旧版 URL 后面直接跟 ID(?:[^/]/)?正好处理了可能有扩展名的这一层。如果有的链接不是标准 32 位 ID正则取不到就直接返回 None后面做任务列表时可以把异常链接筛掉。3.4 详情页字段抓取实现详情页是整条链路里最需要耐心的地方。同一个字段在不同扩展页上可能结构完全不一样有的扩展没有官网有的权限列表为空有的评分人数为 0。所以我建议写一个高度防御性的解析函数每个字段都走 try/except拿不到就用默认值。from playwright.sync_api import TimeoutError as PlaywrightTimeoutError def safe_text(page, selector, default): try: return page.locator(selector).first.inner_text().strip() except PlaywrightTimeoutError: return default def parse_detail(page, url): page.goto(url, wait_untildomcontentloaded) page.locator(h1).first.wait_for() name safe_text(page, h1) desc safe_text(page, div[data-blog-componentextension-description]) rating_text safe_text(page, div[aria-label*评分]) users_text safe_text(page, div:has-text(位用户)) developer safe_text(page, a[href*devsite] span) return { id: extract_ext_id(url), name: name, description: desc, rating: parse_rating(rating_text), rating_count: parse_count(rating_text), users: parse_count(users_text), developer: developer, url: url, }这段代码有几个经验点想单独说。一是page.locator(...).first很实用。一个页面里经常有多个元素匹配同一个选择器比如页面头部和页脚可能各有一个扩展名称用.first直接取页面里第一个避免strict mode violation报错。二是所有文本取出后必须strip()。DOM 里文本经常带前后换行和空格尤其描述区块不清理的话存进数据库之后做关键词检索会很难受。三是wait_untildomcontentloaded比默认的load更快。扩展商店详情页会加载大量脚本和埋点资源等全部 load 可能要等好几秒而我们要的 DOM 结构在 domcontentloaded 之后基本已经就绪后面再用locator(h1).first.wait_for()保证标题渲染完成这样速度和安全兼得。3.5 用户数与评分的清洗函数页面上的文本格式大致是这样评分区域评分 4.5/5 1,234 条评分用户数1,234 位用户更新时间2024年5月20日清洗函数直接写import re def parse_count(text): nums re.findall(r[\d,], text) if not nums: return 0 return int(nums[0].replace(,, )) def parse_rating(text): m re.search(r(\d(?:\.\d)?)\s*/\s*5, text) return float(m.group(1)) if m else 0.0parse_count会从文本里取出第一段数字比如 1,234 位用户 取出 1,234 转成 1234。需要注意如果页面上出现了 1.2万位用户这段逻辑就处理不了所以我才强调要通过localezh-CN统一格式。假如你抓到的商店版本开始返回中文数字单位就得再补一段 万 和 亿 的换算逻辑。3.6 权限列表的抓取权限列表藏在查看权限按钮后面。我的方案是用expect_popup处理弹窗def fetch_permissions(page): try: with page.expect_popup(timeout8000) as popup_info: page.get_by_role(button, name查看权限).click() popup popup_info.value popup.wait_for_load_state() perms popup.locator(li, [class*permission]).all_inner_texts() popup.close() return [p.strip() for p in perms if p.strip()] except Exception: return []这里的逻辑是如果点按钮后打开的是新标签页expect_popup会准确捕获它如果商店界面改版后变成展开式面板这个函数会超时我们直接返回空列表后续再针对新界面调整。实际开发中我遇到过按钮点击被底部的 cookie 弹窗遮住的情况如果点击失败可以先按一下Escape或关闭弹窗组件再点。这个细节在代码里加个try就能覆盖。4. 数据落库SQLAlchemy 模型与增量更新4.1 建表与模型设计数据清洗完总要落库我用 SQLAlchemy 管理数据库模型。为什么不用裸 SQL因为后期字段要加索引、要改类型用 ORM 迁移起来省事。模型这样定义from sqlalchemy import ( create_engine, Column, String, Float, Integer, Date, Text, DateTime, func ) from sqlalchemy.orm import declarative_base, sessionmaker Base declarative_base() class ChromeExtension(Base): __tablename__ chrome_extensions id Column(String(32), primary_keyTrue) name Column(String(255), nullableFalse) description Column(Text, default) rating Column(Float, default0.0) rating_count Column(Integer, default0) users Column(Integer, default0) developer Column(String(255), default) version Column(String(64), default) updated_date Column(Date, nullableTrue) category Column(String(128), default) permissions Column(Text, default[]) homepage Column(String(512), default) crawled_at Column(DateTime, server_defaultfunc.now(), onupdatefunc.now()) DATABASE_URL sqlite:///extensions.db engine create_engine(DATABASE_URL, echoFalse) Base.metadata.create_all(engine) Session sessionmaker(bindengine)主键直接用扩展 ID有两个原因。第一ID 天然唯一不会出现同一个扩展因为 URL 多一个参数被重复插入第二后期做增量抓取时用主键判断这个扩展是否已存在非常快。permissions字段我存的是 JSON 序列化后的字符串读出时再用json.loads还原这样既兼容 SQLite 这类轻量库又保留列表结构。4.2 幂等写入去重、更新与 merge增量抓取的核心目标是跑第二次时不会产生脏数据。最简单的幂等写法是def save_item(session, item: dict): existing session.get(ChromeExtension, item[id]) if existing: for key, value in item.items(): setattr(existing, key, value) else: session.add(ChromeExtension(**item)) session.commit()这个写法逻辑清楚查询主键存在就逐字段更新不存在就新增。但如果你追求代码更短可以直接用 SQLAlchemy 的mergedef save_item_merge(session, item: dict): session.merge(ChromeExtension(**item)) session.commit()merge的行为是如果主键存在执行更新否则插入。实现上一行搞定但我个人更推荐前一种写法因为更新时你可以控制哪些字段需要覆盖比如第一次抓取时某些字段为空第二次抓取补全了只更新有实际值的字段避免别人声明的数据被空字符串冲掉。4.3 断点续爬与任务队列动辄几万个详情页不可能一次跑完。中断恢复是必须考虑的问题。我常用的方案是维护一个简单的任务列表crawled_ids {r[0] for r in session.query(ChromeExtension.id).all()} todo [] for url in unique_links: ext_id extract_ext_id(url) if ext_id and ext_id not in crawled_ids: todo.append(url)跑完之后把todo清空。如果中途断网、进程被杀重跑一次脚本只会处理没抓过的 URL。更稳妥的工程化做法是建一张独立的任务表记录每个 URL 的状态pending / done / failed配合失败重试。但如果只是个人分析用内存集合加数据库主键判断完全够用没必要为一个小项目引入 Celery。4.4 并发加速与限速单页面串行抓取肯定慢但也不能直接起 20 个浏览器进程硬冲。我的经验是用 Playwright 的异步 API asyncio.Semaphore控制并发数。import asyncio import random from playwright.async_api import async_playwright async def fetch_one(sem, browser, url): async with sem: page await browser.new_page() try: await page.goto(url, wait_untildomcontentloaded) await page.locator(h1).first.wait_for() name await page.locator(h1).first.inner_text() # 这里复用前面解析函数中的文本提取逻辑 await page.wait_for_timeout(random.randint(800, 2000)) return {id: extract_ext_id(url), name: name.strip(), url: url} finally: await page.close() async def run_batch(urls, concurrency5): async with async_playwright() as p: browser await p.chromium.launch(headlessTrue) sem asyncio.Semaphore(concurrency) results await asyncio.gather(*[fetch_one(sem, browser, u) for u in urls]) await browser.close() return results在同一浏览器实例里开多个 page 并行加载比开多个浏览器进程节省资源。Semaphore把并发数限制在 5实测下来速度和稳定性取得一个比较好的平衡。每个请求之间强制加 800 到 2000 毫秒的随机延迟这个延迟既是礼貌也是给目标站点一个喘息空间。有个细节比较重要异步代码里所有 Playwright 调用都要await内部脚本执行、元素定位都是异步的写循环的时候尤其要注意不要漏掉await。如果对 asyncio 不熟也可以用同步 API 加上ThreadPoolExecutor但请务必让每个线程使用独立的BrowserContext别让多个线程共享同一个 page 实例否则会有各种诡异的并发问题。5. 常见问题与排查技巧5.1 出现人机验证或访问受限怎么办扩展商店对自动化流量不是完全无感的。如果你并发开太高、请求频率太密集页面大概率会跳出一个验证页面。我的原则是不跟它硬碰硬直接让程序识别这种状态并主动降温。可以在代码里加一个检测函数def detect_blocked(page): text page.locator(body).inner_text().lower() return unusual traffic in text or 验证 in text一旦检测到验证页面果断停止任务等待 60 到 120 分钟再继续。很多人觉得这是浪费时间但比起账号被封、IP 被拉黑等待是成本最低的选择。实际操作中我还会在每次抓取完 200 个扩展后强制让整个程序睡 10 分钟尽量把单次会话的流量摊平。5.2 页面元素定位选择器失效最好用的排查工具是playwright codegen和 Python 端的page.locator(selector).all_text_contents()。每次升级浏览器或商店改版后选择器都可能变。我的建议是选择器尽量少用绝对路径多用文本特征比如div:has-text(位用户)、a[href*devsite]。文本特征比div#root div.main div.card span这类层级结构稳定得多。如果页面里有多个元素匹配同一个选择器尽量用.first或者通过locator.filter(has_text...)缩小范围。不要对所有定位都靠猜把页面截图打印出来代码里加一行page.screenshot(pathdebug.png)往往比反复试选择器快得多。5.3 抓取时间久了内存占用越来越高浏览器跑久了内存膨胀是很正常的。特别是无限滚动列表页开了一堆标签页、每个详情页还带着一堆资源Python 进程本身可能没事Chromium 子进程却可能吃掉几个 GB。解决办法是给抓取过程按批次拆分。比如每抓 500 个扩展关闭当前 context重新创建一批新的 page。在 Playwright 里这就是一个循环里重新browser.new_context()和context.close()的事。另外及时page.close()是关键别把 page 对象存在列表里一直不释放。5.4 SQLAlchemy 写入报错与数据矛盾几个常见问题提前打个预防针字符串超长会报DataError日期解析失败会报ValueErrorJSON 序列化字段忘记json.dumps会直接写成一个带引号的字符串。我的建议是模型里字符串字段留长一点描述用Text类型所有用户数、评分在进入模型前用解析函数包一层任何异常都给默认值所有列表型字段一律先json.dumps再入库取出用json.loads。如果用的是 SQLite并发写入偶尔会出现database is locked这是因为多个连接同时写库。解决办法是把写入操作收敛到一个连接里或者改用check_same_threadFalse参数。我在项目里保证只有主线程做最终 commit其他 worker 只负责返回数据从根上规避这个锁。5.5 Playwright 安装与运行的环境坑在服务器上最容易遇到三类报错一是Executable doesnt exist说明你还没执行playwright install二是Host system is missing dependencies说明缺系统库执行playwright install-deps即可三是中文内容显示为方块通常是服务器缺少中文字体Debian/Ubuntu 上装fonts-noto-cjk能解决。调试时可以开详细日志观察浏览器行为DEBUGpw:browser* python crawl.py这个日志会输出浏览器每个关键行为节点有时候比 DevTools 还好用。另一个实用命令是playwright show-trace trace.zip抓取时可以开启 tracecontext browser.new_context() context.tracing.start(screenshotsTrue, snapshotsTrue) # 抓取若干页面后 context.tracing.stop(pathtrace.zip)trace 文件包含了页面截图、DOM 快照和网络请求复现偶发问题特别有帮助。6. 最后再分享一点个人的实践体会这个项目跑完一遍我最深的感受是爬虫本身只占整个工作量的三分之一页面分析和数据清洗才是真正磨人的部分。扩展商店看起来结构规范但每个扩展页面都会给你来点意外——有的没有官网有的权限列表空荡荡有的评分人数写成 0还有的把描述挤在一个高度受限的容器里造成内容截断。这些脏数据不会让程序崩溃但会让后续分析结果失真。所以我的建议是先用一两百个扩展把整个流程跑通把解析规则调稳了再考虑全量铺开。虽然标题写的是千万级数据但真正的工程里数据不是抢来的是稳稳当当一点一点攒出来的。这个项目抓下来的数据可以用来做扩展市场画像、做竞品趋势跟踪也可以定期重跑增量更新观察用户数和评分的变化曲线。数据采集只是开始数据经过加工和分析之后才有资格叫资产。最后再补一句。运行这种长耗时任务日志记录一定要做扎实。我习惯每抓完一批打印一行进度序号、扩展名、用户数、耗时。这样半夜任务挂了你也能第一时间从日志里看出它卡在了哪里。稳比快重要。
做爬虫这行最怕遇到什么不是验证码是那种 URL 往里一怼requests 连响应头都拿不全页面内容全靠 JavaScript 现场渲染的站点。你翻遍返回的 HTML 找到的只有一堆 script 标签和空壳 div。Chrome 扩展商店就是这类页面里很典型的一个目标——列表无限滚动详情页动态渲染每个扩展还嵌套着一堆异步接口。这篇实战笔记就是记录我用 Playwright 硬啃 Chrome 扩展商店的全过程从环境搭建、页面分析、字段解析到 SQLAlchemy 落库、增量去重、并发加速最后把常见坑一次性列清楚。适合已经会用 requests 写简单爬虫、想进阶动态渲染抓取的开发者参考。在动手之前先把规矩说清楚这个项目只采集公开页面上的扩展元数据不是去破解任何接口或者搞批量滥用。请求频率我会控制在很保守的水平代码里也会强制加随机延时。你要拿去商用或者大规模采集务必先读目标站点的服务条款和 robots 协议出了合规问题别怪我没提醒。技术本身是中性的用在哪、怎么用是我们自己得负责的事。1. 为什么盯上 Chrome 扩展商店数据价值与目标拆解1.1 扩展商店里的数据到底有什么用Chrome 扩展商店可能是很多人忽略的一个高质量数据源。作为最大的浏览器扩展分发渠道之一它上面每一个扩展都携带了非常完整的结构化信息扩展名称、简介、评分、评分人数、累计用户数、开发者名称、当前版本、最近更新时间、隐私权属声明、所需权限、分类、官网链接等等。把这些字段拼起来能做的事情非常多。从产品角度看可以按分类统计扩展市场的用户规模分布看哪些细分赛道已经拥挤、哪些还空着从市场分析角度看可以追踪某个竞品扩展的用户量变化曲线判断它最近一次更新后评分是涨是跌从安全研究角度看可以通过权限列表寻找那些只要一个很小功能却申请了海量权限的可疑扩展。标题里提到的千万级数据并不夸张商店里扩展数量庞大如果再把历史版本、权限变更、评论内容这些维度加进来单个扩展就能衍生出几十条甚至上百条记录。但千万级是最终目标不是一上来就闷头跑全量先把单条数据抓深抓稳再谈扩展规模。1.2 为什么不用 requests 硬怼很多人拿到这类页面的第一反应是打开 DevTools 找 XHR 接口找到那个返回 JSON 的 URL然后用 requests 直接请求。我也这么干过但实际情况是Chrome 扩展商店的页面容器非常重列表页的滚动加载数据确实来自异步请求可这些请求携带了复杂的动态参数和令牌签名逻辑藏在压缩后的 JS 代码里调试成本极高。而且一旦请求方式跟页面实际行为不一致响应内容随时可能变成一段带验证逻辑的 HTML。换个思路看问题我要的数据最终都渲染在 DOM 上了那与其逆向接口不如直接驱动一个真实浏览器去取渲染完成后的结果。这样做的优点是逻辑直观——你眼睛看到什么代码就能抓到什么缺点是需要承担浏览器进程的开销但在批量抓取场景下这个代价完全值得。Playwright 就是冲这个场景来的。1.3 Playwright 比 Selenium 强在哪Selenium 用了这么多年生态成熟但用起来总有一种老派的感觉。首先是 driver 管理麻烦Chrome 一升级chromedriver 版本对不上就开始报错其次定位和等待的 API 设计得绕显式等待写起来啰嗦。Playwright 最大的优势是把这一切收敛了pip install playwright一条命令装完再playwright install chromium就能拉起浏览器不用单独维护 driver。页面操作用的是天然带自动等待的 Locator API而且原生支持异步写并发比 Selenium 顺手太多。另外一个我特别喜欢的功能是playwright codegen。它可以记录你在页面上的真实操作自动生成对应的选择器和代码分析目标页面结构的时候效率非常高。后面讲页面分析时我会再细说。2. 环境准备与目标页面拆解2.1 安装 Playwright 和浏览器内核老规矩先建虚拟环境再装依赖免得把系统 Python 搞乱python -m venv venv source venv/bin/activate # Windows 上执行 venv\Scripts\activate pip install playwright sqlalchemy pandas playwright install chromium这里说明几个关键点。playwright install chromium只会装 Playwright 自己管理的 Chromium 内核跟系统里的 Chrome 互不干扰这样版本可控也避免项目跑到服务器上时找不到浏览器。如果你希望 Playwright 直接驱动已经安装好的正式版 Chrome可以在启动时传channelchrome比如p.chromium.launch(channelchrome)但不是必须的默认的 Chromium 已经完全够用。如果部署到 Linux 服务器上还需要装浏览器运行依赖playwright install-deps chromium这一步是为了解决类似缺少 libnss3 这些底层库的问题。另外几个我常加的启动参数后面代码里再给。2.2 商店页面结构分析Chrome 扩展商店现在的域名已经统一到chromewebstore.google.com老域名chrome.google.com/webstore会自动跳转。列表页和搜索页都是典型的无限滚动设计页面底部有一个加载触发器滚到那里就会通过异步请求追加新的扩展卡片但卡片是渲染成 DOM 的所以我们只需要模拟滚动然后从 DOM 里取链接和卡片信息。详情页的 URL 结构通常是这样的https://chromewebstore.google.com/detail/扩展名slug/一串32位十六进制ID这串 32 位 ID 是扩展在商店内部的主键天然适合作数据库的主键。所有详情页链接都可以从列表页卡片元素的a[href*/detail/]选择器里提取。用playwright codegen可以快速确认关键元素playwright codegen https://chromewebstore.google.com/打开工具后点击任意一个扩展进入详情页codegen 会实时把对应的选择器生成在右侧面板里。比如标题元素通常是h1评分在div[aria-label*评分]里用户数在包含位用户的文本节点中。这些信息不是一成不变的建议每次项目开始前花十分钟重新确认一遍别守着一年前的选择器跑今天的页面。2.3 核心字段建模在开始写爬虫之前先把要抓的字段列清楚方便后面设计数据库表。我这次最终确定的字段如下字段名来源示例目标类型扩展 IDURL 中的 32 位串字符串主键名称页面h1字符串简介详情页描述段文本评分评分区域文本如 4.5/5浮点数评分人数评分区括号里的数字整数累计用户数1,234 位用户整数开发者开发者栏链接文本字符串当前版本版本号文本字符串更新日期详情信息里的日期日期类型分类分类标签字符串权限列表查看权限弹层中的列表JSON 文本官网链接相关链接字符串抓取时间系统当前时间时间戳这里有个小经验抓取之前先在本地跑三五个详情页把所有文本导出成纯文本看一看实际格式再做清洗逻辑。因为页面上可能把用户数显示成1.2万位用户或1,234 位用户不同地区语言环境下格式完全不同。我一律通过设置localezh-CN来保证页面返回简体中文清洗规则相对固定。3. 动态页面抓取核心实现3.1 初始化浏览器上下文Playwright 里有一个容易被新手忽略的概念Browser Context浏览器上下文。它相当于一个独立的浏览器会话cookie、缓存、localStorage 都是隔离的。多个任务并行时让每个任务使用独立 context可以避免登录状态、地理位置这些脏数据互相污染。我通常会封装一个初始化函数from playwright.sync_api import sync_playwright def init_browser(playwright): browser playwright.chromium.launch( headlessTrue, args[--disable-dev-shm-usage], ) context browser.new_context( user_agent( Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36 ), viewport{width: 1920, height: 1080}, localezh-CN, ) page context.new_page() page.set_default_timeout(15000) return browser, context, page解释几个参数headlessTrue保证在服务器上稳定运行调试时可以临时改成 False 看实际渲染过程--disable-dev-shm-usage是容器环境里非常关键的参数避免/dev/shm太小导致 Chromium 崩溃viewport 设置成 1920x1080 是因为有些懒加载逻辑会判断你看到了哪里窗口太小可能导致页面认为某些卡片不在视野内而不加载localezh-CN则让页面输出中文字段解析时统一用位用户、分这些关键字。3.2 列表页滚动加载的稳定实现无限滚动页面的通关口诀是一边滚动一边数卡片连续几次没有新卡片出现就停止。我的实现是这样的def scroll_list(page, max_scrolls60): prev_count 0 stable_rounds 0 for _ in range(max_scrolls): cards page.locator(a[href*/detail/]) count cards.count() page.mouse.wheel(0, 1200) page.wait_for_timeout(1200) if count prev_count: stable_rounds 1 if stable_rounds 3: break else: stable_rounds 0 prev_count count return page.locator(a[href*/detail/]).count()注意几个细节。第一滚动用的是page.mouse.wheel(0, 1200)而不是page.evaluate(window.scrollTo(...))。虽然两者都能触发滚动但鼠标滚轮事件更接近真实用户行为很多页面会根据事件类型决定是否加载后续内容用鼠标滚轮踩坑少。第二每次滚动后等待 1200 毫秒给异步加载留下时间等待太短会漏数据等待太长会拖慢整体进度。第三连续 3 轮没有新增就退出这个阈值可以根据网络情况调整网络差的站点可能需要连续 5 轮才算真到底。如果你抓的是某个分类页可以在 URL 上追加排序参数比如按评分最高或用户数最多排序这样同一页的扩展质量更高也更容易控制数据规模。3.3 提取详情页链接滚动结束后把所有详情页链接一次性取出来links page.locator(a[href*/detail/]).evaluate_all( els els.map(e e.href) ) unique_links list(dict.fromkeys(links))这里用evaluate_all直接在浏览器上下文里把 href 全部映射出来比 Python 端逐个get_attribute快很多。dict.fromkeys去重保序比set更能保持原始顺序调试时方便比对。从链接里提取扩展 ID 用正则import re def extract_ext_id(url): m re.search(r/detail/(?:[^/]/)?([a-z0-9]{32}), url) return m.group(1) if m else None这个正则可同时兼容新旧两种 URL 格式。新版 URL 里扩展名和 ID 之间有个斜杠旧版 URL 后面直接跟 ID(?:[^/]/)?正好处理了可能有扩展名的这一层。如果有的链接不是标准 32 位 ID正则取不到就直接返回 None后面做任务列表时可以把异常链接筛掉。3.4 详情页字段抓取实现详情页是整条链路里最需要耐心的地方。同一个字段在不同扩展页上可能结构完全不一样有的扩展没有官网有的权限列表为空有的评分人数为 0。所以我建议写一个高度防御性的解析函数每个字段都走 try/except拿不到就用默认值。from playwright.sync_api import TimeoutError as PlaywrightTimeoutError def safe_text(page, selector, default): try: return page.locator(selector).first.inner_text().strip() except PlaywrightTimeoutError: return default def parse_detail(page, url): page.goto(url, wait_untildomcontentloaded) page.locator(h1).first.wait_for() name safe_text(page, h1) desc safe_text(page, div[data-blog-componentextension-description]) rating_text safe_text(page, div[aria-label*评分]) users_text safe_text(page, div:has-text(位用户)) developer safe_text(page, a[href*devsite] span) return { id: extract_ext_id(url), name: name, description: desc, rating: parse_rating(rating_text), rating_count: parse_count(rating_text), users: parse_count(users_text), developer: developer, url: url, }这段代码有几个经验点想单独说。一是page.locator(...).first很实用。一个页面里经常有多个元素匹配同一个选择器比如页面头部和页脚可能各有一个扩展名称用.first直接取页面里第一个避免strict mode violation报错。二是所有文本取出后必须strip()。DOM 里文本经常带前后换行和空格尤其描述区块不清理的话存进数据库之后做关键词检索会很难受。三是wait_untildomcontentloaded比默认的load更快。扩展商店详情页会加载大量脚本和埋点资源等全部 load 可能要等好几秒而我们要的 DOM 结构在 domcontentloaded 之后基本已经就绪后面再用locator(h1).first.wait_for()保证标题渲染完成这样速度和安全兼得。3.5 用户数与评分的清洗函数页面上的文本格式大致是这样评分区域评分 4.5/5 1,234 条评分用户数1,234 位用户更新时间2024年5月20日清洗函数直接写import re def parse_count(text): nums re.findall(r[\d,], text) if not nums: return 0 return int(nums[0].replace(,, )) def parse_rating(text): m re.search(r(\d(?:\.\d)?)\s*/\s*5, text) return float(m.group(1)) if m else 0.0parse_count会从文本里取出第一段数字比如 1,234 位用户 取出 1,234 转成 1234。需要注意如果页面上出现了 1.2万位用户这段逻辑就处理不了所以我才强调要通过localezh-CN统一格式。假如你抓到的商店版本开始返回中文数字单位就得再补一段 万 和 亿 的换算逻辑。3.6 权限列表的抓取权限列表藏在查看权限按钮后面。我的方案是用expect_popup处理弹窗def fetch_permissions(page): try: with page.expect_popup(timeout8000) as popup_info: page.get_by_role(button, name查看权限).click() popup popup_info.value popup.wait_for_load_state() perms popup.locator(li, [class*permission]).all_inner_texts() popup.close() return [p.strip() for p in perms if p.strip()] except Exception: return []这里的逻辑是如果点按钮后打开的是新标签页expect_popup会准确捕获它如果商店界面改版后变成展开式面板这个函数会超时我们直接返回空列表后续再针对新界面调整。实际开发中我遇到过按钮点击被底部的 cookie 弹窗遮住的情况如果点击失败可以先按一下Escape或关闭弹窗组件再点。这个细节在代码里加个try就能覆盖。4. 数据落库SQLAlchemy 模型与增量更新4.1 建表与模型设计数据清洗完总要落库我用 SQLAlchemy 管理数据库模型。为什么不用裸 SQL因为后期字段要加索引、要改类型用 ORM 迁移起来省事。模型这样定义from sqlalchemy import ( create_engine, Column, String, Float, Integer, Date, Text, DateTime, func ) from sqlalchemy.orm import declarative_base, sessionmaker Base declarative_base() class ChromeExtension(Base): __tablename__ chrome_extensions id Column(String(32), primary_keyTrue) name Column(String(255), nullableFalse) description Column(Text, default) rating Column(Float, default0.0) rating_count Column(Integer, default0) users Column(Integer, default0) developer Column(String(255), default) version Column(String(64), default) updated_date Column(Date, nullableTrue) category Column(String(128), default) permissions Column(Text, default[]) homepage Column(String(512), default) crawled_at Column(DateTime, server_defaultfunc.now(), onupdatefunc.now()) DATABASE_URL sqlite:///extensions.db engine create_engine(DATABASE_URL, echoFalse) Base.metadata.create_all(engine) Session sessionmaker(bindengine)主键直接用扩展 ID有两个原因。第一ID 天然唯一不会出现同一个扩展因为 URL 多一个参数被重复插入第二后期做增量抓取时用主键判断这个扩展是否已存在非常快。permissions字段我存的是 JSON 序列化后的字符串读出时再用json.loads还原这样既兼容 SQLite 这类轻量库又保留列表结构。4.2 幂等写入去重、更新与 merge增量抓取的核心目标是跑第二次时不会产生脏数据。最简单的幂等写法是def save_item(session, item: dict): existing session.get(ChromeExtension, item[id]) if existing: for key, value in item.items(): setattr(existing, key, value) else: session.add(ChromeExtension(**item)) session.commit()这个写法逻辑清楚查询主键存在就逐字段更新不存在就新增。但如果你追求代码更短可以直接用 SQLAlchemy 的mergedef save_item_merge(session, item: dict): session.merge(ChromeExtension(**item)) session.commit()merge的行为是如果主键存在执行更新否则插入。实现上一行搞定但我个人更推荐前一种写法因为更新时你可以控制哪些字段需要覆盖比如第一次抓取时某些字段为空第二次抓取补全了只更新有实际值的字段避免别人声明的数据被空字符串冲掉。4.3 断点续爬与任务队列动辄几万个详情页不可能一次跑完。中断恢复是必须考虑的问题。我常用的方案是维护一个简单的任务列表crawled_ids {r[0] for r in session.query(ChromeExtension.id).all()} todo [] for url in unique_links: ext_id extract_ext_id(url) if ext_id and ext_id not in crawled_ids: todo.append(url)跑完之后把todo清空。如果中途断网、进程被杀重跑一次脚本只会处理没抓过的 URL。更稳妥的工程化做法是建一张独立的任务表记录每个 URL 的状态pending / done / failed配合失败重试。但如果只是个人分析用内存集合加数据库主键判断完全够用没必要为一个小项目引入 Celery。4.4 并发加速与限速单页面串行抓取肯定慢但也不能直接起 20 个浏览器进程硬冲。我的经验是用 Playwright 的异步 API asyncio.Semaphore控制并发数。import asyncio import random from playwright.async_api import async_playwright async def fetch_one(sem, browser, url): async with sem: page await browser.new_page() try: await page.goto(url, wait_untildomcontentloaded) await page.locator(h1).first.wait_for() name await page.locator(h1).first.inner_text() # 这里复用前面解析函数中的文本提取逻辑 await page.wait_for_timeout(random.randint(800, 2000)) return {id: extract_ext_id(url), name: name.strip(), url: url} finally: await page.close() async def run_batch(urls, concurrency5): async with async_playwright() as p: browser await p.chromium.launch(headlessTrue) sem asyncio.Semaphore(concurrency) results await asyncio.gather(*[fetch_one(sem, browser, u) for u in urls]) await browser.close() return results在同一浏览器实例里开多个 page 并行加载比开多个浏览器进程节省资源。Semaphore把并发数限制在 5实测下来速度和稳定性取得一个比较好的平衡。每个请求之间强制加 800 到 2000 毫秒的随机延迟这个延迟既是礼貌也是给目标站点一个喘息空间。有个细节比较重要异步代码里所有 Playwright 调用都要await内部脚本执行、元素定位都是异步的写循环的时候尤其要注意不要漏掉await。如果对 asyncio 不熟也可以用同步 API 加上ThreadPoolExecutor但请务必让每个线程使用独立的BrowserContext别让多个线程共享同一个 page 实例否则会有各种诡异的并发问题。5. 常见问题与排查技巧5.1 出现人机验证或访问受限怎么办扩展商店对自动化流量不是完全无感的。如果你并发开太高、请求频率太密集页面大概率会跳出一个验证页面。我的原则是不跟它硬碰硬直接让程序识别这种状态并主动降温。可以在代码里加一个检测函数def detect_blocked(page): text page.locator(body).inner_text().lower() return unusual traffic in text or 验证 in text一旦检测到验证页面果断停止任务等待 60 到 120 分钟再继续。很多人觉得这是浪费时间但比起账号被封、IP 被拉黑等待是成本最低的选择。实际操作中我还会在每次抓取完 200 个扩展后强制让整个程序睡 10 分钟尽量把单次会话的流量摊平。5.2 页面元素定位选择器失效最好用的排查工具是playwright codegen和 Python 端的page.locator(selector).all_text_contents()。每次升级浏览器或商店改版后选择器都可能变。我的建议是选择器尽量少用绝对路径多用文本特征比如div:has-text(位用户)、a[href*devsite]。文本特征比div#root div.main div.card span这类层级结构稳定得多。如果页面里有多个元素匹配同一个选择器尽量用.first或者通过locator.filter(has_text...)缩小范围。不要对所有定位都靠猜把页面截图打印出来代码里加一行page.screenshot(pathdebug.png)往往比反复试选择器快得多。5.3 抓取时间久了内存占用越来越高浏览器跑久了内存膨胀是很正常的。特别是无限滚动列表页开了一堆标签页、每个详情页还带着一堆资源Python 进程本身可能没事Chromium 子进程却可能吃掉几个 GB。解决办法是给抓取过程按批次拆分。比如每抓 500 个扩展关闭当前 context重新创建一批新的 page。在 Playwright 里这就是一个循环里重新browser.new_context()和context.close()的事。另外及时page.close()是关键别把 page 对象存在列表里一直不释放。5.4 SQLAlchemy 写入报错与数据矛盾几个常见问题提前打个预防针字符串超长会报DataError日期解析失败会报ValueErrorJSON 序列化字段忘记json.dumps会直接写成一个带引号的字符串。我的建议是模型里字符串字段留长一点描述用Text类型所有用户数、评分在进入模型前用解析函数包一层任何异常都给默认值所有列表型字段一律先json.dumps再入库取出用json.loads。如果用的是 SQLite并发写入偶尔会出现database is locked这是因为多个连接同时写库。解决办法是把写入操作收敛到一个连接里或者改用check_same_threadFalse参数。我在项目里保证只有主线程做最终 commit其他 worker 只负责返回数据从根上规避这个锁。5.5 Playwright 安装与运行的环境坑在服务器上最容易遇到三类报错一是Executable doesnt exist说明你还没执行playwright install二是Host system is missing dependencies说明缺系统库执行playwright install-deps即可三是中文内容显示为方块通常是服务器缺少中文字体Debian/Ubuntu 上装fonts-noto-cjk能解决。调试时可以开详细日志观察浏览器行为DEBUGpw:browser* python crawl.py这个日志会输出浏览器每个关键行为节点有时候比 DevTools 还好用。另一个实用命令是playwright show-trace trace.zip抓取时可以开启 tracecontext browser.new_context() context.tracing.start(screenshotsTrue, snapshotsTrue) # 抓取若干页面后 context.tracing.stop(pathtrace.zip)trace 文件包含了页面截图、DOM 快照和网络请求复现偶发问题特别有帮助。6. 最后再分享一点个人的实践体会这个项目跑完一遍我最深的感受是爬虫本身只占整个工作量的三分之一页面分析和数据清洗才是真正磨人的部分。扩展商店看起来结构规范但每个扩展页面都会给你来点意外——有的没有官网有的权限列表空荡荡有的评分人数写成 0还有的把描述挤在一个高度受限的容器里造成内容截断。这些脏数据不会让程序崩溃但会让后续分析结果失真。所以我的建议是先用一两百个扩展把整个流程跑通把解析规则调稳了再考虑全量铺开。虽然标题写的是千万级数据但真正的工程里数据不是抢来的是稳稳当当一点一点攒出来的。这个项目抓下来的数据可以用来做扩展市场画像、做竞品趋势跟踪也可以定期重跑增量更新观察用户数和评分的变化曲线。数据采集只是开始数据经过加工和分析之后才有资格叫资产。最后再补一句。运行这种长耗时任务日志记录一定要做扎实。我习惯每抓完一批打印一行进度序号、扩展名、用户数、耗时。这样半夜任务挂了你也能第一时间从日志里看出它卡在了哪里。稳比快重要。
