微信公众号文章自动化处理实战：Python + WeChat Official Account API 深度整合

2次阅读

没有评论

共计 3120 个字符，预计需要花费 8 分钟才能阅读完成。

做公众号运营的开发者肯定深有体会，手动管理文章简直是效率黑洞。我总结了三类最头疼的问题：

内容同步困难：写好的文章要在后台反复复制粘贴，格式还经常错乱
发布时间不可控：熬夜等凌晨推送是常态，错过热点是常事
数据统计滞后：阅读量、分享数要手动记录，无法实时分析

先对比两种常见方案：

直接调用官方 API（Requests 库）
优点：灵活可控，适合定制化需求
缺点：要自己处理 OAuth2.0、错误重试等基础逻辑
使用第三方 SDK（如 WeChatPY）
优点：开箱即用，快速上手
缺点：版本更新可能滞后于官方 API

建议：重度使用者建议封装自己的 SDK，下面就以裸调 API 为例演示核心功能。

微信 API 的通行证是 access_token，官方要求全局缓存。这段代码实现了：

失效自动刷新
错误码 429（频繁调用）的指数退避重试
多进程安全

def get_access_token():
    """
    获取带自动刷新的 access_token
    返回: tuple(token, expires_in)
    """redis_key ="wx_access_token"
    cached = redis.get(redis_key)

    if cached:
        return cached.decode(), 7200  # 默认 2 小时

    retry = 3
    while retry > 0:
        try:
            url = f"https://api.weixin.qq.com/cgi-bin/token?grant_type=client_credential&appid={APPID}&secret={APPSECRET}"
            resp = requests.get(url, timeout=5).json()

            if "access_token" in resp:
                redis.setex(redis_key, 7000, resp["access_token"])  # 提前过期
                return resp["access_token"], resp["expires_in"]
            else:
                handle_error(resp)  # 处理错误码

        except requests.exceptions.RequestException as e:
            retry -= 1
            time.sleep(2 ** (3 - retry))  # 指数退避

    raise Exception("获取 access_token 失败")

群发接口要求特定 JSON 结构，注意几个坑：

首图大图尺寸必须 900×500
正文需过滤特殊字符
摘要不超过 120 字

def build_news(articles):
    """
    构建多图文消息体
    :param articles: List[dict] 每个元素包含 title,thumb_media_id 等字段
    """return {"articles": [{"title": escape_html(article["title"]),  # 关键转义！"thumb_media_id": article["thumb_id"],"author": article.get("author",""),
            "digest": article["digest"][:117] + "..." if len(article["digest"]) > 120 else article["digest"],
            "show_cover_pic": 1 if idx == 0 else 0,  # 首图显示封面
            "content": html_escape(article["content"]),  # 处理 <> 等符号
            "content_source_url": article["origin_url"]
        } for idx, article in enumerate(articles)]
    }

大文件上传容易超时，需要分块处理：

先计算文件 MD5
查询是否已存在相同素材
分块上传（每块 2MB）

def upload_media(file_path, media_type="image"):
    """支持断点续传的素材上传"""
    chunk_size = 2 * 1024 * 1024  # 2MB 分块
    file_md5 = calculate_md5(file_path)

    # 检查素材是否已存在
    if media_id := check_exist(file_md5):
        return media_id

    # 初始化上传
    upload_id = init_upload(file_md5, os.path.getsize(file_path))

    # 分块上传
    with open(file_path, "rb") as f:
        for chunk in iter(lambda: f.read(chunk_size), b""):
            upload_chunk(upload_id, chunk)

    # 确认提交
    return confirm_upload(upload_id)

用 aiohttp 替代 requests，查询速度提升 5 倍 +：

async def batch_get_material(media_type, offset=0, count=20):
    """异步批量获取素材列表"""
    async with aiohttp.ClientSession() as session:
        tasks = []
        for i in range(0, count, 20):  # 每次最多 20 条
            params = {
                "type": media_type,
                "offset": offset + i,
                "count": min(20, count - i)
            }
            tasks.append(fetch_one_batch(session, params))

        return await asyncio.gather(*tasks)

推荐双重缓存：

内存缓存（快速读取）
Redis 缓存（进程间共享）

class TokenCache:
    _memory_cache = {}

    @classmethod
    def get(cls, key):
        # 内存优先
        if val := cls._memory_cache.get(key):
            if val["expire"] > time.time():
                return val["token"]

        # 读取 Redis
        if val := redis.get(key):
            cls._memory_cache[key] = {
                "token": val,
                "expire": time.time() + 300  # 内存缓存 5 分钟}
            return val

        return None