Mercurial > libervia-backend
annotate libervia/cli/cmd_notifications.py @ 4270:0d7bb4df2343
Reformatted code base using black.
| author | Goffi <goffi@goffi.org> |
|---|---|
| date | Wed, 19 Jun 2024 18:44:57 +0200 |
| parents | 8d361adf0ee1 |
| children |
| rev | line source |
|---|---|
| 4134 | 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, | |
|
4270
0d7bb4df2343
Reformatted code base using black.
Goffi <goffi@goffi.org>
parents:
4134
diff
changeset
|
101 "", # TODO: self.args.body_rich or "", |
| 4134 | 102 self.args.title or "", |
| 103 self.args.is_global, | |
|
4270
0d7bb4df2343
Reformatted code base using black.
Goffi <goffi@goffi.org>
parents:
4134
diff
changeset
|
104 False, # TODO: self.args.requires_action, |
| 4134 | 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( | |
|
4270
0d7bb4df2343
Reformatted code base using black.
Goffi <goffi@goffi.org>
parents:
4134
diff
changeset
|
303 "-g", |
|
0d7bb4df2343
Reformatted code base using black.
Goffi <goffi@goffi.org>
parents:
4134
diff
changeset
|
304 "--is-global", |
| 4134 | 305 action="store_true", |
| 306 help=_("true if the notification is a global one"), | |
| 307 ) | |
| 308 | |
| 309 self.parser.add_argument( | |
| 310 "--profile-key", | |
| 311 default="@ALL@", | |
| 312 help=_("Profile key (use '@ALL@' for all profiles, default: %(default)s)"), | |
| 313 ) | |
| 314 | |
| 315 async def start(self): | |
| 316 try: | |
| 317 await self.host.bridge.notification_delete( | |
| 318 self.args.id, self.args.is_global, self.profile | |
| 319 ) | |
| 320 except Exception as e: | |
| 321 self.disp(f"can't delete notification: {e}", error=True) | |
| 322 self.host.quit(C.EXIT_BRIDGE_ERRBACK) | |
| 323 else: | |
| 324 self.disp("Notification deleted.") | |
| 325 self.host.quit() | |
| 326 | |
| 327 | |
| 328 class Expire(base.CommandBase): | |
| 329 """Clean expired notifications""" | |
| 330 | |
| 331 def __init__(self, host): | |
| 332 super(Expire, self).__init__( | |
| 333 host, "expire", use_verbose=True, help=_("clean expired notifications") | |
| 334 ) | |
| 335 | |
| 336 def add_parser_options(self): | |
| 337 self.parser.add_argument( | |
| 338 "-l", | |
| 339 "--limit", | |
| 340 type=base.date_decoder, | |
| 341 metavar="TIME_PATTERN", | |
| 342 help=_("time limit for older notifications. default: no limit used)"), | |
| 343 ) | |
| 344 self.parser.add_argument( | |
| 345 "-a", | |
| 346 "--all", | |
| 347 action="store_true", | |
| 348 help=_( | |
| 349 "expire notifications for all profiles (default: use current profile)" | |
| 350 ), | |
| 351 ) | |
| 352 | |
| 353 async def start(self): | |
| 354 try: | |
| 355 await self.host.bridge.notifications_expired_clean( | |
| 356 -1.0 if self.args.limit is None else self.args.limit, | |
| 357 C.PROF_KEY_NONE if self.args.all else self.profile, | |
| 358 ) | |
| 359 except Exception as e: | |
| 360 self.disp(f"can't clean expired notifications: {e}", error=True) | |
| 361 self.host.quit(C.EXIT_BRIDGE_ERRBACK) | |
| 362 else: | |
| 363 self.disp("Expired notifications cleaned.") | |
| 364 self.host.quit() | |
| 365 | |
| 366 | |
| 367 class Notification(base.CommandBase): | |
| 368 subcommands = (Add, Get, Delete, Expire) | |
| 369 | |
| 370 def __init__(self, host): | |
| 371 super(Notification, self).__init__( | |
| 372 host, "notification", use_profile=False, help=_("Notifications handling") | |
| 373 ) |