Pushplus 微信推送 Skill

简介

本 Skill 用于通过 Pushplus（推送加）服务向微信及其他渠道发送消息推送。Pushplus 是一个集成了微信、企业微信、邮件、钉钉、飞书、短信、语音、App、浏览器插件、桌面应用等多种渠道的实时消息推送平台，只需调用简单的 API 即可完成消息推送。

官网: https://www.pushplus.plus
API 文档: https://www.pushplus.plus/doc/guide/api.html

调用时机

• 用户需要发送微信消息通知
• 需要实现消息提醒功能（告警、通知、日报等）
• 需要将信息推送到手机微信
• 需要向群组或好友发送消息
• 需要多渠道同时推送（微信 + 邮件 + 钉钉等）
• 需要发送短信或语音通知

前置条件

1. 在 https://www.pushplus.plus 注册账号并获取 Token
2. 微信搜索关注「pushplus 推送加」公众号（接收微信消息）
3. 完成实名认证（否则无法发送消息）

Token 配置说明

本 Skill 支持以下 Token 配置方式：

1. 全局配置（推荐）: 首次使用时设置全局 Token，后续自动使用
2. 临时指定: 发送时临时指定 Token，覆盖全局配置
3. 环境变量: 通过 PUSHPLUS_TOKEN 环境变量设置

首次使用:
- 如果未配置全局 Token，首次调用时会自动提示设置
- 设置后自动保存到本地配置文件，后续无需重复设置

API 接口概览

请求方式支持：GET、POST、PUT、DELETE
Content-Type: application/json

发送消息接口参数

发送渠道（channel）

消息模板（template）

完整封装客户端

以下是一个功能完整的 PushplusClient 封装类，覆盖所有常用功能，并支持全局 Token 管理：

try:
    import requests  # type: ignore
    HAS_REQUESTS = True
except ImportError:  # 纯净环境回退到标准库
    requests = None  # type: ignore
    HAS_REQUESTS = False
import json
import os
from pathlib import Path
from typing import Optional, List


def _post_json(url: str, payload: dict, timeout: int = 10) -> dict:
    """POST JSON：优先 requests，未安装时回退 urllib（无第三方依赖也能跑）"""
    data = json.dumps(payload).encode("utf-8")
    headers = {"Content-Type": "application/json"}
    if HAS_REQUESTS:
        try:
            resp = requests.post(url, data=data, headers=headers, timeout=timeout)
            return resp.json()
        except requests.exceptions.Timeout:
            return {"code": -1, "msg": "请求超时"}
        except Exception as e:  # noqa: BLE001
            return {"code": -1, "msg": str(e)}
    import urllib.request
    import urllib.error
    req = urllib.request.Request(url, data=data, headers=headers, method="POST")
    try:
        with urllib.request.urlopen(req, timeout=timeout) as r:
            return json.loads(r.read().decode("utf-8"))
    except urllib.error.HTTPError as e:
        return {"code": -1, "msg": f"HTTP {e.code}"}
    except Exception as e:  # noqa: BLE001
        return {"code": -1, "msg": str(e)}


class PushplusClient:
    """Pushplus 推送客户端 - 完整封装（支持全局 Token 管理）"""

    SEND_URL = "http://www.pushplus.plus/send"
    BATCH_SEND_URL = "http://www.pushplus.plus/batchSend"
    CONFIG_FILE = str(Path.home() / ".pushplus_token.json")

    def __init__(self, token: Optional[str] = None, auto_setup: bool = True):
        """
        初始化客户端

        参数:
            token: Pushplus Token，不传则使用全局配置
            auto_setup: 是否自动提示设置 Token（默认 True）
        """
        self._global_token = None
        self._load_global_token()

        # 优先使用传入的 token，否则使用全局配置
        if token:
            self.token = token
        elif self._global_token:
            self.token = self._global_token
        elif auto_setup:
            self.token = self._setup_token_interactive()
        else:
            self.token = None

    def _load_global_token(self) -> None:
        """从配置文件或环境变量加载全局 Token"""
        # 1. 从环境变量加载
        self._global_token = os.environ.get("PUSHPLUS_TOKEN")

        # 2. 从配置文件加载（优先级更高）
        try:
            if os.path.exists(self.CONFIG_FILE):
                with open(self.CONFIG_FILE, "r", encoding="utf-8") as f:
                    config = json.load(f)
                    self._global_token = config.get("token", self._global_token)
        except Exception:
            pass

    def _save_global_token(self, token: str) -> bool:
        """保存 Token 到配置文件"""
        try:
            with open(self.CONFIG_FILE, "w", encoding="utf-8") as f:
                json.dump({"token": token}, f, ensure_ascii=False, indent=2)
            return True
        except Exception as e:
            print(f"[Pushplus] 保存 Token 失败: {e}")
            return False

    def _setup_token_interactive(self) -> str:
        """交互式设置 Token（首次使用时调用）"""
        print("\n" + "=" * 50)
        print("Pushplus Token 未配置")
        print("=" * 50)
        print("\n请按以下步骤获取 Token：")
        print("1. 访问 https://www.pushplus.plus 注册/登录")
        print("2. 进入个人中心 → 一对一消息")
        print("3. 复制你的 Token")
        print("\n" + "-" * 50)

        token = input("\n请输入你的 Pushplus Token: ").strip()

        if not token:
            print("[Pushplus] Token 不能为空，未设置")
            return None

        # 询问是否保存为全局配置
        save = input("是否保存为全局配置？(Y/n): ").strip().lower()
        if save != "n":
            if self._save_global_token(token):
                print(f"[Pushplus] Token 已保存到 {self.CONFIG_FILE}")
            else:
                print("[Pushplus] Token 保存失败，本次会话仍可使用")

        return token

    @staticmethod
    def set_global_token(token: str) -> bool:
        """
        设置全局 Token（类方法，无需实例化即可调用）

        参数:
            token: Pushplus Token

        返回:
            bool: 是否保存成功
        """
        try:
            config_file = str(Path.home() / ".pushplus_token.json")
            with open(config_file, "w", encoding="utf-8") as f:
                json.dump({"token": token}, f, ensure_ascii=False, indent=2)
            print(f"[Pushplus] 全局 Token 已设置")
            return True
        except Exception as e:
            print(f"[Pushplus] 设置 Token 失败: {e}")
            return False

    @staticmethod
    def get_global_token() -> Optional[str]:
        """
        获取当前全局 Token（类方法）

        返回:
            str: 当前全局 Token，未设置则返回 None
        """
        # 1. 从环境变量获取
        token = os.environ.get("PUSHPLUS_TOKEN")

        # 2. 从配置文件获取（优先级更高）
        try:
            config_file = str(Path.home() / ".pushplus_token.json")
            if os.path.exists(config_file):
                with open(config_file, "r", encoding="utf-8") as f:
                    config = json.load(f)
                    token = config.get("token", token)
        except Exception:
            pass

        return token

    @staticmethod
    def clear_global_token() -> bool:
        """
        清除全局 Token（类方法）

        返回:
            bool: 是否清除成功
        """
        try:
            config_file = str(Path.home() / ".pushplus_token.json")
            if os.path.exists(config_file):
                os.remove(config_file)
            print("[Pushplus] 全局 Token 已清除")
            return True
        except Exception as e:
            print(f"[Pushplus] 清除 Token 失败: {e}")
            return False

    def _get_token(self, token: Optional[str] = None) -> str:
        """获取有效的 Token（优先使用传入的，否则使用实例的）"""
        if token:
            return token
        if self.token:
            return self.token
        raise ValueError("Pushplus Token 未配置，请先调用 set_global_token() 或传入 token 参数")

    def send(
        self,
        content: str,
        title: Optional[str] = None,
        template: str = "html",
        channel: str = "wechat",
        topic: Optional[str] = None,
        to: Optional[str] = None,
        option: Optional[str] = None,
        callback_url: Optional[str] = None,
        timestamp: Optional[int] = None,
        pre: Optional[str] = None,
        timeout: int = 10,
        token: Optional[str] = None,
    ) -> dict:
        """
        发送消息（单渠道）

        参数:
            content: 消息内容（必填）
            title: 消息标题
            template: 消息模板 (html/txt/json/markdown/cloudMonitor/jenkins/route/pay)
            channel: 发送渠道 (wechat/webhook/cp/mail/sms/voice/extension/app/clawbot)
            topic: 群组编码（一对多消息）
            to: 好友令牌，多人用逗号隔开
            option: 渠道配置参数（webhook/邮件/企业微信需填渠道编码）
            callback_url: 异步回调地址
            timestamp: 毫秒时间戳（防重复发送）
            pre: 预处理编码（会员功能）
            timeout: 请求超时时间（秒）
            token: 临时指定的 Token（覆盖全局配置）

        返回:
            dict: API 响应结果
                成功: {"code": 200, "msg": "请求成功", "data": "消息流水号"}
                失败: {"code": xxx, "msg": "错误信息", "data": null}
        """
        data = {
            "token": self._get_token(token),
            "content": content,
            "template": template,
            "channel": channel,
        }

        # 可选参数
        if title:
            data["title"] = title
        if topic:
            data["topic"] = topic
        if to:
            data["to"] = to
        if option:
            data["option"] = option
        if callback_url:
            data["callbackUrl"] = callback_url
        if timestamp:
            data["timestamp"] = timestamp
        if pre:
            data["pre"] = pre

        try:
            result = _post_json(self.SEND_URL, data, timeout=timeout)

            if result.get("code") == 200:
                print(f"[Pushplus] 推送请求成功, 流水号: {result.get('data')}")
            else:
                print(f"[Pushplus] 推送失败: code={result.get('code')}, msg={result.get('msg')}")

            return result

        except Exception as e:
            error_msg = {"code": -1, "msg": str(e)}
            print(f"[Pushplus] 请求异常: {e}")
            return error_msg

    def batch_send(
        self,
        content: str,
        channels: List[str],
        title: Optional[str] = None,
        template: str = "html",
        topic: Optional[str] = None,
        to: Optional[str] = None,
        options: Optional[List[str]] = None,
        callback_url: Optional[str] = None,
        timestamp: Optional[int] = None,
        pre: Optional[str] = None,
        timeout: int = 10,
        token: Optional[str] = None,
    ) -> dict:
        """
        多渠道同时发送消息

        参数:
            content: 消息内容（必填）
            channels: 渠道列表，如 ["wechat", "webhook", "mail"]
            title: 消息标题
            template: 消息模板
            topic: 群组编码
            to: 好友令牌
            options: 各渠道的 option 参数列表，与 channels 一一对应，无则填空字符串
            callback_url: 异步回调地址
            timestamp: 毫秒时间戳
            pre: 预处理编码
            timeout: 请求超时时间（秒）
            token: 临时指定的 Token（覆盖全局配置）

        返回:
            dict: 包含每个渠道的发送结果列表
        """
        data = {
            "token": self._get_token(token),
            "content": content,
            "template": template,
            "channel": ",".join(channels),
        }

        if title:
            data["title"] = title
        if topic:
            data["topic"] = topic
        if to:
            data["to"] = to
        if options:
            data["option"] = ",".join(options)
        if callback_url:
            data["callbackUrl"] = callback_url
        if timestamp:
            data["timestamp"] = timestamp
        if pre:
            data["pre"] = pre

        try:
            result = _post_json(self.BATCH_SEND_URL, data, timeout=timeout)

            if result.get("code") == 200:
                print(f"[Pushplus] 多渠道推送请求成功")
                for item in result.get("data", []):
                    print(f"  - 渠道 {item.get('channel')}: 流水号 {item.get('shortCode')}")
            else:
                print(f"[Pushplus] 多渠道推送失败: {result.get('msg')}")

            return result

        except Exception as e:
            error_msg = {"code": -1, "msg": str(e)}
            print(f"[Pushplus] 请求异常: {e}")
            return error_msg

    def send_to_self(self, content: str, title: Optional[str] = None, template: str = "html", token: Optional[str] = None) -> dict:
        """发送消息给自己（最简用法）"""
        return self.send(content=content, title=title, template=template, token=token)

    def send_to_group(
        self, content: str, topic: str, title: Optional[str] = None, template: str = "html", token: Optional[str] = None
    ) -> dict:
        """
        发送群组消息（一对多）

        参数:
            content: 消息内容
            topic: 群组编码（在 pushplus 官网创建群组时自定义）
            title: 消息标题
            template: 消息模板
            token: 临时指定的 Token
        """
        return self.send(content=content, title=title, template=template, topic=topic, token=token)

    def send_to_friend(
        self, content: str, to: str, title: Optional[str] = None, template: str = "html", token: Optional[str] = None
    ) -> dict:
        """
        发送好友消息

        参数:
            content: 消息内容
            to: 好友令牌（在 pushplus 好友列表中查看），多人用逗号隔开
            title: 消息标题
            template: 消息模板
            token: 临时指定的 Token

        注意:
            - 好友消息和群组消息不能同时使用（topic 优先级更高）
            - 实名用户一次最多 10 人，会员最多 100 人
        """
        return self.send(content=content, title=title, template=template, to=to, token=token)

    def send_webhook(
        self,
        content: str,
        webhook_code: str,
        title: Optional[str] = None,
        template: str = "html",
        token: Optional[str] = None,
    ) -> dict:
        """
        发送消息到 Webhook（企业微信/钉钉/飞书等机器人）

        参数:
            content: 消息内容
            webhook_code: 在 pushplus 渠道配置中新增的 webhook 编码
            title: 消息标题
            template: 消息模板
            token: 临时指定的 Token
        """
        return self.send(
            content=content,
            title=title,
            template=template,
            channel="webhook",
            option=webhook_code,
            token=token,
        )

    def send_mail(
        self,
        content: str,
        title: Optional[str] = None,
        mail_code: Optional[str] = None,
        template: str = "html",
        token: Optional[str] = None,
    ) -> dict:
        """
        发送邮件

        参数:
            content: 邮件正文
            title: 邮件标题
            mail_code: 邮件渠道编码（不填使用官网默认邮件发送）
            template: 消息模板
            token: 临时指定的 Token
        """
        return self.send(
            content=content,
            title=title,
            template=template,
            channel="mail",
            option=mail_code or "",
            token=token,
        )

    def send_sms(self, title: str, content: str, token: Optional[str] = None) -> dict:
        """
        发送短信（收费，10 积分/条）

        参数:
            title: 短信内容（短信渠道只展示标题）
            content: 附加内容
            token: 临时指定的 Token
        """
        return self.send(content=content, title=title, channel="sms", token=token)

    def send_voice(self, title: str, content: str, token: Optional[str] = None) -> dict:
        """
        发送语音通知（收费，30 积分/条）

        参数:
            title: 语音通知标题
            content: 附加内容
            token: 临时指定的 Token
        """
        return self.send(content=content, title=title, channel="voice", token=token)

    def send_with_callback(
        self,
        content: str,
        callback_url: str,
        title: Optional[str] = None,
        template: str = "html",
        token: Optional[str] = None,
    ) -> dict:
        """
        发送消息并设置异步回调

        消息处理完成后会 POST 回调结果到 callback_url，回调 body 示例:
        {
            "event": "message_complate",
            "messageInfo": {
                "message": "",
                "shortCode": "88*********50fe",
                "sendStatus": 2  // 0-未发送, 1-发送中, 2-成功, 3-失败
            }
        }
        """
        return self.send(
            content=content,
            title=title,
            template=template,
            callback_url=callback_url,
            token=token,
        )

    def send_with_timestamp(
        self,
        content: str,
        expire_timestamp: int,
        title: Optional[str] = None,
        template: str = "html",
        token: Optional[str] = None,
    ) -> dict:
        """
        发送带时效性的消息（防重复发送）

        参数:
            content: 消息内容
            expire_timestamp: 毫秒时间戳，服务器时间大于此值则不发送
            title: 消息标题
            template: 消息模板
            token: 临时指定的 Token
        """
        return self.send(
            content=content,
            title=title,
            template=template,
            timestamp=expire_timestamp,
            token=token,
        )

    def send_markdown(self, title: str, content: str, token: Optional[str] = None, **kwargs) -> dict:
        """发送 Markdown 格式消息"""
        return self.send(content=content, title=title, template="markdown", token=token, **kwargs)

    def send_json(self, title: str, content: str, token: Optional[str] = None, **kwargs) -> dict:
        """发送 JSON 格式消息"""
        return self.send(content=content, title=title, template="json", token=token, **kwargs)

    def send_txt(self, title: str, content: str, token: Optional[str] = None, **kwargs) -> dict:
        """发送纯文本消息"""
        return self.send(content=content, title=title, template="txt", token=token, **kwargs)

使用示例

1. 首次使用（自动提示设置 Token）

# 首次使用会自动提示输入 Token
client = PushplusClient()
client.send_to_self(title="测试通知", content="这是一条测试消息")

2. 手动设置全局 Token

# 手动设置全局 Token（后续自动使用）
PushplusClient.set_global_token("你的Token")

# 之后无需再传 Token
client = PushplusClient()
client.send_to_self(title="测试通知", content="自动使用全局 Token")

3. 临时使用其他 Token 发送

client = PushplusClient()

# 使用全局 Token 发送
client.send_to_self(title="通知1", content="使用全局 Token")

# 临时使用其他 Token 发送
client.send_to_self(title="通知2", content="使用临时 Token", token="其他用户的Token")

4. 查看和清除全局 Token

# 查看当前全局 Token
token = PushplusClient.get_global_token()
print(f"当前全局 Token: {token}")

# 清除全局 Token
PushplusClient.clear_global_token()

5. 发送 Markdown 格式的日报

content = """
# 系统日报

## 今日统计
- 用户访问: 1,234 次
- 新增订单: 56 单
- 系统运行: 正常

## 异常记录
无异常
"""

client.send_markdown(title="系统日报", content=content)

6. 发送群组消息

# topic 为在 pushplus 官网创建群组时自定义的群组编码
client.send_to_group(
    title="群组通知",
    content="这是一条群组消息，所有群组成员都会收到",
    topic="你的群组编码",
    template="markdown",
)

7. 发送好友消息

# to 为好友令牌，在 pushplus 好友列表中查看
client.send_to_friend(
    title="好友通知",
    content="这是一条好友消息",
    to="好友令牌A,好友令牌B",  # 多人用逗号隔开
)

8. 多渠道同时推送

# 同时推送到微信、企业微信机器人、邮件
client.batch_send(
    title="重要通知",
    content="这是一条多渠道消息",
    channels=["wechat", "webhook", "mail"],
    options=["", "企业微信机器人编码", "163"],  # 与 channels 一一对应
)

9. 发送到企业微信/钉钉/飞书机器人

# 需先在 pushplus 渠道配置中新增 webhook
client.send_webhook(
    title="机器人通知",
    content="通过 Webhook 发送到企业微信/钉钉/飞书",
    webhook_code="你的webhook编码",
)

10. 发送邮件

client.send_mail(
    title="邮件标题",
    content="邮件正文内容",
    mail_code="163",  # 可选，不填使用默认邮件
)

11. 发送短信（收费）

client.send_sms(
    title="验证码: 123456",
    content="5分钟内有效",
)

12. 带回调的消息推送

client.send_with_callback(
    title="异步通知",
    content="消息处理完成后会回调通知",
    callback_url="https://your-server.com/callback",
)

13. 防重复发送（时效性消息）

import time

# 设置 5 分钟后过期
expire_ts = int(time.time() * 1000) + 5 * 60 * 1000

client.send_with_timestamp(
    title="限时通知",
    content="这条消息 5 分钟后失效",
    expire_timestamp=expire_ts,
)

14. 发送 JSON 结构化数据

import json

data = {
    "server": "web-01",
    "cpu_usage": "45%",
    "memory_usage": "72%",
    "disk_usage": "30%",
    "status": "正常"
}

client.send_json(
    title="服务器监控",
    content=json.dumps(data, ensure_ascii=False),
)

返回码说明

> 注意: 接口为异步模式，code=200 仅代表服务端收到请求，不代表消息发送成功。最终结果需通过消息流水号查询或回调获取。

消息回调事件

设置 callbackUrl 后，系统会异步 POST 回调结果：

消息完成回调

{
    "event": "message_complate",
    "messageInfo": {
        "message": "",
        "shortCode": "88*********50fe",
        "sendStatus": 2
    }
}

sendStatus: 0-未发送, 1-发送中, 2-发送成功, 3-发送失败

群组新增用户回调

{
    "event": "add_topic_user",
    "topicUserInfo": {
        "id": 25,
        "openId": "oEdHX******aWDg",
        "topicId": 2,
        "nickName": "用户昵称",
        "topicCode": "群组编码",
        "topicName": "群组名称"
    }
}

新增好友回调

{
    "event": "add_friend",
    "friendInfo": {
        "token": "好友令牌",
        "friendId": 5,
        "nickName": "好友昵称",
        "createTime": "2022-09-23 17:02:21"
    }
}

最佳实践

1. Token 安全:
   - 全局 Token 存储在 ~/.pushplus_token.json，请勿分享此文件
   - 临时 Token 优先级高于全局 Token，适合多账号场景

2. 错误处理: 始终检查返回的 code，遇到 900 应立即停止请求避免账号被封

3. 频率控制: 免费版有推送次数限制，避免高频调用。单日请求超过 1000 次会被封 7 天

4. 异步确认: 接口是异步的，code=200 不代表消息已送达，重要消息建议使用 callbackUrl 确认

5. 内容长度: 消息内容建议控制在合理长度，避免过长影响展示效果

6. 模板选择:
   - 简单文本 → txt
   - 富文本/带样式 → html
   - 文档/报告 → markdown
   - 结构化数据 → json

7. 渠道选择: 微信公众号渠道最常用；如需推送到团队协作工具，使用 webhook 渠道配置对应机器人

8. 防重复: 时效性强的消息使用 timestamp 参数，避免过期消息重复推送

快速参考

# 设置全局 Token（仅需一次）
PushplusClient.set_global_token("你的Token")

# 初始化客户端（自动使用全局 Token）
client = PushplusClient()

# 发给自己
client.send_to_self("内容", "标题")

# 临时使用其他 Token
client.send_to_self("内容", "标题", token="其他Token")

# Markdown 日报
client.send_markdown("标题", "# 报告内容")

# 群组推送
client.send_to_group("内容", topic="群组编码")

# 好友推送
client.send_to_friend("内容", to="好友令牌")

# 多渠道推送
client.batch_send("内容", channels=["wechat", "mail"])

# Webhook 机器人
client.send_webhook("内容", webhook_code="编码")

# 邮件
client.send_mail("内容", title="标题")

# 短信（收费）
client.send_sms("标题", "内容")

# 查看全局 Token
token = PushplusClient.get_global_token()

# 清除全局 Token
PushplusClient.clear_global_token()