Feature/ntfy.sh

example/ntfy_demo.py

@@ -0,0 +1,216 @@
+# -*- coding: utf-8 -*-
+"""
+Ntfy.sh 通知渠道使用示例
+
+本示例展示了如何使用 ntfy.sh 通知渠道发送通知，包括：
+1. 基础配置和使用
+2. 高级功能配置
+3. 同步和异步发送
+4. 与其他渠道的组合使用
+"""
+import asyncio
+
+from use_notify import useNotify
+from use_notify.channels import Ntfy
+
+
+def basic_usage_example():
+    """基础使用示例"""
+    print("=== 基础使用示例 ===")
+
+    # 最简配置，只需要指定主题
+    ntfy = Ntfy({
+        "topic": "my-notifications"
+    })
+
+    # 发送简单消息
+    try:
+        ntfy.send("这是一条测试消息")
+        print("✓ 基础消息发送成功")
+    except Exception as e:
+        print(f"✗ 发送失败: {e}")
+
+    # 发送带标题的消息
+    try:
+        ntfy.send("这是消息内容", "消息标题")
+        print("✓ 带标题消息发送成功")
+    except Exception as e:
+        print(f"✗ 发送失败: {e}")
+
+
+def advanced_features_example():
+    """高级功能使用示例"""
+    print("\n=== 高级功能使用示例 ===")
+
+    # 配置高级功能
+    ntfy = Ntfy({
+        "topic": "my-notifications",
+        "priority": 4,  # 高优先级 (1-5)
+        "tags": ["warning", "computer"],  # 标签
+        "click": "https://example.com",  # 点击跳转链接
+        "attach": "https://example.com/image.jpg"  # 附件
+    })
+
+    try:
+        ntfy.send("这是一条高优先级警告消息", "系统警告")
+        print("✓ 高级功能消息发送成功")
+    except Exception as e:
+        print(f"✗ 发送失败: {e}")
+
+
+def custom_server_example():
+    """自定义服务器示例"""
+    print("\n=== 自定义服务器示例 ===")
+
+    # 使用自托管的 ntfy.sh 服务器
+    ntfy = Ntfy({
+        "topic": "my-notifications",
+        "base_url": "https://ntfy.example.com"  # 自定义服务器
+    })
+
+    try:
+        ntfy.send("来自自定义服务器的消息", "自定义服务器")
+        print("✓ 自定义服务器消息发送成功")
+    except Exception as e:
+        print(f"✗ 发送失败: {e}")
+
+
+async def async_usage_example():
+    """异步使用示例"""
+    print("\n=== 异步使用示例 ===")
+
+    ntfy = Ntfy({
+        "topic": "my-notifications",
+        "priority": 3
+    })
+
+    try:
+        await ntfy.send_async("这是异步发送的消息", "异步消息")
+        print("✓ 异步消息发送成功")
+    except Exception as e:
+        print(f"✗ 异步发送失败: {e}")
+
+
+def publisher_integration_example():
+    """与 Publisher 集成使用示例"""
+    print("\n=== Publisher 集成示例 ===")
+
+    # 创建通知发布器
+    notify = useNotify()
+
+    # 添加 ntfy 渠道
+    notify.add(
+        Ntfy({
+            "topic": "my-notifications",
+            "priority": 3,
+            "tags": ["app", "notification"]
+        })
+    )
+
+    try:
+        notify.publish("通过 Publisher 发送的消息", "Publisher 消息")
+        print("✓ Publisher 集成消息发送成功")
+    except Exception as e:
+        print(f"✗ Publisher 发送失败: {e}")
+
+
+async def async_publisher_example():
+    """异步 Publisher 示例"""
+    print("\n=== 异步 Publisher 示例 ===")
+
+    notify = useNotify()
+    notify.add(
+        Ntfy({
+            "topic": "my-notifications",
+            "priority": 2
+        })
+    )
+
+    try:
+        await notify.publish_async("异步 Publisher 消息", "异步 Publisher")
+        print("✓ 异步 Publisher 消息发送成功")
+    except Exception as e:
+        print(f"✗ 异步 Publisher 发送失败: {e}")
+
+
+def from_settings_example():
+    """从配置创建示例"""
+    print("\n=== 从配置创建示例 ===")
+
+    # 配置字典
+    settings = {
+        "NTFY": {
+            "topic": "my-notifications",
+            "priority": 3,
+            "tags": ["config", "demo"],
+            "click": "https://github.com/ntfy-sh/ntfy"
+        }
+    }
+
+    # 从配置创建通知器
+    notify = useNotify.from_settings(settings)
+
+    try:
+        notify.publish("从配置创建的消息", "配置消息")
+        print("✓ 从配置创建消息发送成功")
+    except Exception as e:
+        print(f"✗ 从配置发送失败: {e}")
+
+
+def actions_example():
+    """交互操作示例"""
+    print("\n=== 交互操作示例 ===")
+
+    # 配置交互操作
+    ntfy = Ntfy({
+        "topic": "my-notifications",
+        "actions": [
+            {
+                "action": "view",
+                "label": "打开网站",
+                "url": "https://ntfy.sh"
+            },
+            {
+                "action": "http",
+                "label": "重启服务",
+                "url": "https://api.example.com/restart",
+                "method": "POST"
+            }
+        ]
+    })
+
+    try:
+        ntfy.send("服务器需要重启，请选择操作", "服务器警告")
+        print("✓ 交互操作消息发送成功")
+    except Exception as e:
+        print(f"✗ 交互操作发送失败: {e}")
+
+
+async def main():
+    """主函数，运行所有示例"""
+    print("Ntfy.sh 通知渠道使用示例")
+    print("=" * 50)
+
+    # 同步示例
+    basic_usage_example()
+    advanced_features_example()
+    custom_server_example()
+    publisher_integration_example()
+    from_settings_example()
+    actions_example()
+
+    # 异步示例
+    await async_usage_example()
+    await async_publisher_example()
+
+    print("\n" + "=" * 50)
+    print("所有示例运行完成！")
+    print("\n注意：")
+    print("1. 请将 'my-notifications' 替换为您的实际主题名称")
+    print("2. 确保您的设备已订阅相应的主题")
+    print("3. 如果使用自定义服务器，请确保服务器地址正确")
+
+
+if __name__ == "__main__":
+    # 运行示例
+    asyncio.run(main())

example/ntfy_from_settings.py

@@ -0,0 +1,50 @@
+# -*- coding: utf-8 -*-
+"""
+使用配置字典创建 Ntfy 通知渠道的简单示例
+"""
+from use_notify import useNotify
+
+# 基础配置
+settings = {
+    "NTFY": {
+        "topic": "my-notifications"
+    }
+}
+
+# 从配置创建通知器
+notify = useNotify.from_settings(settings)
+
+# 发送通知
+try:
+    notify.publish("这是一条测试消息", "测试标题")
+    print("✓ 消息发送成功")
+except Exception as e:
+    print(f"✗ 发送失败: {e}")
+
+
+# 高级配置示例
+advanced_settings = {
+    "NTFY": {
+        "topic": "my-notifications",
+        "base_url": "https://ntfy.sh",  # 可选：自定义服务器
+        "priority": 3,                   # 可选：优先级 (1-5)
+        "tags": ["python", "demo"],     # 可选：标签
+        "click": "https://github.com/ntfy-sh/ntfy",  # 可选：点击链接
+        "actions": [                     # 可选：交互操作
+            {
+                "action": "view",
+                "label": "查看详情",
+                "url": "https://example.com"
+            }
+        ]
+    }
+}
+
+# 使用高级配置
+advanced_notify = useNotify.from_settings(advanced_settings)
+
+try:
+    advanced_notify.publish("这是一条高级配置的消息", "高级消息")
+    print("✓ 高级消息发送成功")
+except Exception as e:
+    print(f"✗ 高级消息发送失败: {e}")

example/ntfy_integration.py

@@ -0,0 +1,52 @@
+# -*- coding: utf-8 -*-
+"""
+Ntfy.sh 与 useNotify 集成使用示例
+"""
+import asyncio
+from use_notify import useNotify
+from use_notify.channels import Ntfy
+
+# 方式1: 直接添加 Ntfy 渠道
+notify = useNotify()
+notify.add(
+    Ntfy({
+        "topic": "my-notifications",
+        "priority": 3,
+        "tags": ["app", "notification"]
+    })
+)
+
+# 同步发送
+try:
+    notify.publish("集成测试消息", "集成测试")
+    print("✓ 集成同步发送成功")
+except Exception as e:
+    print(f"✗ 集成同步发送失败: {e}")
+
+# 异步发送
+async def async_integration():
+    try:
+        await notify.publish_async("异步集成消息", "异步集成")
+        print("✓ 集成异步发送成功")
+    except Exception as e:
+        print(f"✗ 集成异步发送失败: {e}")
+
+asyncio.run(async_integration())
+
+# 方式2: 从配置创建
+settings = {
+    "NTFY": {
+        "topic": "my-notifications",
+        "priority": 4,
+        "tags": ["config", "demo"],
+        "click": "https://ntfy.sh"
+    }
+}
+
+config_notify = useNotify.from_settings(settings)
+
+try:
+    config_notify.publish("配置创建的消息", "配置测试")
+    print("✓ 配置方式发送成功")
+except Exception as e:
+    print(f"✗ 配置方式发送失败: {e}")

example/ntfy_simple.py

@@ -0,0 +1,46 @@
+# -*- coding: utf-8 -*-
+"""
+Ntfy.sh 通知渠道简单使用示例
+"""
+
+import asyncio
+from use_notify.channels import Ntfy
+
+# 基础使用
+ntfy = Ntfy({"topic": "my-notifications"})
+
+# 同步发送
+try:
+    ntfy.send("这是一条测试消息", "测试标题")
+    print("✓ 同步消息发送成功")
+except Exception as e:
+    print(f"✗ 同步发送失败: {e}")
+
+
+# 异步发送
+async def async_example():
+    await ntfy.send_async("这是异步消息", "异步标题")
+    try:
+        print("✓ 异步消息发送成功")
+    except Exception as e:
+        print(f"✗ 异步发送失败: {e}")
+
+
+# 运行异步示例
+asyncio.run(async_example())
+
+# 高级功能示例
+advanced_ntfy = Ntfy(
+    {
+        "topic": "my-notifications",
+        "priority": 4,
+        "tags": ["urgent", "demo"],
+        "click": "https://example.com",
+    }
+)
+
+try:
+    advanced_ntfy.send("高优先级消息", "重要通知")
+    print("✓ 高级功能消息发送成功")
+except Exception as e:
+    print(f"✗ 高级功能发送失败: {e}")

pyproject.toml

@@ -4,9 +4,7 @@ version = "0.3.44"
 description = "一个简单可扩展的异步消息通知库"
 authors = ["miclon <jcnd@163.com>"]
 readme = "README.md"
-packages = [
-    { include = 'use_notify', from = 'src' }
-]
+packages = [{ include = 'use_notify', from = 'src' }]
 
 [tool.poetry.dependencies]
 python = "^3.8"

src/use_notify/channels/__init__.py

@@ -5,6 +5,7 @@
 from .chanify import Chanify
 from .ding import Ding
 from .email import Email
+from .ntfy import Ntfy
 from .pushdeer import PushDeer
 from .pushover import PushOver
 from .wechat import WeChat