我有个小脚本,专门在半夜运行一些定时任务:备份数据库、爬取监控数据、顺便检查服务器磁盘空间。以前总是等到第二天早上打开电脑,面对一屏幕的日志,心里默默祈祷别出现红色的 error。后来想想,这种方式太原始了。我的手机微信天天都在响,为什么不能让任务跑完后直接给我发条微信消息呢?比如:“兄弟,备份完成了!” 或者 “糟了,脚本挂了!”。
于是我折腾了一下 pushplus,用 Python 写了个小工具。现在,任何脚本只需添加一行代码就能把消息推送到我的微信上,非常方便。今天就把完整的实现过程分享出来,代码也已准备好,你可以直接拿去用。
先简单了解一下 pushplus 是什么
别被名字唬住,它本质上就是一个“消息通知转发器”:
- 你通过 HTTP 接口将“消息”发送给 pushplus。
- 它再帮你将消息转发到微信、企业微信、邮件、钉钉、飞书等渠道。
对于我们 Python 开发者而言,核心就是发送一个 HTTP 请求。
它的核心接口非常简单:
- 地址:
http://www.pushplus.plus/send
- 请求方式:支持 GET 或 POST
- 关键参数:
token、title、content,以及可选的 topic(群组)、template(消息模板)、channel(推送渠道)
所以,你可以先建立一个概念:pushplus 就是一个接收 HTTP 请求,然后向微信等平台发送消息的中间件。
开始前的准备工作
在编写代码之前,有三件必要的前置工作:
- 用微信搜索并关注公众号 “pushplus推送加”。
- 访问 pushplus 官网登录,在“个人中心”或“发送消息”页面找到你自己的
token。这个 token 至关重要,它相当于消息接收者的身份标识。建议将其存储在环境变量中,避免直接硬编码在代码里导致泄露。
- 在你的本地环境中安装
requests 库:
pip install requests
最基础的用法:发送单条微信文本消息
让我们从最小的可工作闭环开始:编写一个函数,给自己发送一条最简单的微信文本消息。如果这条消息能成功接收,就意味着整个链路是通的。
下面是我日常使用的一个基础版本,你可以直接替换掉 token 和消息内容进行测试:
import os
import requests
# 建议使用环境变量存储,不要将 token 硬编码在代码中
PUSHPLUS_TOKEN = os.getenv("PUSHPLUS_TOKEN", "这里填你的token")
def pushplus_send(
content: str,
title: str = "Python 小通知",
template: str = "html",
) -> None:
"""
使用 pushplus 向自己发送一条微信消息
"""
if not PUSHPLUS_TOKEN:
raise RuntimeError("缺少 PUSHPLUS_TOKEN,请先配置环境变量")
url = "http://www.pushplus.plus/send"
params = {
"token": PUSHPLUS_TOKEN,
"title": title,
"content": content,
"template": template,
}
try:
resp = requests.get(url, params=params, timeout=5)
data = resp.json()
except Exception as exc:
# 这里可以根据你项目的日志体系进行记录
print(f"[pushplus] 请求失败:{exc}")
return
# pushplus 是异步发送的,这里的 code=200 仅代表“请求已被接收”
# 实际发送成功或失败会通过回调或后续查询返回结果
code = data.get("code")
if code != 200:
print(f"[pushplus] 请求返回异常 code={code}, msg={data.get('msg')}")
else:
# data 中会包含一个消息流水号,可用于后续问题排查
print(f"[pushplus] 已提交消息,流水号={data.get('data')}")
使用一个最简单的调用来测试:
if __name__ == "__main__":
pushplus_send("脚本执行完毕,一切正常 😄", title="定时任务通知")
运行后,查看你微信中 pushplus 公众号的消息。如果收到了这条消息,说明:
- 你的
token 配置正确。
- 你的 Python 环境可以正常访问 pushplus 服务。
- 微信端的接收也一切正常。
链路一旦打通,后面的玩法就多了。
参数简要说明
上面代码中用到的几个参数,已经能满足80%的日常需求:
token:你的个人令牌,pushplus 凭此识别消息接收者。
title:消息标题(可选),但有了标题在微信中查看会更清晰。
content:消息正文,支持纯文本、HTML 或 Markdown 格式。
template:消息展示模板,默认为 html,也可设为 txt、markdown 等。
例如,如果你想发送一个格式更优美的 Markdown 消息,可以这样写:
def send_markdown_demo():
lines = [
"# 每日任务报告",
"",
"- 备份:✅ 成功",
"- 爬虫:✅ 成功",
"- CPU 峰值:42%",
]
content = "\n".join(lines)
pushplus_send(content, title="每日报告", template="markdown")
这样,在微信中打开的就是一条格式清晰的 Markdown 消息,比纯文本的阅读体验好很多。
进阶功能:使用 topic 进行群组推送
另一个常见需求是:不仅给自己发,还要把消息同步给整个团队。例如,“线上接口异常,请相关人员立即查看”。
pushplus 的解决方案是创建“群组”。每个群组有一个唯一的 topicCode,也就是接口参数中的 topic。你可以在官网创建群组并生成二维码,让团队成员扫码订阅。
简化流程如下:
- 在 pushplus 官网创建一个群组,记下群组编码,例如
dev-team。
- 生成该群组的订阅二维码,分享到团队群中,让成员扫码订阅。
- 此后在代码中发送消息时带上参数
topic="dev-team",所有订阅了该群组的成员都会收到消息。
我们对之前的函数稍作扩展,以支持群组推送:
def pushplus_send_group(
content: str,
title: str,
topic: str,
template: str = "markdown",
) -> None:
"""
向指定群组发送消息(一对多)
"""
if not PUSHPLUS_TOKEN:
raise RuntimeError("缺少 PUSHPLUS_TOKEN,请先配置环境变量")
url = "http://www.pushplus.plus/send"
payload = {
"token": PUSHPLUS_TOKEN,
"title": title,
"content": content,
"topic": topic,
"template": template,
}
try:
# 这里使用 POST 请求,参数放在 JSON body 中
resp = requests.post(url, json=payload, timeout=5)
data = resp.json()
except Exception as exc:
print(f"[pushplus] 群组推送失败:{exc}")
return
if data.get("code") != 200:
print(f"[pushplus] 群组推送异常:{data}")
例如,需要向 dev-team 群组发送一条线上告警:
if __name__ == "__main__":
content = """
# 线上告警
- 接口:/api/order/list
- 状态:500
- 错误:数据库连接超时
- 时间:2025-12-12 23:30
"""
pushplus_send_group(content, title="【告警】订单接口异常", topic="dev-team")
这样,即使你不在电脑前,整个团队也能第一时间在微信上收到告警信息,便于快速响应。
封装成可复用的工具类
在实际工作中,更佳实践是将推送逻辑封装成一个工具类,便于在各个脚本中统一调用和管理。
下面是一个简单的封装示例:
from dataclasses import dataclass
from typing import Optional, Literal
import os
import requests
TemplateType = Literal["html", "txt", "markdown", "json"]
@dataclass
class PushplusClient:
token: str = os.getenv("PUSHPLUS_TOKEN", "")
def _ensure_token(self):
if not self.token:
raise RuntimeError("pushplus token 未配置")
def send(
self,
content: str,
title: str = "",
topic: Optional[str] = None,
template: TemplateType = "html",
channel: str = "wechat",
) -> bool:
"""
统一的消息发送方法,返回 True/False 表示请求是否成功
"""
self._ensure_token()
url = "http://www.pushplus.plus/send"
payload = {
"token": self.token,
"title": title,
"content": content,
"template": template,
"channel": channel,
}
if topic:
payload["topic"] = topic
try:
resp = requests.post(url, json=payload, timeout=5)
data = resp.json()
except Exception as exc:
print(f"[pushplus] 请求异常:{exc}")
return False
if data.get("code") != 200:
print(f"[pushplus] 发送失败:{data}")
return False
return True
在任何脚本中,都可以这样使用:
pp = PushplusClient()
def run_job():
# 这里是你的业务逻辑
...
# 成功或失败后发送通知
pp.send("定时任务执行成功", title="任务正常结束")
如果需要发送给群组,只需额外传入 topic 参数即可。
需要注意的两个要点:额度和频率限制
pushplus 作为一项公共服务,对使用有一定限制,不注意可能会踩坑:
- 根据用户身份(未实名、实名、会员),每日可用的请求次数和推送渠道有所不同。
- 例如,通过微信渠道,实名用户每天大约有 200 次免费额度,会员则会更高。接口调用频率也有限制,例如每分钟最多调用几次。
因此有两点建议:
- 不要滥用:避免在死循环或高频日志中疯狂调用 pushplus,它不是日志系统。
- 做好限流:对于可能高频触发的场景(如监控告警),最好在应用层做一层限流。例如,同一类错误在5分钟内只推送一次。可以在工具类外层结合 Redis 或本地文件做一个“已发送”标记,这里就不展开详述了。
一个完整示例:脚本异常时自动推送
一个非常典型的需求是:当脚本运行出错时,立即发送微信通知。实现起来很简单,利用 try/except 捕获异常,并将堆栈信息发送给 pushplus。
import traceback
from pushplus_tool import PushplusClient # 假设上面的类保存在此模块中
pp = PushplusClient()
def main():
# 这里放置你的主业务逻辑
1 / 0 # 故意制造一个异常
if __name__ == "__main__":
try:
main()
pp.send("脚本执行成功 ✅", title="脚本状态")
except Exception:
err = traceback.format_exc()
text = f"脚本执行失败 ❌\n\n```python\n{err}\n```"
pp.send(text, title="脚本报错", template="markdown")
运行这个脚本,一旦发生错误,你的手机微信就会立刻收到包含完整错误堆栈的 Markdown 消息,这基本满足了大多数“远程监控脚本”的需求。
总结
总的来说,使用 pushplus 实现 Python 脚本的微信通知上手门槛很低,重点在于:
- 妥善保管你的
token,避免泄露。
- 使用一个工具类进行统一封装,方便管理和调用。
- 合理控制消息发送频率,避免触发限制。
无论是服务器监控、抢票提醒还是定时生成日报,只要你能用 Python 实现核心逻辑,最后加上一行 pp.send(...) 即可将结果推送到微信。剩下的,就交给 pushplus 和微信来处理吧。
希望这篇指南能帮助你更高效地进行 运维 和脚本管理工作。欢迎在云栈社区分享你的使用心得或更酷的自动化方案。