使用 requests + lxml XPath 爬取 lihkg.com/category/1?order=hot 的熱門帖子。代碼分為 8 格，可逐格複製到 Google Colab 執行，或點擊右上角「複製全部代碼」一次貼入。
tree.xpath("//text()")— 提取正文 HTML 中所有文字節點（格 3、格 5 的核心）lxml_html.fromstring(html)— 將 HTML 字符串解析為可用 XPath 查詢的元素樹msg 字段）是 HTML 格式，需要 XPath 解析提取純文字在 Google Colab 中安裝所需的 Python 套件。lxml 用於 XPath 解析，BeautifulSoup 用於清洗正文 HTML。
# 安裝所需套件（Google Colab 已預裝 requests，只需安裝 lxml）
!pip install lxml beautifulsoup4 pandas --quiet導入所有套件，並設定爬蟲的核心參數：目標板塊、最大頁數、休息時間、請求頭。修改這裡的參數即可調整爬蟲行為。
import requests # 發送 HTTP 請求
import time, random # 控制休息時間（防反爬）
import csv, os, json # 文件讀寫與斷點記錄
from lxml import html as lxml_html # XPath 解析 HTML 正文
from bs4 import BeautifulSoup # 清洗 HTML 為純文字
from datetime import datetime, timezone, timedelta
# ── 爬蟲參數設定 ──────────────────────────────────────────────
CATEGORY_ID = 1 # 板塊 ID：1=吹水台，5=時事台，32=娛樂台
ORDER = "hot" # 排序：hot=熱門，new=最新
MAX_PAGES = 5 # 最多爬幾頁（每頁最多 60 帖）
FETCH_BODY = True # True=同時爬正文，False=只爬標題等列表字段
OUTPUT_FILE = "lihkg_hot.csv"
CHECKPOINT = "lihkg_checkpoint.json"
# ── 防反爬休息時間（秒）────────────────────────────────────────
SLEEP_LIST = (1.5, 3.0) # 翻頁之間的休息時間範圍
SLEEP_BODY = (0.8, 1.8) # 爬正文之間的休息時間範圍
MAX_RETRIES = 3 # 請求失敗最多重試次數
# ── 請求頭（模擬真實瀏覽器，LIHKG 需要這兩個自定義頭部）────────
HEADERS = {
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
"AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36",
"X-LI-DEVICE": "web", # LIHKG 必需：標識設備類型
"X-LI-DEVICE-TYPE": "browser", # LIHKG 必需：標識為瀏覽器
"Referer": "https://lihkg.com/",
"Accept": "application/json",
"Accept-Language": "zh-HK,zh;q=0.9",
}
print("✅ 套件導入完成，參數設定完成")定義三個輔助函數：時間轉換、HTML 清洗（用 XPath 提取正文純文字）、帶重試的 HTTP 請求。
def unix_to_hkt(ts):
"""將 Unix 時間戳轉換為香港時間字符串（UTC+8）"""
hkt = timezone(timedelta(hours=8))
return datetime.fromtimestamp(ts, tz=hkt).strftime("%Y-%m-%d %H:%M:%S")
def clean_body_html(raw_html):
"""
用 lxml XPath 從帖子正文 HTML 中提取純文字。
LIHKG 正文是 HTML 格式，包含 <a>、<img>、<br> 等標籤。
XPath 策略：
1. 用 //text() 提取所有文字節點
2. 過濾掉空白節點
3. 用換行符拼接
"""
if not raw_html:
return ""
# 將 HTML 字符串解析為 lxml 元素樹
tree = lxml_html.fromstring(f"<div>{raw_html}</div>")
# ── XPath 核心：提取所有文字節點 ──────────────────────────
# //text() → 選取所有後代的文字節點
# normalize-space() 在 Python 端處理，避免空白節點干擾
text_nodes = tree.xpath("//text()")
# 過濾空白節點，拼接成純文字
lines = [t.strip() for t in text_nodes if t.strip()]
return " ".join(lines)
def safe_get(url, retries=MAX_RETRIES):
"""帶自動重試的 HTTP GET 請求，返回解析後的 JSON 或 None"""
for attempt in range(1, retries + 1):
try:
r = requests.get(url, headers=HEADERS, timeout=10)
if r.status_code == 200:
return r.json() # 成功返回 JSON
elif r.status_code == 429: # 被限速
print(f" ⚠️ 限速(429)，等待 20 秒..."); time.sleep(20)
else:
print(f" ⚠️ HTTP {r.status_code}，第{attempt}次重試..."); time.sleep(5)
except Exception as e:
print(f" ⚠️ 請求錯誤：{e}，第{attempt}次重試..."); time.sleep(5)
return None # 超過重試次數，返回 None
print("✅ 輔助函數定義完成")爬取帖子列表 API，並用 XPath 從每個帖子的 JSON 轉換的 HTML 中提取字段。這是本節課的核心 XPath 練習部分。
def fetch_thread_list(page):
"""
爬取指定頁碼的帖子列表。
API：/api_v2/thread/latest?cat_id={id}&page={n}&count=60&type=now&order=hot
返回：(帖子列表, 是否還有下一頁)
"""
url = (f"https://lihkg.com/api_v2/thread/latest"
f"?cat_id={CATEGORY_ID}&page={page}&count=60&type=now&order={ORDER}")
data = safe_get(url)
if not data or data.get("success") != 1:
return [], False
resp = data["response"]
items = resp.get("items", [])
has_more = resp.get("is_pagination", False) # True=還有下一頁
return items, has_more
def parse_thread(thread):
"""
從帖子列表的單個 JSON 對象中提取所有字段。
LIHKG API 返回的是 JSON，但帖子正文（msg）是 HTML 字符串。
我們將 JSON 字段直接讀取，正文部分用 XPath 解析。
── JSON 字段映射 ─────────────────────────────────────────
thread["user"]["nickname"] → 用戶名
thread["create_time"] → 發帖時間（Unix 時間戳）
thread["like_count"] → 帖子點讚數
thread["dislike_count"] → 帖子不點讚數
thread["total_page"] → 帖子頁數
thread["category"]["name"] → 板塊名稱
"""
user = thread.get("user", {})
category = thread.get("category", {})
return {
"thread_id": thread.get("thread_id", ""),
"title": thread.get("title", ""),
# 用戶名：從嵌套的 user 對象中提取
"username": user.get("nickname", thread.get("user_nickname", "")),
"user_gender": thread.get("user_gender", ""),
# 發帖時間：Unix 時間戳 → 香港時間
"create_time": unix_to_hkt(thread.get("create_time", 0)),
# 點讚 / 不點讚（帖子標題層面）
"like_count": thread.get("like_count", 0),
"dislike_count":thread.get("dislike_count", 0),
"reply_count": thread.get("no_of_reply", 0),
# 帖子頁數（每頁 25 條回覆）
"total_page": thread.get("total_page", 1),
# 板塊名稱：從嵌套的 category 對象中提取
"cat_name": category.get("name", ""),
"thread_url": f"https://lihkg.com/thread/{thread.get('thread_id','')}/page/1",
# 正文字段（後續爬取後填入）
"body_text": "",
"body_like": 0,
"body_dislike": 0,
}
print("✅ 帖子列表爬取函數定義完成")爬取帖子詳情 API，取得樓主第一樓的 HTML 正文，再用 XPath 提取純文字。重點示範 XPath 在真實 HTML 清洗中的應用。
def fetch_body(thread_id):
"""
爬取帖子正文（樓主第一樓）。
API：/api_v2/thread/{thread_id}/page/1?order=reply_time
正文字段說明：
item_data[0].msg → HTML 格式正文
item_data[0].like_count → 正文點讚數
item_data[0].dislike_count→ 正文不點讚數
item_data[0].msg_num → 樓層號（1 = 樓主）
item_data[0].page → 所在頁碼
"""
url = f"https://lihkg.com/api_v2/thread/{thread_id}/page/1?order=reply_time"
data = safe_get(url)
if not data or data.get("success") != 1:
return "", 0, 0
item_data = data["response"].get("item_data", [])
if not item_data:
return "", 0, 0
# 找到樓主第一樓（msg_num == 1）
first = next((p for p in item_data if p.get("msg_num") == 1), item_data[0])
# ── XPath 核心：從 HTML 正文提取純文字 ────────────────────
# first["msg"] 是 HTML 字符串，例如：
# '<a href="...">連結</a>今日天氣好好<br>出去行山'
#
# 用 clean_body_html() 中的 XPath //text() 提取所有文字節點
body_text = clean_body_html(first.get("msg", ""))
return body_text, first.get("like_count", 0), first.get("dislike_count", 0)
print("✅ 正文爬取函數定義完成")斷點記錄功能：每爬完一頁就保存進度到 JSON 文件。中途中斷後再次運行，會自動從上次停止的地方繼續，不重複爬取。
def load_checkpoint():
"""讀取斷點記錄，返回上次爬到的頁碼和已爬帖子 ID 集合"""
if os.path.exists(CHECKPOINT):
with open(CHECKPOINT, "r", encoding="utf-8") as f:
d = json.load(f)
print(f"📌 發現斷點：上次爬到第 {d['last_page']} 頁，已爬 {len(d['ids'])} 帖")
return d["last_page"], set(d["ids"])
print("🆕 無斷點記錄，從第 1 頁開始")
return 0, set()
def save_checkpoint(page, ids):
"""保存當前進度到斷點記錄文件"""
with open(CHECKPOINT, "w", encoding="utf-8") as f:
json.dump({"last_page": page, "ids": list(ids),
"saved_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S")},
f, ensure_ascii=False, indent=2)
print("✅ 斷點函數定義完成")主函數：自動翻頁爬取所有帖子，寫入 CSV，每頁保存斷點。爬完後自動在 Google Colab 中下載 CSV 文件。
def run():
"""
主爬蟲函數。流程：
1. 讀取斷點 → 2. 逐頁爬列表 → 3. 爬正文 → 4. 寫 CSV → 5. 保存斷點
"""
last_page, done_ids = load_checkpoint()
start_page = last_page + 1
# ── CSV 字段定義 ──────────────────────────────────────────
fields = ["thread_id","title","username","user_gender","create_time",
"like_count","dislike_count","reply_count","total_page",
"cat_name","body_text","body_like","body_dislike","thread_url"]
# 新文件寫表頭，舊文件追加（斷點續爬）
mode = "a" if os.path.exists(OUTPUT_FILE) else "w"
f_csv = open(OUTPUT_FILE, mode, newline="", encoding="utf-8-sig")
writer = csv.DictWriter(f_csv, fieldnames=fields)
if mode == "w":
writer.writeheader() # 只在新文件時寫表頭
total = 0
try:
for page in range(start_page, (MAX_PAGES or 999) + 1):
print(f"
{'─'*50}")
print(f"📖 第 {page} 頁...")
threads, has_more = fetch_thread_list(page)
if not threads:
print("⚠️ 列表為空，停止"); break
for i, t in enumerate(threads):
tid = t.get("thread_id")
if tid in done_ids:
continue # 跳過已爬取的帖子
row = parse_thread(t) # 解析列表字段
if FETCH_BODY:
print(f" [{i+1}/{len(threads)}] 爬正文：{row['title'][:25]}...")
body, bl, bd = fetch_body(tid)
row["body_text"] = body
row["body_like"] = bl
row["body_dislike"] = bd
# 爬正文後隨機休息（防反爬）
time.sleep(random.uniform(*SLEEP_BODY))
writer.writerow(row) # 寫入 CSV
done_ids.add(tid)
total += 1
f_csv.flush() # 刷新緩衝區，確保數據寫入磁盤
save_checkpoint(page, done_ids) # 每頁保存斷點
print(f"✅ 第 {page} 頁完成，累計 {total} 帖")
if not has_more:
print("✅ 已爬完所有頁面"); break
# 翻頁前隨機休息（防反爬）
sleep = random.uniform(*SLEEP_LIST)
print(f"⏸️ 休息 {sleep:.1f} 秒...")
time.sleep(sleep)
except KeyboardInterrupt:
print(f"
⚠️ 用戶中斷，已保存斷點（第 {page} 頁）")
save_checkpoint(page - 1, done_ids)
finally:
f_csv.close()
print(f"
{'='*50}")
print(f"🎉 爬取完成！共 {total} 帖 → {OUTPUT_FILE}")
return total
# ── 執行爬蟲 ─────────────────────────────────────────────────
run()用 pandas 查看爬取結果，顯示統計摘要，並自動觸發 Google Colab 下載 CSV 文件到本地。
import pandas as pd
# 讀取 CSV 並顯示基本統計
df = pd.read_csv(OUTPUT_FILE, encoding="utf-8-sig")
print(f"📊 共爬取 {len(df)} 個帖子")
print(f"
前 5 行預覽：")
print(df[["title","username","like_count","cat_name"]].head())
print(f"
各板塊帖子數：")
print(df["cat_name"].value_counts().head(10))
print(f"
點讚最高 5 帖：")
print(df.nlargest(5,"like_count")[["title","username","like_count","cat_name"]].to_string(index=False))
# ── 自動下載 CSV（Google Colab 專用）────────────────────────
try:
from google.colab import files
files.download(OUTPUT_FILE) # 自動觸發瀏覽器下載
print(f"
✅ {OUTPUT_FILE} 已開始下載！")
except ImportError:
# 非 Colab 環境（本地 Jupyter）
print(f"
💡 文件已保存至：{os.path.abspath(OUTPUT_FILE)}")
print(" 在 Colab 左側文件面板可找到並手動下載")SLEEP_LIST 設為低於 1 秒，否則可能被封 IPFETCH_BODY=True 時，每帖需要額外一次 API 請求，爬取速度較慢utf-8-sig 編碼，在 Excel 中可正確顯示中文