通知渠道配置
use-notify 支持多种通知渠道,每个渠道都有其特定的配置参数。本指南将详细介绍各个渠道的配置方法。
支持的通知渠道
Bark(iOS 推送)
Bark 是一个 iOS 推送通知应用。
python
from use_notify import useNotify, useNotifyChannel
notify = useNotify()
notify.add(useNotifyChannel.Bark({
"token": "your_bark_token", # 必需
"base_url": "https://api.day.app", # 可选,默认官方服务器
"sound": "default", # 可选,通知声音
"group": "myapp", # 可选,通知分组
"url": "https://example.com", # 可选,自定义 URL
"icon": "https://example.com/icon.png" # 可选,自定义图标
}))
配置参数:
token
(必需): Bark 应用中的设备令牌base_url
(可选): Bark 服务器地址,默认为官方服务器sound
(可选): 通知声音,如 "default", "bell", "silence" 等group
(可选): 通知分组名称url
(可选): 自定义 URLicon
(可选): 自定义通知图标 URL
钉钉(DingTalk)
钉钉群机器人通知。
python
notify.add(useNotifyChannel.Ding({
"token": "your_dingtalk_token", # 必需
"at_all": False, # 可选,是否@所有人
"at_mobiles": ["13800138000"], # 可选,@指定手机号
"at_user_ids": ["user123"] # 可选,@指定用户ID
}))
配置参数:
token
(必需): 钉钉机器人的 Webhook Tokenat_all
(可选): 是否@所有人,默认 Falseat_mobiles
(可选): @指定手机号列表at_user_ids
(可选): @指定用户ID列表
企业微信(WeChat Work)
企业微信群机器人通知。
python
notify.add(useNotifyChannel.WeChat({
"token": "your_key", # 必需
"mentioned_list": ["@all"], # 可选,@成员列表
"mentioned_mobile_list": ["13800138000"] # 可选,@手机号列表
}))
配置参数:
token
(必需): 企业微信机器人的 Webhook Tokenmentioned_list
(可选): @成员列表,使用 "@all" 可@所有人mentioned_mobile_list
(可选): @手机号列表
邮件(Email)
SMTP 邮件通知。
python
notify.add(useNotifyChannel.Email({
"server": "smtp.gmail.com", # 必需
"port": 587, # 必需
"username": "[email protected]", # 必需
"password": "your_password", # 必需
"to_emails": ["[email protected]"], # 必需
"from_email": "[email protected]", # 可选,默认使用username
}))
配置参数:
server
(必需): SMTP 服务器地址port
(必需): SMTP 服务器端口username
(必需): 邮箱用户名password
(必需): 邮箱密码或应用密码to_emails
(必需): 收件人邮箱列表from_email
(可选): 发件人邮箱,默认使用 username
Pushover
Pushover 推送服务。
python
notify.add(useNotifyChannel.Pushover({
"token": "your_app_token", # 必需
"user": "your_user_key", # 必需
"device": "your_device", # 可选
"priority": 0, # 可选,优先级 (-2 到 2)
"sound": "default" # 可选,通知声音
}))
配置参数:
token
(必需): Pushover 应用令牌user
(必需): Pushover 用户密钥device
(可选): 特定设备名称priority
(可选): 消息优先级,范围 -2 到 2sound
(可选): 通知声音
Pushdeer
Pushdeer 推送服务。
python
notify.add(useNotifyChannel.Pushdeer({
"token": "your_pushdeer_token", # 必需
"base_url": "https://api2.pushdeer.com" # 可选,默认官方服务器
}))
配置参数:
token
(必需): Pushdeer 设备令牌base_url
(可选): Pushdeer 服务器地址
Chanify
Chanify 推送服务。
python
notify.add(useNotifyChannel.Chanify({
"token": "your_chanify_token", # 必需
"base_url": "https://api.chanify.net" # 可选,默认官方服务器
}))
配置参数:
token
(必需): Chanify 设备令牌base_url
(可选): Chanify 服务器地址
飞书(Feishu)
飞书自定义机器人通知。
python
notify.add(useNotifyChannel.Feishu({
"token": "your_feishu_token", # 必需
"at_all": True, # 可选,是否@所有人
"at_user_ids": ["ou_f9fa825035541c7e205e0377b3111111"] # 可选,@指定用户ID
}))
配置参数:
token
(必需): 飞书自定义机器人的 Webhook Tokenat_all
(可选): 是否@所有人,默认 Falseat_user_ids
(可选): @指定用户ID列表
Ntfy
Ntfy 是一个简单、轻量级的推送通知服务。
python
from use_notify import useNotify, useNotifyChannel
notify = useNotify()
notify.add(useNotifyChannel.Ntfy({
"topic": "my-notifications", # 必需
"base_url": "https://ntfy.sh", # 可选,默认官方服务器
"priority": 3, # 可选,优先级 (1-5)
"tags": ["warning", "skull"], # 可选,通知标签
"click": "https://example.com", # 可选,点击跳转链接
"attach": "https://example.com/image.jpg" # 可选,附件链接
}))
配置参数:
topic
(必需): Ntfy 通知主题,用于标识通知目标base_url
(可选): Ntfy 服务器地址,默认为官方服务器 https://ntfy.shpriority
(可选): 通知优先级,范围 1-5(1 最低,5 最高)tags
(可选): 通知标签列表,用于显示图标click
(可选): 点击通知后跳转的 URLattach
(可选): 附件链接,可以是图片或其他文件actions
(可选): 交互操作按钮列表,例如:python"actions": [ { "action": "view", "label": "打开网站", "url": "https://ntfy.sh" }, { "action": "http", "label": "重启服务", "url": "https://api.example.com/restart", "method": "POST" } ]
多渠道配置
添加多个渠道
python
from use_notify import useNotify, useNotifyChannel
notify = useNotify()
# 一次添加多个渠道
notify.add(
useNotifyChannel.Bark({"token": "bark_token"}),
useNotifyChannel.Ding({"token": "ding_token"}),
useNotifyChannel.Email({
"server": "smtp.gmail.com",
"port": 587,
"username": "[email protected]",
"password": "password",
"to_emails": ["[email protected]"]
})
)
# 或者分别添加
notify.add(useNotifyChannel.Bark({"token": "bark_token"}))
notify.add(useNotifyChannel.Ding({"token": "ding_token"}))
不同场景使用不同渠道
python
# 创建不同用途的通知实例
# 日常通知 - 使用轻量级渠道
daily_notify = useNotify()
daily_notify.add(useNotifyChannel.Bark({"token": "bark_token"}))
# 重要警报 - 使用多个渠道确保送达
alert_notify = useNotify()
alert_notify.add(
useNotifyChannel.Ding({"token": "ding_token", "at_all": True}),
useNotifyChannel.Email({
"server": "smtp.company.com",
"port": 587,
"username": "[email protected]",
"password": "password",
"to_emails": ["[email protected]", "[email protected]"]
}),
useNotifyChannel.Pushover({
"token": "pushover_token",
"user": "admin_user",
"priority": 2 # 高优先级
})
)
# 在装饰器中使用不同实例
@notify(notify_instance=daily_notify)
def routine_task():
return "日常任务完成"
@notify(notify_instance=alert_notify, title="系统警报")
def critical_check():
return "关键检查完成"
自定义通知渠道
如果内置渠道不满足需求,可以创建自定义通知渠道:
python
from use_notify.channels import BaseChannel
import requests
class CustomChannel(BaseChannel):
"""自定义通知渠道"""
def __init__(self, config):
super().__init__()
self.api_url = config.get("api_url")
self.api_key = config.get("api_key")
def send(self, title, content, **kwargs):
"""同步发送通知"""
try:
response = requests.post(
self.api_url,
headers={"Authorization": f"Bearer {self.api_key}"},
json={"title": title, "message": content}
)
response.raise_for_status()
return True
except Exception as e:
print(f"发送通知失败: {e}")
return False
async def send_async(self, title, content, **kwargs):
"""异步发送通知"""
import aiohttp
try:
async with aiohttp.ClientSession() as session:
async with session.post(
self.api_url,
headers={"Authorization": f"Bearer {self.api_key}"},
json={"title": title, "message": content}
) as response:
response.raise_for_status()
return True
except Exception as e:
print(f"发送通知失败: {e}")
return False
# 使用自定义渠道
notify = useNotify()
notify.add(CustomChannel({
"api_url": "https://api.example.com/notify",
"api_key": "your_api_key"
}))
配置最佳实践
1. 环境变量管理
python
import os
from use_notify import useNotify, useNotifyChannel
notify = useNotify()
# 使用环境变量管理敏感信息
notify.add(useNotifyChannel.Bark({
"token": os.getenv("BARK_TOKEN")
}))
notify.add(useNotifyChannel.Ding({
"token": os.getenv("DING_TOKEN"),
"secret": os.getenv("DING_SECRET")
}))
notify.add(useNotifyChannel.Email({
"server": os.getenv("server", "smtp.gmail.com"),
"port": int(os.getenv("port", "587")),
"username": os.getenv("EMAIL_USERNAME"),
"password": os.getenv("EMAIL_PASSWORD"),
"to_emails": os.getenv("EMAIL_RECIPIENTS", "").split(",")
}))
2. 配置文件管理
python
import json
from use_notify import useNotify, useNotifyChannel
# 从配置文件加载
with open("notify_config.json", "r") as f:
config = json.load(f)
notify = useNotify()
# 根据配置动态添加渠道
for channel_config in config["channels"]:
channel_type = channel_config["type"]
channel_params = channel_config["params"]
if channel_type == "bark":
notify.add(useNotifyChannel.Bark(channel_params))
elif channel_type == "ding":
notify.add(useNotifyChannel.Ding(channel_params))
elif channel_type == "email":
notify.add(useNotifyChannel.Email(channel_params))
配置文件示例 (notify_config.json
):
json
{
"channels": [
{
"type": "bark",
"params": {
"token": "your_bark_token",
"sound": "default"
}
},
{
"type": "ding",
"params": {
"token": "your_ding_token",
"at_all": false
}
},
{
"type": "email",
"params": {
"server": "smtp.gmail.com",
"port": 587,
"username": "[email protected]",
"password": "your_password",
"to_emails": ["[email protected]"]
}
}
]
}
3. 错误处理和重试
python
from use_notify import useNotify, useNotifyChannel
import time
class ReliableNotify:
def __init__(self):
self.notify = useNotify()
self.setup_channels()
def setup_channels(self):
# 添加多个备用渠道
self.notify.add(
useNotifyChannel.Bark({"token": os.getenv("BARK_TOKEN")}),
useNotifyChannel.Ding({"token": os.getenv("DING_TOKEN")}),
useNotifyChannel.Email({
"server": "smtp.gmail.com",
"port": 587,
"username": os.getenv("EMAIL_USERNAME"),
"password": os.getenv("EMAIL_PASSWORD"),
"to_emails": [os.getenv("EMAIL_RECIPIENT")]
})
)
def send_with_retry(self, title, content, max_retries=3):
for attempt in range(max_retries):
try:
self.notify.publish(title=title, content=content)
return True
except Exception as e:
print(f"发送失败 (尝试 {attempt + 1}/{max_retries}): {e}")
if attempt < max_retries - 1:
time.sleep(2 ** attempt) # 指数退避
return False
# 使用可靠的通知服务
reliable_notify = ReliableNotify()
reliable_notify.send_with_retry("重要通知", "这是一条重要消息")
4. 测试配置
python
def test_notification_channels():
"""测试所有配置的通知渠道"""
notify = useNotify()
# 添加所有渠道
channels = [
("Bark", useNotifyChannel.Bark({"token": os.getenv("BARK_TOKEN")})),
("钉钉", useNotifyChannel.Ding({"token": os.getenv("DING_TOKEN")})),
("邮件", useNotifyChannel.Email({
"server": "smtp.gmail.com",
"port": 587,
"username": os.getenv("EMAIL_USERNAME"),
"password": os.getenv("EMAIL_PASSWORD"),
"to_emails": [os.getenv("EMAIL_RECIPIENT")]
}))
]
for name, channel in channels:
test_notify = useNotify()
test_notify.add(channel)
try:
test_notify.publish(
title=f"{name} 测试通知",
content=f"这是来自 {name} 渠道的测试消息"
)
print(f"✅ {name} 渠道测试成功")
except Exception as e:
print(f"❌ {name} 渠道测试失败: {e}")
if __name__ == "__main__":
test_notification_channels()