Files
Modus/modules/launcher/plugins/notifications.py
T
2025-06-22 19:07:27 +05:45

611 lines
23 KiB
Python

import time
from datetime import datetime
from typing import List, Optional
import utils.icons as icons
from fabric.widgets.box import Box
from fabric.widgets.button import Button
from fabric.widgets.label import Label
from gi.repository import GdkPixbuf
from modules.launcher.plugin_base import PluginBase
from modules.launcher.result import Result
from services import notification_service
class NotificationDetailWidget(Box):
"""Widget for displaying detailed notification content."""
def __init__(self, notification_data: dict, plugin_instance):
super().__init__(name="notification-detail", orientation="v", spacing=8)
self.notification_data = notification_data
self.plugin = plugin_instance
# Set size constraints to fit within launcher
self.set_size_request(530, 200)
self.set_vexpand(False)
self.set_hexpand(True)
self.setup_ui()
def setup_ui(self):
"""Setup the detail view UI."""
# Header with title and close button
header_box = Box(orientation="h", spacing=8)
# Title
title_label = Label(
label=f"📧 {self.notification_data.get('app_name', 'Unknown')}",
name="notification-detail-title",
h_align="start",
h_expand=True,
)
header_box.add(title_label)
# Close button
close_btn = Button(
child=Label(markup=icons.cancel, name="notification-close-icon"),
name="notification-close-btn",
on_clicked=lambda *_: self.plugin._close_detail_view(),
)
header_box.add(close_btn)
self.add(header_box)
# Content area
content_box = Box(orientation="v", spacing=4)
# Summary
summary = self.notification_data.get("summary", "No title")
summary_label = Label(
label=f"Subject: {summary}",
name="notification-detail-summary",
h_align="start",
ellipsize="end",
)
content_box.add(summary_label)
# Body content
body = self.notification_data.get("body", "")
if body:
body_label = Label(
label=body,
name="notification-detail-body",
h_align="start",
v_align="start",
ellipsize="none",
)
# Set line wrapping using properties
body_label.set_property("wrap", True)
body_label.set_property("wrap-mode", 2) # WORD wrap mode
content_box.add(body_label)
# Timestamp
timestamp = self.notification_data.get("timestamp", time.time())
time_str = self.plugin._format_timestamp(timestamp)
time_label = Label(
label=f"Time: {time_str}", name="notification-detail-time", h_align="start"
)
content_box.add(time_label)
self.add(content_box)
# Action buttons
actions_box = Box(orientation="h", spacing=8)
# Remove button
remove_btn = Button(
child=Label(
markup=f"{icons.trash} Remove", name="notification-action-label"
),
name="notification-action-btn",
on_clicked=lambda *_: self._remove_notification(),
)
actions_box.add(remove_btn)
self.add(actions_box)
def _remove_notification(self):
"""Remove this notification."""
notif_id = self.notification_data.get("notification_id", 0)
self.plugin._clear_notification(notif_id)
class NotificationsPlugin(PluginBase):
def __init__(self):
super().__init__()
self.name = "notifications"
self.display_name = "Notification History"
self.description = "Search and manage notification history"
# Settings
self.max_results = 15
self.notification_service = notification_service
self.showing_detail_for = None # Track which notification is showing details
def initialize(self):
"""Initialize the plugin."""
self.set_triggers(["notif"])
# Connect to notification service signals to refresh when notifications change
self.notification_service.connect(
"notification_count", self._on_notification_count_changed
)
def cleanup(self):
"""Cleanup the plugin."""
# Disconnect from notification service signals
try:
self.notification_service.disconnect(
"notification_count", self._on_notification_count_changed
)
except:
pass
def _format_timestamp(self, timestamp: float) -> str:
"""Format timestamp to human-readable string."""
try:
dt = datetime.fromtimestamp(timestamp)
now = datetime.now()
# If today, show time only
if dt.date() == now.date():
return dt.strftime("%H:%M")
# If this week, show day and time
days_diff = (now.date() - dt.date()).days
if days_diff < 7:
return dt.strftime("%a %H:%M")
# Otherwise show date
return dt.strftime("%m/%d %H:%M")
except:
return "Unknown"
def _get_notification_icon(self, notification) -> Optional[GdkPixbuf.Pixbuf]:
"""Get icon for notification."""
try:
# Try to use notification's image first
if hasattr(notification, "image_pixbuf") and notification.image_pixbuf:
return notification.image_pixbuf.scale_simple(
32, 32, GdkPixbuf.InterpType.BILINEAR
)
# Try app icon
if hasattr(notification, "app_icon") and notification.app_icon:
if notification.app_icon.startswith("file://"):
icon_path = notification.app_icon[7:]
try:
pixbuf = GdkPixbuf.Pixbuf.new_from_file(icon_path)
return pixbuf.scale_simple(
32, 32, GdkPixbuf.InterpType.BILINEAR
)
except:
pass
return None
except:
return None
def _truncate_text(self, text: str, max_length: int = 60) -> str:
"""Truncate text to specified length."""
if not text:
return ""
if len(text) <= max_length:
return text
return text[: max_length - 3] + "..."
def _trigger_refresh(self):
"""Trigger launcher refresh to return to default notification view."""
try:
from gi.repository import GLib
def trigger_refresh():
try:
# Try to access the launcher through the fabric Application
from fabric import Application
app = Application.get_default()
if app and hasattr(app, "launcher"):
launcher = app.launcher
if launcher and hasattr(launcher, "search_entry"):
# Get current search text to preserve the query
current_text = launcher.search_entry.get_text()
# Trigger the search to refresh results
if hasattr(launcher, "_perform_search"):
launcher._perform_search(current_text)
return False
# Fallback: try to find launcher instance through other means
import gc
for obj in gc.get_objects():
if (
hasattr(obj, "__class__")
and obj.__class__.__name__ == "Launcher"
):
if hasattr(obj, "search_entry") and hasattr(
obj, "_perform_search"
):
current_text = obj.search_entry.get_text()
obj._perform_search(current_text)
return False
except Exception as e:
print(f"Error forcing launcher refresh: {e}")
return False # Don't repeat
# Use immediate refresh
GLib.timeout_add(10, trigger_refresh)
except Exception as e:
print(f"Could not trigger refresh: {e}")
def _clear_notification(self, notification_id: int):
"""Remove a specific notification from history."""
try:
# Debug: Print notification removal
print(f"[NotificationPlugin] Removing notification ID: {notification_id}")
# Get count before removal for comparison
count_before = len(self.notification_service.all_notifications)
# Remove the notification
self.notification_service.remove_notification(notification_id)
# Get count after removal
count_after = len(self.notification_service.all_notifications)
# Debug: Print removal result
print(
f"[NotificationPlugin] Notifications before: {count_before}, after: {
count_after
}"
)
if count_after < count_before:
print(
f"[NotificationPlugin] Successfully removed notification {
notification_id
}"
)
else:
print(
f"[NotificationPlugin] Warning: Notification {
notification_id
} may not have been removed"
)
# Trigger immediate refresh using the same pattern as OTP plugin
self._trigger_refresh()
except Exception as e:
print(
f"[NotificationPlugin] Error removing notification {notification_id}: {
e
}"
)
# Still try to refresh even if there was an error
self._trigger_refresh()
def _clear_all_notifications(self):
"""Clear all notifications from history."""
self.notification_service.clear_all_notifications()
def _toggle_dnd(self, enable: bool = None):
"""Toggle or set DND mode."""
try:
if enable is None:
# Toggle current state
new_state = not self.notification_service.dont_disturb
else:
# Set specific state
new_state = enable
self.notification_service.dont_disturb = new_state
status = "enabled" if new_state else "disabled"
print(f"[NotificationPlugin] Do Not Disturb {status}")
# Trigger refresh to show updated status
self._trigger_refresh()
except Exception as e:
print(f"[NotificationPlugin] Error toggling DND: {e}")
def _on_notification_count_changed(self, service, count: int):
"""Handle notification count changes (new notifications added or removed)."""
try:
print(
f"[NotificationPlugin] Notification count changed to {
count
}, triggering refresh"
)
self._trigger_refresh()
except Exception as e:
print(f"[NotificationPlugin] Error handling notification count change: {e}")
def _get_notification_count(self) -> int:
"""Get total number of notifications."""
try:
return len(self.notification_service.all_notifications)
except:
return 0
def _show_notification_details(self, notification_data: dict):
"""Show detailed view of a notification."""
notif_id = notification_data.get("notification_id", 0)
self.showing_detail_for = notif_id
self._trigger_refresh()
def _close_detail_view(self):
"""Close the detail view and return to list."""
self.showing_detail_for = None
self._trigger_refresh()
def query(self, query_string: str) -> List[Result]:
"""Search notification history."""
results = []
# Check if we should show detail view for a specific notification
if self.showing_detail_for is not None:
try:
notifications = self.notification_service.get_deserialized()
for notif in notifications:
notif_data = (
notif.serialize() if hasattr(notif, "serialize") else {}
)
notif_id = notif_data.get("id", 0)
if notif_id == self.showing_detail_for:
# Create detail widget for this notification
detail_data = {
"notification_id": notif_id,
"app_name": getattr(notif, "app_name", "Unknown"),
"timestamp": getattr(notif, "timestamp", time.time()),
"summary": getattr(notif, "summary", "No title"),
"body": getattr(notif, "body", ""),
}
detail_widget = NotificationDetailWidget(detail_data, self)
results.append(
Result(
title="Notification Details",
subtitle="Press Escape to go back to list",
icon_markup=icons.notifications,
action=lambda: None,
relevance=1.0,
plugin_name=self.name,
custom_widget=detail_widget,
data={
"type": "detail_view",
"keep_launcher_open": True,
},
)
)
return results
except Exception as e:
print(f"Error showing notification details: {e}")
self.showing_detail_for = None
# Handle special commands
if query_string.lower() in ["clear"]:
results.append(
Result(
title="Clear All Notifications",
subtitle="Remove all notifications from history",
description="This will permanently delete all notification history",
icon_markup=icons.trash,
relevance=1.0,
plugin_name=self.name,
action=self._clear_all_notifications,
)
)
return results
# Handle DND commands
if query_string.lower() in ["off"]:
current_dnd = self.notification_service.dont_disturb
results.append(
Result(
title="Enable Do Not Disturb",
subtitle="Turn off notification popups"
+ (" (already enabled)" if current_dnd else ""),
description="Notifications will still be saved to history but won't show popups",
icon_markup=icons.notifications_off,
relevance=1.0,
plugin_name=self.name,
action=lambda: self._toggle_dnd(True),
data={"keep_launcher_open": True},
)
)
return results
if query_string.lower() in ["on"]:
current_dnd = self.notification_service.dont_disturb
results.append(
Result(
title="Disable Do Not Disturb",
subtitle="Turn on notification popups"
+ (" (already disabled)" if not current_dnd else ""),
description="Notifications will show popups normally",
icon_markup=icons.notifications,
relevance=1.0,
plugin_name=self.name,
action=lambda: self._toggle_dnd(False),
data={"keep_launcher_open": True},
)
)
return results
# Handle remove command
if query_string.lower().startswith(
"remove "
) or query_string.lower().startswith("delete "):
search_term = (
query_string[7:].strip()
if query_string.lower().startswith("remove ")
else query_string[7:].strip()
)
if search_term:
results.append(
Result(
title=f"Remove notifications containing '{search_term}'",
subtitle="Search for notifications to remove",
description="This will show notifications matching your search term for removal",
icon_markup=icons.trash,
relevance=1.0,
plugin_name=self.name,
action=lambda: None, # Will be handled by showing matching notifications
)
)
try:
# Get notifications from service
notifications = self.notification_service.get_deserialized()
if not notifications:
results.append(
Result(
title="No notifications found",
subtitle="Your notification history is empty",
icon_markup=icons.notifications,
relevance=0.5,
plugin_name=self.name,
)
)
return results
# Show count if no query
if not query_string:
count = len(notifications)
dnd_status = self.notification_service.dont_disturb
# Add DND status result
if dnd_status:
results.append(
Result(
title="Do Not Disturb: ON",
subtitle="Notification popups are disabled • Type 'on' to enable",
icon_markup=icons.notifications_off,
relevance=0.9,
plugin_name=self.name,
action=lambda: self._toggle_dnd(False),
data={"keep_launcher_open": True},
)
)
else:
results.append(
Result(
title="Do Not Disturb: OFF",
subtitle="Notification popups are enabled • Type 'off' to disable",
icon_markup=icons.notifications,
relevance=0.9,
plugin_name=self.name,
action=lambda: self._toggle_dnd(True),
data={"keep_launcher_open": True},
)
)
results.append(
Result(
title=f"Notification History ({count} notifications)",
subtitle="Type to search notifications or 'clear' to remove all",
icon_markup=icons.notifications,
relevance=0.8,
plugin_name=self.name,
)
)
# Filter notifications based on query
query_lower = query_string.lower() if query_string else ""
filtered_notifications = []
for notif in notifications:
# Get notification data
notif_data = notif.serialize() if hasattr(notif, "serialize") else {}
notif_id = notif_data.get("id", 0)
timestamp = getattr(notif, "timestamp", time.time())
# Search in summary, body, and app name
searchable_text = " ".join(
[
getattr(notif, "summary", ""),
getattr(notif, "body", ""),
getattr(notif, "app_name", ""),
]
).lower()
if not query_lower or query_lower in searchable_text:
filtered_notifications.append((notif, notif_id, timestamp))
# Sort by timestamp (newest first)
filtered_notifications.sort(key=lambda x: x[2], reverse=True)
# Limit results (leave room for header if no query)
max_notif_results = self.max_results - (1 if not query_string else 0)
filtered_notifications = filtered_notifications[:max_notif_results]
# Create results
for notif, notif_id, timestamp in filtered_notifications:
summary = getattr(notif, "summary", "No title")
body = getattr(notif, "body", "")
app_name = getattr(notif, "app_name", "Unknown")
# Create title and subtitle
title = self._truncate_text(summary, 50)
subtitle_parts = []
if body:
subtitle_parts.append(self._truncate_text(body, 40))
subtitle_parts.append(f"from {app_name}")
subtitle_parts.append(self._format_timestamp(timestamp))
subtitle = " • ".join(subtitle_parts)
# Get icon
icon = self._get_notification_icon(notif)
result = Result(
title=title,
subtitle=subtitle,
description="", # Don't show description by default
icon=icon,
icon_markup=icons.notifications if not icon else None,
relevance=1.0,
plugin_name=self.name,
action=lambda nd={
"notification_id": notif_id,
"app_name": app_name,
"timestamp": timestamp,
"summary": summary,
"body": body,
}: self._show_notification_details(nd),
data={
"notification_id": notif_id,
"app_name": app_name,
"timestamp": timestamp,
"can_remove": True,
"alt_action": lambda nid=notif_id: self._clear_notification(
nid
),
"keep_launcher_open": True,
},
)
results.append(result)
except Exception as e:
results.append(
Result(
title="Error loading notifications",
subtitle=str(e),
icon_markup=icons.alert,
relevance=0.0,
plugin_name=self.name,
)
)
return results