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 "type": "stream",
46 "to": self.config.stream,
47 "topic": topic,
48 "content": content,
49 })
50
51 if result['result'] == 'success':
52 logger.info(
53 "Message sent to Zulip",
54 stream=self.config.stream,
55 topic=topic,
56 message_id=result.get('id')
57 )
58 return True
59 else:
60 logger.error(
61 "Failed to send message to Zulip",
62 stream=self.config.stream,
63 topic=topic,
64 error=result.get('msg', 'Unknown error')
65 )
66 return False
67
68 except Exception as e:
69 logger.error(
70 "Exception sending message to Zulip",
71 stream=self.config.stream,
72 topic=topic,
73 error=str(e)
74 )
75 return False