簡介本資源是一套面向 Python 爬蟲學(xué)習(xí)者與數(shù)據(jù)分析人員的微信公眾號文章采集工具包聚焦于從微信公眾平臺批量獲取公眾號所有文章的 URL 地址適合具備一定 Python 基礎(chǔ)、希望練習(xí)真實(shí)站點(diǎn)抓取與接口調(diào)試的開發(fā)者。壓縮包共 30 個(gè)文件約 17.22MB以 13 個(gè) py 腳本為核心輔以 5 個(gè) md 說明文檔、若干 jpg/png 截圖以及 chromedriver 驅(qū)動(dòng)文件覆蓋文章鏈接抓取、文章信息解析、結(jié)果讀取輸出等模塊并配有 requirements.txt 便于快速搭建運(yùn)行環(huán)境。文檔部分整理了獲取 appmsg_token、cookie_token 等關(guān)鍵參數(shù)的思路以及微信公眾號接口使用與自動(dòng)獲取微信參數(shù)的說明能幫助讀者理解抓取流程中的鑒權(quán)與請求構(gòu)造環(huán)節(jié)。目前已有 583 人學(xué)習(xí)下載可作為微信公眾號爬蟲入門與接口調(diào)試的實(shí)踐參考。1. 公眾號內(nèi)容采集一個(gè)被低估的爬蟲練手項(xiàng)目很多人第一次接觸爬蟲都是從公眾號文章開始的。原因很直接頁面結(jié)構(gòu)相對穩(wěn)定數(shù)據(jù)量適中反爬強(qiáng)度不算極端非常適合拿來練手。但真上手之后你會發(fā)現(xiàn)事情沒那么簡單——文章鏈接藏在歷史消息里正文被各種內(nèi)聯(lián)樣式包裹圖片還做了防盜鏈處理。我見過不少開發(fā)者在這個(gè)項(xiàng)目上翻車不是卡在請求上就是卡在解析上。這個(gè)項(xiàng)目要解決的問題很明確給定一個(gè)公眾號批量抓取它的歷史文章列表再逐篇提取標(biāo)題、作者、發(fā)布時(shí)間、正文內(nèi)容和封面圖。適合誰剛學(xué)完 requests 和 BeautifulSoup、想找一個(gè)完整項(xiàng)目練手的 Python 開發(fā)者也適合需要做內(nèi)容歸檔、數(shù)據(jù)分析的運(yùn)營或研究人員。技術(shù)棧不復(fù)雜核心就是 requests BeautifulSoup 正則存儲用 CSV 或 SQLite 都行。2. 項(xiàng)目架構(gòu)拆解從入口請求到數(shù)據(jù)落盤的完整鏈路2.1 整體模塊劃分與數(shù)據(jù)流向這個(gè)項(xiàng)目的架構(gòu)不復(fù)雜但該有的分層不能少。我一般會把它拆成四個(gè)模塊請求層、解析層、存儲層、調(diào)度層。請求層負(fù)責(zé)模擬客戶端發(fā)送 HTTP 請求拿到原始 HTML解析層從 HTML 里提取結(jié)構(gòu)化數(shù)據(jù)存儲層把數(shù)據(jù)寫到本地文件或數(shù)據(jù)庫調(diào)度層控制翻頁邏輯和請求頻率。數(shù)據(jù)流向是這樣的調(diào)度層發(fā)起第一頁請求 → 請求層返回 HTML → 解析層提取文章列表 → 存儲層保存列表數(shù)據(jù) → 調(diào)度層根據(jù)分頁信息發(fā)起下一頁請求 → 重復(fù)直到?jīng)]有更多數(shù)據(jù) → 對每篇文章詳情頁發(fā)起請求 → 解析正文 → 存儲。為什么這么分因?yàn)楣娞柕牧斜眄摵驮斍轫摻Y(jié)構(gòu)完全不同解析邏輯必須分開寫。而且請求頻率控制、Cookie 管理這些通用邏輯放在請求層后面改起來方便。2.2 請求層Cookie 與請求頭的處理公眾號的很多接口需要攜帶有效的 Cookie 才能返回?cái)?shù)據(jù)尤其是歷史文章列表。常見做法是手動(dòng)從瀏覽器開發(fā)者工具里復(fù)制 Cookie 和 User-Agent寫進(jìn)配置里。import requests import time import random # 配置請求頭Cookie 需要從瀏覽器手動(dòng)獲取 HEADERS { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36, Cookie: 你的Cookie字符串, # 從瀏覽器開發(fā)者工具復(fù)制 Referer: https://mp.weixin.qq.com/ } def fetch_page(url, retry3): 發(fā)送 GET 請求帶重試機(jī)制 for i in range(retry): try: resp requests.get(url, headersHEADERS, timeout10) if resp.status_code 200: return resp.text else: print(f狀態(tài)碼 {resp.status_code}第 {i1} 次重試) except requests.RequestException as e: print(f請求異常{e}第 {i1} 次重試) time.sleep(random.uniform(2, 5)) # 隨機(jī)延時(shí)降低被封風(fēng)險(xiǎn) return None這段代碼的關(guān)鍵點(diǎn)有三個(gè)。第一HEADERS里的Cookie是必須的沒有它很多接口直接返回空數(shù)據(jù)或者跳轉(zhuǎn)到驗(yàn)證頁。第二retry參數(shù)控制重試次數(shù)網(wǎng)絡(luò)抖動(dòng)時(shí)自動(dòng)重試比直接報(bào)錯(cuò)友好得多。第三time.sleep(random.uniform(2, 5))是保命操作固定間隔的請求很容易被識別為機(jī)器行為隨機(jī)延時(shí)能有效降低風(fēng)險(xiǎn)。參數(shù)怎么改timeout根據(jù)你的網(wǎng)絡(luò)情況調(diào)整一般 10 秒夠用。retry建議不超過 5 次太多會浪費(fèi)時(shí)間。Referer字段有些接口會校驗(yàn)帶上更穩(wěn)妥。2.3 解析層列表頁與詳情頁的提取邏輯列表頁返回的通常是 JSON 或者嵌在 HTML 里的 JSON 片段詳情頁則是完整的 HTML。兩種解析方式不一樣。from bs4 import BeautifulSoup import re import json def parse_article_list(html): 從列表頁 HTML 中提取文章條目 articles [] soup BeautifulSoup(html, html.parser) # 文章列表通常在特定的 div 或 script 標(biāo)簽中 items soup.select(div.article-item) # 根據(jù)實(shí)際頁面結(jié)構(gòu)調(diào)整 for item in items: title item.select_one(h3).get_text(stripTrue) link item.select_one(a)[href] # 補(bǔ)全相對鏈接 if link.startswith(/): link https://mp.weixin.qq.com link articles.append({title: title, url: link}) return articles def parse_article_detail(html): 從詳情頁提取正文、作者、發(fā)布時(shí)間 soup BeautifulSoup(html, html.parser) # 標(biāo)題 title_tag soup.select_one(h1.rich_media_title) title title_tag.get_text(stripTrue) if title_tag else # 作者 author_tag soup.select_one(span.rich_media_meta_text) author author_tag.get_text(stripTrue) if author_tag else # 發(fā)布時(shí)間 time_tag soup.select_one(em#publish_time) publish_time time_tag.get_text(stripTrue) if time_tag else # 正文內(nèi)容 content_tag soup.select_one(div#js_content) content content_tag.get_text(stripTrue) if content_tag else # 封面圖 cover_tag soup.select_one(img.rich_media_thumb) cover cover_tag[src] if cover_tag else return { title: title, author: author, publish_time: publish_time, content: content, cover: cover }解析層的核心是選擇器要寫對。公眾號頁面的 class 名經(jīng)常變所以選擇器不能寫死最好用多個(gè)備選方案。比如正文容器可能是div#js_content也可能是div.rich_media_content可以兩個(gè)都試一下。get_text(stripTrue)會把標(biāo)簽內(nèi)的文本提取出來并去掉首尾空白。正文里如果有換行和段落get_text默認(rèn)會丟掉格式如果需要保留段落結(jié)構(gòu)可以用get_text(separator\n)。2.4 存儲層CSV 與 SQLite 的選擇數(shù)據(jù)量小的時(shí)候 CSV 夠用數(shù)據(jù)量大或者需要頻繁查詢就用 SQLite。我一般兩個(gè)都寫方便切換。import csv import sqlite3 def save_to_csv(articles, filenamearticles.csv): 將文章列表寫入 CSV 文件 if not articles: return keys articles[0].keys() with open(filename, a, newline, encodingutf-8-sig) as f: writer csv.DictWriter(f, fieldnameskeys) # 文件為空時(shí)寫入表頭 if f.tell() 0: writer.writeheader() writer.writerows(articles) def save_to_sqlite(articles, db_patharticles.db): 將文章寫入 SQLite 數(shù)據(jù)庫 conn sqlite3.connect(db_path) cursor conn.cursor() cursor.execute( CREATE TABLE IF NOT EXISTS articles ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT, author TEXT, publish_time TEXT, content TEXT, cover TEXT, url TEXT UNIQUE ) ) for art in articles: try: cursor.execute( INSERT OR IGNORE INTO articles (title, author, publish_time, content, cover, url) VALUES (?, ?, ?, ?, ?, ?), (art.get(title), art.get(author), art.get(publish_time), art.get(content), art.get(cover), art.get(url)) ) except sqlite3.Error as e: print(f寫入失敗{e}) conn.commit() conn.close()CSV 用utf-8-sig編碼是為了 Excel 打開時(shí)不亂碼這個(gè)坑我踩過好幾次。SQLite 的url字段加了UNIQUE約束配合INSERT OR IGNORE可以自動(dòng)去重避免重復(fù)抓取同一篇文章。3. 運(yùn)行環(huán)境與依賴安裝從零把項(xiàng)目跑起來3.1 Python 版本與依賴清單這個(gè)項(xiàng)目對 Python 版本要求不高3.8 以上都能跑。依賴就三個(gè)核心庫pip install requests beautifulsoup4 lxmlrequests負(fù)責(zé)發(fā)請求beautifulsoup4負(fù)責(zé)解析 HTMLlxml是解析器后端比 Python 自帶的html.parser快不少。如果要用 SQLitePython 標(biāo)準(zhǔn)庫自帶sqlite3不用額外裝。我一般會再裝一個(gè)fake-useragent用來隨機(jī)生成 User-Agent降低被封的概率pip install fake-useragent3.2 配置文件與參數(shù)說明把 Cookie、目標(biāo)公眾號標(biāo)識、請求間隔這些可變參數(shù)抽到一個(gè)config.py里改起來不用翻主邏輯代碼。# config.py COOKIE 你的Cookie字符串 TARGET_BIZ 目標(biāo)公眾號的biz參數(shù) # 從文章鏈接里提取 MAX_PAGES 50 # 最大翻頁數(shù)防止無限循環(huán) REQUEST_DELAY (2, 5) # 隨機(jī)延時(shí)范圍單位秒 OUTPUT_FORMAT sqlite # 可選 csv 或 sqlite DB_PATH articles.db CSV_PATH articles.csvTARGET_BIZ是公眾號的唯一標(biāo)識通常藏在文章鏈接的__biz參數(shù)里。MAX_PAGES是保險(xiǎn)絲萬一翻頁邏輯出問題不會無限跑下去。REQUEST_DELAY控制請求頻率數(shù)值越大越安全但越慢。3.3 主流程串聯(lián)與運(yùn)行把前面幾個(gè)模塊串起來主流程大概長這樣from config import MAX_PAGES, OUTPUT_FORMAT from fetcher import fetch_page from parser import parse_article_list, parse_article_detail from storage import save_to_csv, save_to_sqlite import time import random def main(): all_articles [] for page in range(MAX_PAGES): # 構(gòu)造列表頁 URL具體參數(shù)根據(jù)實(shí)際情況調(diào)整 list_url fhttps://mp.weixin.qq.com/mp/profile_ext?actiongetmsgpage{page} html fetch_page(list_url) if not html: print(f第 {page} 頁請求失敗停止) break articles parse_article_list(html) if not articles: print(沒有更多文章了) break # 逐篇抓取詳情 for art in articles: detail_html fetch_page(art[url]) if detail_html: detail parse_article_detail(detail_html) detail[url] art[url] all_articles.append(detail) time.sleep(random.uniform(2, 5)) print(f第 {page} 頁完成累計(jì) {len(all_articles)} 篇) # 落盤 if OUTPUT_FORMAT csv: save_to_csv(all_articles) else: save_to_sqlite(all_articles) print(f抓取完成共 {len(all_articles)} 篇) if __name__ __main__: main()運(yùn)行方式就是python main.py。第一次跑建議把MAX_PAGES設(shè)成 2 或者 3先驗(yàn)證流程通不通再放開跑全量。4. 避坑與常見問題那些讓我熬夜的報(bào)錯(cuò)4.1 請求返回空數(shù)據(jù)或跳轉(zhuǎn)驗(yàn)證頁現(xiàn)象請求列表頁返回的 HTML 里沒有文章條目或者直接跳到一個(gè)驗(yàn)證頁面。原因Cookie 過期了或者請求頭里的 User-Agent 和 Cookie 不匹配。公眾號對 Cookie 的有效期管得比較嚴(yán)一般幾小時(shí)到一天就會失效。解決重新從瀏覽器復(fù)制 Cookie確保 User-Agent 和復(fù)制 Cookie 時(shí)用的瀏覽器一致。如果還是不行檢查Referer字段是否缺失。4.2 正文提取為空或只有部分內(nèi)容現(xiàn)象parse_article_detail返回的content字段是空字符串或者只有開頭一小段。原因公眾號正文可能是懶加載的初始 HTML 里只有占位符。另外有些文章正文在 iframe 里直接解析外層 HTML 拿不到。解決先確認(rèn)頁面是否完整加載。如果是懶加載需要找到實(shí)際的數(shù)據(jù)接口。如果是 iframe用requests請求 iframe 的src地址再解析。我一般會先把 HTML 存到本地用瀏覽器打開看看結(jié)構(gòu)再決定怎么解析。4.3 圖片鏈接失效或防盜鏈現(xiàn)象抓下來的封面圖鏈接在瀏覽器里打不開返回 403。原因公眾號圖片有防盜鏈機(jī)制直接請求圖片 URL 會被拒絕。解決下載圖片時(shí)帶上Referer頭值設(shè)為https://mp.weixin.qq.com/。如果還不行可能需要把圖片 URL 里的wx_fmt參數(shù)去掉再試。4.4 翻頁邏輯死循環(huán)現(xiàn)象程序一直跑但抓到的文章列表始終是同一批。原因翻頁參數(shù)沒變或者接口返回的數(shù)據(jù)結(jié)構(gòu)在最后一頁時(shí)沒有正確判斷。解決在循環(huán)里加一個(gè)判斷如果本次解析出的文章列表和上一次完全一樣就break。另外MAX_PAGES一定要設(shè)這是最后的保險(xiǎn)。4.5 寫入 CSV 時(shí)中文亂碼現(xiàn)象用 Excel 打開 CSV 文件中文全是亂碼。原因Excel 默認(rèn)用 GBK 編碼打開 CSV而文件是 UTF-8 編碼。解決寫入時(shí)用utf-8-sig編碼這個(gè)編碼會在文件開頭加 BOM 標(biāo)記Excel 能正確識別。5. 二次開發(fā)與改造思路從能跑到好用5.1 增量抓取與去重策略第一次跑全量之后后面再跑就應(yīng)該只抓新文章。最簡單的做法是在存儲層加一個(gè)url唯一索引每次插入前先查一下是否已存在。SQLite 的INSERT OR IGNORE配合UNIQUE約束就能做到。如果數(shù)據(jù)量大可以再加一個(gè)last_fetch_time字段記錄每篇文章的抓取時(shí)間增量抓取時(shí)只請求比這個(gè)時(shí)間新的文章。不過公眾號列表頁一般按時(shí)間倒序排列所以更簡單的做法是每次從第一頁開始翻遇到已存在的 URL 就停止翻頁。def is_url_exists(url, db_patharticles.db): 檢查 URL 是否已存在于數(shù)據(jù)庫 conn sqlite3.connect(db_path) cursor conn.cursor() cursor.execute(SELECT 1 FROM articles WHERE url ? LIMIT 1, (url,)) result cursor.fetchone() conn.close() return result is not None在翻頁循環(huán)里每解析出一篇文章就先調(diào)這個(gè)函數(shù)判斷如果已存在就跳過連續(xù)跳過 N 篇就終止翻頁。5.2 正文清洗與結(jié)構(gòu)化輸出原始正文里通常夾雜著大量無關(guān)內(nèi)容廣告、推薦閱讀、二維碼提示等。我一般會做一輪清洗把正文按段落切分過濾掉長度過短的段落和包含特定關(guān)鍵詞的段落。def clean_content(raw_text): 清洗正文去掉廣告和無關(guān)段落 lines raw_text.split(\n) cleaned [] skip_keywords [點(diǎn)擊上方, 關(guān)注我們, 長按識別, 推薦閱讀] for line in lines: line line.strip() if len(line) 10: continue if any(kw in line for kw in skip_keywords): continue cleaned.append(line) return \n.join(cleaned)清洗規(guī)則可以根據(jù)實(shí)際抓取結(jié)果不斷調(diào)整。我習(xí)慣先把原始正文存一份清洗后的另存一份方便對比和回滾。5.3 多公眾號批量采集的調(diào)度改造單個(gè)公眾號跑通之后很自然會想擴(kuò)展到多個(gè)公眾號。這時(shí)候需要把TARGET_BIZ改成列表外層再加一層循環(huán)。但要注意不同公眾號的請求間隔要獨(dú)立控制不能共用一個(gè)延時(shí)否則總請求頻率會翻倍。TARGET_LIST [ {biz: 公眾號A的biz, name: 賬號A}, {biz: 公眾號B的biz, name: 賬號B}, ] for target in TARGET_LIST: print(f開始抓取 {target[name]}) crawl_one(target[biz]) time.sleep(random.uniform(10, 20)) # 賬號之間多等一會兒賬號之間的間隔建議比單賬號內(nèi)的間隔更長給服務(wù)器一個(gè)緩沖時(shí)間。我一般設(shè) 10 到 20 秒。5.4 驗(yàn)證改造效果的方法改完之后怎么確認(rèn)沒改壞我的習(xí)慣是準(zhǔn)備一個(gè)小型測試集選一個(gè)文章數(shù)量少的公眾號手動(dòng)記錄前 5 篇文章的標(biāo)題和發(fā)布時(shí)間然后跑一遍程序?qū)Ρ茸ト〗Y(jié)果是否一致。如果一致再放開跑全量。另外每次改完解析邏輯先把 HTML 存到本地用本地文件跑解析函數(shù)這樣不用反復(fù)發(fā)請求調(diào)試效率高很多。# 調(diào)試用從本地文件讀取 HTML 并解析 with open(debug_page.html, r, encodingutf-8) as f: html f.read() result parse_article_detail(html) print(result)從那以后我每次改解析規(guī)則都強(qiáng)制先走一遍本地 HTML 調(diào)試確認(rèn)沒問題再發(fā)真實(shí)請求。這個(gè)習(xí)慣幫我省了大量等待時(shí)間也避免了因?yàn)檎{(diào)試請求太頻繁被限制。希望幫到你。本文還有配套的精品資源點(diǎn)擊獲取