From e5471eb771309611ce60479eab1d090c1d42a5dc Mon Sep 17 00:00:00 2001 From: S4NKALP Date: Mon, 30 Jun 2025 01:11:08 +0545 Subject: [PATCH] =?UTF-8?q?=F0=9F=94=A7=20chore(dock/components):=20import?= =?UTF-8?q?=20cleanup=20in=20=5F=5Finit=5F=5F=20files,=20refactor=20System?= =?UTF-8?q?TrayItem=20monkey=20patching?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- modules/dock/components/__init__.py | 16 +- modules/dock/components/music_player.py | 164 ++++++++--- modules/dock/components/notifications.py | 70 ++--- modules/dock/components/player.py | 62 +++-- modules/launcher/plugins/otp.py | 47 +++- services/auth.py | 39 ++- services/qrcode.py | 335 ----------------------- styles/components.css | 3 + utils/notification_utils.py | 56 ++-- 9 files changed, 293 insertions(+), 499 deletions(-) delete mode 100644 services/qrcode.py diff --git a/modules/dock/components/__init__.py b/modules/dock/components/__init__.py index d2bbbb8..25236e0 100644 --- a/modules/dock/components/__init__.py +++ b/modules/dock/components/__init__.py @@ -1,6 +1,9 @@ +from gi.repository import Gtk +from fabric.system_tray.widgets import SystemTrayItem import json import os +import gi from fabric.hyprland.service import HyprlandEvent from fabric.hyprland.widgets import Language, get_hyprland_connection from fabric.system_tray.widgets import SystemTray @@ -21,12 +24,9 @@ from .metrics import Metrics from .music_player import MusicPlayer from .workspaces import workspace -import gi gi.require_version("Gtk", "3.0") -from gi.repository import Gtk - # Monkey patch SystemTray to use GTK theme icons -from fabric.system_tray.widgets import SystemTrayItem + def _patched_do_update_properties(self, *_): """Enhanced update method that prioritizes GTK theme icons""" @@ -65,10 +65,13 @@ def _patched_do_update_properties(self, *_): # Set tooltip (same as original) tooltip = self._item.tooltip self.set_tooltip_markup( - tooltip.description or tooltip.title or self._item.title.title() if self._item.title else "Unknown" + tooltip.description or tooltip.title or self._item.title.title() + if self._item.title + else "Unknown" ) return + # Apply the patch SystemTrayItem.do_update_properties = _patched_do_update_properties @@ -142,7 +145,8 @@ class DockComponents(Box): # Apply invert style for specific themes and positions should_invert = data.DOCK_THEME in ["Dense", "Edge"] or ( - data.DOCK_THEME == "Pills" and data.DOCK_POSITION in ["Left", "Right", "Top"] + data.DOCK_THEME == "Pills" + and data.DOCK_POSITION in ["Left", "Right", "Top"] ) if should_invert: diff --git a/modules/dock/components/music_player.py b/modules/dock/components/music_player.py index 3346ad6..9c95826 100644 --- a/modules/dock/components/music_player.py +++ b/modules/dock/components/music_player.py @@ -8,9 +8,9 @@ from gi.repository import Gdk import config.data as data import utils.icons as icons +from modules.dock.components.player import PlayerContainer from services.mpris import PlayerManager, PlayerService from utils.wayland import WaylandWindow as Window -from modules.dock.components.player import PlayerContainer class MusicPlayerPopup(Window): @@ -40,7 +40,9 @@ class MusicPlayerPopup(Window): # Connect player container events to sync with dock if self.dock_component: - self.player_container.connect("active-player-changed", self.on_active_player_changed) + self.player_container.connect( + "active-player-changed", self.on_active_player_changed + ) # Connect to escape key to close self.connect("key-press-event", self.on_key_press) @@ -152,15 +154,19 @@ class MusicPlayer(Box): # Set up periodic check for active player from gi.repository import GLib - self.active_player_check_timeout = GLib.timeout_add_seconds(2, self.check_active_player) + + self.active_player_check_timeout = GLib.timeout_add_seconds( + 2, self.check_active_player + ) # Connect to destroy signal for cleanup self.connect("destroy", self.on_destroy) def on_destroy(self, widget): """Clean up popup when component is destroyed""" - if hasattr(self, 'active_player_check_timeout'): + if hasattr(self, "active_player_check_timeout"): from gi.repository import GLib + GLib.source_remove(self.active_player_check_timeout) if self.popup: self.popup.destroy() @@ -175,8 +181,12 @@ class MusicPlayer(Box): self.set_current_player(player) else: # If the new player is playing, switch to it - if (player.props.playback_status.value_name == "PLAYERCTL_PLAYBACK_STATUS_PLAYING" and - self.current_player.props.playback_status.value_name != "PLAYERCTL_PLAYBACK_STATUS_PLAYING"): + if ( + player.props.playback_status.value_name + == "PLAYERCTL_PLAYBACK_STATUS_PLAYING" + and self.current_player.props.playback_status.value_name + != "PLAYERCTL_PLAYBACK_STATUS_PLAYING" + ): self.set_current_player(player) self.update_visibility() @@ -194,26 +204,40 @@ class MusicPlayer(Box): self.current_player_service = None # Try to find another player with media content - if hasattr(self.manager, '_manager') and hasattr(self.manager._manager, 'props'): + if hasattr(self.manager, "_manager") and hasattr( + self.manager._manager, "props" + ): for player_name in self.manager._manager.props.player_names: try: from gi.repository import Playerctl + candidate_player = Playerctl.Player.new_from_name(player_name) # Check if this player has media content - if (hasattr(candidate_player, 'props') and - hasattr(candidate_player.props, 'metadata')): + if hasattr(candidate_player, "props") and hasattr( + candidate_player.props, "metadata" + ): metadata = candidate_player.props.metadata if metadata and len(metadata.keys()) > 0: keys = metadata.keys() - if ("xesam:title" in keys or - "xesam:artist" in keys or - "mpris:length" in keys): - print(f"[DEBUG] Switching to player with media: {candidate_player.props.player_name}") + if ( + "xesam:title" in keys + or "xesam:artist" in keys + or "mpris:length" in keys + ): + print( + f"[DEBUG] Switching to player with media: { + candidate_player.props.player_name + }" + ) self.set_current_player(candidate_player) break except Exception as e: - print(f"[DEBUG] Error checking candidate player {player_name}: {e}") + print( + f"[DEBUG] Error checking candidate player {player_name}: { + e + }" + ) continue self.update_visibility() @@ -221,15 +245,25 @@ class MusicPlayer(Box): def get_playing_player(self): """Find a currently playing player from the manager""" try: - if hasattr(self.manager, '_manager') and hasattr(self.manager._manager, 'props'): + if hasattr(self.manager, "_manager") and hasattr( + self.manager._manager, "props" + ): for player_name in self.manager._manager.props.player_names: try: from gi.repository import Playerctl + player = Playerctl.Player.new_from_name(player_name) - if (hasattr(player, 'props') and - hasattr(player.props, 'playback_status') and - player.props.playback_status.value_name == "PLAYERCTL_PLAYBACK_STATUS_PLAYING"): - print(f"[DEBUG] Found playing player: {player.props.player_name}") + if ( + hasattr(player, "props") + and hasattr(player.props, "playback_status") + and player.props.playback_status.value_name + == "PLAYERCTL_PLAYBACK_STATUS_PLAYING" + ): + print( + f"[DEBUG] Found playing player: { + player.props.player_name + }" + ) return player except Exception as e: print(f"[DEBUG] Error checking player {player_name}: {e}") @@ -246,22 +280,42 @@ class MusicPlayer(Box): if playing_player: # If we found a playing player and it's different from current, switch to it - if (not self.current_player or - self.current_player.props.player_name != playing_player.props.player_name): - print(f"[DEBUG] Switching to playing player: {playing_player.props.player_name}") + if ( + not self.current_player + or self.current_player.props.player_name + != playing_player.props.player_name + ): + print( + f"[DEBUG] Switching to playing player: { + playing_player.props.player_name + }" + ) self.set_current_player(playing_player) elif self.current_player: # If no player is playing, check if current player is paused - if (self.current_player.props.playback_status.value_name != "PLAYERCTL_PLAYBACK_STATUS_PLAYING"): + if ( + self.current_player.props.playback_status.value_name + != "PLAYERCTL_PLAYBACK_STATUS_PLAYING" + ): # Current player is paused, see if there's another player available - if hasattr(self.manager, '_manager') and hasattr(self.manager._manager, 'props'): + if hasattr(self.manager, "_manager") and hasattr( + self.manager._manager, "props" + ): for player_name in self.manager._manager.props.player_names: try: from gi.repository import Playerctl + player = Playerctl.Player.new_from_name(player_name) - if player.props.player_name != self.current_player.props.player_name: + if ( + player.props.player_name + != self.current_player.props.player_name + ): # Switch to a different available player - print(f"[DEBUG] Switching to available player: {player.props.player_name}") + print( + f"[DEBUG] Switching to available player: { + player.props.player_name + }" + ) self.set_current_player(player) break except: @@ -320,10 +374,12 @@ class MusicPlayer(Box): pos_sec = int(pos % 60) dur_min = int(dur // 60) dur_sec = int(dur % 60) - timestamp_text = f"{pos_min:02d}:{pos_sec:02d} / {dur_min:02d}:{dur_sec:02d}" + timestamp_text = ( + f"{pos_min:02d}:{pos_sec:02d} / {dur_min:02d}:{dur_sec:02d}" + ) # Get current track info if available - if hasattr(self, '_current_track_info'): + if hasattr(self, "_current_track_info"): tooltip = f"{self._current_track_info}\n{timestamp_text}" else: tooltip = f"Music Player\n{timestamp_text}" @@ -337,7 +393,7 @@ class MusicPlayer(Box): self.music_label.set_markup(icons.music) self.progress_bar.remove_style_class("paused") # Update tooltip to show playing state - if hasattr(self, '_current_track_info'): + if hasattr(self, "_current_track_info"): self.event_box.set_tooltip_text(f"{self._current_track_info} (Playing)") # Trigger immediate check for active player when play state changes @@ -351,7 +407,7 @@ class MusicPlayer(Box): self.music_label.set_markup(icons.music) self.progress_bar.add_style_class("paused") # Update tooltip to show paused state - if hasattr(self, '_current_track_info'): + if hasattr(self, "_current_track_info"): self.event_box.set_tooltip_text(f"{self._current_track_info} (Paused)") # Trigger immediate check for active player when pause state changes @@ -373,12 +429,18 @@ class MusicPlayer(Box): self._current_track_info = f"{artist} - {title}" # If we have position info, include timestamp - if hasattr(self, 'position') and hasattr(self, 'duration') and self.duration > 0: + if ( + hasattr(self, "position") + and hasattr(self, "duration") + and self.duration > 0 + ): pos_min = int(self.position // 60) pos_sec = int(self.position % 60) dur_min = int(self.duration // 60) dur_sec = int(self.duration % 60) - timestamp_text = f"{pos_min:02d}:{pos_sec:02d} / {dur_min:02d}:{dur_sec:02d}" + timestamp_text = ( + f"{pos_min:02d}:{pos_sec:02d} / {dur_min:02d}:{dur_sec:02d}" + ) tooltip = f"{self._current_track_info}\n{timestamp_text}" else: tooltip = self._current_track_info @@ -438,29 +500,47 @@ class MusicPlayer(Box): def has_running_players(self): """Check if there are any players that actually have media content running""" try: - if hasattr(self.manager, '_manager') and hasattr(self.manager._manager, 'props'): + if hasattr(self.manager, "_manager") and hasattr( + self.manager._manager, "props" + ): for player_name in self.manager._manager.props.player_names: try: from gi.repository import Playerctl + player = Playerctl.Player.new_from_name(player_name) # Check if player has metadata indicating actual media content - if hasattr(player, 'props') and hasattr(player.props, 'metadata'): + if hasattr(player, "props") and hasattr( + player.props, "metadata" + ): metadata = player.props.metadata if metadata and len(metadata.keys()) > 0: # Check for essential metadata that indicates real media keys = metadata.keys() - if ("xesam:title" in keys or - "xesam:artist" in keys or - "mpris:length" in keys): - print(f"[DEBUG] Found player with media content: {player.props.player_name}") + if ( + "xesam:title" in keys + or "xesam:artist" in keys + or "mpris:length" in keys + ): + print( + f"[DEBUG] Found player with media content: { + player.props.player_name + }" + ) return True # Also check if player is currently playing (even without full metadata) - if (hasattr(player, 'props') and - hasattr(player.props, 'playback_status') and - player.props.playback_status.value_name == "PLAYERCTL_PLAYBACK_STATUS_PLAYING"): - print(f"[DEBUG] Found playing player: {player.props.player_name}") + if ( + hasattr(player, "props") + and hasattr(player.props, "playback_status") + and player.props.playback_status.value_name + == "PLAYERCTL_PLAYBACK_STATUS_PLAYING" + ): + print( + f"[DEBUG] Found playing player: { + player.props.player_name + }" + ) return True except Exception as e: diff --git a/modules/dock/components/notifications.py b/modules/dock/components/notifications.py index 39e41ac..e5756d8 100644 --- a/modules/dock/components/notifications.py +++ b/modules/dock/components/notifications.py @@ -7,33 +7,32 @@ from fabric.widgets.button import Button from fabric.widgets.centerbox import CenterBox from fabric.widgets.label import Label from fabric.widgets.scrolledwindow import ScrolledWindow -from gi.repository import GLib, Gtk, Gdk +from gi.repository import Gdk, GLib, Gtk import config.data as data import utils.icons as icons from modules.notification_popup import NotificationBox from utils.custom_image import CustomImage -from utils.wayland import WaylandWindow as Window from utils.notification_utils import ( + CONFIG_FILE, PERSISTENT_DIR, PERSISTENT_HISTORY_FILE, - CONFIG_FILE, - load_scaled_pixbuf, - get_shared_notification_history, cleanup_orphan_cached_images, - get_date_header, - schedule_midnight_update, compute_time_label, create_historical_notification_from_data, - save_persistent_history + get_date_header, + get_shared_notification_history, + load_scaled_pixbuf, + save_persistent_history, + schedule_midnight_update, ) +from utils.wayland import WaylandWindow as Window class NotificationIndicator(Button): def __init__(self, **kwargs): super().__init__(name="button-bar-notifications", **kwargs) - self.notification_icon = Label( name="notification-icon", markup=icons.notifications ) @@ -46,7 +45,6 @@ class NotificationIndicator(Button): self.popup_window = None - self._connect_to_dnd_state() def on_clicked(self, _button): @@ -64,14 +62,15 @@ class NotificationIndicator(Button): def _connect_to_dnd_state(self): try: - self.notification_history = get_shared_notification_history() + self.notification_history.connect( + "dnd-state-changed", self._on_dnd_state_changed + ) - self.notification_history.connect('dnd-state-changed', self._on_dnd_state_changed) - - - self._update_icon_for_dnd_state(self.notification_history.do_not_disturb_enabled) + self._update_icon_for_dnd_state( + self.notification_history.do_not_disturb_enabled + ) except Exception as e: print(f"Could not connect to DND state: {e}") @@ -87,9 +86,6 @@ class NotificationIndicator(Button): self.set_tooltip_text("Notifications") - - - class NotificationHistory(Box): def __init__(self, **kwargs): super().__init__(name="notification-history", orientation="v", **kwargs) @@ -169,10 +165,8 @@ class NotificationHistory(Box): cleanup_orphan_cached_images(self.persistent_notifications) self.schedule_midnight_update() - self._connect_to_shared_history() - self._load_and_sync_dnd_state() self.LIMITED_APPS_HISTORY = ["Spotify"] @@ -180,7 +174,9 @@ class NotificationHistory(Box): def _connect_to_shared_history(self): try: shared_history = get_shared_notification_history() - shared_history.connect('notification-added', self._on_shared_notification_added) + shared_history.connect( + "notification-added", self._on_shared_notification_added + ) except Exception as e: print(f"Could not connect to shared notification history: {e}") @@ -189,20 +185,17 @@ class NotificationHistory(Box): def _refresh_history(self): try: - for child in self.notifications_list.get_children()[:]: self.notifications_list.remove(child) - if hasattr(child, 'destroy'): + if hasattr(child, "destroy"): child.destroy() self.containers.clear() - if os.path.exists(PERSISTENT_HISTORY_FILE): with open(PERSISTENT_HISTORY_FILE, "r") as f: self.persistent_notifications = json.load(f) - for note in reversed(self.persistent_notifications[-50:]): self._add_historical_notification(note) @@ -212,18 +205,15 @@ class NotificationHistory(Box): def _load_and_sync_dnd_state(self): try: - dnd_enabled = False if os.path.exists(CONFIG_FILE): with open(CONFIG_FILE, "r") as f: config_data = json.load(f) dnd_enabled = config_data.get("dnd_enabled", False) - shared_history = get_shared_notification_history() shared_history.do_not_disturb_enabled = dnd_enabled - self.header_switch.set_active(dnd_enabled) self.do_not_disturb_enabled = dnd_enabled @@ -284,8 +274,6 @@ class NotificationHistory(Box): def on_do_not_disturb_changed(self, switch, pspec): self.do_not_disturb_enabled = switch.get_active() - - try: shared_history = get_shared_notification_history() shared_history.set_do_not_disturb_enabled(self.do_not_disturb_enabled) @@ -305,8 +293,6 @@ class NotificationHistory(Box): self.notifications_list.remove(child) child.destroy() - - if os.path.exists(PERSISTENT_HISTORY_FILE): try: os.remove(PERSISTENT_HISTORY_FILE) @@ -322,14 +308,11 @@ class NotificationHistory(Box): if not os.path.exists(PERSISTENT_DIR): os.makedirs(PERSISTENT_DIR, exist_ok=True) - if os.path.exists(PERSISTENT_HISTORY_FILE): try: with open(PERSISTENT_HISTORY_FILE, "r") as f: self.persistent_notifications = json.load(f) - - for note in reversed(self.persistent_notifications[-50:]): self._add_historical_notification(note) @@ -349,8 +332,6 @@ class NotificationHistory(Box): self.no_notifications_box.set_visible(not has_notifications) self.notifications_list.set_visible(has_notifications) - - def _add_historical_notification(self, note): hist_notif = create_historical_notification_from_data(note) @@ -497,8 +478,6 @@ class NotificationHistory(Box): if removed_from_list: self.persistent_notifications = new_persistent_notifications - - self._save_persistent_history() container.destroy() self.containers = [c for c in self.containers if c != container] @@ -507,10 +486,8 @@ class NotificationHistory(Box): class NotificationHistoryWindow(Window): def __init__(self): - self.notification_history = NotificationHistory() - scrolled = ScrolledWindow( name="notifications-scrolled", child=self.notification_history, @@ -533,11 +510,9 @@ class NotificationHistoryWindow(Window): children=[scrolled], ) - dock_position = data.DOCK_POSITION popup_anchor = self._get_popup_anchor(dock_position) - margin = self._get_popup_margin(dock_position) super().__init__( @@ -551,15 +526,12 @@ class NotificationHistoryWindow(Window): all_visible=True, ) - self.connect("button-press-event", self.on_button_press) self.connect("key-press-event", self.on_key_press) - self.set_can_focus(True) self.grab_focus() - self.set_tooltip_text( "Keyboard shortcuts:\n• Escape: Close popup\n• Ctrl+D: Toggle DND\n• Ctrl+A: Clear all notifications" ) @@ -586,22 +558,16 @@ class NotificationHistoryWindow(Window): pass def on_key_press(self, _widget, event): - if event.keyval == Gdk.KEY_Escape: - self.set_visible(False) return True - elif event.state & Gdk.ModifierType.CONTROL_MASK and event.keyval == Gdk.KEY_d: - current_dnd = self.notification_history.do_not_disturb_enabled self.notification_history.header_switch.set_active(not current_dnd) return True - elif event.state & Gdk.ModifierType.CONTROL_MASK and event.keyval == Gdk.KEY_a: - self.notification_history.clear_history() return True diff --git a/modules/dock/components/player.py b/modules/dock/components/player.py index 01b2290..5026a2b 100644 --- a/modules/dock/components/player.py +++ b/modules/dock/components/player.py @@ -1,25 +1,27 @@ -from fabric.widgets.box import Box -from fabric.widgets.label import Label -from fabric.widgets.centerbox import CenterBox -from fabric.widgets.button import Button -from fabric.widgets.stack import Stack from fabric.core.service import Signal +from fabric.widgets.box import Box +from fabric.widgets.button import Button +from fabric.widgets.centerbox import CenterBox +from fabric.widgets.label import Label +from fabric.widgets.stack import Stack +from loguru import logger +import config.data as data +import utils.icons as icons from services.mpris import PlayerManager, PlayerService from utils.wiggle_bar import WigglyWidget -import utils.icons as icons -import config.data as data - -from loguru import logger - class Player(Box): def __init__(self, player, **kwargs): super().__init__(style_classes="player", orientation="v", **kwargs) # Check if we're in vertical mode based on dock position - vertical_mode = data.DOCK_POSITION in ["Left", "Right"] if hasattr(data, 'DOCK_POSITION') else False + vertical_mode = ( + data.DOCK_POSITION in ["Left", "Right"] + if hasattr(data, "DOCK_POSITION") + else False + ) if not vertical_mode: # vertical class binding from unknown source @@ -42,9 +44,7 @@ class Player(Box): ) # Use a fallback background image - self.set_style( - f"background-image:url('{data.HOME_DIR}/.current.wall')" - ) + self.set_style(f"background-image:url('{data.HOME_DIR}/.current.wall')") self.song = Label( name="song", @@ -182,7 +182,11 @@ class Player(Box): keys = metadata.keys() if "xesam:artist" in keys and "xesam:title" in keys: # Check if we're in vertical mode - vertical_mode = data.DOCK_POSITION in ["Left", "Right"] if hasattr(data, 'DOCK_POSITION') else False + vertical_mode = ( + data.DOCK_POSITION in ["Left", "Right"] + if hasattr(data, "DOCK_POSITION") + else False + ) _max_chars = 43 if not vertical_mode else 30 song_title = metadata["xesam:title"] @@ -201,16 +205,21 @@ class Player(Box): elif "xesam:url" in keys and "youtube.com" in metadata["xesam:url"]: # Simple YouTube thumbnail extraction for Firefox import re + url = metadata["xesam:url"] - match = re.search(r'(?:v=|/)([a-zA-Z0-9_-]{11})', url) + match = re.search(r"(?:v=|/)([a-zA-Z0-9_-]{11})", url) if match: video_id = match.group(1) - thumbnail_url = f"https://img.youtube.com/vi/{video_id}/hqdefault.jpg" + thumbnail_url = ( + f"https://img.youtube.com/vi/{video_id}/hqdefault.jpg" + ) self.set_style(f"background-image:url('{thumbnail_url}')") print(f"Using YouTube thumbnail: {thumbnail_url}") else: # Fallback to default wallpaper - self.set_style(f"background-image:url('{data.HOME_DIR}/.current.wall')") + self.set_style( + f"background-image:url('{data.HOME_DIR}/.current.wall')" + ) else: # Fallback to default wallpaper when no artwork is available self.set_style(f"background-image:url('{data.HOME_DIR}/.current.wall')") @@ -222,7 +231,9 @@ class Player(Box): self.on_play(self._player) if player.props.shuffle == True: - self.shuffle_button.get_child().set_markup(icons.shuffle) # Use same icon for now + self.shuffle_button.get_child().set_markup( + icons.shuffle + ) # Use same icon for now self.shuffle_button.get_child().set_name("disable-shuffle") else: self.shuffle_button.get_child().set_markup(icons.shuffle) @@ -250,7 +261,9 @@ class Player(Box): self.shuffle_button.get_child().set_markup(icons.shuffle) self.shuffle_button.get_child().set_name("shuffle") else: - self.shuffle_button.get_child().set_markup(icons.shuffle) # Use same icon for now + self.shuffle_button.get_child().set_markup( + icons.shuffle + ) # Use same icon for now self.shuffle_button.get_child().set_name("disable-shuffle") self.shuffle_button.get_child().set_style("color: white") @@ -324,7 +337,11 @@ class PlayerContainer(Box): ) # Check if we're in vertical mode - vertical_mode = data.DOCK_POSITION in ["Left", "Right"] if hasattr(data, 'DOCK_POSITION') else False + vertical_mode = ( + data.DOCK_POSITION in ["Left", "Right"] + if hasattr(data, "DOCK_POSITION") + else False + ) self.player_switch_container = CenterBox( name="player-switch-container", @@ -347,7 +364,8 @@ class PlayerContainer(Box): new_player.gtk_wrapper.queue_draw() new_player.set_name(player.props.player_name) self.player.append(new_player) - self.player_objects[player.props.player_name] = player # Store the player object + # Store the player object + self.player_objects[player.props.player_name] = player print("stacking player") self.stack.add_named(new_player, player.props.player_name) diff --git a/modules/launcher/plugins/otp.py b/modules/launcher/plugins/otp.py index 84b42e7..d9f95c4 100644 --- a/modules/launcher/plugins/otp.py +++ b/modules/launcher/plugins/otp.py @@ -13,9 +13,9 @@ from modules.launcher.result import Result from services.auth import ( generate_totp, get_time_remaining_with_blink, - validate_base32_secret, parse_otpauth_uri, scan_qr_and_add_account, + validate_base32_secret, ) @@ -421,7 +421,9 @@ class OTPPlugin(PluginBase): results.append( Result( title=f"{totp_code}", - subtitle_markup=f"{display_name} • {time_display} remaining • Shift+Enter: remove", + subtitle_markup=f"{display_name} • { + time_display + } remaining • Shift+Enter: remove", icon_markup=icons.key, action=lambda code=totp_code: self._copy_to_clipboard(code), relevance=1.0, @@ -430,7 +432,9 @@ class OTPPlugin(PluginBase): "type": "totp", "account": account_name, "code": totp_code, - "alt_action": lambda acc=account_name: self._remove_account_and_refresh(acc), + "alt_action": lambda acc=account_name: self._remove_account_and_refresh( + acc + ), }, ) ) @@ -488,7 +492,9 @@ class OTPPlugin(PluginBase): "type": "totp", "account": account_name, "code": totp_code, - "alt_action": lambda acc=account_name: self._remove_account_and_refresh(acc), + "alt_action": lambda acc=account_name: self._remove_account_and_refresh( + acc + ), }, ) ) @@ -606,7 +612,9 @@ class OTPPlugin(PluginBase): title=f"Scan QR Code for '{account_name}'", subtitle="Click to scan QR code from screen", icon_markup=icons.connect, - action=lambda name=account_name: self._scan_qr_and_add_account(name), + action=lambda name=account_name: self._scan_qr_and_add_account( + name + ), relevance=1.0, plugin_name=self.display_name, data={"type": "qr_scan"}, @@ -620,6 +628,7 @@ class OTPPlugin(PluginBase): # Run QR scanning asynchronously so launcher closes immediately import threading + thread = threading.Thread(target=self._scan_qr_async, args=(account_name,)) thread.daemon = True thread.start() @@ -688,14 +697,16 @@ class OTPPlugin(PluginBase): time_display } remaining", icon_markup=icons.trash, - action=lambda acc=acc_name: self._remove_account_and_refresh(acc), + action=lambda acc=acc_name: self._remove_account_and_refresh( + acc + ), relevance=0.9, plugin_name=self.display_name, data={ "type": "remove_instruction", "account": acc_name, "code": totp_code, - "keep_launcher_open": True + "keep_launcher_open": True, }, ) ) @@ -703,9 +714,11 @@ class OTPPlugin(PluginBase): results.append( Result( title=f"Error: {acc_name}", - subtitle=f"Press Enter to remove (Invalid secret)", + subtitle="Press Enter to remove (Invalid secret)", icon_markup=icons.trash, - action=lambda acc=acc_name: self._remove_account_and_refresh(acc), + action=lambda acc=acc_name: self._remove_account_and_refresh( + acc + ), relevance=0.8, plugin_name=self.display_name, data={ @@ -746,7 +759,11 @@ class OTPPlugin(PluginBase): action=lambda acc=account_name: self._remove_account_and_refresh(acc), relevance=1.0, plugin_name=self.display_name, - data={"type": "remove_confirm", "account": account_name,"keep_launcher_open": True,}, + data={ + "type": "remove_confirm", + "account": account_name, + "keep_launcher_open": True, + }, ) ] @@ -779,7 +796,9 @@ class OTPPlugin(PluginBase): return [ Result( title=f"✓ Added '{account_name}'", - subtitle=f"OTP account added successfully (secret: {result['secret'][:4]}...)", + subtitle=f"OTP account added successfully (secret: { + result['secret'][:4] + }...)", icon_markup=icons.check, action=lambda: self._trigger_refresh(), relevance=1.0, @@ -814,7 +833,11 @@ class OTPPlugin(PluginBase): } self._save_secrets() - display_name = f"{result['issuer']} - {result['account_name']}" if result['issuer'] else result['account_name'] + display_name = ( + f"{result['issuer']} - {result['account_name']}" + if result["issuer"] + else result["account_name"] + ) return [ Result( title=f"✓ Added '{display_name}'", diff --git a/services/auth.py b/services/auth.py index c003288..3a7b070 100644 --- a/services/auth.py +++ b/services/auth.py @@ -1,13 +1,15 @@ -import subprocess -import time import json import os +import subprocess +import time from datetime import datetime +from pathlib import Path +from urllib.parse import parse_qs, urlparse + +import pyotp from PIL import Image from pyzbar.pyzbar import decode -import pyotp -from urllib.parse import urlparse, parse_qs -from pathlib import Path + import config.data as data @@ -197,9 +199,7 @@ def validate_base32_secret(secret: str) -> dict: try: # Clean up the secret - remove spaces, dashes, and convert to uppercase - clean_secret = ( - secret.replace(" ", "").replace("-", "").replace("_", "").upper() - ) + clean_secret = secret.replace(" ", "").replace("-", "").replace("_", "").upper() # Remove any non-base32 characters clean_secret = re.sub(r"[^A-Z2-7]", "", clean_secret) @@ -233,7 +233,10 @@ def parse_otpauth_uri(uri: str, account_name: str = "") -> dict: try: parsed = urlparse(uri) if parsed.scheme != "otpauth" or parsed.netloc != "totp": - return {"success": False, "error": "Only otpauth://totp/ URIs are supported"} + return { + "success": False, + "error": "Only otpauth://totp/ URIs are supported", + } if not account_name: account_path = parsed.path.lstrip("/") @@ -280,7 +283,10 @@ def scan_qr_and_add_account(account_name: str, secrets_file_path: str) -> dict: decoded_objects = decode(img) if not decoded_objects: - return {"success": False, "error": "No QR code detected in selected area"} + return { + "success": False, + "error": "No QR code detected in selected area", + } # Process the first QR code found qr_data = decoded_objects[0].data.decode("utf-8") @@ -316,14 +322,21 @@ def scan_qr_and_add_account(account_name: str, secrets_file_path: str) -> dict: with open(secrets_file_path, "w", encoding="utf-8") as f: json.dump(secrets, f, indent=2) except Exception as e: - return {"success": False, "error": f"Error saving secrets: {str(e)}"} + return { + "success": False, + "error": f"Error saving secrets: {str(e)}", + } - display_name = f"{result['issuer']} - {result['account_name']}" if result['issuer'] else result['account_name'] + display_name = ( + f"{result['issuer']} - {result['account_name']}" + if result["issuer"] + else result["account_name"] + ) return { "success": True, "account_name": result["account_name"], "display_name": display_name, - "message": f"Successfully added OTP account: {display_name}" + "message": f"Successfully added OTP account: {display_name}", } else: return {"success": False, "error": "QR code is not an otpauth URI"} diff --git a/services/qrcode.py b/services/qrcode.py deleted file mode 100644 index c003288..0000000 --- a/services/qrcode.py +++ /dev/null @@ -1,335 +0,0 @@ -import subprocess -import time -import json -import os -from datetime import datetime -from PIL import Image -from pyzbar.pyzbar import decode -import pyotp -from urllib.parse import urlparse, parse_qs -from pathlib import Path -import config.data as data - - -def get_otp_file_path(): - """Returns the path to the OTP file inside the cache directory.""" - cache_dir = Path(data.CACHE_DIR) / "otp" - # Create the directory if it doesn't exist - cache_dir.mkdir(parents=True, exist_ok=True) - - file_path = cache_dir / "otp_codes.json" - - # Create the file if it doesn't exist - if not file_path.exists(): - with open(file_path, "w") as f: - json.dump([], f) - print(f"File {file_path} created successfully!") - - return file_path - - -def capture_selected_area(filename="/tmp/screenshot.png"): - """ - Uses slurp to let the user select an area of the screen, - then captures that area using grim. - """ - try: - # slurp returns coordinates in format: x,y widthxheight (e.g. 100,200 300x400) - result = subprocess.run(["slurp"], check=True, capture_output=True, text=True) - geometry = result.stdout.strip() - if not geometry: - print("No area selected with slurp.") - return None - except subprocess.CalledProcessError as e: - print("Error selecting area with slurp:", e) - return None - - try: - # grim uses -g to capture a specific region - subprocess.run(["grim", "-g", geometry, filename], check=True) - except subprocess.CalledProcessError as e: - print("Error capturing screenshot with grim:", e) - return None - - return filename - - -def read_and_save_to_json(): - # Get the full path to the JSON file - json_file = get_otp_file_path() - - screenshot = capture_selected_area() - if screenshot is None: - print("Failed to capture the selected area.") - return False - - # Short delay to ensure the file is written - time.sleep(1) - - # Open the captured image - try: - img = Image.open(screenshot) - except Exception as e: - print("Error opening the image:", e) - return False - - # Decode QR Code(s) from the image - decoded_objects = decode(img) - if not decoded_objects: - print("No QR Code detected in the selected area.") - return False - - results = [] - - for obj in decoded_objects: - data = obj.data.decode("utf-8") - print("QR Code detected:", data) - - result_entry = {"timestamp": datetime.now().isoformat(), "qr_data": data} - - if data.startswith("otpauth://"): - parsed = urlparse(data) - query = parse_qs(parsed.query) - - # Extract the secret properly - secret = query.get("secret", [None])[0] - issuer_from_query = query.get("issuer", [None])[0] - - # Extract the label (path), which may contain issuer and account - label = parsed.path.lstrip("/") if parsed.path else "" - account_name = label - issuer_from_path = None - - if ":" in label: - parts = label.split(":", 1) - issuer_from_path = parts[0] - account_name = parts[1] - - # Prefer issuer from path if available, otherwise use from query - issuer = issuer_from_path or issuer_from_query - - # Extract the period (default is 30 seconds) - period = int(query.get("period", ["30"])[0]) - - # Create TOTP object with the correct interval - totp = pyotp.TOTP(secret, interval=period) - current_otp = totp.now() - print(f"Generated OTP: {current_otp} (valid for {period} seconds)") - - result_entry.update( - { - "type": "otp", - "secret": secret, - "issuer": issuer, - "account_name": account_name, - } - ) - else: - result_entry["type"] = "unknown" - print("Unrecognized format. Expected a URI like otpauth://") - return False - - results.append(result_entry) - - # Load existing data if the file exists - existing_data = [] - if os.path.exists(json_file): - try: - with open(json_file, "r") as f: - existing_data = json.load(f) - except json.JSONDecodeError: - print(f"Error reading the file {json_file}. Creating a new one.") - - # Append new results - existing_data.extend(results) - - # Save to JSON file - with open(json_file, "w") as f: - json.dump(existing_data, f, indent=4) - - print(f"OTP data saved to {json_file}") - return True - - -def CodeOTP(uri): - parsed = urlparse(uri) - query = parse_qs(parsed.query) - secret = query.get("secret", [None])[0] - - if secret is None: - return None - else: - totp = pyotp.TOTP(secret) - return totp.now() - - -# TOTP/OTP utility functions -def generate_totp(secret: str) -> str: - """Generate TOTP code from secret.""" - try: - return pyotp.TOTP(secret).now() - except Exception as e: - print(f"Error generating TOTP: {e}") - return None - - -def get_time_remaining() -> int: - """Get seconds remaining until next token refresh.""" - return 30 - (int(time.time()) % 30) - - -def get_time_remaining_with_blink() -> str: - """Get time remaining with blinking effect.""" - time_remaining = get_time_remaining() - current_second = int(time.time()) - should_blink = current_second % 2 == 0 - - if should_blink: - return f"{time_remaining}s" - else: - return f"{time_remaining}s" - - -def validate_base32_secret(secret: str) -> dict: - """Validate and clean Base32 secret.""" - import base64 - import re - - try: - # Clean up the secret - remove spaces, dashes, and convert to uppercase - clean_secret = ( - secret.replace(" ", "").replace("-", "").replace("_", "").upper() - ) - - # Remove any non-base32 characters - clean_secret = re.sub(r"[^A-Z2-7]", "", clean_secret) - - # Add padding if needed (Base32 requires padding to multiple of 8) - while len(clean_secret) % 8 != 0: - clean_secret += "=" - - # Validate Base32 format - try: - base64.b32decode(clean_secret) - except Exception as e: - return {"success": False, "error": f"Invalid Base32 secret: {str(e)}"} - - # Test if the secret can generate a valid TOTP - try: - test_totp = pyotp.TOTP(clean_secret) - test_code = test_totp.now() - if not test_code or len(test_code) != 6: - raise ValueError("Generated invalid TOTP code") - except Exception as e: - return {"success": False, "error": f"Cannot generate TOTP: {str(e)}"} - - return {"success": True, "secret": clean_secret} - except Exception as e: - return {"success": False, "error": f"Unexpected error: {str(e)}"} - - -def parse_otpauth_uri(uri: str, account_name: str = "") -> dict: - """Parse otpauth URI and extract account information.""" - try: - parsed = urlparse(uri) - if parsed.scheme != "otpauth" or parsed.netloc != "totp": - return {"success": False, "error": "Only otpauth://totp/ URIs are supported"} - - if not account_name: - account_path = parsed.path.lstrip("/") - if ":" in account_path: - issuer, extracted_name = account_path.split(":", 1) - account_name = extracted_name - else: - account_name = account_path - - params = parse_qs(parsed.query) - secret = params.get("secret", [""])[0] - issuer = params.get("issuer", [""])[0] - algorithm = params.get("algorithm", ["SHA1"])[0] - digits = int(params.get("digits", ["6"])[0]) - period = int(params.get("period", ["30"])[0]) - - if not secret: - return {"success": False, "error": "No secret found in URI"} - - return { - "success": True, - "account_name": account_name, - "secret": secret, - "issuer": issuer, - "algorithm": algorithm, - "digits": digits, - "period": period, - } - except Exception as e: - return {"success": False, "error": f"Error parsing otpauth URI: {str(e)}"} - - -def scan_qr_and_add_account(account_name: str, secrets_file_path: str) -> dict: - """Scan QR code and add OTP account to secrets file.""" - try: - # Capture QR code from screen - screenshot_path = capture_selected_area() - if not screenshot_path: - return {"success": False, "error": "QR scan cancelled or failed"} - - # Decode QR code - try: - img = Image.open(screenshot_path) - decoded_objects = decode(img) - - if not decoded_objects: - return {"success": False, "error": "No QR code detected in selected area"} - - # Process the first QR code found - qr_data = decoded_objects[0].data.decode("utf-8") - print(f"QR Code detected: {qr_data}") - - if qr_data.startswith("otpauth://"): - # Parse otpauth URI - result = parse_otpauth_uri(qr_data, account_name) - if not result["success"]: - return result - - # Load existing secrets - secrets = {} - if os.path.exists(secrets_file_path): - try: - with open(secrets_file_path, "r", encoding="utf-8") as f: - secrets = json.load(f) - except Exception as e: - print(f"Error loading secrets: {e}") - - # Add new account - secrets[result["account_name"]] = { - "secret": result["secret"], - "issuer": result["issuer"], - "algorithm": result["algorithm"], - "digits": result["digits"], - "period": result["period"], - } - - # Save secrets - try: - os.makedirs(os.path.dirname(secrets_file_path), exist_ok=True) - with open(secrets_file_path, "w", encoding="utf-8") as f: - json.dump(secrets, f, indent=2) - except Exception as e: - return {"success": False, "error": f"Error saving secrets: {str(e)}"} - - display_name = f"{result['issuer']} - {result['account_name']}" if result['issuer'] else result['account_name'] - return { - "success": True, - "account_name": result["account_name"], - "display_name": display_name, - "message": f"Successfully added OTP account: {display_name}" - } - else: - return {"success": False, "error": "QR code is not an otpauth URI"} - - except Exception as e: - return {"success": False, "error": f"Error processing QR code: {str(e)}"} - - except Exception as e: - return {"success": False, "error": f"Error during QR scan: {str(e)}"} diff --git a/styles/components.css b/styles/components.css index 9d0621f..96ad81a 100644 --- a/styles/components.css +++ b/styles/components.css @@ -12,6 +12,9 @@ color: var(--surface-bright); border: 3px solid var(--primary); } +#music-label{ + font-size: 16x; +} /* * ── language indicator ──────────────────────────────────────────────── diff --git a/utils/notification_utils.py b/utils/notification_utils.py index 30cbd8c..b15aec0 100644 --- a/utils/notification_utils.py +++ b/utils/notification_utils.py @@ -1,10 +1,6 @@ -""" -Common notification utilities and shared functionality -""" import json import locale import os -import uuid from datetime import datetime, timedelta from fabric.utils.helpers import get_relative_path @@ -31,7 +27,9 @@ _shared_notification_history = None class HistoricalNotification: """Data structure for historical notification data""" - def __init__(self, id, app_icon, summary, body, app_name, timestamp, cached_image_path=None): + def __init__( + self, id, app_icon, summary, body, app_name, timestamp, cached_image_path=None + ): self.id = id self.app_icon = app_icon self.summary = summary @@ -51,12 +49,17 @@ def cache_notification_pixbuf(notification_box): if notification.image_pixbuf: os.makedirs(PERSISTENT_DIR, exist_ok=True) cache_file = os.path.join( - PERSISTENT_DIR, f"{NOTIFICATION_IMAGE_PREFIX}{notification_box.uuid}{NOTIFICATION_IMAGE_SUFFIX}" + PERSISTENT_DIR, + f"{NOTIFICATION_IMAGE_PREFIX}{notification_box.uuid}{ + NOTIFICATION_IMAGE_SUFFIX + }", ) try: scaled = notification.image_pixbuf.scale_simple( - DEFAULT_NOTIFICATION_IMAGE_SIZE, DEFAULT_NOTIFICATION_IMAGE_SIZE, GdkPixbuf.InterpType.BILINEAR + DEFAULT_NOTIFICATION_IMAGE_SIZE, + DEFAULT_NOTIFICATION_IMAGE_SIZE, + GdkPixbuf.InterpType.BILINEAR, ) scaled.savev(cache_file, "png", [], []) return cache_file @@ -84,23 +87,33 @@ def load_scaled_pixbuf(notification_box, width, height): ): try: logger.debug( - f"Attempting to load cached image from: {notification_box.cached_image_path} " + f"Attempting to load cached image from: { + notification_box.cached_image_path + } " f"for notification {notification.id}" ) pixbuf = GdkPixbuf.Pixbuf.new_from_file(notification_box.cached_image_path) if pixbuf: - pixbuf = pixbuf.scale_simple(width, height, GdkPixbuf.InterpType.BILINEAR) + pixbuf = pixbuf.scale_simple( + width, height, GdkPixbuf.InterpType.BILINEAR + ) return pixbuf except Exception as e: logger.error( f"Error loading cached image from {notification_box.cached_image_path} " f"for notification {notification.id}: {e}" ) - logger.warning(f"Falling back to notification.image_pixbuf for notification {notification.id}") + logger.warning( + f"Falling back to notification.image_pixbuf for notification { + notification.id + }" + ) # Fall back to notification image_pixbuf if notification.image_pixbuf: - pixbuf = notification.image_pixbuf.scale_simple(width, height, GdkPixbuf.InterpType.BILINEAR) + pixbuf = notification.image_pixbuf.scale_simple( + width, height, GdkPixbuf.InterpType.BILINEAR + ) return pixbuf # Fall back to app icon @@ -130,8 +143,10 @@ def cleanup_orphan_cached_images(persistent_notifications): return cached_files = [ - f for f in os.listdir(PERSISTENT_DIR) - if f.startswith(NOTIFICATION_IMAGE_PREFIX) and f.endswith(NOTIFICATION_IMAGE_SUFFIX) + f + for f in os.listdir(PERSISTENT_DIR) + if f.startswith(NOTIFICATION_IMAGE_PREFIX) + and f.endswith(NOTIFICATION_IMAGE_SUFFIX) ] if not cached_files: return @@ -141,7 +156,9 @@ def cleanup_orphan_cached_images(persistent_notifications): } for cached_file in cached_files: try: - uuid_from_filename = cached_file[len(NOTIFICATION_IMAGE_PREFIX):-len(NOTIFICATION_IMAGE_SUFFIX)] + uuid_from_filename = cached_file[ + len(NOTIFICATION_IMAGE_PREFIX) : -len(NOTIFICATION_IMAGE_SUFFIX) + ] if uuid_from_filename not in history_uuids: cache_file_path = os.path.join(PERSISTENT_DIR, cached_file) os.remove(cache_file_path) @@ -256,7 +273,9 @@ def load_persistent_history(): try: with open(PERSISTENT_HISTORY_FILE, "r") as f: notifications = json.load(f) - logger.info(f"Loaded {len(notifications)} notifications from persistent history") + logger.info( + f"Loaded {len(notifications)} notifications from persistent history" + ) return notifications except Exception as e: logger.error(f"Error loading persistent history: {e}") @@ -363,13 +382,16 @@ class NotificationHistory(Box): try: original_count = len(self.persistent_notifications) self.persistent_notifications = [ - notif for notif in self.persistent_notifications + notif + for notif in self.persistent_notifications if notif.get("app_name") != app_name ] removed_count = original_count - len(self.persistent_notifications) if removed_count > 0: self._save_persistent_history() - logger.info(f"Cleared {removed_count} notifications for app: {app_name}") + logger.info( + f"Cleared {removed_count} notifications for app: {app_name}" + ) except Exception as e: print(f"Error clearing history for app {app_name}: {e}")