微信公众号历史文章怎么获取？文章列表、正文与分页采集教程

python3 -m pip install requests
export JUSTONEAPI_TOKEN='your_token'
export WECHAT_ARTICLE_URL='https://mp.weixin.qq.com/s/your_article_id'

import json
import os
from datetime import datetime, timezone
from pathlib import Path

import requests

BASE_URL = "https://api.justoneapi.com"
TOKEN = os.environ.get("JUSTONEAPI_TOKEN", "")


def call_api(path, method="GET", accepted_codes=(0,), **params):
    if not TOKEN or TOKEN == "your_token":
        raise RuntimeError("Set JUSTONEAPI_TOKEN before running")
    values = {"token": TOKEN, **params}
    arguments = {"data": values} if method == "POST" else {"params": values}
    try:
        response = requests.request(
            method, BASE_URL + path, timeout=120, **arguments,
        )
    except requests.RequestException:
        raise RuntimeError("Network error or timeout") from None
    if response.status_code != 200:
        raise RuntimeError(f"HTTP {response.status_code}")
    try:
        result = response.json()
    except ValueError:
        raise RuntimeError("Response is not valid JSON") from None
    if not isinstance(result, dict) or result.get("code") not in accepted_codes:
        code = result.get("code") if isinstance(result, dict) else None
        raise RuntimeError(f"API failed: code={code}; check the docs")
    return result


def output_dir(prefix):
    stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S%fZ")
    folder = Path(prefix) / stamp
    folder.mkdir(parents=True, exist_ok=False)
    return folder


def save_json(folder, filename, value):
    (folder / filename).write_text(
        json.dumps(value, ensure_ascii=False, indent=2), encoding="utf-8",
    )


SOURCE_ARTICLE_URL = os.environ.get("WECHAT_ARTICLE_URL", "")
MAX_HISTORY_PAGES = 2
MAX_ARTICLES = 3


def main():
    if not SOURCE_ARTICLE_URL.startswith("https://mp.weixin.qq.com/"):
        raise RuntimeError("Set WECHAT_ARTICLE_URL to a WeChat article URL")
    folder = output_dir("wechat-output")
    offset = ""
    seen_offsets = set()
    article_urls = []
    seen_urls = set()

    for page in range(1, MAX_HISTORY_PAGES + 1):
        history = call_api(
            "/api/weixin/get-account-history-articles/v2", method="POST",
            url=SOURCE_ARTICLE_URL, offset=offset,
        )
        save_json(folder, f"history-{page}.json", history)
        data = history["data"]
        for message in (data.get("MsgList") or {}).get("Msg") or []:
            entries = (message.get("AppMsg") or {}).get("DetailInfo") or []
            for article in entries:
                article_url = article.get("ContentUrl")
                if article_url and article_url not in seen_urls:
                    seen_urls.add(article_url)
                    article_urls.append(article_url)
        paging = data.get("PagingInfo") or {}
        is_end = paging.get("IsEnd")
        if is_end in (True, 1, "1"):
            break
        if is_end not in (False, 0, "0"):
            raise RuntimeError("Missing IsEnd; inspect saved history JSON")
        next_offset = paging.get("Offset")
        if page < MAX_HISTORY_PAGES:
            if not next_offset or next_offset in seen_offsets:
                raise RuntimeError("Missing or repeated history Offset")
            seen_offsets.add(next_offset)
            offset = next_offset

    selected = article_urls[:MAX_ARTICLES]
    save_json(folder, "selected-article-urls.json", selected)
    for index, article_url in enumerate(selected, start=1):
        detail = call_api(
            "/api/weixin/get-article-detail/v1",
            articleUrl=article_url, mode="TEXT",
        )
        save_json(folder, f"article-{index}-detail.json", detail)
    print(f"Saved {len(selected)} article samples to {folder}")


if __name__ == "__main__":
    try:
        main()
    except RuntimeError as error:
        raise SystemExit(str(error)) from None