如何高效、稳健地批量抓取 Yahoo Finance 社区论坛全部评论与回复-Python教程-PHP中文网

如何高效、稳健地批量抓取 Yahoo Finance 社区论坛全部评论与回复

本文介绍一种基于 spot.im api 的轻量级、可恢复式抓取方案，绕过 selenium 交互瓶颈，通过分页请求获取 yahoo finance 股票社区（如 tsla）的全部评论及嵌套回复，并集成断点续爬与防封策略。

本文介绍一种基于 spot.im api 的轻量级、可恢复式抓取方案，绕过 selenium 交互瓶颈，通过分页请求获取 yahoo finance 股票社区（如 tsla）的全部评论及嵌套回复，并集成断点续爬与防封策略。

Yahoo Finance 社区论坛实际由第三方评论系统 Spot.im 提供支持，其前端页面虽动态加载、无稳定 DOM ID，但底层通信完全依赖公开的 RESTful API（https://api-2-0.spot.im/v1.0.0/conversation/read）。这意味着我们无需启动浏览器、模拟点击“展开回复”或处理无限滚动，而应直接调用该 API —— 这不仅大幅提升效率，也显著降低被反爬拦截的风险。

以下是一个生产就绪的 Python 抓取脚本核心逻辑，已整合分页遍历、增量存储与故障恢复能力：

import requests
import json
import sqlite3
import time
from datetime import datetime

# === 配置区（按目标股票替换）===
CONVERSATION_ID = "sp_Rba9aFpG_finmb$27444752"  # 格式：sp_{spotId}_{postID}
SPOT_ID = "sp_Rba9aFpG"
POST_ID = "finmb$27444752"
API_URL = "https://api-2-0.spot.im/v1.0.0/conversation/read"
BATCH_SIZE = 25  # 建议 20–50；过大易触发限流，过小增加请求数

# === 数据库初始化（SQLite）===
def init_db(db_path="yahoo_comments.db"):
    conn = sqlite3.connect(db_path)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS comments (
            id TEXT PRIMARY KEY,
            author TEXT,
            content TEXT,
            created_at INTEGER,
            likes INTEGER,
            is_deleted BOOLEAN DEFAULT 0,
            parent_id TEXT NULL,
            offset INTEGER NOT NULL
        )
    """)
    conn.execute("CREATE INDEX IF NOT EXISTS idx_offset ON comments(offset)")
    conn.commit()
    return conn

# === 主抓取循环（支持断点续传）===
def fetch_all_comments(db_path="yahoo_comments.db", delay=1.2):
    conn = init_db(db_path)
    cursor = conn.cursor()

    # 读取最后成功保存的 offset（checkpoint）
    cursor.execute("SELECT MAX(offset) FROM comments")
    last_offset = cursor.fetchone()[0] or 0

    payload = {
        "conversation_id": CONVERSATION_ID,
        "count": BATCH_SIZE,
        "offset": last_offset,
        "sort_by": "oldest"  # 推荐 oldest + 正向 offset，避免新评论插入导致漏采
    }

    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
        "Content-Type": "application/json",
        "x-spot-id": SPOT_ID,
        "x-post-id": POST_ID
    }

    total_fetched = 0
    while True:
        try:
            resp = requests.post(API_URL, headers=headers, data=json.dumps(payload), timeout=15)
            resp.raise_for_status()
            data = resp.json()

            comments = data["conversation"]["comments"]
            if not comments:
                print(f"[INFO] No more comments at offset {payload['offset']}. Done.")
                break

            # 批量写入数据库（含去重 & deleted 处理）
            for c in comments:
                cid = c.get("id") or str(hash(json.dumps(c, sort_keys=True)))
                cursor.execute(
                    "INSERT OR REPLACE INTO comments VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
                    (
                        cid,
                        c.get("author", {}).get("name", "anonymous"),
                        c.get("content", "").strip(),
                        int(c.get("created_at", 0)),
                        c.get("likes", 0),
                        c.get("is_deleted", False),
                        c.get("parent_id"),
                        payload["offset"]
                    )
                )
            conn.commit()

            total_fetched += len(comments)
            print(f"[FETCHED] {len(comments)} comments @ offset {payload['offset']} | Total: {total_fetched}")

            # 更新 offset 并检查是否还有下一页
            if not data["conversation"].get("has_next", False):
                print("[SUCCESS] All pages fetched.")
                break

            payload["offset"] = data["conversation"]["offset"]
            time.sleep(delay)  # 必须！缓解请求频率，降低 429 风险

        except requests.exceptions.RequestException as e:
            print(f"[ERROR] Request failed: {e}. Saving checkpoint at offset {payload['offset']}...")
            conn.commit()
            time.sleep(5)  # 暂停后重试
            continue
        except Exception as e:
            print(f"[FATAL] Unexpected error: {e}")
            break

    conn.close()

# 启动抓取
if __name__ == "__main__":
    fetch_all_comments(delay=1.5)

✅ 关键优化说明：

断点续爬：每次成功写入即更新 offset，程序中断后自动从最新 MAX(offset) 继续，无需手动维护状态文件；
防封策略：delay=1.5s 是实测安全阈值（低于 1s 易触发 429 Too Many Requests），配合真实 UA 与必要 headers；
数据健壮性：显式处理 is_deleted 字段，过滤空内容，兼容缺失字段（如 parent_id 为空表示顶层评论）；
性能权衡：使用 sort_by="oldest" + 正向 offset 可规避新评论插入导致的重复或遗漏（newest 在高活跃帖中风险更高）；
扩展建议：如需获取嵌套回复（reply-to-reply），需递归请求 replies 子接口（/v1.0.0/comment/{id}/replies），但注意其分页逻辑独立。

⚠️ 重要提醒：

Yahoo Finance 社区数据受其服务条款约束，仅限个人学习与非商业分析用途；
Spot.im API 未公开文档，参数可能随版本变更，请定期验证 conversation_id 和 headers 结构（可通过浏览器 DevTools → Network → XHR 捕获真实请求）；
超大规模抓取（如 90 万+ 条）建议加入随机 jitter（如 time.sleep(delay * (0.8 + 0.4*random())）、分布式任务队列（Celery + Redis）及监控告警。

该方案将原本需数小时、高失败率的 Selenium 交互式爬虫，压缩为稳定、可审计、可中断恢复的 API 批处理流程，是面向金融社区数据采集的专业实践范式。