sankalp/Modus
1
0 Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases 1 Wiki Activity

♻️ refactor(notification): improve notification caching and history management

Browse Source
This commit is contained in:
sankalp committed 2025-08-11 22:14:02 +05:45
1 parent 018fd9184a
commit ad6eded2cb
5 files changed
+312 -192
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

config/assets/config.json
+14 -35
View File
@@ -1,37 +1,16 @@
{
"wallpapers_dir": "~/Pictures/Wallpapers/",
"dock_position": "Bottom",
"terminal_command": "kitty -e",
"dock_enabled": true,
"dock_auto_hide": true,
"dock_always_occluded": false,
"dock_icon_size": 52,
"window_switcher_items_per_row": 10,
"hide_special_workspace": true,
"dock_hide_special_workspace_apps": true,
"dock_preview_apps": false,
"notification_timeout": "5s",
"notification_ignored_apps": [
"Hyprshot"
],
"dock_workspace_visible": true,
"dock_metrics_visible": true,
"dock_battery_visible": true,
"dock_date_time_visible": true,
"dock_controls_visible": true,
"dock_indicators_visible": true,
"dock_notifications_visible": true,
"dock_tray_visible": true,
"dock_applications_visible": true,
"dock_language_visible": true,
"metrics_disks": [
"/"
],
"metrics_visible": {
"cpu": true,
"ram": true,
"disk": true,
"swap": true,
"gpu": false
}
"wallpapers_dir": "~/Pictures/Wallpapers/",
"dock_position": "Bottom",
"terminal_command": "kitty -e",
"dock_enabled": true,
"dock_auto_hide": true,
"dock_always_occluded": false,
"dock_icon_size": 52,
"window_switcher_items_per_row": 10,
"hide_special_workspace": true,
"dock_hide_special_workspace_apps": true,
"dock_preview_apps": false,
"notification_timeout": "5s",
"notification_ignored_apps": ["Hyprshot"],
"notification_limited_apps_history": ["Spotify"]
}
config/data.py
+9 -47
View File
@@ -10,8 +10,6 @@ gi.require_version("Gtk", "3.0")
APP_NAME = "modus"
APP_NAME_CAP = "Modus"
ALLOWED_PLAYERS = ["vlc", "cmus", "firefox", "spotify", "chromium", "vivaldi", "brave"]
def parse_timeout_string(timeout_str):
"""
@@ -90,30 +88,16 @@ if os.path.exists(CONFIG_FILE):
DOCK_HIDE_SPECIAL_WORKSPACE_APPS = config.get(
"dock_hide_special_workspace_apps", True
)
DOCK_PREVIEW_APPS = config.get("dock_preview_apps", False)
NOTIFICATION_TIMEOUT_STR = config.get("notification_timeout", "5s")
NOTIFICATION_TIMEOUT = parse_timeout_string(NOTIFICATION_TIMEOUT_STR)
NOTIFICATION_IGNORED_APPS = config.get("notification_ignored_apps", ["Hyprshot"])
DOCK_COMPONENTS_VISIBILITY = {
"workspace": config.get("dock_workspace_visible", True),
"metrics": config.get("dock_metrics_visible", True),
"battery": config.get("dock_battery_visible", True),
"date_time": config.get("dock_date_time_visible", True),
"controls": config.get("dock_controls_visible", True),
"indicators": config.get("dock_indicators_visible", True),
"notifications": config.get("dock_notifications_visible", True),
"systray": config.get("dock_tray_visible", True),
"applications": config.get("dock_applications_visible", True),
"language": config.get("dock_language_visible", True),
}
METRICS_DISKS = config.get("metrics_disks", ["/"])
METRICS_VISIBLE = config.get(
"metrics_visible",
{"cpu": True, "ram": True, "disk": True, "swap": True, "gpu": False},
NOTIFICATION_IGNORED_APPS_HISTORY = config.get(
"notification_ignored_apps_history", ["Hyprshot"]
)
NOTIFICATION_LIMITED_APPS_HISTORY = config.get(
"notification_limited_apps_history", ["Spotify"]
)
else:
WALLPAPERS_DIR = WALLPAPERS_DIR_DEFAULT
DOCK_POSITION = "Bottom"
@@ -122,34 +106,12 @@ else:
DOCK_AUTO_HIDE = True
TERMINAL_COMMAND = "kitty -e"
DOCK_THEME = "Pills"
DOCK_ICON_SIZE = 52
DOCK_ICON_SIZE = 60
WINDOW_SWITCHER_ITEMS_PER_ROW = 10
HIDE_SPECIAL_WORKSPACE = True
DOCK_HIDE_SPECIAL_WORKSPACE_APPS = True
DOCK_PREVIEW_APPS = False
NOTIFICATION_TIMEOUT_STR = "5s"
NOTIFICATION_TIMEOUT = parse_timeout_string(NOTIFICATION_TIMEOUT_STR)
NOTIFICATION_IGNORED_APPS = ["Hyprshot"]
DOCK_COMPONENTS_VISIBILITY = {
"workspace": True,
"metrics": True,
"battery": True,
"date_time": True,
"controls": True,
"indicators": True,
"notifications": True,
"systray": True,
"applications": True,
"language": True,
}
METRICS_DISKS = ["/"]
METRICS_VISIBLE = {
"cpu": True,
"ram": True,
"disk": True,
"swap": True,
"gpu": False,
}
NOTIFICATION_IGNORED_APPS_HISTORY = ["Hyprshot"]
NOTIFICATION_LIMITED_APPS_HISTORY = ["Spotify"]
modules/notification/notification.py
+168 -70
View File
@@ -240,22 +240,35 @@ def find_existing_cached_image(notification_id):
"""Find existing cached image for a notification ID, regardless of hash - prioritizes restart scenarios"""
try:
if not os.path.exists(NOTIFICATION_IMAGE_CACHE_DIR):
logger.debug(f"Notification image cache directory does not exist: {NOTIFICATION_IMAGE_CACHE_DIR}")
logger.debug(
f"Notification image cache directory does not exist: {
NOTIFICATION_IMAGE_CACHE_DIR
}"
)
return None
# Look for any cached image that starts with this notification ID
matching_files = []
for filename in os.listdir(NOTIFICATION_IMAGE_CACHE_DIR):
if filename.startswith(f"notif_{notification_id}_") and filename.endswith(".png"):
if filename.startswith(f"notif_{notification_id}_") and filename.endswith(
".png"
):
matching_files.append(filename)
if not matching_files:
logger.debug(f"No cached images found for notification ID: {notification_id}")
logger.debug(
f"No cached images found for notification ID: {notification_id}"
)
return None
# Sort by modification time to get the most recent cache file
matching_files.sort(key=lambda f: os.path.getmtime(os.path.join(NOTIFICATION_IMAGE_CACHE_DIR, f)), reverse=True)
matching_files.sort(
key=lambda f: os.path.getmtime(
os.path.join(NOTIFICATION_IMAGE_CACHE_DIR, f)
),
reverse=True,
)
# Try to load the most recent cached image
for filename in matching_files:
cache_path = os.path.join(NOTIFICATION_IMAGE_CACHE_DIR, filename)
@@ -263,16 +276,28 @@ def find_existing_cached_image(notification_id):
cached_pixbuf = GdkPixbuf.Pixbuf.new_from_file(cache_path)
if cached_pixbuf:
cache_key = filename[:-4] # Remove .png extension
logger.debug(f"Found existing cached image for notification {notification_id}: {cache_key}")
logger.debug(
f"Found existing cached image for notification {
notification_id
}: {cache_key}"
)
return cached_pixbuf
except Exception as e:
logger.debug(f"Failed to load cached image {filename}: {e}")
continue
logger.debug(f"Found {len(matching_files)} cache files for notification {notification_id} but none could be loaded")
logger.debug(
f"Found {len(matching_files)} cache files for notification {
notification_id
} but none could be loaded"
)
return None
except Exception as e:
logger.debug(f"Failed to find existing cached image for notification {notification_id}: {e}")
logger.debug(
f"Failed to find existing cached image for notification {notification_id}: {
e
}"
)
return None
@@ -286,12 +311,18 @@ def get_notification_image_cache_key(notification_id, image_pixbuf):
image_hash = hashlib.md5(pixel_data).hexdigest()[:8]
return f"notif_{notification_id}_{image_hash}"
except Exception as pixel_error:
logger.debug(f"Failed to get pixel data for cache key (temp file likely gone): {pixel_error}")
logger.debug(
f"Failed to get pixel data for cache key (temp file likely gone): {
pixel_error
}"
)
# Fallback to notification ID + dimensions if pixel data fails
try:
width = image_pixbuf.get_width()
height = image_pixbuf.get_height()
dimension_hash = hashlib.md5(f"{width}x{height}".encode()).hexdigest()[:8]
dimension_hash = hashlib.md5(
f"{width}x{height}".encode()
).hexdigest()[:8]
return f"notif_{notification_id}_{dimension_hash}"
except Exception:
# Ultimate fallback using just notification ID and timestamp
@@ -327,26 +358,30 @@ def cache_notification_image(notification_id, image_pixbuf, size=(64, 64)):
return cache_path, cache_key
except Exception as scale_error:
# Handle cases where pixbuf is corrupted or references invalid files
logger.debug(f"Failed to scale/save pixbuf (likely temp file gone): {scale_error}")
logger.debug(
f"Failed to scale/save pixbuf (likely temp file gone): {scale_error}"
)
# Try to get pixel data directly to create a valid cache key even if file is gone
try:
# If we can't scale but pixbuf exists, create a fallback cache entry
fallback_cache_key = f"fallback_{notification_id}_{int(time.time())}"
fallback_cache_path = os.path.join(NOTIFICATION_IMAGE_CACHE_DIR, f"{fallback_cache_key}.png")
fallback_cache_path = os.path.join(
NOTIFICATION_IMAGE_CACHE_DIR, f"{fallback_cache_key}.png"
)
# Create a fallback image based on notification ID
fallback_pixbuf = get_fallback_notification_icon(size)
if fallback_pixbuf:
fallback_pixbuf.savev(fallback_cache_path, "png", [], [])
logger.debug(f"Created fallback cached image: {fallback_cache_key}")
return fallback_cache_path, fallback_cache_key
except Exception as fallback_error:
logger.debug(f"Fallback cache creation failed: {fallback_error}")
return None, None
except Exception as e:
logger.warning(f"Failed to cache notification image: {e}")
return None, None
@@ -376,19 +411,33 @@ def cleanup_notification_image_cache(cache_key=None):
notification_id_pattern = cache_key[:-2] # Remove "_*"
cleaned_count = 0
for filename in os.listdir(NOTIFICATION_IMAGE_CACHE_DIR):
if filename.startswith(notification_id_pattern) and filename.endswith(".png"):
if filename.startswith(
notification_id_pattern
) and filename.endswith(".png"):
filepath = os.path.join(NOTIFICATION_IMAGE_CACHE_DIR, filename)
try:
os.unlink(filepath)
logger.debug(f"Cleaned up cached notification image: {filename}")
logger.debug(
f"Cleaned up cached notification image: {filename}"
)
cleaned_count += 1
except Exception as e:
logger.warning(f"Failed to cleanup cache file {filename}: {e}")
logger.warning(
f"Failed to cleanup cache file {filename}: {e}"
)
if cleaned_count > 0:
logger.info(f"Cleaned up {cleaned_count} cached images for notification ID pattern: {notification_id_pattern}")
logger.info(
f"Cleaned up {
cleaned_count
} cached images for notification ID pattern: {
notification_id_pattern
}"
)
else:
# Remove specific cached image
cache_path = os.path.join(NOTIFICATION_IMAGE_CACHE_DIR, f"{cache_key}.png")
cache_path = os.path.join(
NOTIFICATION_IMAGE_CACHE_DIR, f"{cache_key}.png"
)
if os.path.exists(cache_path):
os.unlink(cache_path)
logger.debug(f"Cleaned up cached notification image: {cache_key}")
@@ -468,15 +517,25 @@ def verify_cache_persistence():
try:
icon_cache_files = []
image_cache_files = []
if os.path.exists(NOTIFICATION_ICON_CACHE_DIR):
icon_cache_files = [f for f in os.listdir(NOTIFICATION_ICON_CACHE_DIR) if f.endswith('.png')]
icon_cache_files = [
f for f in os.listdir(NOTIFICATION_ICON_CACHE_DIR) if f.endswith(".png")
]
if os.path.exists(NOTIFICATION_IMAGE_CACHE_DIR):
image_cache_files = [f for f in os.listdir(NOTIFICATION_IMAGE_CACHE_DIR) if f.endswith('.png')]
logger.info(f"Cache persistence check: {len(icon_cache_files)} icons, {len(image_cache_files)} images cached")
image_cache_files = [
f
for f in os.listdir(NOTIFICATION_IMAGE_CACHE_DIR)
if f.endswith(".png")
]
logger.info(
f"Cache persistence check: {len(icon_cache_files)} icons, {
len(image_cache_files)
} images cached"
)
# Test loading a few cached items to verify they work
for cache_file in icon_cache_files[:2]: # Test first 2 icon files
try:
@@ -486,7 +545,7 @@ def verify_cache_persistence():
logger.debug(f"Successfully verified cached icon: {cache_file}")
except Exception as e:
logger.warning(f"Failed to load cached icon {cache_file}: {e}")
for cache_file in image_cache_files[:2]: # Test first 2 image files
try:
cache_path = os.path.join(NOTIFICATION_IMAGE_CACHE_DIR, cache_file)
@@ -495,9 +554,9 @@ def verify_cache_persistence():
logger.debug(f"Successfully verified cached image: {cache_file}")
except Exception as e:
logger.warning(f"Failed to load cached image {cache_file}: {e}")
return len(icon_cache_files) + len(image_cache_files) > 0
except Exception as e:
logger.error(f"Failed to verify cache persistence: {e}")
return False
@@ -507,11 +566,11 @@ def migrate_persistent_notifications():
"""Migrate persistent notifications to use cached assets when temp files are gone"""
try:
from services.modus import notification_service
migrated_count = 0
for cached_notification in notification_service.cached_notifications:
notification = cached_notification._notification
# Check if notification has image_pixbuf but temp file might be gone
if hasattr(notification, "image_pixbuf") and notification.image_pixbuf:
try:
@@ -523,13 +582,23 @@ def migrate_persistent_notifications():
if hasattr(notification, "app_icon") and notification.app_icon:
cache_notification_icon(notification.app_icon, (35, 35))
migrated_count += 1
logger.debug(f"Migrated notification for {notification.app_name} to use cached app icon")
logger.debug(
f"Migrated notification for {
notification.app_name
} to use cached app icon"
)
except Exception as cache_error:
logger.debug(f"Failed to cache app icon for {notification.app_name}: {cache_error}")
logger.debug(
f"Failed to cache app icon for {notification.app_name}: {
cache_error
}"
)
if migrated_count > 0:
logger.info(f"Migrated {migrated_count} persistent notifications to use cached assets")
logger.info(
f"Migrated {migrated_count} persistent notifications to use cached assets"
)
except Exception as e:
logger.warning(f"Failed to migrate persistent notifications: {e}")
@@ -557,22 +626,38 @@ def preload_notification_assets(notification):
cache_notification_icon(notification.app_icon, (35, 35))
logger.debug(f"Preloaded app icon for: {notification.app_name}")
except Exception as icon_error:
logger.debug(f"Failed to preload app icon for {notification.app_name}: {icon_error}")
logger.debug(
f"Failed to preload app icon for {notification.app_name}: {
icon_error
}"
)
# Preload notification image if present (with resilient handling)
if hasattr(notification, "image_pixbuf") and notification.image_pixbuf:
try:
notification_id = getattr(notification, "id", int(time.time()))
# Try to cache the image, but don't fail if temp files are gone
cache_path, cache_key = cache_notification_image(notification_id, notification.image_pixbuf, (35, 35))
cache_path, cache_key = cache_notification_image(
notification_id, notification.image_pixbuf, (35, 35)
)
if cache_path and cache_key:
logger.debug(f"Preloaded notification image for: {notification.app_name}")
logger.debug(
f"Preloaded notification image for: {notification.app_name}"
)
else:
logger.debug(f"Skipped preloading notification image for {notification.app_name} (temp file likely gone)")
logger.debug(
f"Skipped preloading notification image for {
notification.app_name
} (temp file likely gone)"
)
except Exception as image_error:
logger.debug(f"Failed to preload notification image for {notification.app_name}: {image_error}")
logger.debug(
f"Failed to preload notification image for {
notification.app_name
}: {image_error}"
)
except Exception as e:
logger.warning(f"Failed to preload notification assets: {e}")
@@ -598,7 +683,9 @@ class ActionButton(Button):
style_class = (
"start-action"
if index == 0
else "end-action" if index == total - 1 else "middle-action"
else "end-action"
if index == total - 1
else "middle-action"
)
self.add_style_class(style_class)
self.connect(
@@ -659,8 +746,10 @@ class NotificationWidget(Box):
"""Create notification header with optimized cached app icon"""
try:
# Use optimized caching for header app icon (24x24 for header display)
cached_app_icon_pixbuf = cache_notification_icon(notification.app_icon, (24, 24))
cached_app_icon_pixbuf = cache_notification_icon(
notification.app_icon, (24, 24)
)
if cached_app_icon_pixbuf:
# Create image from cached pixbuf for consistent display
app_icon = CustomImage(pixbuf=cached_app_icon_pixbuf)
@@ -810,26 +899,34 @@ class NotificationWidget(Box):
def _get_notification_pixbuf(self, notification):
"""Optimized notification pixbuf with cached-first approach for restart scenarios"""
notification_id = getattr(notification, "id", int(time.time()))
# Priority 1: ALWAYS check for existing cached image first (especially important for restarts)
if notification_id:
existing_cached_image = find_existing_cached_image(notification_id)
if existing_cached_image:
# Store the cache key for cleanup when manually closed
self.notification_image_cache_key = f"notif_{notification_id}_*" # Wildcard for cleanup
logger.debug(f"Using existing cached image for notification: {notification_id}, cache_key: {self.notification_image_cache_key}")
self.notification_image_cache_key = (
f"notif_{notification_id}_*" # Wildcard for cleanup
)
logger.debug(
f"Using existing cached image for notification: {
notification_id
}, cache_key: {self.notification_image_cache_key}"
)
return existing_cached_image
try:
# Priority 2: Try to process current pixbuf only if we have no cached version
if hasattr(notification, "image_pixbuf") and notification.image_pixbuf:
try:
cache_key = get_notification_image_cache_key(notification_id, notification.image_pixbuf)
cache_key = get_notification_image_cache_key(
notification_id, notification.image_pixbuf
)
cached_image = get_cached_notification_image(cache_key)
if cached_image:
logger.debug(f"Using cached notification image: {cache_key}")
return cached_image
# Try to cache the image now
cache_path, cache_key = cache_notification_image(
notification_id, notification.image_pixbuf, (35, 35)
@@ -841,7 +938,9 @@ class NotificationWidget(Box):
# Load and return the newly cached image
cached_pixbuf = get_cached_notification_image(cache_key)
if cached_pixbuf:
logger.debug(f"Cached and using notification image: {cache_key}")
logger.debug(
f"Cached and using notification image: {cache_key}"
)
return cached_pixbuf
# Direct fallback if caching succeeded but loading failed
@@ -850,9 +949,11 @@ class NotificationWidget(Box):
35, 35, GdkPixbuf.InterpType.BILINEAR
)
except Exception as image_error:
logger.debug(f"Notification image processing failed (temp file likely gone): {image_error}")
logger.debug(
f"Notification image processing failed (temp file likely gone): {image_error}"
)
# Continue to app icon fallback
except Exception as e:
logger.debug(f"Failed to process notification image: {e}")
@@ -908,7 +1009,11 @@ class NotificationWidget(Box):
def _manual_close(self):
"""Handle manual close button click - mark for cache cleanup"""
self._should_cleanup_cache = True
logger.debug(f"Manual close triggered for notification with cache_key: {getattr(self, 'notification_image_cache_key', 'None')} and app_icon: {getattr(self, 'app_icon_source', 'None')}")
logger.debug(
f"Manual close triggered for notification with cache_key: {
getattr(self, 'notification_image_cache_key', 'None')
} and app_icon: {getattr(self, 'app_icon_source', 'None')}"
)
self.notification.close("dismissed-by-user")
def destroy(self):
@@ -1171,9 +1276,6 @@ class ModusNoti(Window):
self.TRANSITION_DELAY = 100 # Smoother transition timing
self.DEBOUNCE_DELAY = 50 # Prevent rapid fire notifications
# Initialize ignored apps list from config
self.ignored_apps = data.NOTIFICATION_IGNORED_APPS
self._server.connect("notification-added", self.on_new_notification)
super().__init__(
anchor="top right",
@@ -1188,10 +1290,6 @@ class ModusNoti(Window):
def on_new_notification(self, fabric_notif, id):
notification: Notification = fabric_notif.get_notification_from_id(id)
# Check if the notification should be ignored completely (ignored apps)
if notification.app_name in self.ignored_apps:
return
if self._server.dont_disturb or modus_service.dont_disturb:
# Notification is already cached by the service, just don't show popup
return
modules/notification/notification_center.py
+110 -37
View File
@@ -26,6 +26,7 @@ from modules.notification.notification import (
from services.modus import notification_service
from widgets.custom_image import CustomImage
from widgets.wayland import WaylandWindow as Window
from config import data
class ExpandableNotificationGroup(Box):
@@ -310,24 +311,36 @@ class ExpandableNotificationGroup(Box):
def _get_notification_pixbuf(self, notification):
"""Use cached-first approach prioritizing existing cache over pixbuf processing"""
notification_id = getattr(notification, "id", None)
# Priority 1: ALWAYS check for existing cached image first (critical for restart scenarios)
if notification_id:
existing_cached_image = find_existing_cached_image(notification_id)
if existing_cached_image:
logger.debug(f"Using existing cached image in center for notification: {notification_id}")
logger.debug(
f"Using existing cached image in center for notification: {
notification_id
}"
)
return existing_cached_image
try:
# Priority 2: Try to process current pixbuf only if we have no cached version
if hasattr(notification, "image_pixbuf") and notification.image_pixbuf and notification_id:
if (
hasattr(notification, "image_pixbuf")
and notification.image_pixbuf
and notification_id
):
try:
cache_key = get_notification_image_cache_key(notification_id, notification.image_pixbuf)
cache_key = get_notification_image_cache_key(
notification_id, notification.image_pixbuf
)
cached_image = get_cached_notification_image(cache_key)
if cached_image:
logger.debug(f"Using cached notification image in center: {cache_key}")
logger.debug(
f"Using cached notification image in center: {cache_key}"
)
return cached_image
# Try to cache the image now
cache_path, cache_key = cache_notification_image(
notification_id, notification.image_pixbuf, (35, 35)
@@ -335,17 +348,26 @@ class ExpandableNotificationGroup(Box):
if cache_path and cache_key:
cached_image = get_cached_notification_image(cache_key)
if cached_image:
logger.debug(f"Cached and using notification image in center: {cache_key}")
logger.debug(
f"Cached and using notification image in center: {cache_key}"
)
return cached_image
# Direct scaling fallback if caching works but loading fails
logger.debug("Using direct scaling for notification image in center")
return notification.image_pixbuf.scale_simple(35, 35, 2) # GdkPixbuf.InterpType.BILINEAR = 2
logger.debug(
"Using direct scaling for notification image in center"
)
# GdkPixbuf.InterpType.BILINEAR = 2
return notification.image_pixbuf.scale_simple(35, 35, 2)
except Exception as image_error:
logger.debug(f"Notification image processing failed in center (temp file likely gone): {image_error}")
logger.debug(
f"Notification image processing failed in center (temp file likely gone): {
image_error
}"
)
# Continue to app icon fallback
except Exception as e:
logger.debug(f"Failed to process notification image in center: {e}")
@@ -356,7 +378,9 @@ class ExpandableNotificationGroup(Box):
if app_icon_source:
cached_app_icon = cache_notification_icon(app_icon_source, (35, 35))
if cached_app_icon:
logger.debug(f"Using cached app icon in center for: {app_icon_source}")
logger.debug(
f"Using cached app icon in center for: {app_icon_source}"
)
return cached_app_icon
except Exception as e:
logger.debug(f"Failed to get cached app icon in center: {e}")
@@ -425,15 +449,17 @@ class ExpandableNotificationGroup(Box):
# Get notification ID for cache cleanup
notification_id = getattr(notification._notification, "id", None)
app_icon_source = getattr(notification._notification, "app_icon", None)
# Clean up caches using notification ID pattern
if notification_id:
cleanup_notification_specific_caches(
app_icon_source=app_icon_source,
notification_image_cache_key=f"notif_{notification_id}_*"
notification_image_cache_key=f"notif_{notification_id}_*",
)
logger.debug(f"Cleaned up caches for notification ID: {notification_id}")
logger.debug(
f"Cleaned up caches for notification ID: {notification_id}"
)
notification_service.remove_cached_notification(notification.cache_id)
except Exception as e:
logger.error(
@@ -446,15 +472,17 @@ class ExpandableNotificationGroup(Box):
# Get notification ID for cache cleanup
notification_id = getattr(notification._notification, "id", None)
app_icon_source = getattr(notification._notification, "app_icon", None)
# Clean up caches using notification ID pattern
if notification_id:
cleanup_notification_specific_caches(
app_icon_source=app_icon_source,
notification_image_cache_key=f"notif_{notification_id}_*"
notification_image_cache_key=f"notif_{notification_id}_*",
)
logger.debug(f"Cleaned up caches for notification ID: {notification_id}")
logger.debug(
f"Cleaned up caches for notification ID: {notification_id}"
)
notification_service.remove_cached_notification(notification.cache_id)
logger.debug(f"Closed single notification: {notification.cache_id}")
except Exception as e:
@@ -583,15 +611,17 @@ class NotificationCenterWidget(NotificationWidget):
# Get notification ID for cache cleanup
notification_id = getattr(self.notification, "id", None)
app_icon_source = getattr(self.notification, "app_icon", None)
# Clean up caches using notification ID pattern
if notification_id:
cleanup_notification_specific_caches(
app_icon_source=app_icon_source,
notification_image_cache_key=f"notif_{notification_id}_*"
notification_image_cache_key=f"notif_{notification_id}_*",
)
logger.debug(f"Cleaned up caches for notification center ID: {notification_id}")
logger.debug(
f"Cleaned up caches for notification center ID: {notification_id}"
)
notification_service.remove_cached_notification(self.notification_id)
except Exception as e:
logger.error(f"Error removing notification {self.notification_id}: {e}")
@@ -696,31 +726,53 @@ class NotificationCenter(Window):
for cached_notification in notification_service.cached_notifications:
app_name = cached_notification._notification.app_name
notification_id = getattr(cached_notification._notification, "id", None)
logger.debug(f"Rebuilding notification for {app_name} with ID: {notification_id}")
# Skip ignored apps during rebuild
if app_name in data.NOTIFICATION_IGNORED_APPS_HISTORY:
continue
# Preload assets for each cached notification to ensure display consistency
preload_notification_assets(cached_notification._notification)
# Check if we can find cached images for this notification
if notification_id:
existing_image = find_existing_cached_image(notification_id)
if existing_image:
logger.debug(f"✓ Found cached image for notification {notification_id} during rebuild")
logger.debug(
f"✓ Found cached image for notification {
notification_id
} during rebuild"
)
else:
logger.debug(f"✗ No cached image found for notification {notification_id} during rebuild")
logger.debug(
f"✗ No cached image found for notification {
notification_id
} during rebuild"
)
self.notification_groups[app_name].append(cached_notification)
rebuild_count += 1
logger.info(f"Rebuilt {rebuild_count} notifications into {len(self.notification_groups)} groups")
logger.info(
f"Rebuilt {rebuild_count} notifications into {
len(self.notification_groups)
} groups"
)
# Create group widgets
# Create group widgets and handle limited apps
for app_name, notifications in self.notification_groups.items():
# Sort notifications by timestamp (newest first)
notifications.sort(key=lambda n: getattr(n, "timestamp", 0), reverse=True)
group_widget = ExpandableNotificationGroup(app_name, notifications)
# Handle limited apps history - only keep 5 notifications during rebuild
if app_name in data.NOTIFICATION_LIMITED_APPS_HISTORY:
if len(notifications) > 5:
# Keep only the 5 most recent notifications
self.notification_groups[app_name] = notifications[:5]
group_widget = ExpandableNotificationGroup(
app_name, self.notification_groups[app_name]
)
self.group_widgets[app_name] = group_widget
self.notifications_box.add(group_widget)
@@ -728,6 +780,10 @@ class NotificationCenter(Window):
try:
app_name = cached_notification._notification.app_name
# Check if this app should be ignored for history (don't add to notification center)
if app_name in data.NOTIFICATION_IGNORED_APPS_HISTORY:
return
# Preload assets for notification center display (ensure caching consistency)
preload_notification_assets(cached_notification._notification)
@@ -736,6 +792,23 @@ class NotificationCenter(Window):
0, cached_notification
) # Insert at beginning (newest first)
# Handle limited apps history - only keep 5 notifications
if app_name in data.NOTIFICATION_LIMITED_APPS_HISTORY:
notifications_for_app = self.notification_groups[app_name]
if len(notifications_for_app) > 5:
# Remove oldest notifications beyond the limit
excess_notifications = notifications_for_app[5:]
self.notification_groups[app_name] = notifications_for_app[:5]
# Remove excess notifications from the service cache
for excess_notification in excess_notifications:
try:
notification_service.remove_cached_notification(
excess_notification.cache_id
)
except Exception as e:
logger.error(f"Error removing excess notification: {e}")
# Update or create group widget
if app_name in self.group_widgets:
# Update existing group
@@ -826,7 +899,7 @@ class NotificationCenter(Window):
# No notifications label removed - only update clear button and scrolled visibility
self.clear_all_button.set_visible(current_count > 0)
self.scrolled.set_visible(current_count > 0)
# Auto-close notification center when no notifications remain
if current_count == 0 and hasattr(self, "mousecapture"):
self.mousecapture.hide_child_window()
services/custom_notification.py
+11 -3
View File
@@ -3,7 +3,7 @@ import json
import os
from typing import List
# Fabric imports
# Fabric imports
import gi
from gi.repository import GdkPixbuf
@@ -224,7 +224,15 @@ class CachedNotifications(Notifications):
notification = self.get_notification_from_id(notification_id)
if notification:
# Always cache notifications, regardless of DND status
# Import here to avoid circular imports
from config import data
# Check if this app should be ignored for history (don't cache)
if notification.app_name in data.NOTIFICATION_IGNORED_APPS_HISTORY:
# Don't cache notifications from ignored apps, but still allow popup display
return
# Cache notifications (except for ignored apps)
# DND only affects popup display, not caching
self._count += 1
cache_id = self._count
@@ -258,7 +266,7 @@ class CachedNotifications(Notifications):
if handler_id:
# Disconnect the signal handler
cached_notification.disconnect(handler_id)
# Emit signal to notify UI that notification was removed
self.emit("cached-notification-removed", cached_notification)