公众号 API Python 教程：把文章转成 Markdown 给 RAG 用（2026）

微信官方接口能做什么？公众号的服务端接口都从 access_token 开始，它用公众号的 AppID 和 AppSecret 换取，有效期 7200 秒。拿到它之后，发布能力里的 freepublish_batchget 能返回你这个号已经发布的文章，其他接口还能管理草稿、素材和看统计数据。第三方平台在公众号授权之后，也能代它调用这些接口。所以如果号是你自己的，就用官方接口：免费、完整，还能拿到原始 HTML。它做不到的是读你不管理的号，而做行业研究或竞品知识库，要的恰恰是这部分。

这篇教程讲的就是这种情况。你会写一个 Python 小客户端：按名称找到公众号，列出最近的文章，拉每篇的全文，存成带 YAML front matter 的 Markdown 文件，开头写着标题、公众号、日期和原文链接。这些文件可以直接交给切块和向量化流程，每个块都能带上出处。

分步操作

第一步：准备环境

需要 Python 3.9 以上、requests，以及放在环境变量里的 key。没有 SDK，也不用登录微信。

pip install requests
export OKDATA_API_KEY=okd_...

每次调用都是对 https://api.okdata.online 加接口路径发 GET 请求，key 放在 Bearer 头里。成功时 "code" 为 "ok"；出错时返回 code、message、request_id 和 retry 提示（yes 表示等 retry_after_seconds 后可以重试）。

第二步：找到公众号

公众号搜索的 query 填公众号名称。每个结果有 id（原始 ID，gh_…）、username（微信号）、name 和 followers。只保留 name 完全一致的那个：公众号名称不唯一，排第一的不一定是你要的号。

搜索按返回的公众号数计费：每个 $0.058，一次最多 $0.58。用完整名称搜，通常只返回一个或几个。第一次跑完就把 gh_ ID 存下来，以后这个号再也不用搜。

curl -G "https://api.okdata.online/v1/wechat_mp/accounts/search" \
  -H "Authorization: Bearer $OKDATA_API_KEY" \
  --data-urlencode 'query=少数派'

第三步：列出最近的文章

公众号历史文章的 id 传 gh_ ID 或微信号，每次返回 10 篇，从新到旧，每次 $0.02。每篇有 id、title、url、published_at 和摘要 text。用 cursor 翻页，直到 page.has_more 为 false 或者到了你设的页数上限。

第四步：把每篇文章拉成文本

公众号文章详情的 url 传文章链接，text 里是纯文本正文，段落和小标题各占一行，正文图片已去掉，每次 $0.02。同时返回 title、published_at 和 author（公众号的 gh_ ID 和名称）。加 view=full 时，ext.author_byline 是署名作者，转载文章的原文地址在 ext.source_url。

curl -G "https://api.okdata.online/v1/wechat_mp/articles/get" \
  -H "Authorization: Bearer $OKDATA_API_KEY" \
  --data-urlencode 'url=https://mp.weixin.qq.com/s/Vt8nR2kLq5XwJ0cYhPa3dE' \
  --data-urlencode 'view=full'

第五步：客户端代码

下面的脚本跑的就是示例：搜索一次，最多读 MAX_PAGES 页历史文章，把列出的所有文章写进索引，再挑出最多 MAX_ARTICLES 篇本地还没有的文章拉全文，存成 kb/wechat/<id>.md。下周再跑一次，只为新文章付钱。

import os
import pathlib
import re
import time
import uuid
import requests

API = "https://api.okdata.online"
SESSION = requests.Session()
SESSION.headers["Authorization"] = f"Bearer {os.environ['OKDATA_API_KEY']}"
OUT = pathlib.Path("kb/wechat")
MAX_PAGES = 3      # 历史文章页数，每页 10 篇
MAX_ARTICLES = 10  # 每次运行最多拉几篇全文


class ApiError(Exception):
    def __init__(self, body):
        super().__init__(f"{body['code']}: {body['message']} (request {body['request_id']})")
        self.code = body["code"]


def call(path, **params):
    key = str(uuid.uuid4())  # 重试沿用同一个 key：重放不会再次扣费
    for attempt in range(3):
        body = SESSION.get(API + path, params=params, headers={"Idempotency-Key": key}, timeout=90).json()
        if body["code"] == "ok":
            return body
        if body.get("retry") == "yes" and attempt < 2:
            time.sleep(body.get("retry_after_seconds") or 5)
            continue
        raise ApiError(body)


def find_account(name):
    accounts = call("/v1/wechat_mp/accounts/search", query=name)["data"]["items"]
    match = next((a for a in accounts if a["name"] == name), None)
    if match is None:
        raise SystemExit(f"no account named {name!r}; got {[a['name'] for a in accounts]}")
    return match


def history(account_id):
    cursor = None
    for _ in range(MAX_PAGES):
        body = call("/v1/wechat_mp/accounts/articles/list", id=account_id, **({"cursor": cursor} if cursor else {}))
        yield from body["data"]["items"]
        if not body["page"]["has_more"]:
            return
        cursor = body["page"]["next_cursor"]


def quoted(s):
    return '"' + (s or "").replace("\\", "\\\\").replace('"', '\\"') + '"'


def to_markdown(article, account):
    ext = article.get("ext", {})
    front = [
        "---",
        f"title: {quoted(article['title'])}",
        f"account: {quoted(account['name'])}",
        f"account_id: {account['id']}",
        f"byline: {quoted(ext.get('author_byline'))}",
        f"published: {article['published_at']}",
        f"source: {article['url']}",
        "---",
    ]
    return "\n".join(front) + f"\n\n# {article['title'] or ''}\n\n{article['text']}\n"


def main(name):
    OUT.mkdir(parents=True, exist_ok=True)
    account = find_account(name)
    listed = list(history(account["id"]))  # 从新到旧
    with open(OUT / "index.tsv", "w", encoding="utf-8") as f:
        f.writelines(f"{a['published_at']}\t{a['title']}\t{a['url']}\n" for a in listed)

    path_of = lambda a: OUT / (re.sub(r"[^\w-]", "_", a["id"]) + ".md")
    todo = [a for a in listed if not path_of(a).exists()][:MAX_ARTICLES]
    for a in todo:
        try:
            full = call("/v1/wechat_mp/articles/get", url=a["url"], view="full")["data"]["item"]
        except ApiError as err:
            if err.code == "not_found":  # 列出之后被删了；不扣费
                continue
            raise
        path_of(a).write_text(to_markdown(full, account), encoding="utf-8")
        print("saved", full["title"])


if __name__ == "__main__":
    main("少数派")

第一次运行就是示例里的 14 次调用：1 次搜索、3 页历史文章、10 篇全文。如果要把这个号最近 300 篇都补齐，还要再花 $6.60。

第六步：切块和入库

每个文件是 front matter 加正文。按空行把正文切成几百字一块，把 title、account、published 和 source 复制到每个块的元数据里，这样回答时能引用原文、附上链接。大多数加载器能直接读 YAML front matter；读不了的话，自己解析两行 --- 之间的内容即可。

注意事项

• 是文本，不是排版。 详情返回纯文本，表格会变成一行行文字，正文图片没了，封面链接在 images 里。要保留文章原本的样子，得用别的办法存 HTML。
• 转载。 很多号会转载别人的文章。详情里的 ext.source_url 和 ext.copyright 可以帮你在入库前去重，免得重复内容干扰检索。
• 删掉的不收钱。 列出之后被删的文章返回 not_found，不扣费，客户端会直接跳过。
• 重试是安全的。 客户端对同一个请求的重试沿用同一个 Idempotency-Key，24 小时内同样的请求配同样的 key，会拿回上次的结果，meta.replayed 为 true，不再扣费。
• 文本不可信。 文章是陌生人写的，要告诉模型把检索到的段落当数据看，不要当指令执行。
• 版权。 只用于内部研究，每个块都带上原文链接，不要把整篇文章原样返回给用户。

只想看 Markdown 那一步的简版，见怎么把公众号文章转成 RAG 能用的干净文本。手上只有名称、要先查 ID，见怎么查公众号的原始 ID。

{
  "endpoint": "wechat_mp.accounts.search",
  "params": {
    "query": "36氪"
  },
  "idempotency_key": "<新的 UUID>"
}