用 Python 实现对微信进行推送消息的工具 -- pushplus

有段时间我经常被自己写的脚本吓醒。

半夜三点，手机突然震一下，是某个爬虫崩了、某个定时任务没跑完，或者服务器磁盘又快满了。以前这些通知都是我自己写邮件、打 webhook，写来写去一堆重复代码，心情非常暴躁。

后来换成 pushplus 之后，基本就一个函数解决问题了，整个世界清净不少。

今天就用一个完整的小工具，讲一下怎么用 Python 把各种消息推到微信上去，用的就是 pushplus 这个服务。

先把 pushplus 是啥说清楚

简单讲，pushplus 就是一个“代发消息”的平台，它帮你对接好了微信、企业微信、邮件、钉钉、飞书之类的渠道，你这边只要调一个 HTTP 接口，把 token、标题、内容丢过去，它就负责送达。官方文档里，最常用的发送接口是 http://www.pushplus.plus/send，支持 GET 和 POST，两种都行，不过长内容官方也更建议用 POST

实际用的时候，你只关心这几个信息：

• 你的用户 token（登录官网、关注公众号之后可以在个人中心看到）
• 一个固定的接口地址
• 要发的标题和内容
• 可选的一些参数，比如模板类型、群组编码 topic、发送渠道 channel 等

所以整体思路就是：Python 里用 requests 调这个接口，包一层函数，之后任何脚本里想通知就调用一下。

环境准备一下

Python 版本随意，3.7 以上就很正常了，核心依赖就一个：

pip install requests

然后找个地方把你的 pushplus token 记下来，等等代码里要用，长得大概像一串 32 位的十六进制字符串。

写一个最小可用的推送函数

先来个能跑起来的最小版本，先不考虑各种异常和高级功能，思路就是：构造一个 JSON，POST 给 pushplus 的 send 接口。

import requests

PUSHPLUS_TOKEN = "替换成你自己的 token"

defpushplus_simple(content: str, title: str = "Python 通知"):
"""
    用 pushplus 推一条最简单的微信消息
    """
    url = "http://www.pushplus.plus/send"
    body = {
"token": PUSHPLUS_TOKEN,
"title": title,
"content": content,
# 不写 template 默认是 html，就先用默认的
    }

    resp = requests.post(url, json=body, timeout=10)
    resp.raise_for_status()
    data = resp.json()
# 文档里说 code=200 只代表“请求收到了”，真正是否送达是异步处理的:contentReference[oaicite:2]{index=2}
if data.get("code") == 200:
        print("pushplus 已接收请求，流水号:", data.get("data"))
else:
        print("pushplus 拒绝了请求:", data)

跑一行试试看：

if __name__ == "__main__":
    pushplus_simple("脚本跑完啦，耗时 3.21s", "测试推送")

如果你公众号关注好了，基本几秒钟微信上就能看到消息。

稍微专业一点：封装成一个小工具类

上面这个函数够用，但一旦你想支持群组、markdown、邮件渠道之类的，就会发现参数开始变多了。可以稍微封装一个类，后面在任何项目里直接 import 复用。

import json
import traceback
from typing import Optional, Dict, Any

import requests


classPushPlusClient:
"""
    一个简单的 pushplus 封装
    """

def__init__(
        self,
        token: str,
        base_url: str = "http://www.pushplus.plus/send",
        timeout: int = 10,
    ):
        self.token = token
        self.base_url = base_url
        self.timeout = timeout

defsend(
        self,
        content: str,
        title: str = "Python 通知",
        template: str = "html",
        topic: Optional[str] = None,
        channel: Optional[str] = None,
        option: Optional[str] = None,
        to: Optional[str] = None,
        extra: Optional[Dict[str, Any]] = None,
    ) -> Dict[str, Any]:
"""
        发送一条消息到 pushplus

        :param content: 消息正文
        :param title: 消息标题
        :param template: 模板类型，html/txt/json/markdown 等:contentReference[oaicite:3]{index=3}
        :param topic: 群组编码，不传就是发给自己
        :param channel: 发送渠道，比如 wechat/mail/webhook/cp 等:contentReference[oaicite:4]{index=4}
        :param option: 某些渠道需要的额外配置编码，比如企业微信机器人编码
        :param to: 好友 token（发给指定好友时用）:contentReference[oaicite:5]{index=5}
        :param extra: 预留的一些扩展字段，比如 callbackUrl、timestamp 等
        """
        payload: Dict[str, Any] = {
"token": self.token,
"title": title,
"content": content,
"template": template,
        }

if topic:
            payload["topic"] = topic
if channel:
            payload["channel"] = channel
if option:
            payload["option"] = option
if to:
            payload["to"] = to
if extra:
            payload.update(extra)

try:
            resp = requests.post(
                self.base_url, json=payload, timeout=self.timeout
            )
            resp.raise_for_status()
            data = resp.json()
except Exception:
# 这里就别再推自己了，打印个本地日志就行
            traceback.print_exc()
return {"code": -1, "msg": "request error"}

# 按文档约定，code=200 表示“请求正常受理”，真正发没发出去需要看异步回调或者用流水号去查:contentReference[oaicite:6]{index=6}
return data

用法大概长这样：

client = PushPlusClient(token="你的 token")

# 发一个普通 html 消息
client.send("构建成功 ✅", title="CI 构建通知")

# 发一个 markdown 消息
md_content = """# 线上发布结果

- 版本：v1.2.3
- 状态：**成功**
- 耗时：15 分钟
"""
client.send(md_content, title="发版结果", template="markdown")

做一个“脚本异常就自动推送”的小工具

最常见的需求就是：脚本里一旦出错，就给自己发个微信，不然半夜炸了你第二天才知道。

可以写一个简单的装饰器，把异常捕获之后丢给 pushplus：

import functools
import traceback

push_client = PushPlusClient(token="你的 token")


defnotify_on_error(title: str = "脚本异常告警"):
"""
    把这个装饰器挂在任意函数上，一旦抛异常就微信通知
    """

defdecorator(func):
        @functools.wraps(func)
defwrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except Exception as e:
                tb = traceback.format_exc()
                content = f"{func.__name__} 运行异常：{e}\n\n```python\n{tb}\n```"
# 用 markdown 模板，这样栈信息好看一点
                push_client.send(
                    content=content,
                    title=title,
                    template="markdown",
                )
# 异常再抛出去，看你业务需要
raise

return wrapper

return decorator

随便找个函数套上：

@notify_on_error("定时任务异常")
defsync_order_data():
# 模拟一下错误
    x = 1 / 0

这个函数每次炸掉的时候，你手机上就会收到一条类似这样的消息：标题“定时任务异常”，正文里是完整的 traceback，定位问题非常舒服。

顺便利用一下 json 模板做“小报表”

pushplus 有一个 json 模板，意思是你传过去的 content 是一段 JSON 字符串，它会按表格风格帮你排一下版，比纯文本好看很多。

比如每天把爬虫成功/失败数量、耗时之类的统计一下，发一个小日报：

from datetime import datetime


defsend_daily_report():
    report = {
"date": datetime.now().strftime("%Y-%m-%d"),
"spider": "job_spider",
"total": 1234,
"success": 1200,
"failed": 34,
"avg_ms": 85,
    }

    content = json.dumps(report, ensure_ascii=False)

    push_client.send(
        content=content,
        title="爬虫日报",
        template="json",  # 让 pushplus 按 json 模板渲染:contentReference[oaicite:8]{index=8}
    )

你会在微信里看到一块类似“字段-值”的结构化展示，比截图发送、美化日志都省事。

如果要群发同事，看下 topic 的用法

自己一个人玩还好，如果你想把消息推给一整个小组，比如“后端技术群”，可以用 pushplus 的群组功能，对应参数叫 topic。

大致流程是：

• 在 pushplus 官网创建一个群组，拿到一个群组编码（topic）
• 把同事拉进这个群
• 代码里把 topic 传进去，消息就会群发给这批人

代码其实跟刚才没啥区别，多传个参数而已：

defsend_team_deploy_msg(ok: bool, topic: str):
    status = "成功 ✅"if ok else"失败 ❌"
    content = f"新版本发布{status}"

    push_client.send(
        content=content,
        title="发布结果通知",
        topic=topic,         # 群组编码
        template="html",
    )

这样一来，CI/CD 跑完之后，整个小组手机上一起震一下，谁也跑不掉。

多渠道的事简单提一嘴

有时候你希望除了微信，还顺便发个企业微信机器人或者邮件，这个在 pushplus 里叫“多渠道”，参数是 channel 和 option。

比如你已经在 pushplus 里配置好了一个企业微信机器人，编码叫 pushplus，那大概是这样：

defsend_to_wechat_robot(msg: str):
    push_client.send(
        content=msg,
        title="机器人测试",
        channel="webhook",  # 指定走 webhook 渠道:contentReference[oaicite:12]{index=12}
        option="pushplus",  # 你在控制台配置的渠道编码
        template="markdown",
    )

邮件、企业微信应用这些也类似，只是 channel、option 的取值不一样，文档里都有列。

最后几个小细节，踩坑少一点

这个工具写完基本就能一直复用了，不过有几个小点提一下，免得你以后骂它：

• 内容很长的时候，尽量用 POST，不要偷懒用 GET，那样 URL 可能会被浏览器或中间代理砍掉一截。
• code == 200 不代表消息百分百送达，只能说明 pushplus 服务端收到了你的请求，真正的失败/成功是异步处理的，如果你业务很关键，可以配一下 callbackUrl 或者用流水号再查一次结果。
• 记得设个合理的 timeout，不要让你的业务线程被一个网络抖动拖死；上面的例子里我用了 10 秒，你可以按场景自己调。
• token 别写死在公开仓库里，最起码搞个环境变量或者配置文件忽略掉。

基本就这些了，你把上面这几个函数拷到自己项目里，改好 token 和一点点业务文案，日常脚本、定时任务、监控告警，通通可以一键推到微信上。

至于推的太多会不会被同事拉黑，那就另说了…