
Playwright 抓取网页时,响应可能是文本(HTML/JSON)或二进制(图片、字体、JS/CSS),直接调用 response.text() 并用 UTF-8 写入会导致 UnicodeDecodeError;应根据响应类型选择 .text() 或 .body(),并动态命名文件以避免覆盖。
playwright 抓取网页时,响应可能是文本(html/json)或二进制(图片、字体、js/css),直接调用 `response.text()` 并用 utf-8 写入会导致 `unicodedecodeerror`;应根据响应类型选择 `.text()` 或 `.body()`,并动态命名文件以避免覆盖。
在使用 Playwright 进行网络抓取时,page.on('response') 会捕获所有 HTTP 响应——不仅包括 HTML 页面,还包括图片(PNG/JPEG)、样式表(CSS)、脚本(JS)、字体(WOFF)、JSON API 数据等。这些资源的数据类型各不相同:HTML 和 JSON 是 UTF-8 文本,而 PNG 文件以字节 0x89 0x50 0x4E 0x47(即 ‰PNG)开头,无法用 str.encode('utf-8') 解码,因此调用 response.text() 会抛出 UnicodeDecodeError。
✅ 正确做法是:
-
先判断响应是否为可读文本:检查
response.headers.get("content-type")是否包含text/、application/json、application/javascript等; -
对二进制资源使用
.body()+"wb"模式写入; -
对文本资源使用
.text()+"w"模式写入(并显式指定encoding="utf-8"); -
为每个响应生成唯一文件名(如基于 URL 路径或哈希),避免反复覆盖
copy.txt; -
添加基础过滤:跳过空响应(
not response.ok)或非关键资源类型(如media,font,other)。
以下是一个健壮的响应处理器示例:
import os
import re
from urllib.parse import urlparse, unquote
from playwright.sync_api import sync_playwright
output_dir = "scraped_responses"
os.makedirs(output_dir, exist_ok=True)
def sanitize_filename(url: str) -> str:
"""从 URL 提取安全、去重的文件名(保留扩展名)"""
parsed = urlparse(url)
path = unquote(parsed.path.strip("/"))
if not path:
path = "index.html"
# 替换非法字符,截断过长路径
name = re.sub(r'[<>:"/\|?*]', "_", path)[-120:]
if not os.path.splitext(name)[1]:
name += ".html"
return name
def handle_response(response):
# 跳过非成功响应或无内容的请求
if not response.ok or not response.url:
return
content_type = response.headers.get("content-type", "").lower()
resource_type = response.request.resource_type
# 忽略明显非目标资源(可按需调整)
if resource_type in ("media", "font", "other"):
return
# 判断是否为文本资源
is_text = (
content_type.startswith("text/") or
"json" in content_type or
"javascript" in content_type or
"xml" in content_type or
resource_type in ("document", "script", "stylesheet")
)
file_path = os.path.join(output_dir, sanitize_filename(response.url))
try:
if is_text:
text_content = response.text()
with open(file_path, "w", encoding="utf-8") as f:
f.write(text_content)
print(f"[TEXT] Saved {file_path} ({len(text_content)} chars)")
else:
binary_content = response.body()
with open(file_path, "wb") as f:
f.write(binary_content)
print(f"[BINARY] Saved {file_path} ({len(binary_content)} bytes)")
except Exception as e:
print(f"[ERROR] Failed to save {response.url}: {e}")
def main():
with sync_playwright() as p:
browser = p.chromium.launch(headless=True)
page = browser.new_page()
page.on("response", handle_response)
# 推荐使用 wait_until="networkidle" 替代固定 timeout
page.goto(
"https://www.booking.com/hotel/it/hotelnordroma.en-gb.html?aid=304142&checkin=2025-05-15&checkout=2025-05-16#map_opened-map_trigger_header_pin",
wait_until="networkidle"
)
browser.close()
if __name__ == "__main__":
main()? 关键注意事项:
- ❌ 不要对所有响应统一调用
.text()—— 这是引发UnicodeDecodeError的根本原因; - ✅ 使用
response.headers["content-type"]和response.request.resource_type双重判断更可靠; - ⚠️
response.text()内部会尝试自动解码(基于Content-Type中的charset),若缺失或错误仍可能失败,生产环境建议配合try/except; - ? 若仅需特定数据(如某 AJAX 返回的 JSON),推荐监听特定 URL 模式(如
page.route("**/api/**", ...))而非捕获全部响应,效率更高、结果更可控。
掌握响应类型识别与分路处理,是构建稳定 Web 抓取流程的基础能力。


















