Netdata.cloud bot for Zulip
anil.recoil.org
1"""Zulip client for sending notifications."""
2
3import structlog
4import zulip
5
6from .models import ZulipConfig
7
8logger = structlog.get_logger()
9
10
11class ZulipNotifier:
12 """Handles sending notifications to Zulip."""
13
14 def __init__(self, config: ZulipConfig):
15 """Initialize Zulip client with configuration."""
16 self.config = config
17 self.client = zulip.Client(
18 site=config.site,
19 email=config.email,
20 api_key=config.api_key,
21 )
22
23 # Test connection
24 try:
25 result = self.client.get_profile()
26 if result["result"] != "success":
27 raise ConnectionError(f"Failed to connect to Zulip: {result}")
28 logger.info("Connected to Zulip", user=result["email"])
29 except Exception as e:
30 logger.error("Failed to initialize Zulip client", error=str(e))
31 raise
32
33 def send_message(self, topic: str, content: str) -> bool:
34 """Send a message to the configured Zulip stream.
35
36 Args:
37 topic: The topic within the stream
38 content: The message content (markdown formatted)
39
40 Returns:
41 True if successful, False otherwise
42 """
43 try:
44 result = self.client.send_message(
45 {
46 "type": "stream",
47 "to": self.config.stream,
48 "topic": topic,
49 "content": content,
50 }
51 )
52
53 if result["result"] == "success":
54 logger.info(
55 "Message sent to Zulip",
56 stream=self.config.stream,
57 topic=topic,
58 message_id=result.get("id"),
59 )
60 return True
61 else:
62 logger.error(
63 "Failed to send message to Zulip",
64 stream=self.config.stream,
65 topic=topic,
66 error=result.get("msg", "Unknown error"),
67 )
68 return False
69
70 except Exception as e:
71 logger.error(
72 "Exception sending message to Zulip",
73 stream=self.config.stream,
74 topic=topic,
75 error=str(e),
76 )
77 return False