feat: 添加 NotificationService 通知服务 + 测试

2026-05-09 14:11:49 +08:00
parent 52bb1f53cc
commit 3616e5799d
2 changed files with 86 additions and 0 deletions
@@ -0,0 +1,46 @@
+from datetime import datetime
+from typing import Any, Dict, List
+from app.config import settings
+from app.wechat.client import WeChatClient
+
+
+class NotificationService:
+    def __init__(self):
+        self.client = WeChatClient()
+
+    async def send(self, announcements: List[Dict[str, Any]]) -> int:
+        if not settings.wechat_enabled:
+            return 0
+        if not announcements:
+            return 0
+
+        sent = 0
+        for ann in announcements:
+            try:
+                title = ann.get("title", "")
+                if len(title) > 128:
+                    title = title[:125] + "..."
+
+                purchase_name = ann.get("purchase_name", "")
+                if len(purchase_name) > 25:
+                    purchase_name = purchase_name[:22] + "..."
+
+                pub_date = ann.get("publish_date")
+                time_str = pub_date.strftime("%Y-%m-%d %H:%M") if pub_date else "时间未知"
+
+                source_name = ann.get("source_name", "")
+
+                description = (
+                    f'<div style="font-size: 14px; margin-top: 8px;">'
+                    f'{source_name} | {purchase_name} | {time_str}'
+                    f'</div>'
+                )
+
+                url = ann.get("content_url", "")
+
+                if await self.client.send_textcard(title, description, url):
+                    sent += 1
+            except Exception:
+                continue
+
+        return sent