1. 项目概述与核心挑战最近在做一个东南亚电商平台的数据采集项目目标站点是Shopee。做过电商数据抓取的朋友都知道这类平台的反爬机制一年比一年复杂从最初的简单User-Agent校验到后来的动态参数加密、行为指纹检测再到现在的接口数据混淆简直是一场没有硝烟的战争。我这次遇到的挑战主要有两个一是Shopee的页面和接口都部署了相当成熟的反爬策略传统的requests库配合简单代理和请求头修改已经很难稳定获取数据二是即便成功请求到接口返回的数据也经常是乱码或者是一堆看似无意义的加密字符串给数据解析带来了巨大麻烦。经过一番技术选型我最终决定采用微软开源的Playwright作为核心工具。它不仅仅是一个自动化测试框架更是一个强大的浏览器自动化库能够模拟真实用户的浏览器环境这对于绕过基于浏览器指纹和JavaScript执行环境的反爬机制非常有效。同时处理接口返回乱码的问题则需要从HTTP请求的编码、响应内容的解码以及Playwright自身的网络拦截和响应处理等多个层面进行综合配置。这篇文章我就来详细拆解一下整个实战流程。我会从环境搭建、Playwright的核心配置、如何监听和拦截接口、再到解决乱码问题的完整解码链条一步步带你走通。无论你是刚接触数据采集的新手还是被反爬和乱码困扰的老手相信都能从中找到可以直接“抄作业”的解决方案。2. 环境准备与Playwright深度配置2.1 为什么选择Playwright而非Selenium或Puppeteer在开始动手之前我们先聊聊工具选型。市面上主流的浏览器自动化工具主要有Selenium、Puppeteer和Playwright。Selenium历史悠久生态庞大但启动速度慢资源占用高且对现代Web应用大量依赖JavaScript的反爬对抗能力逐渐力不从心。Puppeteer是Chrome官方团队出品对Chrome/Chromium的控制力极强速度快但最初只支持Chrome且API设计上更偏向于Chrome DevTools Protocol的原生映射。Playwright则可以看作是Puppeteer的“升级版”和“跨平台版”。它由微软团队开发原生支持Chromium、Firefox和WebKitSafari内核三大浏览器引擎。这意味着你可以用同一套脚本在不同浏览器上运行对于测试不同环境下的反爬策略很有帮助。更重要的是Playwright在反爬对抗方面有天然优势真实的浏览器上下文它能创建完全隔离的浏览器上下文Browser Context每个上下文都有独立的Cookie、LocalStorage等可以模拟多个独立用户会话避免因Cookie关联被识别为爬虫。自动等待与智能选择器内置了强大的自动等待机制能等待元素出现、可操作、网络空闲等减少了编写复杂等待逻辑的代码让脚本更稳定更像真人操作。强大的网络拦截Route/Abort这是本次实战的核心。我们可以精确地拦截特定模式的网络请求比如XHR/Fetch请求即API接口然后直接修改请求参数、获取原始响应甚至模拟返回数据这比用Selenium后再通过DevTools获取网络日志高效和直接得多。移动端模拟与设备预设可以轻松模拟手机、平板等设备的User-Agent和屏幕尺寸这对于Shopee这类移动端流量占主导的电商平台尤为重要。基于以上几点Playwright在应对复杂反爬场景时提供了更现代、更强大的武器库。2.2 基础环境搭建与核心依赖安装我的开发环境是Windows 11但步骤在macOS和Linux上大同小异。首先确保你安装了Python建议3.8及以上版本和Node.jsPlaywright本身是Node.js库但Python绑定非常完善。第一步创建项目并安装Playwright Python包打开你的命令行工具创建一个新的项目目录并安装playwright的Python库。# 创建项目目录 mkdir shopee-data-crawler cd shopee-data-crawler # 创建虚拟环境推荐避免包冲突 python -m venv venv # Windows激活 venv\Scripts\activate # macOS/Linux激活 source venv/bin/activate # 安装playwright pip install playwright第二步安装浏览器二进制文件Playwright需要下载它自己管理的浏览器内核才能工作。运行以下命令这会下载Chromium、Firefox和WebKit。虽然会占用一些磁盘空间约1GB但保证了环境的一致性。playwright install注意这一步可能会因为网络问题下载缓慢或失败。可以尝试设置镜像源例如在命令前添加PLAYWRIGHT_DOWNLOAD_HOSThttps://npmmirror.com/mirrors/playwright/。如果只想安装Chromium以节省空间可以使用playwright install chromium。第三步验证安装创建一个简单的测试脚本test_install.pyimport asyncio from playwright.async_api import async_playwright async def main(): async with async_playwright() as p: # 启动Chromium浏览器headlessFalse表示有界面方便调试 browser await p.chromium.launch(headlessFalse) page await browser.new_page() await page.goto(https://www.google.com) print(await page.title()) await browser.close() asyncio.run(main())运行这个脚本如果能看到浏览器打开并打印出“Google”的标题说明环境配置成功。3. 核心反爬策略分析与Playwright应对方案3.1 Shopee反爬机制深度剖析在编写代码之前我们必须先理解对手。通过对Shopee网站以新加坡站shopee.sg为例进行手动浏览和网络抓包分析我发现了以下几层主要的反爬措施请求头校验这是最基础的一层。检查User-Agent是否来自主流浏览器检查Accept、Accept-Language、Accept-Encoding、Referer等头是否完整且合理。缺少或使用Pythonrequests库的默认头会立刻被识别。Cookie与会话管理首次访问会设置一系列Cookie如SPC_EC、SPC_F等这些Cookie参与了后续几乎所有接口请求的签名或验证。单纯复制Cookie字符串其生命周期和关联性很容易失效。JavaScript环境检测这是重头戏。页面会通过JavaScript检测浏览器环境例如navigator.webdriver属性在自动化工具控制下此属性通常为true而真实浏览器为undefined或false。早期Selenium很容易被这个检测到。window.chrome对象检测Chrome特定属性和方法。Notification、Permissions等Web API的存在性和行为。鼠标移动轨迹、点击事件的精确坐标和间隔时间。接口参数加密商品列表、搜索建议等核心数据接口通常路径包含/api/v4/的URL或请求体POST数据中包含加密参数如signature、bundle等。这些参数由前端JavaScript根据当前时间、Cookie、页面状态等实时计算生成直接构造请求非常困难。频率与行为限制短时间内发起大量相同模式的请求或浏览行为不符合人类模式如页面停留时间为0秒无限下拉会触发IP限制或要求验证码。3.2 Playwright的破解之道模拟真实浏览器会话Playwright如何应对上述挑战核心思路是“以真乱真”即创建一个与普通用户无异的浏览器会话。对抗环境检测Playwright启动的浏览器上下文默认已经处理了navigator.webdriver等常见属性的暴露问题。我们还可以通过add_init_script方法在页面加载前注入JavaScript进一步抹平与真实浏览器的差异。获取动态参数我们不需要去逆向JavaScript加密逻辑。我们让Playwright浏览器正常加载页面、执行所有JS。当页面上的JavaScript发起数据请求时我们通过网络拦截Route功能“截胡”这个请求直接拿到它已经构造好的、包含正确加密参数的请求URL和请求体。这就是“借力打力”。管理会话状态使用browser.new_context()创建一个独立的浏览器上下文。在这个上下文中产生的Cookie、本地存储都是隔离的并且会随着浏览操作自动管理。我们只需要像真人一样操作页面点击、输入、滚动所有的状态维护都由浏览器自动完成。接下来的章节我们就将围绕“网络拦截”和“编码处理”这两个核心点展开具体的代码实现。4. 实战配置Playwright进行数据采集4.1 初始化浏览器与上下文细节决定成败一个健壮的初始化设置能避免很多后续的坑。下面是我经过多次调试后总结的最佳实践配置。import asyncio from playwright.async_api import async_playwright import json async def create_browser_context(): async with async_playwright() as p: # 1. 启动浏览器 # 使用 headlessFalse 便于调试生产环境可设为 True # args 参数用于传递额外的浏览器启动参数非常重要 browser await p.chromium.launch( headlessFalse, args[ --disable-blink-featuresAutomationControlled, # 关键隐藏自动化控制痕迹 --no-sandbox, --disable-setuid-sandbox, --disable-dev-shm-usage, --disable-accelerated-2d-canvas, --disable-gpu, # 在某些虚拟环境中可能需要 --start-maximized # 启动时最大化更符合用户行为 ] ) # 2. 创建浏览器上下文 (Context) # Context 比单纯的 Page 提供了更好的隔离性相当于一个独立的隐身会话。 context await browser.new_context( viewport{width: 1920, height: 1080}, user_agentMozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36, # 使用一个常见的UA localeen-SG, # 针对新加坡站设置语言地区 timezone_idAsia/Singapore, # 可以设置代理如果需要的话 # proxy{server: http://your-proxy:port} ) # 3. 在页面加载前执行初始化脚本进一步隐藏自动化特征 await context.add_init_script( // 覆盖 webdriver 属性 Object.defineProperty(navigator, webdriver, { get: () undefined }); // 覆盖 languages 和 plugins 属性 Object.defineProperty(navigator, languages, { get: () [en-US, en] }); Object.defineProperty(navigator, plugins, { get: () [1, 2, 3, 4, 5] }); // 模拟 Chrome 的 runtime 对象 window.chrome { runtime: {} }; ) # 4. 创建页面 page await context.new_page() # 设置默认超时时间 page.set_default_timeout(60000) # 60秒 return browser, context, page # 使用示例 async def main(): browser, context, page await create_browser_context() try: # 你的采集逻辑将在这里执行 await page.goto(https://shopee.sg) await page.wait_for_load_state(networkidle) # 等待网络基本空闲 # ... 更多操作 finally: # 最后务必关闭浏览器释放资源 await browser.close() asyncio.run(main())实操心得args中的--disable-blink-featuresAutomationControlled是隐藏自动化标志的关键参数。user_agent、locale、timezone_id的设置要与目标站点的常见用户保持一致形成一个自洽的“数字指纹”。4.2 关键步骤监听与拦截目标API接口这是整个采集流程的灵魂。我们不通过解析HTML来获取数据而是直接“窃听”浏览器与服务器之间的数据通信。Shopee的商品列表数据通常通过类似https://shopee.sg/api/v4/search/search_items?...这样的接口返回。我们的目标是拦截这类请求拿到原始的JSON响应。import asyncio from playwright.async_api import async_playwright, Route, Request import json # 存储拦截到的数据 captured_data [] async def handle_route(route: Route, request: Request): 网络请求拦截处理函数 # 打印请求信息用于调试和确认拦截目标 print(f拦截到请求: {request.url}) print(f请求方法: {request.method}) # print(f请求头: {request.headers}) # 调试时可查看 # 1. 判断是否为目标接口 # 这里以搜索商品接口为例你可以根据实际需要调整URL模式 if /api/v4/search/search_items in request.url: print( 发现目标API接口准备获取响应...) # 2. 继续发送请求并获取响应 response await route.fetch() # 关键使用fetch获取原始响应 # 3. 获取响应体 body await response.body() # 4. 尝试解码响应体乱码问题在此处解决 try: # 首先尝试用UTF-8解码 decoded_body body.decode(utf-8) except UnicodeDecodeError: # 如果UTF-8失败尝试常见的其他编码 try: decoded_body body.decode(gbk) print( 警告响应使用GBK编码解码) except UnicodeDecodeError: # 如果还失败可以尝试从响应头获取编码或使用chardet库检测 print( 错误无法用UTF-8或GBK解码响应尝试原始字节或检测编码) # 使用chardet检测编码需安装 pip install chardet import chardet encoding chardet.detect(body)[encoding] if encoding: decoded_body body.decode(encoding, errorsignore) print(f 使用检测到的编码 {encoding} 解码可能不准确) else: decoded_body body.hex() # 最后手段转为十六进制字符串 print( 无法解码已转为十六进制) # 5. 将解码后的文本解析为JSON如果是JSON接口 try: data json.loads(decoded_body) captured_data.append(data) # 存储数据 print(f 成功解析JSON数据已保存。本次包含条目数{len(data.get(items, []))}) except json.JSONDecodeError as e: print(f 响应不是有效的JSON: {e}) # 可能是HTML或其他格式保存文本供分析 with open(ferror_response_{int(time.time())}.txt, w, encodingutf-8) as f: f.write(decoded_body[:1000]) # 保存前1000字符 # 6. 继续路由让页面收到响应如果页面需要 # 如果你想修改响应返回给页面可以用 route.fulfill(body...) # 这里我们直接使用原始响应完成路由 await route.fulfill(responseresponse) else: # 对于非目标请求直接继续不做处理 await route.continue_() async def main_crawl(): browser, context, page await create_browser_context() # 启用网络请求监听对所有请求应用处理函数 await page.route(**/*, handle_route) # ‘**/*’ 匹配所有请求可根据需要缩小范围如 ‘**/api/v4/**’ try: # 模拟用户访问Shopee并搜索商品 await page.goto(https://shopee.sg) await page.wait_for_load_state(networkidle) # 定位搜索框输入关键词并搜索 search_input page.locator(input[placeholder*Search]).first # 根据实际placeholder调整 await search_input.fill(laptop) await search_input.press(Enter) # 等待搜索结果页面加载和接口请求完成 # 搜索后页面会触发AJAX请求加载商品列表我们的拦截器会捕获它 await page.wait_for_timeout(5000) # 等待5秒确保网络请求完成。更优做法是等待特定元素出现。 # 打印捕获到的数据摘要 print(f\n 数据采集完成 ) print(f共捕获到 {len(captured_data)} 个目标API响应。) if captured_data: first_item captured_data[0].get(items, [{}])[0] print(f示例商品: {first_item.get(item_basic, {}).get(name, N/A)}) # 可以将数据保存到文件 with open(shopee_items.json, w, encodingutf-8) as f: json.dump(captured_data, f, ensure_asciiFalse, indent2) print(数据已保存至 shopee_items.json) except Exception as e: print(f爬取过程中发生错误: {e}) import traceback traceback.print_exc() finally: await browser.close() asyncio.run(main_crawl())这段代码的核心是page.route(**/*, handle_route)和handle_route函数。route.fetch()方法是关键它允许我们获取请求的响应而不影响页面的正常流程。我们在handle_route函数中完成了对目标接口的识别、响应的获取、乱码解码和JSON解析。5. 乱码问题的根源与系统性解决方案5.1 乱码产生的原因剖析在数据采集中乱码通常源于“编码”与“解码”使用的字符集不匹配。具体到我们拦截HTTP响应的情况服务器编码不明确服务器在返回响应时会在HTTP响应头Content-Type中指定字符集如Content-Type: application/json; charsetutf-8。但有些服务器可能配置不当未指定字符集或者指定了错误的字符集如明明是UTF-8编码的数据却声明为charsetgbk。Playwright/Response.body()的默认处理await response.body()返回的是原始的字节流bytes。如何将这些字节转换成字符串str就需要正确的解码。如果我们直接用body.decode(utf-8)但服务器实际用的是gbk或iso-8859-1编码就会产生乱码。响应内容被压缩或加密服务器可能返回gzip压缩的数据Playwright的fetch()默认会自动解压。但如果响应体本身是某种自定义的二进制格式或加密数据直接解码成字符串自然会是乱码。5.2 构建健壮的解码流程基于以上分析我们不能假设所有响应都是UTF-8。需要构建一个层次化的解码策略。上面的handle_route函数中已经体现了一部分这里再系统化一下async def decode_response_body(body_bytes: bytes, response_headers: dict) - str: 一个健壮的响应体解码函数。 :param body_bytes: 原始的响应字节流 :param response_headers: 响应头字典 :return: 解码后的字符串 # 策略1优先使用HTTP响应头中声明的编码 content_type response_headers.get(content-type, ).lower() charset None if charset in content_type: # 从类似 application/json; charsetgbk 中提取 gbk charset content_type.split(charset)[-1].split(;)[0].strip() # 处理一些常见的不规范写法 charset charset.replace(, ).replace(, ) if charset: try: return body_bytes.decode(charset) except (UnicodeDecodeError, LookupError): # LookupError处理不支持的编码名 print(f 警告使用响应头指定的编码 {charset} 解码失败尝试其他方法。) # 策略2尝试常见编码 (UTF-8, GBK, GB2312, ISO-8859-1) common_encodings [utf-8, gbk, gb2312, latin-1, iso-8859-1] for encoding in common_encodings: try: return body_bytes.decode(encoding) except UnicodeDecodeError: continue # 策略3使用chardet进行编码检测作为备选因为可能不准且慢 try: import chardet detection_result chardet.detect(body_bytes) if detection_result[confidence] 0.7: # 置信度高于70% detected_encoding detection_result[encoding] print(f 检测到编码: {detected_encoding} (置信度: {detection_result[confidence]:.2f})) try: return body_bytes.decode(detected_encoding, errorsignore) except (UnicodeDecodeError, LookupError): pass except ImportError: print( 未安装chardet库跳过编码检测。) # 策略4终极备选方案 - 忽略错误或返回字节表示 # 如果数据本身不是文本如图片、加密数据则不应解码为字符串 print( 错误所有解码尝试均失败响应体可能不是文本数据。) # 返回十六进制字符串或直接抛出异常取决于你的业务逻辑 return body_bytes.hex() # 或者 raise ValueError(无法解码响应体) # 在 handle_route 函数中这样使用 async def handle_route_improved(route: Route, request: Request): if /api/v4/ in request.url: response await route.fetch() body_bytes await response.body() headers response.headers # 使用改进的解码函数 decoded_text decode_response_body(body_bytes, headers) # ... 后续的JSON解析等操作 await route.fulfill(responseresponse)注意事项chardet库并非百分百准确且对于短文本检测效果差。对于明确的接口如Shopee的API其编码通常是固定的UTF-8。在长期运行中你应该根据目标站点的实际情况将最可能的编码如UTF-8放在common_encodings列表的最前面并可以移除不必要的检测以减少开销。6. 高级技巧与稳定性优化6.1 处理动态加载与无限滚动Shopee的商品列表是动态加载的滚动到底部会加载更多。我们需要模拟这个行为来获取大量数据。async def scroll_to_load_more(page, scroll_pause_time2.0, max_scrolls20): 模拟滚动页面以触发动态加载。 last_height await page.evaluate(document.body.scrollHeight) scroll_attempts 0 while scroll_attempts max_scrolls: # 滚动到页面底部 await page.evaluate(window.scrollTo(0, document.body.scrollHeight)) # 等待新内容加载 await page.wait_for_timeout(int(scroll_pause_time * 1000)) # 计算新的滚动高度 new_height await page.evaluate(document.body.scrollHeight) if new_height last_height: # 高度未变可能已加载完毕或遇到加载失败 print(f滚动 {scroll_attempts1} 次后页面高度未变化停止滚动。) break last_height new_height scroll_attempts 1 print(f已滚动 {scroll_attempts} 次当前页面高度: {new_height}) # 在搜索后调用 await scroll_to_load_more(page, scroll_pause_time3.0, max_scrolls10) # 滚动过程中拦截器会自动捕获每次滚动触发的API请求6.2 请求过滤与性能优化拦截所有请求**/*虽然简单但会处理大量无关请求如图片、CSS、JS影响性能。我们应该精确过滤。# 更精确的拦截模式只拦截包含特定路径的请求 api_patterns [ **/api/v4/search/search_items**, **/api/v4/item/get**, **/api/v4/recommend/recommend**, # 添加其他你需要的数据接口模式 ] async def setup_routes(page): 为页面设置多个精确的路由处理器 for pattern in api_patterns: await page.route(pattern, handle_route) # 对于其他请求可以设置一个快速通过的默认路由避免阻塞 # await page.route(**/*, lambda route: route.continue_()) # 注意这会覆盖前面的路由需要更精细的控制 # 在主函数中用 setup_routes(page) 代替 page.route(**/*, handle_route)6.3 错误处理与重试机制网络请求不稳定需要添加重试逻辑。import random async def robust_goto(page, url, max_retries3): 带重试的页面访问函数 for attempt in range(max_retries): try: response await page.goto(url, wait_untilnetworkidle, timeout60000) if response and response.ok: return response else: print(f第{attempt1}次访问响应状态码异常: {response.status if response else 无响应}) except Exception as e: print(f第{attempt1}次访问失败: {e}) if attempt max_retries - 1: wait_time random.uniform(5, 15) # 随机等待5-15秒 print(f等待{wait_time:.1f}秒后重试...) await asyncio.sleep(wait_time) raise Exception(f访问 {url} 失败已达最大重试次数 {max_retries}) # 使用它 try: await robust_goto(page, https://shopee.sg) except Exception as e: print(f无法访问首页: {e}) # 可能需要更换IP或检查网络7. 常见问题排查与实战心得7.1 问题速查表问题现象可能原因排查步骤与解决方案拦截不到目标API请求1. 路由模式不匹配。2. 请求在页面加载前就已发起如SSR。3. 请求来自iframe或Worker。1. 使用page.on(request, lambda req: print(req.url))监听所有请求URL确认目标URL的确切模式。2. 在page.goto()之前就设置路由(page.route)。3. 检查请求来源可能需要为iframe页面单独设置路由。route.fetch()抛出异常或超时1. 请求被浏览器策略如CORS阻止。2. 服务器端中断了连接。1. 尝试使用route.continue_()代替route.fetch()然后通过page.on(response)事件来获取响应但这可能拿不到响应体。2. 检查网络稳定性增加超时时间。解码后的JSON解析失败1. 响应不是JSON格式可能是HTML错误页。2. 解码仍然不正确残留乱码破坏了JSON结构。3. 响应是gzip压缩但未正确处理。1. 将解码后的文本前500字符打印出来看是否是HTML。2. 检查decode_response_body函数尝试latin-1编码它不会解码失败但可能产生乱码字符可用于查看结构。3. Playwright的fetch()通常会自动解压无需手动处理。被网站识别并屏蔽1. 浏览器指纹不完整。2. 行为模式过于规律。3. IP地址被标记。1. 完善add_init_script中的脚本添加更多属性覆盖。2. 在操作间添加随机延迟(await page.wait_for_timeout(random.randint(1000, 3000)))。3. 使用高质量的住宅代理IP池并在Playwright上下文中配置。页面卡住或元素找不到1. 页面加载未完成。2. 元素选择器不对或动态生成。3. 被弹窗如登录框、验证码遮挡。1. 使用page.wait_for_load_state(networkidle)或等待特定元素出现(page.wait_for_selector)。2. 使用Playwright的调试工具(playwright codegen)重新生成选择器。3. 编写代码处理常见弹窗或手动处理一次后保存Cookie。7.2 实战心得与避坑指南调试是王道在开发阶段务必使用headlessFalse模式并配合page.pause()方法或者使用Playwright Inspector (playwright codegen)让你可以实时看到浏览器操作和网络请求极大提升效率。Cookie管理如果遇到需要登录的页面可以先手动登录一次然后使用context.storage_state(pathstate.json)保存会话状态包括Cookie、LocalStorage。下次启动时用browser.new_context(storage_statestate.json)恢复避免重复登录和触发风控。控制请求节奏即使是模拟真人也不要过于频繁地触发搜索或翻页。在关键操作之间设置随机的、人性化的等待时间是长期稳定运行的关键。关注接口变化电商平台的接口路径和参数可能随时变更。定期运行你的脚本并关注captured_data是否为空或结构是否改变建立简单的监控告警。数据去重与存储拦截到的接口可能会被重复调用如滚动加载。在存储数据时根据商品ID等唯一标识进行去重。考虑使用数据库如SQLite、MongoDB而非单纯写入JSON文件便于后续管理和分析。尊重robots.txt在开始大规模采集前检查目标网站的robots.txt文件如https://shopee.sg/robots.txt了解网站允许和禁止爬取的路径。合规采集是长期项目的基础。通过以上从环境配置、反爬分析、核心代码实现到问题排查的完整流程你应该已经能够搭建一个稳定、高效的Shopee数据采集系统。这套以Playwright网络拦截为核心的方案其思想同样适用于其他复杂JavaScript渲染且接口加密的现代Web应用希望这套“组合拳”能成为你数据采集工具箱中的利器。
