Feat pushdeer

example/demo.py

@@ -6,7 +6,7 @@
 notify = useNotify()
 notify.add(
     # 添加多个通知渠道
-    useNotifyChannel.Bark({"token": "your token"}),
+    useNotifyChannel.PushDeer({"token": "Your Token"}),
 )
 
 asyncio.run(notify.publish_async(title="消息标题", content="消息正文"))

pyproject.toml

@@ -22,6 +22,7 @@ flake8 = "*"
 isort = "*"
 pre-commit = "*"
 pre-commit-hooks = "*"
+pytest-asyncio = "0.18.3"
 
 [build-system]
 requires = ["poetry-core>=1.0.0"]

src/use_notify/channels/pushdeer.py

@@ -1,5 +1,6 @@
 # -*- coding: utf-8 -*-
 import logging
+from urllib.parse import quote
 
 import httpx
 
@@ -9,27 +10,101 @@
 
 
 class PushDeer(BaseChannel):
-    """pushdeer app 消息通知"""
+    """pushdeer app 消息通知
+
+    支持三种消息类型:
+    - text: 纯文本消息 (默认)
+    - image: 图片消息
+    - markdown: Markdown格式消息
+
+    配置参数:
+    - token: PushDeer的pushkey
+    - base_url: 可选，自建PushDeer服务的URL，默认为"https://api2.pushdeer.com"
+    - type: 可选，消息类型，可选值为text、markdown、image，默认为text
+    """
 
     @property
     def api_url(self):
-        return f"https://api2.pushdeer.com/message/push?pushkey={self.config.token}&text={{text}}"
+        """获取PushDeer API基础URL"""
+        # 确保token存在
+        if not self.config.token:
+            raise ValueError("PushDeer token (pushkey) is required")
+
+        # Check if base_url exists in config, otherwise use default
+        if self.config.base_url:
+            base_url = self.config.base_url.rstrip("/")
+        else:
+            base_url = "https://api2.pushdeer.com"
+
+        return f"{base_url}/message/push"
+
+    def _prepare_params(self, content, title=None):
+        """准备请求参数
+
+        Args:
+            content: 消息内容
+            title: 消息标题
+
+        Returns:
+            dict: 请求参数
+        """
+        # 默认标题
+        if not title:
+            title = "消息提醒"
+
+        # 确定消息类型
+        msg_type = getattr(self.config, "type", "markdown")
+        if msg_type not in ["text", "markdown", "image"]:
+            if msg_type:  # 只有当type不为空且无效时才记录警告
+                logger.warning(f"Invalid message type: {msg_type}, fallback to text")
+            msg_type = "markdown"
+
+        params = {
+            "pushkey": self.config.token,
+            "text": title,
+            "type": msg_type
+        }
+
+        # 根据消息类型处理内容
+        if msg_type == "text":
+            # 对于text类型，content直接作为text参数
+            params["text"] = content
+        elif msg_type == "markdown":
+            # 对于markdown类型，content作为desp参数
+            params["desp"] = content
+        elif msg_type == "image":
+            # 对于image类型，content应该是图片URL
+            params["desp"] = content
+
+        return params
 
     @property
     def headers(self):
+        """请求头"""
         return {"Content-Type": "application/x-www-form-urlencoded"}
 
-    def build_api_body(self, content, title=None):
-        return self.api_url.format_map({"text": f"{title}\n{content}"})
-
     def send(self, content, title=None):
-        api_url = self.build_api_body(content, title)
+        """发送PushDeer消息
+
+        Args:
+            content: 消息内容
+            title: 消息标题
+        """
+        params = self._prepare_params(content, title)
+
         with httpx.Client() as client:
-            client.get(api_url, headers=self.headers)
-        logger.debug("`pushdeer` send successfully")
+            client.get(self.api_url, params=params, headers=self.headers)
+        logger.debug(f"`pushdeer` send message successfully")
 
     async def send_async(self, content, title=None):
-        api_url = self.build_api_body(content, title)
+        """异步发送PushDeer消息
+
+        Args:
+            content: 消息内容
+            title: 消息标题
+        """
+        params = self._prepare_params(content, title)
+
         async with httpx.AsyncClient() as client:
-            await client.get(api_url, headers=self.headers)
-        logger.debug("`pushdeer` send successfully")
+            await client.get(self.api_url, params=params, headers=self.headers)
+        logger.debug(f"`pushdeer` send message successfully")

tests/test_pushdeer.py

@@ -0,0 +1,134 @@
+# -*- coding: utf-8 -*-
+import pytest
+import httpx
+import json
+from unittest.mock import patch, MagicMock, AsyncMock
+
+from use_notify.channels.pushdeer import PushDeer
+
+
+class TestPushDeer:
+    """PushDeer通知渠道测试"""
+
+    @pytest.fixture
+    def pushdeer_config(self):
+        """基本配置"""
+        return {
+            "token": "test_token"
+        }
+
+    @pytest.fixture
+    def pushdeer_instance(self, pushdeer_config):
+        """PushDeer实例"""
+        return PushDeer(pushdeer_config)
+
+    def test_api_url(self, pushdeer_instance):
+        """测试API URL构建"""
+        assert pushdeer_instance.api_url == "https://api2.pushdeer.com/message/push"
+
+    def test_api_url_custom_base(self):
+        """测试自定义base_url"""
+        pushdeer = PushDeer({
+            "token": "test_token",
+            "base_url": "https://custom.pushdeer.com/"
+        })
+        assert pushdeer.api_url == "https://custom.pushdeer.com/message/push"
+
+    def test_api_url_no_token(self):
+        """测试没有token的情况"""
+        pushdeer = PushDeer({})
+        with pytest.raises(ValueError, match="PushDeer token .* required"):
+            _ = pushdeer.api_url
+
+    def test_prepare_params_text(self, pushdeer_instance):
+        """测试文本消息参数准备"""
+        params = pushdeer_instance._prepare_params("Hello World")
+        assert params["pushkey"] == "test_token"
+        assert params["desp"] == "Hello World"
+        assert params["type"] == "markdown"
+
+    def test_prepare_params_markdown(self):
+        """测试Markdown消息参数准备"""
+        pushdeer = PushDeer({
+            "token": "test_token",
+            "type": "markdown"
+        })
+        params = pushdeer._prepare_params("# Hello\n\nWorld", "Test Title")
+        assert params["pushkey"] == "test_token"
+        assert params["text"] == "Test Title"
+        assert params["desp"] == "# Hello\n\nWorld"
+        assert params["type"] == "markdown"
+
+    def test_prepare_params_image(self):
+        """测试图片消息参数准备"""
+        pushdeer = PushDeer({
+            "token": "test_token",
+            "type": "image"
+        })
+        params = pushdeer._prepare_params("https://example.com/image.jpg")
+        assert params["pushkey"] == "test_token"
+        assert params["desp"] == "https://example.com/image.jpg"
+        assert params["type"] == "image"
+
+    def test_prepare_params_invalid_type(self):
+        """测试无效消息类型"""
+        pushdeer = PushDeer({
+            "token": "test_token",
+            "type": "invalid"
+        })
+        params = pushdeer._prepare_params("Hello World")
+        assert params["type"] == "markdown"  # 应该回退到text类型
+
+    @patch("httpx.Client")
+    def test_send_success(self, mock_client, pushdeer_instance):
+        """测试发送成功"""
+        # 模拟响应
+        mock_response = MagicMock()
+        mock_response.raise_for_status.return_value = None
+        mock_response.json.return_value = {"code": 0, "message": "success"}
+
+        # 模拟客户端
+        mock_client_instance = MagicMock()
+        mock_client_instance.get.return_value = mock_response
+        mock_client.return_value.__enter__.return_value = mock_client_instance
+
+        # 执行发送
+        pushdeer_instance.send("Hello World", "Test Title")
+
+        # 验证调用
+        mock_client_instance.get.assert_called_once()
+        args, kwargs = mock_client_instance.get.call_args
+        assert args[0] == pushdeer_instance.api_url
+        assert "params" in kwargs
+        # 对于text类型，content会覆盖title作为text参数
+        assert kwargs["params"]["desp"] == "Hello World"
+
+    @patch('httpx.AsyncClient')
+    @pytest.mark.asyncio
+    async def test_send_async_success(self, mock_async_client, pushdeer_instance):
+        """测试异步发送成功"""
+        # 模拟响应
+        mock_response = AsyncMock()
+        mock_response.raise_for_status = AsyncMock()
+        # 设置json方法为AsyncMock，返回值为字典
+        mock_json = AsyncMock()
+        mock_json.return_value = {"code": 0, "message": "success"}
+        mock_response.json = mock_json
+
+        # 模拟客户端
+        mock_client = AsyncMock()
+        mock_client.get.return_value = mock_response
+
+        # 替换httpx.AsyncClient
+        mock_async_client.return_value.__aenter__.return_value = mock_client
+
+        # 执行发送
+        await pushdeer_instance.send_async("Hello World", "Test Title")
+
+        # 验证调用
+        mock_client.get.assert_called_once()
+        args, kwargs = mock_client.get.call_args
+        assert args[0] == pushdeer_instance.api_url
+        assert "params" in kwargs
+        # 对于text类型，content会覆盖title作为text参数
+        assert kwargs["params"]["desp"] == "Hello World"