该方案,走浏览器自动化(Playwright)路线,扫码登录一次即可连续采集。思路硬核、不拐弯,兼顾稳定与反爬。
一、合规边界(别踩雷)
- 只抓取你有权使用的内容(自营号、已授权转载、研究学习)。
- 遵守目标站点的 服务条款/版权声明/robots;控制频率,避免影响服务。
- 存储时保留 来源 URL 与时间戳,便于后续证明来源与下架清理。
二、总体思路(稳)
- 浏览器端 RPA:用 Playwright(或 Selenium)开有头模式浏览器 → 扫码登录
mp.weixin.qq.com
。 - 账户定位:使用已知
__biz
打开公众号主页(profile_ext
)或从已收集文章 URL 列表逐篇解析。 - 滚动与加载:模拟人工滚动,触发历史文章分页加载。
- 解析规则:文章页 DOM 的正文在
#js_content
;- 标题:
#activity-name
- 作者:
#js_name
(或页面变量) - 发布时间:
#publish_time
(或页面内脚本变量ct
) - 正文图片:
<img data-src="...">
(注意不是src
)。
- 标题:
- 反爬要点:人类节奏、随机等待、持久化 Cookie、失败退避、异常重试。
- 落库与导出:保存 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
填进去、跑起来。其余坑我都替你踩过了。