Plombery + pushplus:Python 任务失败推到微信
效果:Plombery pipeline 跑失败或结束时,在任务里用 requests 调 pushplus。
前置条件
- 一个 pushplus token(官网扫码获取)
- 可改 pipeline 代码的 Plombery 项目
配置步骤
在失败分支或收尾任务里 POST JSON。环境变量用 PUSHPLUS_TOKEN:
import os
import requests
from plombery import task
def notify(title: str, content: str) -> None:
requests.post(
"https://www.pushplus.plus/send",
json={
"token": os.environ["PUSHPLUS_TOKEN"],
"title": title,
"content": content,
"template": "markdown",
},
timeout=10,
).raise_for_status()
@task
async def on_failure_notify():
notify("Plombery 任务失败", "请打开 Plombery 查看本次 run 日志")
把 on_failure_notify 挂到 pipeline 的失败路径,或在业务 task 的 except 里调用 notify。不要把 token 写进仓库。
若界面提供 Webhook / 通知 URL 且不能自定义 JSON,用固定文案:
https://www.pushplus.plus/send?token=你的token&title=Plombery通知&content=请打开Plombery查看任务
验证
curl -X POST "https://www.pushplus.plus/send" \
-H "Content-Type: application/json" \
-d '{"token":"你的token","title":"Plombery 配置测试","content":"token 可用","template":"markdown"}'
再手动跑一条必失败的 pipeline,确认微信能收到。
常见问题
成功也推? 只在 except / 失败 task 调用,避免每次成功占额度。
容器里 requests 超时? 确认调度环境能访问 www.pushplus.plus。