Mercurial > libervia-backend

comparison libervia/cli/cmd_notifications.py @ 4134:8d361adf0ee1

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
cli: add `notification` commands
author Goffi <goffi@goffi.org>
date Wed, 18 Oct 2023 15:33:45 +0200
parents
children 0d7bb4df2343
comparison
equal deleted inserted replaced
4133:33fd658d9d00 4134:8d361adf0ee1
1 #!/usr/bin/env python3
2
3 # Libervia CLI
4 # Copyright (C) 2009-2023 JΓ©rΓ΄me Poisson (goffi@goffi.org)
5
6 # This program is free software: you can redistribute it and/or modify
7 # it under the terms of the GNU Affero General Public License as published by
8 # the Free Software Foundation, either version 3 of the License, or
9 # (at your option) any later version.
10
11 # This program is distributed in the hope that it will be useful,
12 # but WITHOUT ANY WARRANTY; without even the implied warranty of
13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14 # GNU Affero General Public License for more details.
15
16 # You should have received a copy of the GNU Affero General Public License
17 # along with this program. If not, see <http://www.gnu.org/licenses/>.
18
19 from libervia.backend.core.i18n import _
20 from libervia.backend.memory.memory import (
21 NotificationPriority,
22 NotificationStatus,
23 NotificationType,
24 )
25 from libervia.backend.tools.common import data_format, date_utils
26 from libervia.cli.constants import Const as C
27 from rich.live import Live
28 from rich.table import Table
29 from rich.text import Text
30
31 from . import base
32
33 __commands__ = ["Notification"]
34
35
36 class Add(base.CommandBase):
37 """Create and broadcast a notification"""
38
39 def __init__(self, host):
40 super(Add, self).__init__(
41 host, "add", use_verbose=True, help=_("create and broadcast a notification")
42 )
43
44 def add_parser_options(self):
45 self.parser.add_argument(
46 "type",
47 choices=[e.name for e in NotificationType],
48 help=_("notification type (default: %(default)s)"),
49 )
50
51 self.parser.add_argument(
52 "body_plain", help=_("plain text body of the notification")
53 )
54
55 # TODO:
56 # self.parser.add_argument(
57 # "-r", "--body-rich", default="", help=_("rich text body of the notification")
58 # )
59
60 self.parser.add_argument(
61 "-t", "--title", default="", help=_("title of the notification")
62 )
63
64 self.parser.add_argument(
65 "-g",
66 "--is-global",
67 action="store_true",
68 help=_("indicates if the notification is for all profiles"),
69 )
70
71 # TODO:
72 # self.parser.add_argument(
73 # "--requires-action",
74 # action="store_true",
75 # help=_("indicates if the notification requires action"),
76 # )
77
78 self.parser.add_argument(
79 "-P",
80 "--priority",
81 default="MEDIUM",
82 choices=[p.name for p in NotificationPriority],
83 help=_("priority level of the notification (default: %(default)s)"),
84 )
85
86 self.parser.add_argument(
87 "-e",
88 "--expire-at",
89 type=base.date_decoder,
90 default=0,
91 help=_(
92 "expiration timestamp for the notification (optional, can be 0 for none)"
93 ),
94 )
95
96 async def start(self):
97 try:
98 await self.host.bridge.notification_add(
99 self.args.type,
100 self.args.body_plain,
101 "", # TODO: self.args.body_rich or "",
102 self.args.title or "",
103 self.args.is_global,
104 False, # TODO: self.args.requires_action,
105 self.args.priority,
106 self.args.expire_at,
107 "",
108 self.profile,
109 )
110 except Exception as e:
111 self.disp(f"can't add notification: {e}", error=True)
112 self.host.quit(C.EXIT_BRIDGE_ERRBACK)
113 else:
114 self.disp("Notification added.")
115
116 self.host.quit()
117
118
119 class Get(base.CommandBase):
120 """Get available notifications"""
121
122 def __init__(self, host):
123 super(Get, self).__init__(
124 host,
125 "get",
126 use_output=C.OUTPUT_LIST_DICT,
127 extra_outputs={"default": self.default_output},
128 help=_("display notifications"),
129 )
130
131 def add_parser_options(self):
132 self.parser.add_argument(
133 "-f",
134 "--follow",
135 action="store_true",
136 help=_("wait and print incoming notifications"),
137 )
138
139 self.parser.add_argument(
140 "-t",
141 "--type",
142 type=str,
143 choices=[t.name for t in NotificationType],
144 help=_("filter by type of the notification"),
145 )
146
147 self.parser.add_argument(
148 "-s",
149 "--status",
150 type=str,
151 choices=[s.name for s in NotificationStatus],
152 help=_("filter by status of the notification"),
153 )
154
155 self.parser.add_argument(
156 "-a",
157 "--requires-action",
158 type=C.bool,
159 default=None,
160 help=_(
161 "filter notifications that require (or not) user action, true by "
162 "default, don't filter if omitted"
163 ),
164 )
165
166 self.parser.add_argument(
167 "-P",
168 "--min-priority",
169 type=str,
170 choices=[p.name for p in NotificationPriority],
171 help=_("filter notifications with at least the specified priority"),
172 )
173
174 def create_table(self):
175 table = Table(box=None, show_header=False, collapse_padding=True)
176 table.add_column("is_new")
177 table.add_column("type")
178 table.add_column("id")
179 table.add_column("priority")
180 table.add_column("timestamp")
181 table.add_column("body")
182 return table
183
184 def default_output(self, notifs):
185 if self.args.follow:
186 if self.live is None:
187 self.table = table = self.create_table()
188 self.live = Live(table, auto_refresh=False, console=self.console)
189 self.host.add_on_quit_callback(self.live.stop)
190 self.live.start()
191 else:
192 table = self.table
193 else:
194 table = self.create_table()
195
196 for notif in notifs:
197 emoji_mapper = {
198 "chat": "πŸ’¬",
199 "blog": "πŸ“",
200 "calendar": "πŸ“…",
201 "file": "πŸ“‚",
202 "call": "πŸ“ž",
203 "service": "πŸ“’",
204 "other": "🟣",
205 }
206 emoji = emoji_mapper[notif.get("type", "other")]
207 notif_id = Text(notif["id"])
208 created = date_utils.date_fmt(notif["timestamp"], tz_info=date_utils.TZ_LOCAL)
209
210 priority_name = NotificationPriority(notif["priority"]).name.lower()
211
212 priority = Text(f"[{priority_name}]", style=f"priority_{priority_name}")
213
214 body_parts = []
215 title = notif.get("title")
216 if title:
217 body_parts.append((f"{title}\n", "notif_title"))
218 body_parts.append(notif["body_plain"])
219 body = Text.assemble(*body_parts)
220
221 new_flag = "🌟 " if notif.get("new") else ""
222 table.add_row(new_flag, emoji, notif_id, created, priority, body)
223
224 if self.args.follow:
225 self.live.refresh()
226 else:
227 self.print(table)
228
229 async def on_notification_new(
230 self,
231 id_: str,
232 timestamp: float,
233 type_: str,
234 body_plain: str,
235 body_rich: str,
236 title: str,
237 requires_action: bool,
238 priority: int,
239 expire_at: float,
240 extra: str,
241 profile: str,
242 ) -> None:
243 """Callback when a new notification is emitted."""
244 notification_data = {
245 "id": id_,
246 "timestamp": timestamp,
247 "type": type_,
248 "body_plain": body_plain,
249 "body_rich": body_rich,
250 "title": title,
251 "requires_action": requires_action,
252 "priority": priority,
253 "expire_at": expire_at,
254 "extra": data_format.deserialise(extra),
255 "profile": profile,
256 "new": True,
257 }
258
259 await self.output([notification_data])
260
261 async def start(self):
262 keys = ["type", "status", "requires_action", "min_priority"]
263 filters = {
264 key: getattr(self.args, key) for key in keys if getattr(self.args, key)
265 }
266 try:
267 notifications = data_format.deserialise(
268 await self.host.bridge.notifications_get(
269 data_format.serialise(filters),
270 self.profile,
271 ),
272 type_check=list,
273 )
274 except Exception as e:
275 self.disp(f"can't get notifications: {e}", error=True)
276 self.host.quit(C.EXIT_BRIDGE_ERRBACK)
277 else:
278 self.live = None
279 await self.output(notifications)
280 if self.args.follow:
281 self.host.bridge.register_signal(
282 "notification_new", self.on_notification_new, "core"
283 )
284 else:
285 self.host.quit()
286
287
288 class Delete(base.CommandBase):
289 """Delete a notification"""
290
291 def __init__(self, host):
292 super(Delete, self).__init__(
293 host, "delete", use_verbose=True, help=_("delete a notification")
294 )
295
296 def add_parser_options(self):
297 self.parser.add_argument(
298 "id",
299 help=_("ID of the notification to delete"),
300 )
301
302 self.parser.add_argument(
303 "-g", "--is-global",
304 action="store_true",
305 help=_("true if the notification is a global one"),
306 )
307
308 self.parser.add_argument(
309 "--profile-key",
310 default="@ALL@",
311 help=_("Profile key (use '@ALL@' for all profiles, default: %(default)s)"),
312 )
313
314 async def start(self):
315 try:
316 await self.host.bridge.notification_delete(
317 self.args.id, self.args.is_global, self.profile
318 )
319 except Exception as e:
320 self.disp(f"can't delete notification: {e}", error=True)
321 self.host.quit(C.EXIT_BRIDGE_ERRBACK)
322 else:
323 self.disp("Notification deleted.")
324 self.host.quit()
325
326
327 class Expire(base.CommandBase):
328 """Clean expired notifications"""
329
330 def __init__(self, host):
331 super(Expire, self).__init__(
332 host, "expire", use_verbose=True, help=_("clean expired notifications")
333 )
334
335 def add_parser_options(self):
336 self.parser.add_argument(
337 "-l",
338 "--limit",
339 type=base.date_decoder,
340 metavar="TIME_PATTERN",
341 help=_("time limit for older notifications. default: no limit used)"),
342 )
343 self.parser.add_argument(
344 "-a",
345 "--all",
346 action="store_true",
347 help=_(
348 "expire notifications for all profiles (default: use current profile)"
349 ),
350 )
351
352 async def start(self):
353 try:
354 await self.host.bridge.notifications_expired_clean(
355 -1.0 if self.args.limit is None else self.args.limit,
356 C.PROF_KEY_NONE if self.args.all else self.profile,
357 )
358 except Exception as e:
359 self.disp(f"can't clean expired notifications: {e}", error=True)
360 self.host.quit(C.EXIT_BRIDGE_ERRBACK)
361 else:
362 self.disp("Expired notifications cleaned.")
363 self.host.quit()
364
365
366 class Notification(base.CommandBase):
367 subcommands = (Add, Get, Delete, Expire)
368
369 def __init__(self, host):
370 super(Notification, self).__init__(
371 host, "notification", use_profile=False, help=_("Notifications handling")
372 )