用 Python 拉 Facebook 主页帖子:先解析 URL,再回传游标

发布于 2026年10月15日 · 更新于2026年10月16日

手里通常只有一条公开主页链接,没有 page_id。pages-posts 不吃 facebook.com/nasa,也不吃 page=2。要先用 page-details 把 URL 换成 id,之后的续读只认上一页带回、你造不出来的 cursor——原样回传,不要自增。把 Facebook 主页当成编号目录,或跳过解析把 handle 塞进 page_id,才是这次采集写错的地方。

下面用一段 Python:解析链接、原样回传游标、自己设页预算,游标缺失或重复就停,窗口重叠按 post_id 去重。打开本文或复制代码不会发请求;换成真实密钥并执行后才会调用、才会扣额度。

Facebook 数据 API 只读公开主页表面,不是 Meta 的 Page Insights,也不保证完整历史。互动数字是采集当下的快照。

主页帖子不是 offset 翻页

pages-posts 没有 page、没有 offset、也不能由客户端指定页长。下一页的唯一合法输入,是同一个操作、同一个 page_id 上一页带回的 cursor。不要拆、不要截断、不要自增,也不要把 search-pages 或 search-post 的游标塞进来。相邻页可能重叠。页大小以实际 results 长度为准,那是观察值,不是合同。停手不等于已经拿到全部历史。

登录后从控制台复制有权使用的 Key,放在服务端环境变量 OPENDATA_API_KEY。查询串里带 key,日志、截图、异常栈都要脱敏,不要写进前端。示例里的 YOUR_API_KEY 只是占位,不是可用密钥。不需要 Facebook 登录 Cookie。生产环境把循环放在服务端跑,浏览器里拼带密钥的 GET 等于把额度公开。

每次响应有两层状态。HTTP 4xx/5xx 是传输、鉴权或额度问题。HTTP 200 不等于成功:网关信封仍是 code / message / data,业务成功看 code=200。然后再看 data 里的业务体:主页详情是 results 对象,主页帖子是 results 数组加 cursor。不要把结构异常改写成空数组当成功。start_date / end_date 在部分路由上有文档,格式和筛选效果未验证,不要套别家产品的日期写法。

先把公开 URL 解析掉

表格里的个性链接不是 page_id。page-details 吃公开 URL。文档示例是 https://www.facebook.com/facebook。先核对 results.name 和 results.page_id,再采集。search-pages 的命中只是候选,不等于品牌官方号,标识字段还叫 facebook_id,和详情的 page_id 不是同一个名字。先在详情上确认身份,再把详情里的 page_id 交给帖子接口。

若详情给出 type=private_page 且没有身份,这是不确定结论:分不清页面不存在、非公开还是暂时不可访问。停。不要自己编 ID,也不要当抖动去重试。

search-post 的关键词命中不是「这个主页自己发的帖」。search-pages 出现 results=[] 且 cursor=null 可以是合法空结果。先看 HTTP 和外层 code;空搜索或含糊的详情,都不能当成全网不存在的证明。

第一页只传 page_id,不要带 cursor,也不要用空字符串占坑。帖子在 data.results,续翻字段就在这份业务体的 cursor。有值就原样带回;缺失、null 或空字符串就结束。采样里出现过 100064860875397 这类 ID,那只是笔记用的对照,生产里用详情真正返回的值。

原样回传游标,限制页数

先装 requests,在服务端导出 OPENDATA_API_KEY。下面的循环最多五页,是客户端预算,不是平台公布的上限或重试合同。遇到空、重复游标或打满页数就停。集合只在本次进程有效;定时任务请把 (page_id, post_id) 做成库唯一键。

import json
import os
from datetime import datetime, timezone

import requests

API_KEY = os.environ["OPENDATA_API_KEY"]
BASE = "https://api.opendata-api.com/api/project/facebook-scraper"
MAX_PAGES = 5  # local cost guard, not an API limit


def read(operation, **params):
    response = requests.get(
        f"{BASE}/{operation}",
        params={"key": API_KEY, **params},
        timeout=30,
    )
    if not 200 <= response.status_code < 300:
        raise RuntimeError(f"HTTP failure: {response.status_code}")
    body = response.json()
    if not isinstance(body, dict) or body.get("code") != 200:
        raise RuntimeError("Gateway business failure; inspect redacted status")
    payload = body.get("data")
    if not isinstance(payload, dict):
        raise RuntimeError("Unexpected operation payload; validate before parsing")
    return payload


def collect(page_url):
    details = read("page-details", url=page_url).get("results")
    if not isinstance(details, dict) or not details.get("page_id"):
        raise RuntimeError("Page identity unresolved; review the source URL")
    if details.get("type") == "private_page":
        raise RuntimeError("Inconclusive Page result; do not invent a page_id")
    page_id = str(details["page_id"])
    seen_posts, seen_cursors = set(), set()
    cursor = None

    for _ in range(MAX_PAGES):
        params = {"page_id": page_id}
        if cursor is not None:
            params["cursor"] = cursor
        payload = read("pages-posts", **params)
        posts = payload.get("results")
        if not isinstance(posts, list):
            raise RuntimeError("Expected the sampled Page-post results array")
        observed_at = datetime.now(timezone.utc).isoformat()
        for post in posts:
            if not isinstance(post, dict) or not post.get("post_id"):
                raise RuntimeError("Post has no usable identifier")
            post_id = str(post["post_id"])
            if post_id in seen_posts:
                continue
            seen_posts.add(post_id)
            timestamp = post.get("timestamp")
            published_at = (
                datetime.fromtimestamp(timestamp, timezone.utc).isoformat()
                if isinstance(timestamp, (int, float))
                else None
            )
            yield {
                "page_id": page_id,
                "post_id": post_id,
                "published_at_utc": published_at,
                "observed_at_utc": observed_at,
                "raw": post,
            }
        next_cursor = payload.get("cursor")
        if next_cursor is None or next_cursor == "":
            break
        if not isinstance(next_cursor, str):
            raise RuntimeError("Unexpected cursor type")
        if next_cursor in seen_cursors:
            break
        seen_cursors.add(next_cursor)
        cursor = next_cursor


if __name__ == "__main__":
    with open("page_posts.jsonl", "a", encoding="utf-8") as output:
        for record in collect("https://www.facebook.com/facebook"):
            output.write(json.dumps(record, ensure_ascii=False) + "\n")

互动计数请另存观察时间,不要覆盖已知帖。不要用「页长 × 页数」估总量,合并只认字符串 post_id。失败运行可能留下半截 JSONL,不要标成采完。这里不假设自动重试:HTTP 或外层 code 失败就停,看脱敏后的状态,不要盲走游标。

评论、嵌套回复、单帖详情是另一条链路(comments / comments-nested / get-post-details),不要在翻页循环里逐条去打——那会把页预算悄悄乘上去,而且讨论键、expansion_token 都不能从列表页臆造。

不要自己编互动指标

帖子里实际有的再留:post_id、url、message / message_rich、数值型 timestamp、author / authors、comments_count、reactions_count、reshare_count、表情分类计数,以及可能为 null 的 image / video / video_files。ID 当字符串。缺了不是零。top_comments 只是预览,不是全部讨论;comments_id、shares_id 是下游标识,不代表你已经采完评论。

粉丝数和互动数是公开观察值,不是触达、曝光或 Page Insights。比发帖节奏时,发布时间与采集时间要分开存:同一条帖过几天再采,评论数会涨,那是观察窗口变了,不是「发布当天就有这么多互动」。媒体引用会变,也不等于转载授权。不要把每条帖压成自制「媒体对象」,更不要把表情分类直接当成情绪结论。竞品复盘只统计你实际入库、且带 page_id 的帖,不要拿关键词搜索的条数去当全网声量。

公开可见不等于可以自动采集。确认条款、合法处理依据和媒体权利,少存个人数据,设好保留期限。存下来的链接不是下载或再分发的许可。

脚本能跑之后仍常见的错:继续传 page=2 / offset / 自制游标;解码或截短 cursor;把搜索接口的游标塞进 pages-posts;跳过详情、把个性链接当成 page_id;把 private_page 当成已证实不存在;HTTP 200 但外层 code 不是 200 仍当帖子列表;缺少 results 数组却写成空成功;停手就宣布历史采全;覆盖互动快照而不打 observed_at_utc;把计数当成 Insights;把带 Key 的完整 URL 打进日志。先把这些停掉,再去接关键词搜索、广告或评论——那些路由的翻页字段本来就不同,混用等于换了件衣服的 page=2。

参考 主页详情 与 主页帖子。单次调用费用见定价。