#!/usr/bin/env python3 """ app.py — OTP Watcher menu bar app (presentation layer only). Responsibilities (per SRP): - Render the menu bar icon and dropdown - React to OTP events from OTPService - Copy code to clipboard and send macOS notification - Open the Settings window as a subprocess Does NOT contain IMAP logic, config parsing, or OTP extraction — those live in otp_service.py and config.py respectively. Run with: python3 app.py """ import os import sys import queue import subprocess import threading from datetime import datetime import rumps sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import config as cfg_module from otp_service import OTPService from otp_watcher import load_seen_uids, save_seen_uids # ── Presentation helpers ────────────────────────────────────────────────────── def _copy_to_clipboard(text): subprocess.run(["pbcopy"], input=text.encode("utf-8"), check=True) def _send_notification(title, message, sound=True): sound_clause = ' sound name "Glass"' if sound else "" script = 'display notification "{}" with title "{}"{}'.format( message.replace('"', '\\"'), title.replace('"', '\\"'), sound_clause, ) subprocess.run(["osascript", "-e", script], check=False) # ── Menu bar app ────────────────────────────────────────────────────────────── class OTPWatcherApp(rumps.App): """ Presentation layer: renders the menu bar and reacts to OTP events. Business logic is delegated to OTPService instances. """ MAX_RECENT = 5 def __init__(self): _icon = os.path.join(os.path.dirname(os.path.abspath(__file__)), "icon.png") super().__init__( name="OTP Watcher", icon=_icon, template=True, # adapts to dark/light menu bar automatically quit_button=None, ) self.seen_uids = load_seen_uids() self._otp_queue = queue.Queue() self._recent = [] # [(code, app_name, time_str)] self._services = [] # List[OTPService] self._settings_proc = None self._cfg_mtime = 0 self._sound = True self._build_menu() self._reload() # ── Menu ────────────────────────────────────────────────────────────────── def _build_menu(self): self._status_item = rumps.MenuItem("Starting…") self._status_item.set_callback(None) # informational, not clickable self._recent_header = rumps.MenuItem("No recent OTPs") self._recent_header.set_callback(None) self.menu = [ self._status_item, None, self._recent_header, None, rumps.MenuItem("Settings…", callback=self._open_settings), rumps.MenuItem("Quit OTP Watcher", callback=self._quit), ] def _refresh_recent_menu(self): """Rebuild the Recent OTPs section after a new OTP arrives.""" # Remove stale recent items (identified by the ·-separator in their label) for key in list(self.menu.keys()): if " · " in key: del self.menu[key] if not self._recent: self._recent_header.title = "No recent OTPs" return self._recent_header.title = "Recent OTPs" for i, (code, app_name, time_str) in enumerate(self._recent): label = "{} · {} ({})".format(code, app_name, time_str) item = rumps.MenuItem( label, callback=lambda _, c=code: self._recopy(c), ) self.menu.insert_after("Recent OTPs", item) def _recopy(self, code): _copy_to_clipboard(code) rumps.notification("OTP Watcher", "", "{} copied again".format(code)) # ── Service lifecycle ───────────────────────────────────────────────────── def _reload(self): """Load config and (re)start OTPService instances.""" for svc in self._services: svc.stop() self._services = [] cfg = cfg_module.load() self._cfg_mtime = cfg_module.mtime() self._sound = cfg.get("sound_enabled", True) accounts = cfg.get("accounts", []) refresh_mins = cfg.get("idle_refresh_minutes", 20) if not accounts: self._status_item.title = "⚠️ No accounts — open Settings" return for acc in accounts: svc = OTPService( email_addr = acc["email"], app_password = acc["app_password"], seen_uids = self.seen_uids, on_otp = self._on_otp, idle_refresh_minutes = refresh_mins, ) svc.start() self._services.append(svc) labels = " ".join(a["email"].split("@")[0] for a in accounts) self._status_item.title = "Connecting… ({})".format(labels) def _on_otp(self, code, app_name, subject): """Called from OTPService threads — push to main thread via queue.""" self._otp_queue.put({ "code": code, "app_name": app_name, "time_str": datetime.now().strftime("%H:%M"), }) # ── Timers ──────────────────────────────────────────────────────────────── @rumps.timer(1) def _drain_queue(self, _): """Process OTP events on the main thread (safe for UI updates).""" while not self._otp_queue.empty(): event = self._otp_queue.get_nowait() code = event["code"] app_name = event["app_name"] time_str = event["time_str"] _copy_to_clipboard(code) _send_notification( title = "OTP for {}".format(app_name), message = "{} copied to clipboard".format(code), sound = self._sound, ) save_seen_uids(self.seen_uids) self._recent.insert(0, (code, app_name, time_str)) self._recent = self._recent[:self.MAX_RECENT] self._refresh_recent_menu() # Brief title flash to signal a new code, then clear self.title = code threading.Timer(4.0, lambda: setattr(self, "title", "")).start() print("[{}] OTP {} — {}".format(time_str, code, app_name)) @rumps.timer(5) def _watch_config(self, _): """Detect config.json changes (e.g. after Settings save) and reload.""" if cfg_module.mtime() != self._cfg_mtime: print("[app] Config changed — reloading services...") self._reload() @rumps.timer(15) def _refresh_status(self, _): """Keep the status line in sync with actual connection state.""" cfg = cfg_module.load() accounts = cfg.get("accounts", []) if not accounts: self._status_item.title = "⚠️ No accounts — open Settings" return alive = sum(1 for s in self._services if s.is_alive) labels = " ".join(a["email"].split("@")[0] for a in accounts) if alive == len(self._services) and alive > 0: self._status_item.title = "✓ Watching {}".format(labels) else: self._status_item.title = "⟳ Reconnecting… ({}/{})".format( alive, len(self._services)) # ── Actions ─────────────────────────────────────────────────────────────── @rumps.clicked("Settings…") def _open_settings(self, _): if self._settings_proc and self._settings_proc.poll() is None: return # already open — bring focus instead of opening another self._settings_proc = subprocess.Popen([ sys.executable, os.path.join(os.path.dirname(os.path.abspath(__file__)), "settings_window.py"), ]) @rumps.clicked("Quit OTP Watcher") def _quit(self, _): for svc in self._services: svc.stop() rumps.quit_application() # ── Entry point ─────────────────────────────────────────────────────────────── if __name__ == "__main__": # Menu bar only — no Dock icon (macOS accessory app policy) try: from AppKit import NSApp, NSApplicationActivationPolicyAccessory NSApp.setActivationPolicy_(NSApplicationActivationPolicyAccessory) except Exception: pass OTPWatcherApp().run()