import tkinter as tk from tkinter import ttk, simpledialog, messagebox, filedialog import json import os import db import recorder import playback import logging import requests import threading import time FONT = ('Segoe UI', 10) HEADER_FONT = ('Segoe UI', 14, 'bold') SIDEBAR_FONT = ('Segoe UI', 11, 'bold') # --- GUI Window Setup --- class ClickRecorderApp(tk.Tk): def __init__(self): super().__init__() self.title('AuditWhizz AutoExtract | Workpapers') self.is_minimized = True # Start minimized # Calculate vertical center for left side screen_height = self.winfo_screenheight() min_height = 280 # Height for minimized view min_y = int((screen_height - min_height) / 2) self.geometry(f'100x280+0+{min_y}') self.configure(bg='#e5e5e5') self.attributes('-topmost', True) self.resizable(False, False) self.playlists = self.load_playlists() self.selected_playlist = tk.StringVar() self.recording = False self.live_clicks = [] self.playback_clicks = [] self._main_frame = None self._min_frame = None self.status_var = tk.StringVar(value='Idle') # Ensure status_var always exists self.debug_window = None # Track debug window self._loader_popup = None # Track loader popup self.create_widgets() self.start_auto_extractor_watcher() def create_widgets(self): if self._main_frame: self._main_frame.destroy() if self._min_frame: self._min_frame.destroy() if self.is_minimized: self.create_minimized_view() else: self.create_full_view() def create_minimized_view(self): # Make minimized window smaller and vertical, remove playlist selection screen_height = self.winfo_screenheight() min_height = 280 min_y = int((screen_height - min_height) / 2) self.geometry(f'100x280+0+{min_y}') self._min_frame = tk.Frame(self, bg='#f8f9fa') self._min_frame.pack(fill='both', expand=True) # Blue indicator if playing if self.status_var.get() == 'Playing': indicator = tk.Frame(self._min_frame, bg='#228be6', height=6) indicator.pack(fill='x', side='top') # Logo from PIL import Image, ImageTk try: logo_img = Image.open('audit_whizz_logo_transparent.png').resize((28, 28), Image.ANTIALIAS) self.logo_photo = ImageTk.PhotoImage(logo_img) logo_label = tk.Label(self._min_frame, image=self.logo_photo, bg='#f8f9fa') logo_label.pack(pady=(10, 8)) except Exception as e: self.logo_photo = None # Smaller vertical button stack btn_frame = tk.Frame(self._min_frame, bg='#f8f9fa') btn_frame.pack(fill='both', expand=True) record_btn = tk.Button(btn_frame, text='⏺', font=('Segoe UI', 12), fg='#d6336c', bg='#fff', relief='flat', command=self.on_start_recording, width=2, height=1) record_btn.pack(pady=2, fill='x') play_btn = tk.Button(btn_frame, text='▶', font=('Segoe UI', 12), fg='#228be6', bg='#fff', relief='flat', command=self.on_play_playlist, width=2, height=1) play_btn.pack(pady=2, fill='x') stop_btn = tk.Button(btn_frame, text='⏹', font=('Segoe UI', 12), fg='#495057', bg='#fff', relief='flat', command=self.on_stop_recording, width=2, height=1) stop_btn.pack(pady=2, fill='x') screenshot_btn = tk.Button(btn_frame, text='📸', font=('Segoe UI', 11), fg='#495057', bg='#fff', relief='flat', command=self.on_manual_screenshot, width=2, height=1) screenshot_btn.pack(pady=2, fill='x') maximize_btn = tk.Button(btn_frame, text='⤡', font=('Segoe UI', 10), fg='#495057', bg='#fff', relief='flat', command=self.toggle_view, width=2, height=1) maximize_btn.pack(pady=2, fill='x') # Tooltips self.add_tooltip(record_btn, 'Start Recording') self.add_tooltip(play_btn, 'Play Playlist') self.add_tooltip(stop_btn, 'Stop Recording/Playback') self.add_tooltip(screenshot_btn, 'Manual Screenshot') self.add_tooltip(maximize_btn, 'Maximize') def create_full_view(self): # Auto-scale height if debug view is open if self.debug_window and self.debug_window.winfo_exists(): full_height = 600 else: full_height = 400 screen_height = self.winfo_screenheight() full_y = int((screen_height - full_height) / 2) self.geometry(f'700x{full_height}+0+{full_y}') self._main_frame = tk.Frame(self, bg='#e5e5e5') self._main_frame.pack(fill='both', expand=True) # Logo at top left from PIL import Image, ImageTk try: logo_img = Image.open('audit_whizz_logo_transparent.png').resize((40, 40), Image.ANTIALIAS) self.logo_photo_full = ImageTk.PhotoImage(logo_img) logo_label = tk.Label(self._main_frame, image=self.logo_photo_full, bg='#e5e5e5') logo_label.place(x=10, y=10) except Exception as e: self.logo_photo_full = None # Card for playlist management card = tk.Frame(self._main_frame, bg='#fff', bd=0, highlightthickness=1, highlightbackground='#ddd') card.pack(padx=30, pady=10, fill='x') card.grid_columnconfigure(0, weight=1) card.grid_columnconfigure(1, weight=1) card.grid_columnconfigure(2, weight=1) # Recording name, playlist, and wait time fields fields_frame = tk.Frame(card, bg='#fff') fields_frame.pack(fill='x', padx=10, pady=(10, 0)) tk.Label(fields_frame, text='Recording Name:', font=FONT, bg='#fff', fg='#222').pack(side='left') self.recording_name_var = tk.StringVar() self.recording_name_entry = ttk.Entry(fields_frame, textvariable=self.recording_name_var, width=20, font=FONT) self.recording_name_entry.pack(side='left', padx=5) tk.Label(fields_frame, text='Playlist:', font=FONT, bg='#fff', fg='#222').pack(side='left', padx=(20, 0)) self.playlist_dropdown = ttk.Combobox(fields_frame, textvariable=self.selected_playlist, values=self.playlists, state='readonly', width=18, font=FONT) self.playlist_dropdown.pack(side='left', padx=5) self.playlist_dropdown.set('Select playlist') self.playlist_dropdown.bind('<>', self.on_playlist_selected) # Wait time input tk.Label(fields_frame, text='Wait (s):', font=FONT, bg='#fff', fg='#222').pack(side='left', padx=(20, 0)) self.wait_time_var = tk.StringVar(value='0') self.wait_time_entry = ttk.Entry(fields_frame, textvariable=self.wait_time_var, width=5, font=FONT) self.wait_time_entry.pack(side='left', padx=5) # Multi-row action buttons (max 4 per row) btn_texts_cmds = [ ('Record', self.on_start_recording, 'Dark.TButton', 'Record'), ('Stop', self.on_stop_recording, 'Dark.TButton', 'Stop'), ('Play', self.on_play_playlist, 'Dark.TButton', 'Play'), ('Save', self.on_save_playlist, 'Dark.TButton', 'Save Playlist'), ('Screenshot', self.on_manual_screenshot, 'Dark.TButton', 'Screenshot'), ('Export', self.on_export_json, 'Dark.TButton', 'Export Playlist'), ('Delete', self.on_delete_playlist_supabase, 'Red.TButton', 'Delete Playlist from Supabase'), # ('Debug', self.toggle_debug_view, 'Dark.TButton', 'Show Supabase Debug'), # Debug button disabled ('Check Queue', self.check_auto_extractor_queue_once, 'Dark.TButton', 'Check Auto Extractor Queue'), ('⤢', self.toggle_view, 'Dark.TButton', 'Minimize/Maximize'), ] btn_rows = [] for i in range(0, len(btn_texts_cmds), 4): btn_rows.append(btn_texts_cmds[i:i+4]) for row in btn_rows: row_frame = tk.Frame(card, bg='#fff') row_frame.pack(fill='x', padx=10, pady=(5, 0)) for text, cmd, style, tooltip in row: btn = ttk.Button(row_frame, text=text, command=cmd, style=style) btn.pack(side='left', padx=6, pady=2) self.add_tooltip(btn, tooltip) # Status label self.status_var = tk.StringVar(value='Idle') tk.Label(card, textvariable=self.status_var, font=FONT, bg='#fff', fg='#007bff').pack(anchor='e', padx=10) # Live clicks view (card style) live_card = tk.Frame(self._main_frame, bg='#fff', bd=0, highlightthickness=1, highlightbackground='#ddd') live_card.pack(padx=30, pady=(0, 10), fill='x') tk.Label(live_card, text='Live Clicks', font=FONT, bg='#fff', fg='#222').pack(anchor='w', padx=10, pady=(10, 0)) self.live_clicks_text = tk.Text(live_card, height=5, bg='#f8f9fa', fg='#222', font=FONT, state='disabled', relief='flat', bd=0) self.live_clicks_text.pack(padx=10, pady=(0, 10), fill='x') # Add Debug button # debug_btn = ttk.Button(self._main_frame, text='Debug', command=self.toggle_debug_view, style='Dark.TButton') # debug_btn.pack(anchor='ne', padx=10, pady=5) # Custom styles style = ttk.Style(self) style.theme_use('clam') style.configure('TButton', font=FONT, padding=6, relief='flat') style.configure('Dark.TButton', background='#495057', foreground='#fff') style.map('Dark.TButton', background=[('active', '#343a40')], foreground=[('active', '#fff')]) style.configure('Red.TButton', background='#dc3545', foreground='#fff') style.map('Red.TButton', background=[('active', '#b52a37')], foreground=[('active', '#fff')]) style.configure('TCombobox', fieldbackground='#fff', background='#fff', foreground='#222') def toggle_view(self): self.is_minimized = not self.is_minimized self.create_widgets() def append_live_click(self, x, y): if not self.recording: return if not hasattr(self, 'live_clicks_text') or not self.live_clicks_text.winfo_exists(): return self.live_clicks.append((x, y)) self.live_clicks_text.config(state='normal') self.live_clicks_text.insert('end', f'Click: x={x}, y={y}\n') self.live_clicks_text.see('end') self.live_clicks_text.config(state='disabled') def show_playback_click(self, x, y, idx=None): if not hasattr(self, 'live_clicks_text') or not self.live_clicks_text.winfo_exists(): return self.live_clicks_text.config(state='normal') self.live_clicks_text.delete('1.0', 'end') if hasattr(self, 'playback_clicks') and self.playback_clicks: for i, (cx, cy, ts) in enumerate(self.playback_clicks): if idx is not None and i == idx: self.live_clicks_text.insert('end', f'>> Click {i+1}: x={cx}, y={cy}, t={ts:.2f}s <<\n', 'highlight') else: self.live_clicks_text.insert('end', f'Click {i+1}: x={cx}, y={cy}, t={ts:.2f}s\n') self.live_clicks_text.tag_config('highlight', background='#ffe066', foreground='#222') else: self.live_clicks_text.insert('end', f'Playing: x={x}, y={y}\n') self.live_clicks_text.see('end') self.live_clicks_text.config(state='disabled') def clear_live_clicks(self): if not hasattr(self, 'live_clicks_text') or not self.live_clicks_text.winfo_exists(): return self.live_clicks = [] self.live_clicks_text.config(state='normal') self.live_clicks_text.delete('1.0', 'end') self.live_clicks_text.config(state='disabled') def append_live_key(self, key, event_type): if not self.recording: return if not hasattr(self, 'live_clicks_text') or not self.live_clicks_text.winfo_exists(): return self.live_clicks_text.config(state='normal') self.live_clicks_text.insert('end', f'Key {event_type}: {key}\n') self.live_clicks_text.see('end') self.live_clicks_text.config(state='disabled') # --- Playlist Data Loading --- def load_playlists(self): # Load playlist names from Supabase only (SQLite disabled) try: from supabase_client import get_playlist_names_from_supabase return get_playlist_names_from_supabase() except Exception as e: import logging logging.error(f"Failed to load playlists from Supabase: {e}") return [] def update_playlist_dropdown_selection(self, playlist_name): """Update the dropdown's values and set the selection to playlist_name.""" self.playlists = self.load_playlists() if hasattr(self, 'playlist_dropdown') and self.playlist_dropdown: self.playlist_dropdown['values'] = self.playlists if playlist_name in self.playlists: self.selected_playlist.set(playlist_name) self.playlist_dropdown.set(playlist_name) else: self.selected_playlist.set('Select playlist') self.playlist_dropdown.set('Select playlist') # --- Button Callbacks (to be implemented) --- def on_start_recording(self): self.status_var.set('Recording') self.clear_live_clicks() self.recording = True pl_name = self.selected_playlist.get() if not pl_name or pl_name == 'Select playlist': # messagebox.showinfo('Start Recording', 'Please select a playlist to record to.') self.status_var.set('Idle') self.recording = False return # Get playlist_id from DB conn = db.get_connection() cur = conn.cursor() cur.execute('SELECT id FROM Playlists WHERE name = ?', (pl_name,)) row = cur.fetchone() conn.close() if not row: # messagebox.showerror('Start Recording', 'Playlist not found in database.') self.status_var.set('Idle') self.recording = False return playlist_id = row[0] # Start recording and pass the live click and key callbacks import recorder recorder.start_recording(playlist_id, on_click_callback=self.append_live_click, on_key_callback=self.append_live_key) # On save, show a summary of both clicks and keyboard events in the live clicks view def on_stop_recording(self): self.status_var.set('Idle') self.recording = False import recorder recorder.stop_recording() import logging logging.basicConfig(filename='app_debug.log', level=logging.INFO, format='%(asctime)s %(levelname)s: %(message)s') # Show summary in live clicks view and save actions to Supabase try: conn = db.get_connection() cur = conn.cursor() pl_name = self.selected_playlist.get() cur.execute('SELECT id FROM Playlists WHERE name = ?', (pl_name,)) row = cur.fetchone() if row: playlist_id = row[0] cur.execute('SELECT x, y, timestamp FROM Clicks WHERE playlist_id = ? ORDER BY timestamp ASC', (playlist_id,)) clicks = cur.fetchall() cur.execute('SELECT key, event_type, timestamp FROM KeyboardEvents WHERE playlist_id = ? ORDER BY timestamp ASC', (playlist_id,)) keys = cur.fetchall() self.live_clicks_text.config(state='normal') self.live_clicks_text.delete('1.0', 'end') for i, (x, y, ts) in enumerate(clicks): self.live_clicks_text.insert('end', f'Click {i+1}: x={x}, y={y}, t={ts:.2f}s\n') for i, (key, event_type, ts) in enumerate(keys): self.live_clicks_text.insert('end', f'Key {event_type}: {key}, t={ts:.2f}s\n') self.live_clicks_text.see('end') self.live_clicks_text.config(state='disabled') # Save actions to Supabase try: from supabase_client import save_action_to_supabase # Save clicks for x, y, ts in clicks: try: save_action_to_supabase(playlist_id, 'click', x, y, None, ts, pl_name) except Exception as e: logging.error(f"Failed to save click to Supabase: {e}") # Save keyboard events for key, event_type, ts in keys: try: action_type = 'key_press' if event_type == 'press' else 'key_release' save_action_to_supabase(playlist_id, action_type, None, None, key, ts, pl_name) except Exception as e: logging.error(f"Failed to save key event to Supabase: {e}") # messagebox.showinfo('Supabase', 'Actions saved to Supabase!') except Exception as e: logging.error(f"Failed to save actions to Supabase: {e}") # messagebox.showerror('Supabase', f'Failed to save actions to Supabase: {e}') conn.close() except Exception as e: logging.error(f"Error in on_stop_recording: {e}") # messagebox.showinfo('Recording Saved', 'Recording has been saved successfully.') def on_play_playlist(self): import pyautogui import threading self.status_var.set('Playing') self.clear_live_clicks() pl_name = self.selected_playlist.get() if not pl_name or pl_name == 'Select playlist': # messagebox.showinfo('Play Playlist', 'Please select a playlist to play.') self.status_var.set('Idle') return # Mapping for special keys from pynput to pyautogui special_key_map = { 'Key.enter': 'enter', 'Key.caps_lock': 'capslock', 'Key.tab': 'tab', 'Key.shift': 'shift', 'Key.shift_r': 'shift', 'Key.ctrl_l': 'ctrl', 'Key.ctrl_r': 'ctrl', 'Key.alt_l': 'alt', 'Key.alt_r': 'alt', 'Key.esc': 'esc', 'Key.backspace': 'backspace', 'Key.space': 'space', 'Key.delete': 'delete', 'Key.up': 'up', 'Key.down': 'down', 'Key.left': 'left', 'Key.right': 'right', # Add more as needed } # Fetch actions from Supabase (disable SQLite logic) try: from supabase_client import supabase res = supabase.table('actions').select('*').eq('playlist_name', pl_name).order('timestamp', desc=False).execute() if not hasattr(res, 'data') or not res.data: # messagebox.showinfo('Play Playlist', f'No actions found for playlist "{pl_name}" in Supabase.') self.status_var.set('Idle') return actions = res.data # Playback thread to avoid blocking UI def playback(): prev_time = 0 for act in actions: if self.status_var.get() != 'Playing': break interval = act.get('timestamp', 0) - prev_time if interval > 0: import time time.sleep(interval) if act.get('action_type') == 'click': x, y = act.get('x'), act.get('y') pyautogui.click(x, y) self.show_playback_click(x, y) elif act.get('action_type') in ('key_press', 'key_release'): key = act.get('key') # Map special keys to pyautogui names key = special_key_map.get(key, key) if act.get('action_type') == 'key_press': pyautogui.keyDown(key) else: pyautogui.keyUp(key) prev_time = act.get('timestamp', 0) self.status_var.set('Idle') threading.Thread(target=playback, daemon=True).start() except Exception as e: import logging logging.error(f"Error fetching or playing actions from Supabase: {e}") # messagebox.showerror('Play Playlist', f'Error fetching or playing actions from Supabase:\n{e}') self.status_var.set('Idle') def on_pause_playlist(self): self.status_var.set('Paused') # TODO: Call playback.pause_playback def on_add_playlist(self): import logging logging.basicConfig(filename='app_debug.log', level=logging.INFO, format='%(asctime)s %(levelname)s: %(message)s') name = simpledialog.askstring('Add Playlist', 'Enter playlist name:') if name: try: from supabase_client import playlist_name_exists_in_supabase, save_playlist_to_supabase except Exception as import_e: logging.error(f"Failed to import supabase_client: {import_e}") # messagebox.showerror('Supabase Import', f'Failed to import supabase_client: {import_e}') return # Cloud-only: check for duplicate in Supabase self.show_loader('Checking playlist in Supabase...') if playlist_name_exists_in_supabase(name): self.hide_loader() # messagebox.showerror('Add Playlist', f'A playlist named "{name}" already exists in Supabase.') return # Cloud-only: insert into Supabase try: self.show_loader('Saving playlist to Supabase...') result = save_playlist_to_supabase(name) self.hide_loader() logging.info(f"Supabase insert result: {result}") # messagebox.showinfo('Supabase', f'Playlist "{name}" saved to Supabase!') except Exception as call_e: self.hide_loader() logging.error(f"Failed to call save_playlist_to_supabase: {call_e}") # messagebox.showerror('Supabase', f'Failed to save playlist to Supabase: {call_e}') return self.playlists.append(name) self.playlist_dropdown['values'] = self.playlists self.selected_playlist.set(name) def on_edit_playlist(self): if not self.selected_playlist.get() or self.selected_playlist.get() == 'Select playlist': messagebox.showinfo('Edit Playlist', 'Please select a playlist to edit.') return new_name = simpledialog.askstring('Edit Playlist', 'Enter new playlist name:', initialvalue=self.selected_playlist.get()) if new_name: # Update in DB conn = db.get_connection() cur = conn.cursor() cur.execute('UPDATE Playlists SET name = ? WHERE name = ?', (new_name, self.selected_playlist.get())) conn.commit() conn.close() idx = self.playlists.index(self.selected_playlist.get()) self.playlists[idx] = new_name self.playlist_dropdown['values'] = self.playlists self.selected_playlist.set(new_name) def on_delete_playlist(self): if not self.selected_playlist.get() or self.selected_playlist.get() == 'Select playlist': messagebox.showinfo('Delete Playlist', 'Please select a playlist to delete.') return if messagebox.askyesno('Delete Playlist', f'Delete playlist "{self.selected_playlist.get()}"?'): # Delete from DB conn = db.get_connection() cur = conn.cursor() cur.execute('DELETE FROM Playlists WHERE name = ?', (self.selected_playlist.get(),)) conn.commit() conn.close() self.playlists.remove(self.selected_playlist.get()) self.playlist_dropdown['values'] = self.playlists self.selected_playlist.set('Select playlist') def analyze_screenshot_with_openai(self, screenshot_path, user_prompt="Describe this screenshot."): import base64 import os import requests api_key = "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" if not api_key: return 'OpenAI API key not set.' with open(screenshot_path, 'rb') as f: img_bytes = f.read() img_b64 = base64.b64encode(img_bytes).decode('utf-8') image_url = f"data:image/png;base64,{img_b64}" headers = { 'Authorization': f'Bearer {api_key}', 'Content-Type': 'application/json' } data = { "model": "gpt-4o", "messages": [ { "role": "user", "content": [ {"type": "text", "text": user_prompt}, {"type": "image_url", "image_url": {"url": image_url}} ] } ], "max_tokens": 1000 } try: response = requests.post( "https://api.openai.com/v1/chat/completions", headers=headers, json=data, timeout=60 ) response.raise_for_status() result = response.json() return result['choices'][0]['message']['content'] except Exception as e: return f"OpenAI API error: {e}" def on_manual_screenshot(self): pl_name = self.selected_playlist.get() if not pl_name or pl_name == 'Select playlist': # messagebox.showinfo('Manual Screenshot', 'Please select a playlist to save the screenshot to.') return try: from supabase_client import supabase # Fetch playlist ID from Supabase self.show_loader('Fetching playlist from Supabase...') res = supabase.table('playlists').select('id').eq('name', pl_name).limit(1).execute() if not hasattr(res, 'data') or not res.data or 'id' not in res.data[0]: self.hide_loader() # messagebox.showerror('Manual Screenshot', f'Playlist "{pl_name}" not found in Supabase.') return playlist_id = res.data[0]['id'] import recorder try: self.show_loader('Capturing screenshot...') screenshot_path = recorder.capture_screenshot(playlist_id, is_manual=True, playlist_name=pl_name) self.hide_loader() # messagebox.showinfo('Manual Screenshot', 'Manual screenshot captured and saved!') # Analyze screenshot with OpenAI self.show_loader('Analyzing screenshot with OpenAI...') result = self.analyze_screenshot_with_openai(screenshot_path) self.hide_loader() # self.show_text_popup('OpenAI Screenshot Analysis', result) # Save analysis to Supabase try: self.show_loader('Saving analysis to Supabase...') from supabase_client import save_screenshot_analysis_to_supabase save_result = save_screenshot_analysis_to_supabase(pl_name, screenshot_path, result) self.hide_loader() if not save_result or (hasattr(save_result, 'error') and save_result.error): # messagebox.showerror('Supabase', 'Failed to save screenshot analysis to Supabase.') pass except Exception as e: self.hide_loader() # messagebox.showerror('Supabase', f'Error saving screenshot analysis to Supabase:\n{e}') pass except Exception as e: self.hide_loader() # messagebox.showerror('Manual Screenshot', f'Failed to capture screenshot:\n{e}') pass except Exception as e: self.hide_loader() # messagebox.showerror('Manual Screenshot', f'Error fetching playlist from Supabase:\n{e}') pass def on_export_json(self): # Export the selected playlist to JSON pl_name = self.selected_playlist.get() if not pl_name or pl_name == 'Select playlist': messagebox.showinfo('Export JSON', 'Please select a playlist to export.') return # Fetch playlist info from DB conn = db.get_connection() cur = conn.cursor() cur.execute('SELECT id, name, created_date FROM Playlists WHERE name = ?', (pl_name,)) row = cur.fetchone() if not row: messagebox.showerror('Export JSON', 'Playlist not found in database.') return playlist_id, name, created_date = row # Fetch clicks cur.execute('SELECT x, y, timestamp FROM Clicks WHERE playlist_id = ? ORDER BY timestamp ASC', (playlist_id,)) clicks = [{'x': x, 'y': y, 'timestamp': ts} for x, y, ts in cur.fetchall()] # Fetch screenshots cur.execute('SELECT path, is_manual FROM Screenshots WHERE playlist_id = ?', (playlist_id,)) screenshots = [{'path': path, 'is_manual': bool(is_manual)} for path, is_manual in cur.fetchall()] conn.close() # Prepare JSON data data = { 'name': name, 'created_date': created_date, 'clicks': clicks, 'screenshots': screenshots } # Ask user for file path os.makedirs('exports', exist_ok=True) default_path = os.path.join('exports', f'{name}.json') file_path = filedialog.asksaveasfilename(defaultextension='.json', initialfile=f'{name}.json', initialdir='exports', filetypes=[('JSON Files', '*.json')]) if not file_path: return with open(file_path, 'w', encoding='utf-8') as f: json.dump(data, f, indent=2) messagebox.showinfo('Export JSON', f'Exported playlist to {file_path}') def on_show_clicks(self): pl_name = self.selected_playlist.get() self.clear_live_clicks() if not pl_name or pl_name == 'Select playlist': messagebox.showinfo('Show Clicks', 'Please select a playlist.') return conn = db.get_connection() cur = conn.cursor() cur.execute('SELECT id FROM Playlists WHERE name = ?', (pl_name,)) row = cur.fetchone() if not row: messagebox.showerror('Show Clicks', 'Playlist not found in database.') return playlist_id = row[0] cur.execute('SELECT x, y, timestamp FROM Clicks WHERE playlist_id = ? ORDER BY timestamp ASC', (playlist_id,)) clicks = cur.fetchall() conn.close() if not clicks: self.live_clicks_text.config(state='normal') self.live_clicks_text.insert('end', 'No clicks recorded for this playlist.\n') self.live_clicks_text.config(state='disabled') return # Show clicks in the live view self.live_clicks_text.config(state='normal') for i, (x, y, timestamp) in enumerate(clicks): self.live_clicks_text.insert('end', f'Click {i+1}: x={x}, y={y}, t={timestamp:.2f}s\n') self.live_clicks_text.see('end') self.live_clicks_text.config(state='disabled') def on_playlist_selected(self, event=None): pl_name = self.selected_playlist.get() self.clear_live_clicks() if not pl_name or pl_name == 'Select playlist': return if not hasattr(self, 'live_clicks_text') or not self.live_clicks_text.winfo_exists(): return # Fetch actions from Supabase instead of SQLite self.live_clicks_text.config(state='normal') self.live_clicks_text.delete('1.0', 'end') try: from supabase_client import supabase res = supabase.table('actions').select('*').eq('playlist_name', pl_name).order('timestamp', desc=False).execute() if hasattr(res, 'data') and res.data: for i, act in enumerate(res.data): if act.get('action_type') == 'click': self.live_clicks_text.insert('end', f"Click {i+1}: x={act.get('x')}, y={act.get('y')}, t={act.get('timestamp'):.2f}s\n") elif act.get('action_type') in ('key_press', 'key_release'): self.live_clicks_text.insert('end', f"Key {act.get('action_type').replace('key_', '')}: {act.get('key')}, t={act.get('timestamp'):.2f}s\n") self.live_clicks_text.see('end') else: self.live_clicks_text.insert('end', f'No actions found for playlist "{pl_name}" in Supabase.\n') except Exception as e: self.live_clicks_text.insert('end', f'Error fetching actions from Supabase:\n{e}\n') self.live_clicks_text.config(state='disabled') # Disabled: Local SQLite fetch logic for actions # conn = db.get_connection() # cur = conn.cursor() # cur.execute('SELECT id FROM Playlists WHERE name = ?', (pl_name,)) # row = cur.fetchone() # if not row: # conn.close() # return # playlist_id = row[0] # cur.execute('SELECT x, y, timestamp FROM Clicks WHERE playlist_id = ? ORDER BY timestamp ASC', (playlist_id,)) # clicks = cur.fetchall() # conn.close() # if not clicks: # self.live_clicks_text.config(state='normal') # self.live_clicks_text.insert('end', 'No clicks recorded for this playlist.\n') # self.live_clicks_text.config(state='disabled') # return # self.live_clicks_text.config(state='normal') # for i, (x, y, timestamp) in enumerate(clicks): # self.live_clicks_text.insert('end', f'Click {i+1}: x={x}, y={y}, t={timestamp:.2f}s\n') # self.live_clicks_text.see('end') # self.live_clicks_text.config(state='disabled') def on_save_playlist(self): import logging logging.basicConfig(filename='app_debug.log', level=logging.INFO, format='%(asctime)s %(levelname)s: %(message)s') name = self.recording_name_var.get().strip() if not name: from tkinter import messagebox messagebox.showinfo('Save Playlist', 'Please enter a recording name.') return # Check for duplicate conn = db.get_connection() cur = conn.cursor() cur.execute('SELECT id FROM Playlists WHERE name = ?', (name,)) if cur.fetchone(): conn.close() from tkinter import messagebox messagebox.showerror('Save Playlist', f'A playlist named "{name}" already exists.') return # Insert new playlist from datetime import datetime cur.execute('INSERT INTO Playlists (name, created_date) VALUES (?, ?)', (name, datetime.now().isoformat())) conn.commit() conn.close() # Insert into Supabase try: logging.info(f"Attempting to import and call save_playlist_to_supabase for playlist: {name}") from tkinter import messagebox try: from supabase_client import save_playlist_to_supabase except Exception as import_e: logging.error(f"Failed to import save_playlist_to_supabase: {import_e}") messagebox.showerror('Supabase Import', f'Failed to import supabase_client: {import_e}') return try: result = save_playlist_to_supabase(name) logging.info(f"Supabase insert result: {result}") # messagebox.showinfo('Supabase', f'Playlist "{name}" saved to Supabase!') except Exception as call_e: logging.error(f"Failed to call save_playlist_to_supabase: {call_e}") messagebox.showerror('Supabase', f'Failed to save playlist to Supabase: {call_e}') except Exception as e: logging.error(f"Unexpected error in Supabase block: {e}") from tkinter import messagebox messagebox.showerror('Supabase', f'Unexpected error: {e}') # Update playlists dropdown self.playlists.append(name) self.playlist_dropdown['values'] = self.playlists self.selected_playlist.set(name) self.status_var.set(f'Playlist "{name}" saved.') from tkinter import messagebox messagebox.showinfo('Save Playlist', f'Playlist "{name}" has been saved.') def on_delete_playlist_supabase(self): from tkinter import messagebox pl_name = self.selected_playlist.get() if not pl_name or pl_name == 'Select playlist': # messagebox.showinfo('Delete Playlist', 'Please select a playlist to delete.') return # if not messagebox.askyesno('Delete Playlist', f'Delete playlist "{pl_name}" and all its actions from Supabase?'): # return try: self.show_loader('Deleting playlist from Supabase...') from supabase_client import supabase # Delete actions first res_actions = supabase.table('actions').delete().eq('playlist_name', pl_name).execute() # Delete playlist res_playlist = supabase.table('playlists').delete().eq('name', pl_name).execute() self.hide_loader() # Remove from local dropdown if pl_name in self.playlists: self.playlists.remove(pl_name) self.playlist_dropdown['values'] = self.playlists self.selected_playlist.set('Select playlist') # messagebox.showinfo('Delete Playlist', f'Playlist "{pl_name}" and its actions deleted from Supabase.') except Exception as e: self.hide_loader() # messagebox.showerror('Delete Playlist', f'Failed to delete playlist from Supabase:\n{e}') pass def show_text_popup(self, title, text): # Disabled all text popups return def add_tooltip(self, widget, text): def on_enter(event): self.tooltip = tk.Toplevel(widget) self.tooltip.wm_overrideredirect(True) x = widget.winfo_rootx() + 20 y = widget.winfo_rooty() + 30 self.tooltip.wm_geometry(f'+{x}+{y}') label = tk.Label(self.tooltip, text=text, bg='#222', fg='#fff', font=('Segoe UI', 9), padx=6, pady=2, relief='solid', borderwidth=1) label.pack() def on_leave(event): if hasattr(self, 'tooltip'): self.tooltip.destroy() self.tooltip = None widget.bind('', on_enter) widget.bind('', on_leave) def toggle_debug_view(self): # Debug view disabled return def start_auto_extractor_watcher(self): def watcher(): while True: time.sleep(10) # Only run if app is idle (not recording or playing) if self.status_var.get() == 'Idle' and not self.recording: try: self.status_var.set('Checking queue...') from supabase_client import get_next_auto_extractor_job, mark_auto_extractor_job_processed job = get_next_auto_extractor_job() if job: playlist_name = job.get('playlist_name') job_id = job.get('id') if playlist_name: # Always update dropdown and selection using helper self.create_widgets() # Ensure widgets are up to date self.update_playlist_dropdown_selection(playlist_name) # Wait for UI update time.sleep(1) # Trigger playback self.after(100, self.on_play_playlist) # Wait for playback to finish while self.status_var.get() == 'Playing': time.sleep(2) # Mark job as processed mark_auto_extractor_job_processed(job_id) # Set status back to Idle if not playing or recording if self.status_var.get() != 'Playing' and not self.recording: self.status_var.set('Idle') except Exception as e: import logging logging.error(f'Auto extractor watcher error: {e}') if self.status_var.get() != 'Playing' and not self.recording: self.status_var.set('Idle') threading.Thread(target=watcher, daemon=True).start() def check_auto_extractor_queue_once(self): # Manual queue check with popup result self.show_loader('Checking queue...') try: from supabase_client import get_next_auto_extractor_job, mark_auto_extractor_job_processed job = get_next_auto_extractor_job() if job: playlist_name = job.get('playlist_name') job_id = job.get('id') # self.show_text_popup('Queue Job Found', f"Job ID: {job_id}\nPlaylist: {playlist_name}\nProcessed: {job.get('processed')}\nCreated: {job.get('created_at')}") # Set playlist selection self.selected_playlist.set(playlist_name) self.create_widgets() # Refresh UI to update dropdown # Wait for UI update time.sleep(1) # Trigger playback self.after(100, self.on_play_playlist) # Wait for playback to finish while self.status_var.get() == 'Playing': time.sleep(2) mark_auto_extractor_job_processed(job_id) else: # self.show_text_popup('Queue Check', 'No unprocessed jobs found in the queue.') pass except Exception as e: # self.show_text_popup('Queue Check Error', f'Error checking queue: {e}') pass finally: self.hide_loader() def show_loader(self, message='Processing...'): if hasattr(self, '_loader_popup') and self._loader_popup and self._loader_popup.winfo_exists(): return # Already showing self._loader_popup = tk.Toplevel(self) self._loader_popup.title('Please wait') self._loader_popup.geometry('220x80') self._loader_popup.configure(bg='#f0f0f0') self._loader_popup.transient(self) self._loader_popup.grab_set() self._loader_popup.resizable(False, False) label = tk.Label(self._loader_popup, text=message, font=('Segoe UI', 11), bg='#f0f0f0', fg='#007bff') label.pack(pady=20) self._loader_popup.update() def hide_loader(self): if hasattr(self, '_loader_popup') and self._loader_popup and self._loader_popup.winfo_exists(): self._loader_popup.grab_release() self._loader_popup.destroy() self._loader_popup = None if __name__ == '__main__': app = ClickRecorderApp() app.mainloop()