Python实现数据爬取（Playwright异步版）

2301_79959211

14754人浏览 · 2026-03-03 11:08:14

2301_79959211 · 2026-03-03 11:08:14 发布

在做短视频数据研究和舆情分析的过程中，我一直在思考一个问题：如何在保证稳定性的前提下，实现一个可长期运行、可扩展、可维护的某音数据采集系统，而不是写一堆临时脚本反复推倒重来。

基于这个目标，我设计并实现了这套爬虫核心代码，基于 Playwright 构建真实浏览器环境，支持 CDP 模式与 stealth 反检测，同时内置代理池、并发控制、时间范围过滤、评论批量抓取、媒体自动识别下载（视频/图文自动判断），三种模式（关键词搜索 / 指定视频 / 创作者主页）统一在一个可扩展框架里，通过 config 控制行为，store 负责数据落地，client 负责接口请求，结构清晰、职责分离，既考虑了稳定性与风控规避，又兼顾了扩展性和可维护性，真正做到了可长期运行、可二次开发、可规模化使用，而不是一次性采集工具，这也是我在设计时最看重的一点。

前言

本文介绍如何基于 Playwright + asyncio 实现爬虫核心模块，支持三种采集模式：

search：按关键词搜索批量采集作品信息，并批量抓评论

detail：给定作品链接/ID，抓作品详情与评论

creator：给定创作者主页链接，抓创作者信息、其作品列表、作品评论

同时，该模块内置多种“工程化能力”：

反检测（stealth 脚本 + 可选 CDP 模式）

代理池（自动获取/刷新代理）

并发控制（Semaphore）

时间范围过滤（按 create_time 过滤）

自动媒体下载（短视频/图文笔记自动识别）

一、环境配置与依赖准备

1.1 安装依赖

该模块使用 Playwright 异步 API（playwright.async_api），运行前需要安装并初始化浏览器驱动：

pip install playwright
playwright install

1.2 导入模块

import asyncio
import os
import random
import time
from asyncio import Task
from typing import Any, Dict, List, Optional

from playwright.async_api import BrowserContext, BrowserType, Page, Playwright, async_playwright

import config
from base.base_crawler import AbstractCrawler
from proxy.proxy_ip_pool import IpInfoModel, create_ip_pool
from store import douyin as douyin_store
from tools import utils
from tools.cdp_browser import CDPBrowserManager
from var import crawler_type_var, source_keyword_var

from .client import DouYinClient
from .exception import DataFetchError
from .field import PublishTimeType
from .help import parse_video_info_from_url, parse_creator_info_from_url
from .login import MouYinLogin

二、爬虫类结构与初始化

2.1 定义爬虫类

class MouYinCrawler(AbstractCrawler):
    context_page: Page
    dy_client: DouYinClient
    browser_context: BrowserContext
    cdp_manager: Optional[CDPBrowserManager]

    def __init__(self) -> None:
        self.index_url = "https://www.douyin.com"
        self.cdp_manager = None
        self.ip_proxy_pool = None

这里 MouYinCrawler 继承 AbstractCrawler，需要把“启动浏览器 → 登录 → 分模式采集 → 关闭资源”串起来。

三、程序入口：start() 全流程解析

start() 是整个爬虫运行的主入口，逻辑非常清晰：

代理池初始化（可选）
启动 Playwright（CDP 模式或标准模式）
打开抖音首页（带超时与重试）
创建 MouYinClient（从浏览器上下文获取 cookies）
检查登录状态，不通过就登录并更新 cookies
根据 CRAWLER_TYPE 分发到 search/detail/creator 模式

核心代码骨架：

async def start(self) -> None:
    playwright_proxy_format, httpx_proxy_format = None, None
    if config.ENABLE_IP_PROXY:
        self.ip_proxy_pool = await create_ip_pool(config.IP_PROXY_POOL_COUNT, enable_validate_ip=True)
        ip_proxy_info: IpInfoModel = await self.ip_proxy_pool.get_proxy()
        playwright_proxy_format, httpx_proxy_format = utils.format_proxy_info(ip_proxy_info)

    async with async_playwright() as playwright:
        if config.ENABLE_CDP_MODE:
            self.browser_context = await self.launch_browser_with_cdp(
                playwright, playwright_proxy_format, None, headless=config.CDP_HEADLESS
            )
        else:
            chromium = playwright.chromium
            self.browser_context = await self.launch_browser(
                chromium, playwright_proxy_format, user_agent=None, headless=config.HEADLESS
            )
            await self.browser_context.add_init_script(path="libs/stealth.min.js")

        self.context_page = await self.browser_context.new_page()
        try:
            await self.context_page.goto(self.index_url, timeout=180000, wait_until="domcontentloaded")
        except Exception:
            await self.context_page.reload(timeout=300000, wait_until="domcontentloaded")

        self.dy_client = await self.create_douyin_client(httpx_proxy_format)

        if not await self.dy_client.pong(browser_context=self.browser_context):
            login_obj = DouYinLogin(
                login_type=config.LOGIN_TYPE,
                login_phone="",
                browser_context=self.browser_context,
                context_page=self.context_page,
                cookie_str=config.COOKIES,
            )
            await login_obj.begin()
            await self.dy_client.update_cookies(browser_context=self.browser_context)

        crawler_type_var.set(config.CRAWLER_TYPE)
        if config.CRAWLER_TYPE == "search":
            await self.search()
        elif config.CRAWLER_TYPE == "detail":
            await self.get_specified_awemes()
        elif config.CRAWLER_TYPE == "creator":
            await self.get_creators_and_videos()

几个关键点需要特别注意：

goto(... timeout=180000)：给首页 3 分钟加载时间，避免网络慢导致失败

pong()：用来判断当前 cookies 是否可用，登录态是否有效

登录成功后：update_cookies() 让 client 立刻用新 cookies 发请求

crawler_type_var / source_keyword_var：用于在日志/存储层区分不同任务上下文

四、模式一：关键词搜索

search() 的核心流程如下所示：

每个关键词循环分页请求搜索结果
把结果中的 aweme_id 收集起来
每条作品：
- 时间范围过滤
- 写入作品信息
- 下载媒体（视频/图文）
- 每条作品后随机延迟 2~5 秒
每页后 asyncio.sleep(config.CRAWLER_MAX_SLEEP_SEC)
最后：批量抓取该关键词下所有作品的评论

核心代码：

async def search(self) -> None:
    dy_limit_count = 10
    if config.CRAWLER_MAX_NOTES_COUNT < dy_limit_count:
        config.CRAWLER_MAX_NOTES_COUNT = dy_limit_count

    start_page = config.START_PAGE
    for keyword in config.KEYWORDS.split(","):
        source_keyword_var.set(keyword)
        aweme_list: List[str] = []
        page = 0
        dy_search_id = ""

        while (page - start_page + 1) * dy_limit_count <= config.CRAWLER_MAX_NOTES_COUNT:
            if page < start_page:
                page += 1
                continue

            posts_res = await self.dy_client.search_info_by_keyword(
                keyword=keyword,
                offset=page * dy_limit_count - dy_limit_count,
                publish_time=PublishTimeType(config.PUBLISH_TIME_TYPE),
                search_id=dy_search_id,
            )
            page += 1

            dy_search_id = posts_res.get("extra", {}).get("logid", "")

            for post_item in posts_res.get("data"):
                aweme_info = post_item.get("aweme_info") or post_item.get("aweme_mix_info", {}).get("mix_items")[0]

                if not self._is_in_time_range(aweme_info):
                    continue

                aweme_list.append(aweme_info.get("aweme_id", ""))
                await douyin_store.update_douyin_aweme(aweme_item=aweme_info)
                await self.get_aweme_media(aweme_item=aweme_info)

                delay_seconds = random.uniform(2, 5)
                time.sleep(delay_seconds)

            await asyncio.sleep(config.CRAWLER_MAX_SLEEP_SEC)

        await self.batch_get_note_comments(aweme_list)

五、模式二：指定作品详情

核心逻辑：

async def get_specified_awemes(self):
    aweme_id_list = []
    for video_url in config.DY_SPECIFIED_ID_LIST:
        video_info = parse_video_info_from_url(video_url)

        if video_info.url_type == "short":
            resolved_url = await self.dy_client.resolve_short_url(video_url)
            if resolved_url:
                video_info = parse_video_info_from_url(resolved_url)
            else:
                continue

        aweme_id_list.append(video_info.aweme_id)

    semaphore = asyncio.Semaphore(config.MAX_CONCURRENCY_NUM)
    task_list = [self.get_aweme_detail(aweme_id=aweme_id, semaphore=semaphore) for aweme_id in aweme_id_list]
    aweme_details = await asyncio.gather(*task_list)

    filtered_aweme_ids = []
    for aweme_detail in aweme_details:
        if aweme_detail is not None and self._is_in_time_range(aweme_detail):
            await douyin_store.update_douyin_aweme(aweme_item=aweme_detail)
            await self.get_aweme_media(aweme_item=aweme_detail)
            filtered_aweme_ids.append(aweme_detail.get("aweme_id"))

            delay_seconds = random.uniform(2, 5)
            time.sleep(delay_seconds)

    await self.batch_get_note_comments(filtered_aweme_ids)

六、模式三：创作者模式

这一模式会实现如下功能：

解析创作者主页 URL 得到 sec_user_id
拉取创作者基本信息并保存
拉取该创作者全部作品列表
并发补齐每条作品详情（并时间过滤）
下载媒体
批量抓评论

关键代码流程：

async def get_creators_and_videos(self) -> None:
    for creator_url in config.DY_CREATOR_ID_LIST:
        creator_info_parsed = parse_creator_info_from_url(creator_url)
        user_id = creator_info_parsed.sec_user_id

        creator_info: Dict = await self.dy_client.get_user_info(user_id)
        if creator_info:
            await douyin_store.save_creator(user_id, creator=creator_info)

        all_video_list = await self.dy_client.get_all_user_aweme_posts(
            sec_user_id=user_id, callback=self.fetch_creator_video_detail
        )

        video_ids = [video_item.get("aweme_id") for video_item in all_video_list]
        await self.batch_get_note_comments(video_ids)