爬取搜狗微信搜索結果,收集含指定關鍵詞的微信公眾號文章。支持翻頁功能, 每頁 10 條,最多 10 頁(100 條)。使用 requests + lxml XPath 解析。
div.txt-box 內的每條文章| 字段名 | XPath 表達式 | 說明 |
|---|---|---|
| title | .//h3/a//text() | 文章標題(去除高亮標籤) |
| article_url | .//h3/a/@href | 文章跳轉鏈接(搜狗中轉) |
| account_name | .//span[@class="all-time-y2"]/text() | 公眾號名稱 |
| pub_date | .//span[@class="s2"]/script/text() | 發布日期(從時間戳轉換) |
page=N 實現。 例如第 2 頁:https://weixin.sogou.com/weixin?type=2&query=關鍵詞&page=2。 代碼中的 MAX_PAGES 控制最多爬幾頁,每頁之間加入 2 秒延遲避免被封 IP。 搜狗微信最多支持 10 頁(page=1 至 page=10)。document.write(timeConvert('1773232740'))。需要用正則表達式提取其中的 Unix 時間戳,再用datetime.fromtimestamp()轉換為可讀日期格式(如 2026-03-11)。安裝爬蟲所需的 Python 套件(requests 和 lxml)。
!pip install requests lxml
導入所有需要的 Python 模組。
import requests import re import csv import datetime from lxml import etree
設定搜索關鍵詞、最大爬取頁數、輸出文件名。搜狗微信每頁顯示 10 條結果,最多支持 10 頁(100 條)。
# ── 參數設定 ────────────────────────────────────────
QUERY = "香港浸会大学" # 搜索關鍵詞
MAX_PAGES = 5 # 最多爬幾頁(每頁 10 條,最多 10 頁)
OUTPUT_CSV = "sogou_weixin.csv"
# HTTP Headers:模擬真實瀏覽器,避免被反爬
HEADERS = {
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/133.0.0.0 Safari/537.36",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": "zh-CN,zh;q=0.9",
"Referer": "https://weixin.sogou.com/",
}
print(f"✅ 參數設定完成")
print(f" 關鍵詞:{QUERY}")
print(f" 最大頁數:{MAX_PAGES}(預計最多 {MAX_PAGES * 10} 條)")循環爬取多頁搜索結果,每頁發送一個 HTTP 請求,收集所有文章的原始 HTML 節點。
import time
def fetch_page(query, page):
"""爬取搜狗微信搜索結果的一頁"""
url = (
f"https://weixin.sogou.com/weixin"
f"?type=2&query={requests.utils.quote(query)}&page={page}"
)
try:
resp = requests.get(url, headers=HEADERS, timeout=15)
if resp.status_code == 200:
return resp.text
else:
print(f" ✗ 第 {page} 頁請求失敗,狀態碼:{resp.status_code}")
return None
except Exception as e:
print(f" ✗ 第 {page} 頁請求異常:{e}")
return None
all_raw_items = [] # 存放每頁的 (page_num, html_text) 元組
for page in range(1, MAX_PAGES + 1):
print(f"正在爬取第 {page} 頁...")
html = fetch_page(QUERY, page)
if html:
tree = etree.HTML(html)
items = tree.xpath('//div[@class="txt-box"]')
print(f" ✓ 找到 {len(items)} 條結果")
all_raw_items.append((page, html))
if len(items) == 0:
print(" ⚠ 本頁無結果,停止翻頁")
break
else:
break
# 禮貌性延遲,避免被封 IP
if page < MAX_PAGES:
time.sleep(2)
print(f"\n✅ 共爬取 {len(all_raw_items)} 頁")用 XPath 從每頁 HTML 中提取文章標題、文章鏈接、公眾號名稱、發布時間。時間以 Unix 時間戳存儲,需轉換為可讀格式。
def parse_timestamp(script_text):
"""從 JavaScript 代碼中提取時間戳並轉換為日期字符串"""
match = re.search(r"timeConvert\('(\d+)'\)", script_text)
if match:
ts = int(match.group(1))
return datetime.datetime.fromtimestamp(ts).strftime('%Y-%m-%d')
return ""
def parse_page(html):
"""解析一頁 HTML,返回文章列表"""
tree = etree.HTML(html)
items = tree.xpath('//div[@class="txt-box"]')
records = []
for item in items:
# ── 文章標題(去除高亮 <em> 標籤,合並所有文字)
title = "".join(item.xpath('.//h3/a//text()')).strip()
title = re.sub(r'\s+', ' ', title)
# ── 文章鏈接(搜狗跳轉鏈接,前綴 weixin.sogou.com)
href_list = item.xpath('.//h3/a/@href')
article_url = ("https://weixin.sogou.com" + href_list[0]) if href_list else ""
# ── 公眾號名稱
account_els = item.xpath('.//span[@class="all-time-y2"]/text()')
account_name = account_els[0].strip() if account_els else ""
# ── 發布時間(從 JavaScript timeConvert 提取時間戳)
time_scripts = item.xpath('.//span[@class="s2"]/script/text()')
pub_date = ""
for script in time_scripts:
pub_date = parse_timestamp(script)
if pub_date:
break
records.append({
"title": title,
"article_url": article_url,
"account_name": account_name,
"pub_date": pub_date,
})
return records
# 解析所有頁面
all_records = []
for page_num, html in all_raw_items:
records = parse_page(html)
all_records.extend(records)
print(f"第 {page_num} 頁:解析 {len(records)} 條")
print(f"\n✅ 共解析 {len(all_records)} 條文章")
if all_records:
print("\n前 3 條預覽:")
for r in all_records[:3]:
print(f" 標題:{r['title'][:40]}...")
print(f" 公眾號:{r['account_name']}")
print(f" 時間:{r['pub_date']}")
print()將所有文章數據保存為 CSV 文件,並自動觸發下載到本地電腦。
if not all_records:
print("❌ 錯誤:all_records 為空!請先執行 Cell 4 和 Cell 5")
else:
# 保存 CSV
with open(OUTPUT_CSV, "w", newline="", encoding="utf-8-sig") as f:
writer = csv.DictWriter(f, fieldnames=["title", "article_url", "account_name", "pub_date"])
writer.writeheader()
writer.writerows(all_records)
print(f"✅ 已保存 {len(all_records)} 條記錄到 {OUTPUT_CSV}")
# 自動下載
from google.colab import files
files.download(OUTPUT_CSV)
print(f"✅ 下載完成:{OUTPUT_CSV}")