找回密码
立即注册
搜索
热搜: Java Python Linux Go
发回帖 发新帖

2410

积分

1

好友

333

主题
发表于 昨天 14:43 | 查看: 11| 回复: 0

我有个小脚本,专门在半夜运行一些定时任务:备份数据库、爬取监控数据、顺便检查服务器磁盘空间。以前总是等到第二天早上打开电脑,面对一屏幕的日志,心里默默祈祷别出现红色的 error。后来想想,这种方式太原始了。我的手机微信天天都在响,为什么不能让任务跑完后直接给我发条微信消息呢?比如:“兄弟,备份完成了!” 或者 “糟了,脚本挂了!”。

于是我折腾了一下 pushplus,用 Python 写了个小工具。现在,任何脚本只需添加一行代码就能把消息推送到我的微信上,非常方便。今天就把完整的实现过程分享出来,代码也已准备好,你可以直接拿去用。

先简单了解一下 pushplus 是什么

别被名字唬住,它本质上就是一个“消息通知转发器”:

  • 你通过 HTTP 接口将“消息”发送给 pushplus。
  • 它再帮你将消息转发到微信、企业微信、邮件、钉钉、飞书等渠道。

对于我们 Python 开发者而言,核心就是发送一个 HTTP 请求。

它的核心接口非常简单:

  • 地址http://www.pushplus.plus/send
  • 请求方式:支持 GET 或 POST
  • 关键参数tokentitlecontent,以及可选的 topic(群组)、template(消息模板)、channel(推送渠道)

所以,你可以先建立一个概念:pushplus 就是一个接收 HTTP 请求,然后向微信等平台发送消息的中间件。

开始前的准备工作

在编写代码之前,有三件必要的前置工作:

  1. 用微信搜索并关注公众号 “pushplus推送加”。
  2. 访问 pushplus 官网登录,在“个人中心”或“发送消息”页面找到你自己的 token。这个 token 至关重要,它相当于消息接收者的身份标识。建议将其存储在环境变量中,避免直接硬编码在代码里导致泄露。
  3. 在你的本地环境中安装 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,也可设为 txtmarkdown 等。

例如,如果你想发送一个格式更优美的 Markdown 消息,可以这样写:

def send_markdown_demo():
    lines = [
        "# 每日任务报告",
        "",
        "- 备份:✅ 成功",
        "- 爬虫:✅ 成功",
        "- CPU 峰值:42%",
    ]
    content = "\n".join(lines)
    pushplus_send(content, title="每日报告", template="markdown")

这样,在微信中打开的就是一条格式清晰的 Markdown 消息,比纯文本的阅读体验好很多。

进阶功能:使用 topic 进行群组推送

另一个常见需求是:不仅给自己发,还要把消息同步给整个团队。例如,“线上接口异常,请相关人员立即查看”。

pushplus 的解决方案是创建“群组”。每个群组有一个唯一的 topicCode,也就是接口参数中的 topic。你可以在官网创建群组并生成二维码,让团队成员扫码订阅。

简化流程如下:

  1. 在 pushplus 官网创建一个群组,记下群组编码,例如 dev-team
  2. 生成该群组的订阅二维码,分享到团队群中,让成员扫码订阅。
  3. 此后在代码中发送消息时带上参数 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 次免费额度,会员则会更高。接口调用频率也有限制,例如每分钟最多调用几次。

因此有两点建议:

  1. 不要滥用:避免在死循环或高频日志中疯狂调用 pushplus,它不是日志系统。
  2. 做好限流:对于可能高频触发的场景(如监控告警),最好在应用层做一层限流。例如,同一类错误在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 和微信来处理吧。

希望这篇指南能帮助你更高效地进行 运维 和脚本管理工作。欢迎在云栈社区分享你的使用心得或更酷的自动化方案。




上一篇:面试40K遭流水压价?从薪资博弈到旅行销售员问题的算法实战
下一篇:Spring Boot 4 升级必读:Jackson 3 的4大核心变化与迁移指南
您需要登录后才可以回帖 登录 | 立即注册

手机版|小黑屋|网站地图|云栈社区 ( 苏ICP备2022046150号-2 )

GMT+8, 2026-1-11 22:19 , Processed in 0.333910 second(s), 39 queries , Gzip On.

Powered by Discuz! X3.5

© 2025-2025 云栈社区.

快速回复 返回顶部 返回列表