| 1 | from dataclasses import dataclass |
| 2 | import uuid |
| 3 | import threading |
| 4 | from datetime import datetime, timedelta |
| 5 | from enum import Enum |
| 6 | |
| 7 | from helpers.localization import Localization |
| 8 | |
| 9 | |
| 10 | class NotificationType(Enum): |
| 11 | INFO = "info" |
| 12 | SUCCESS = "success" |
| 13 | WARNING = "warning" |
| 14 | ERROR = "error" |
| 15 | PROGRESS = "progress" |
| 16 | |
| 17 | |
| 18 | class NotificationPriority(Enum): |
| 19 | NORMAL = 10 |
| 20 | HIGH = 20 |
| 21 | |
| 22 | |
| 23 | @dataclass |
| 24 | class NotificationItem: |
| 25 | manager: "NotificationManager" |
| 26 | no: int |
| 27 | type: NotificationType |
| 28 | priority: NotificationPriority |
| 29 | title: str |
| 30 | message: str |
| 31 | detail: str # HTML content for expandable details |
| 32 | timestamp: datetime |
| 33 | display_time: int = 3 # Display duration in seconds, default 3 seconds |
| 34 | read: bool = False |
| 35 | id: str = "" |
| 36 | group: str = "" # Group identifier for grouping related notifications |
| 37 | |
| 38 | def __post_init__(self): |
| 39 | if not self.id: |
| 40 | self.id = str(uuid.uuid4()) |
| 41 | # Ensure type is always NotificationType |
| 42 | if isinstance(self.type, str): |
| 43 | self.type = NotificationType(self.type) |
| 44 | |
| 45 | def mark_read(self): |
| 46 | self.read = True |
| 47 | self.manager.update_item(self.no, read=True) |
| 48 | |
| 49 | def output(self): |
| 50 | return { |
| 51 | "no": self.no, |
| 52 | "id": self.id, |
| 53 | "type": self.type.value if isinstance(self.type, NotificationType) else self.type, |
| 54 | "priority": self.priority.value if isinstance(self.priority, NotificationPriority) else self.priority, |
| 55 | "title": self.title, |
| 56 | "message": self.message, |
| 57 | "detail": self.detail, |
| 58 | "timestamp": Localization.get().serialize_datetime(self.timestamp), |
| 59 | "display_time": self.display_time, |
| 60 | "read": self.read, |
| 61 | "group": self.group, |
| 62 | } |
| 63 | |
| 64 | |
| 65 | class NotificationManager: |
| 66 | def __init__(self, max_notifications: int = 100): |
| 67 | self._lock = threading.RLock() |
| 68 | self.guid: str = str(uuid.uuid4()) |
| 69 | self.updates: list[int] = [] |
| 70 | self.notifications: list[NotificationItem] = [] |
| 71 | self.max_notifications = max_notifications |
| 72 | |
| 73 | @staticmethod |
| 74 | def send_notification( |
| 75 | type: NotificationType, |
| 76 | priority: NotificationPriority, |
| 77 | message: str, |
| 78 | title: str = "", |
| 79 | detail: str = "", |
| 80 | display_time: int = 3, |
| 81 | group: str = "", |
| 82 | id: str = "", |
| 83 | ) -> NotificationItem: |
| 84 | from agent import AgentContext |
| 85 | return AgentContext.get_notification_manager().add_notification( |
| 86 | type, priority, message, title, detail, display_time, group, id |
| 87 | ) |
| 88 | |
| 89 | def add_notification( |
| 90 | self, |
| 91 | type: NotificationType, |
| 92 | priority: NotificationPriority, |
| 93 | message: str, |
| 94 | title: str = "", |
| 95 | detail: str = "", |
| 96 | display_time: int = 3, |
| 97 | group: str = "", |
| 98 | id: str = "", |
| 99 | ) -> NotificationItem: |
| 100 | with self._lock: |
| 101 | existing = None |
| 102 | if id: |
| 103 | existing = next((n for n in self.notifications if n.id == id), None) |
| 104 | |
| 105 | if existing: |
| 106 | existing.type = NotificationType(type) |
| 107 | existing.priority = NotificationPriority(priority) |
| 108 | existing.title = title |
| 109 | existing.message = message |
| 110 | existing.detail = detail |
| 111 | existing.timestamp = Localization.get().now() |
| 112 | existing.display_time = display_time |
| 113 | existing.group = group |
| 114 | existing.read = False |
| 115 | self.updates.append(existing.no) |
| 116 | item = existing |
| 117 | else: |
| 118 | # Create notification item |
| 119 | item = NotificationItem( |
| 120 | manager=self, |
| 121 | no=len(self.notifications), |
| 122 | type=NotificationType(type), |
| 123 | priority=NotificationPriority(priority), |
| 124 | title=title, |
| 125 | message=message, |
| 126 | detail=detail, |
| 127 | timestamp=Localization.get().now(), |
| 128 | display_time=display_time, |
| 129 | id=id, |
| 130 | group=group, |
| 131 | ) |
| 132 | |
| 133 | self.notifications.append(item) |
| 134 | self.updates.append(item.no) |
| 135 | self._enforce_limit() |
| 136 | |
| 137 | from helpers.state_monitor_integration import mark_dirty_all |
| 138 | mark_dirty_all(reason="notification.NotificationManager.add_notification") |
| 139 | return item |
| 140 | |
| 141 | def _enforce_limit(self): |
| 142 | with self._lock: |
| 143 | if len(self.notifications) > self.max_notifications: |
| 144 | # Remove oldest notifications |
| 145 | to_remove = len(self.notifications) - self.max_notifications |
| 146 | self.notifications = self.notifications[to_remove:] |
| 147 | # Adjust notification numbers |
| 148 | for i, notification in enumerate(self.notifications): |
| 149 | notification.no = i |
| 150 | # Adjust updates list |
| 151 | self.updates = [no - to_remove for no in self.updates if no >= to_remove] |
| 152 | |
| 153 | def get_recent_notifications(self, seconds: int = 30) -> list[NotificationItem]: |
| 154 | cutoff = Localization.get().now() - timedelta(seconds=seconds) |
| 155 | with self._lock: |
| 156 | return [n for n in self.notifications if n.timestamp >= cutoff] |
| 157 | |
| 158 | def output(self, start: int | None = None, end: int | None = None) -> list[dict]: |
| 159 | return self.output_with_state(start, end)[0] |
| 160 | |
| 161 | def output_with_state( |
| 162 | self, start: int | None = None, end: int | None = None |
| 163 | ) -> tuple[list[dict], str, int]: |
| 164 | with self._lock: |
| 165 | if start is None: |
| 166 | start = 0 |
| 167 | if end is None: |
| 168 | end = len(self.updates) |
| 169 | updates = self.updates[start:end] |
| 170 | out = [] |
| 171 | seen = set() |
| 172 | for update in updates: |
| 173 | if update not in seen and update < len(self.notifications): |
| 174 | out.append(self.notifications[update].output()) |
| 175 | seen.add(update) |
| 176 | return out, self.guid, len(self.updates) |
| 177 | |
| 178 | def output_all(self) -> list[dict]: |
| 179 | with self._lock: |
| 180 | notifications = list(self.notifications) |
| 181 | return [n.output() for n in notifications] |
| 182 | |
| 183 | def mark_read_by_ids(self, notification_ids: list[str]) -> int: |
| 184 | ids = {nid for nid in notification_ids if isinstance(nid, str) and nid.strip()} |
| 185 | if not ids: |
| 186 | return 0 |
| 187 | |
| 188 | changed_nos: list[int] = [] |
| 189 | with self._lock: |
| 190 | for notification in self.notifications: |
| 191 | if notification.id in ids and not notification.read: |
| 192 | notification.read = True |
| 193 | changed_nos.append(notification.no) |
| 194 | if changed_nos: |
| 195 | self.updates.extend(changed_nos) |
| 196 | |
| 197 | if not changed_nos: |
| 198 | return 0 |
| 199 | |
| 200 | from helpers.state_monitor_integration import mark_dirty_all |
| 201 | mark_dirty_all(reason="notification.NotificationManager.mark_read_by_ids") |
| 202 | return len(changed_nos) |
| 203 | |
| 204 | def update_item(self, no: int, **kwargs) -> None: |
| 205 | self._update_item(no, **kwargs) |
| 206 | |
| 207 | def _update_item(self, no: int, **kwargs): |
| 208 | changed = False |
| 209 | with self._lock: |
| 210 | if no < len(self.notifications): |
| 211 | item = self.notifications[no] |
| 212 | for key, value in kwargs.items(): |
| 213 | if hasattr(item, key): |
| 214 | setattr(item, key, value) |
| 215 | self.updates.append(no) |
| 216 | changed = True |
| 217 | |
| 218 | if not changed: |
| 219 | return |
| 220 | |
| 221 | from helpers.state_monitor_integration import mark_dirty_all |
| 222 | mark_dirty_all(reason="notification.NotificationManager._update_item") |
| 223 | |
| 224 | def mark_all_read(self): |
| 225 | changed_nos: list[int] = [] |
| 226 | with self._lock: |
| 227 | for notification in self.notifications: |
| 228 | if not notification.read: |
| 229 | notification.read = True |
| 230 | changed_nos.append(notification.no) |
| 231 | if changed_nos: |
| 232 | self.updates.extend(changed_nos) |
| 233 | |
| 234 | if not changed_nos: |
| 235 | return |
| 236 | |
| 237 | from helpers.state_monitor_integration import mark_dirty_all |
| 238 | mark_dirty_all(reason="notification.NotificationManager.mark_all_read") |
| 239 | |
| 240 | def clear_all(self): |
| 241 | with self._lock: |
| 242 | self.notifications = [] |
| 243 | self.updates = [] |
| 244 | self.guid = str(uuid.uuid4()) |
| 245 | from helpers.state_monitor_integration import mark_dirty_all |
| 246 | mark_dirty_all(reason="notification.NotificationManager.clear_all") |
| 247 | |
| 248 | def get_notifications_by_type(self, type: NotificationType) -> list[NotificationItem]: |
| 249 | with self._lock: |
| 250 | return [n for n in self.notifications if n.type == type] |