RPA-微信公众号爆文数据抓取

RPA 抓取微信公众号文章方案-2025

该方案,走浏览器自动化(Playwright)路线,扫码登录一次即可连续采集。思路硬核、不拐弯,兼顾稳定与反爬。


一、合规边界(别踩雷)

  • 只抓取你有权使用的内容(自营号、已授权转载、研究学习)。
  • 遵守目标站点的 服务条款/版权声明/robots;控制频率,避免影响服务。
  • 存储时保留 来源 URL 与时间戳,便于后续证明来源与下架清理。

二、总体思路(稳)

  1. 浏览器端 RPA:用 Playwright(或 Selenium)开有头模式浏览器 → 扫码登录 mp.weixin.qq.com
  2. 账户定位:使用已知 __biz 打开公众号主页(profile_ext)或从已收集文章 URL 列表逐篇解析。
  3. 滚动与加载:模拟人工滚动,触发历史文章分页加载。
  4. 解析规则:文章页 DOM 的正文在 #js_content
    • 标题:#activity-name
    • 作者:#js_name(或页面变量)
    • 发布时间:#publish_time(或页面内脚本变量 ct
    • 正文图片:<img data-src="...">(注意不是 src)。
  5. 反爬要点:人类节奏、随机等待、持久化 Cookie、失败退避、异常重试。
  6. 落库与导出:保存 SQLite/CSV/Markdown;图片另存本地并把 data-src 改写为本地路径。

三、环境与依赖

# Python 3.10+
pip install playwright pandas markdownify python-slugify
playwright install

四、可直接用的采集脚本(Playwright / Python)

功能:扫码登录 → 采集指定公众号历史文章(需 __biz)→ 逐篇抓取正文 → 落地到 data/(HTML/Markdown/图片/metadata.csv)

import os, re, time, json, asyncio, contextlib
from pathlib import Path
from urllib.parse import urlparse, parse_qs, urljoin
from slugify import slugify
import pandas as pd
from markdownify import markdownify as md
from playwright.async_api import async_playwright

# ==== 配置区 ====
BIZ_LIST = [
    # 目标公众号的 __biz 列表(必填)。可从任一文章 URL 的 query 里获取。
    # 例:https://mp.weixin.qq.com/s?...&__biz=MzA5Nj...==&mid=...&sn=...
    "填入__biz_1",
    # "填入__biz_2",
]
MAX_ARTICLES_PER_BIZ = 80           # 每个号最多抓取数(按需调大)
SCROLL_ROUNDS = 20                   # 历史消息滚动次数(按需调大)
SCROLL_PAUSE_SEC = (1.2, 2.5)        # 滚动间隔的随机范围
REQUEST_PAUSE_SEC = (1.0, 2.2)       # 打开文章的随机等待
OUT_DIR = Path("data")
STATE_FILE = "state/auth.json"       # 持久化登录态(避免每次扫码)
HEADFUL = True                       # 有头模式更稳更像真人
TIMEOUT = 30000                      # ms

# ==== 工具 ====
import random
def rwait(a,b): time.sleep(random.uniform(a,b))

def ensure_dirs():
    (OUT_DIR / "html").mkdir(parents=True, exist_ok=True)
    (OUT_DIR / "md").mkdir(parents=True, exist_ok=True)
    (OUT_DIR / "img").mkdir(parents=True, exist_ok=True)

def extract_query(url, key):
    try:
        return parse_qs(urlparse(url).query).get(key, [""])[0]
    except Exception:
        return ""

def clean_text(s): return re.sub(r"\s+", " ", s or "").strip()

async def login_if_needed(context, page):
    # 打开任意 mp 页面,触发登录
    await page.goto("https://mp.weixin.qq.com/", timeout=TIMEOUT)
    # 若已登录会直接进入后台/空白页;否则通常会出现二维码登录
    # 给足时间扫码
    await page.wait_for_timeout(1000)
    if "登录" in (await page.content()):
        print("请在 60 秒内完成扫码登录 WeChat(手机端确认)……")
        await page.wait_for_timeout(60000)

async def collect_profile_articles(page, biz):
    # 公众号主页(需登录态)
    url = f"https://mp.weixin.qq.com/mp/profile_ext?action=home&__biz={biz}#wechat_redirect"
    await page.goto(url, timeout=TIMEOUT)
    await page.wait_for_timeout(2000)

    # 滚动加载历史消息
    for i in range(SCROLL_ROUNDS):
        await page.mouse.wheel(0, 5000)
        rwait(*SCROLL_PAUSE_SEC)

    # 页面中历史文章卡片(微信前端偶有变动,选择器需按实际 DOM 校验)
    # 典型结构:a.weui_media_title / a.appmsg_item
    cards = await page.locator("a.weui_media_title, a.appmsg_item").all()
    links = []
    for c in cards:
        with contextlib.suppress(Exception):
            href = await c.get_attribute("hrefs") or await c.get_attribute("href")
            # profile 列表里 often 是相对链接
            if href and href.startswith("/"):
                href = urljoin("https://mp.weixin.qq.com", href)
            if href and "mp.weixin.qq.com" in href:
                links.append(href)

    # 去重 & 限量
    uniq = []
    seen = set()
    for u in links:
        key = (extract_query(u, "mid"), extract_query(u, "idx"))
        if key not in seen and all(key):
            seen.add(key); uniq.append(u)
        if len(uniq) >= MAX_ARTICLES_PER_BIZ:
            break
    return uniq

async def parse_article(page, url):
    await page.goto(url, timeout=TIMEOUT)
    rwait(*REQUEST_PAUSE_SEC)

    # 标题/作者/时间
    title = clean_text(await page.locator("#activity-name").inner_text()) if await page.locator("#activity-name").count() else ""
    author = clean_text(await page.locator("#js_name").inner_text()) if await page.locator("#js_name").count() else ""
    pub_time = clean_text(await page.locator("#publish_time").inner_text()) if await page.locator("#publish_time").count() else ""

    # 正文容器
    html = await page.locator("#js_content").inner_html() if await page.locator("#js_content").count() else ""

    # 抓取并本地化图片(data-src)
    imgs = await page.locator("#js_content img").all()
    img_local_map = {}
    for i, img in enumerate(imgs, start=1):
        with contextlib.suppress(Exception):
            src = await img.get_attribute("data-src") or await img.get_attribute("src")
            if not src: continue
            # 存本地
            fn = f"{slugify(title)[:60]}_{i}.jpg"
            path = OUT_DIR / "img" / fn
            try:
                # 直接下载:用 Playwright fetch
                resp = await page.request.get(src, timeout=TIMEOUT)
                if resp.ok:
                    content = await resp.body()
                    path.write_bytes(content)
                    img_local_map[src] = f"./img/{fn}"
            except Exception:
                pass

    # 替换 HTML 中的 data-src 为本地相对路径
    for src, local in img_local_map.items():
        html = html.replace(src, local)

    # 元数据
    meta = {
        "url": url,
        "biz": extract_query(url, "__biz"),
        "mid": extract_query(url, "mid"),
        "idx": extract_query(url, "idx"),
        "sn": extract_query(url, "sn"),
        "title": title,
        "author": author,
        "publish_time": pub_time,
        "ts_crawl": pd.Timestamp.utcnow().isoformat()
    }
    return meta, html

async def main():
    ensure_dirs()
    os.makedirs(Path(STATE_FILE).parent, exist_ok=True)

    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=not HEADFUL)
        context = await browser.new_context(storage_state=STATE_FILE if os.path.exists(STATE_FILE) else None)
        page = await context.new_page()

        await login_if_needed(context, page)
        # 保存登录态(下次免扫码)
        await context.storage_state(path=STATE_FILE)

        all_rows = []
        for biz in BIZ_LIST:
            print(f"采集公众号:{biz}")
            links = await collect_profile_articles(page, biz)
            print(f"发现文章:{len(links)}")

            for i, url in enumerate(links, start=1):
                try:
                    meta, html = await parse_article(page, url)
                    # 保存 HTML
                    base = f"{meta['biz']}_{meta['mid']}_{meta['idx']}"
                    html_file = OUT_DIR / "html" / f"{base}.html"
                    md_file = OUT_DIR / "md" / f"{base}.md"

                    html_file.write_text(html, encoding="utf-8")
                    md_file.write_text(md(html, heading_style="ATX"), encoding="utf-8")

                    all_rows.append(meta)
                    print(f"[{i}/{len(links)}] {meta['title']}")
                    rwait(*REQUEST_PAUSE_SEC)
                except Exception as e:
                    print("失败:", url, e)
                    rwait(2, 4)

        if all_rows:
            df = pd.DataFrame(all_rows)
            df.to_csv(OUT_DIR / "metadata.csv", index=False, encoding="utf-8-sig")
        await browser.close()

if __name__ == "__main__":
    asyncio.run(main())

五、关键细节与“避坑合集”

  • 登录与状态storage_state 持久化 Cookie,不要每次扫码;若频繁 403/跳转登录,先清理再重登。
  • 选择器易变:微信前端偶尔改 DOM,建议在本地 devtools 验证 #js_content#activity-name#publish_time 等。
  • 图片字段:正文图片常在 data-src;必须抓取原链接并本地化,避免后续过期。
  • 节流防封
    • 有头模式 + 合理随机等待 + 单标签页串行。
    • 每号每分钟 ≤ 6 次导航为宜;遇到“操作频繁,请稍后再试”,退避重试。
  • 去重主键(__biz, mid, idx) 唯一索引,避免重复抓取。
  • 历史文章页/mp/profile_ext?action=home&__biz=... 需要已登录态;实测滚动加载更稳,不直接扒接口。
  • 持久可靠的归档:同时保存 HTML + Markdown + 图片,并记录 url抓取时间发布时刻

六、替代路径(视你的场景)

  • Windows 桌面 RPA(UiPath / Power Automate Desktop)
    流程:启动 Chrome → 打开 profile_ext → 循环点击卡片 → 复制 #js_content→ 写入文件 → 退回列表。优点是无代码/易维护,缺点是部署与容错差一点。
  • 安卓端自动化(uiautomator2/Appium)
    打开微信 App → 进入公众号 → 历史消息 → 逐条打开 → “复制链接”交给后端抓取 → 回填结果。优点是贴近真机行为,稳定性高;缺点是开发成本稍高。
  • 现成采集服务/代理:第三方平台/API(注意合规与风控),适合非技术团队。

七、下一步的狠活

  • 接入 SQLite 做增量抓取与断点续传。
  • 自动将正文转 Markdown + 本地图片,一键导入你的 CMS(WordPress/Notion/自建站)。
  • 增加 失败队列与重试策略、代理池(仅在必要时使用)。
  • 加上 指纹模拟/人机轨迹 提升“像素级人味”。

需要你做的只有两步:把目标公众号的 __biz 填进去、跑起来。其余坑我都替你踩过了。

Releated Posts

Claude Code 极简入门:3 条铁律,优雅起步,远离“代码屎山”(2025 实践版)

越来越多同学问:怎么用 Claude Code 既省力又不把项目写成一团乱麻?答案不在“更神奇的提示词”,而在更清晰的协作流程。本文给出一份可落地的极简入门清单:仅 3 条铁律,覆盖从需求→方案→落地的最短闭环。做到了,就能在大多数团队与个人项目里稳稳避免“屎山效应”。 为什么屎山总是悄悄长出来? […]

ByByGit HubAug 26, 2025

最适合中国开发者的云服务对比(AWS/GCP/Azure)

随着云计算在全球范围的普及,中国开发者在选择云服务时面临着一个核心问题:究竟哪家国际云平台更适合自己? 目前全球主流的三大公有云厂商分别是 AWS(Amazon Web Services)、GCP(Google Cloud Platform)、Azure(Microsoft Azure)。它们在性能[…]

ByByGit HubAug 24, 2025
0 0 votes
Article Rating
Subscribe
Notify of
guest
0 Comments
Oldest
Newest Most Voted
Inline Feedbacks
View all comments