.*?)(?P<gap>\s+)(?P<old_version>v?[0-9]+(?:\.[0-9]+)*)(?P<trailing>\s*)(?P<suffix>##\s*)$', re.MULTILINE ) def replace_banner_line(match): title_text = match.group("title").strip() title_text = re.sub(r'\s+', ' ', title_text) old_version = match.group("old_version").strip() trailing_spaces = match.group("trailing") if old_version.lower().startswith("v"): trailing_spaces = trailing_spaces + " " return "%s%s%s%s%s%s" % ( match.group("prefix"), title_text, match.group("gap"), new_version, trailing_spaces, match.group("suffix") ) updated_content, count = banner_pattern.subn(replace_banner_line, content, 1) if count: return updated_content website_pattern = re.compile( r'^(?P<line>\s*##\s*www\.tormachtips\.com\s*##\s*)$', re.MULTILINE | re.IGNORECASE ) website_match = website_pattern.search(content) if website_match: start_pos = website_match.start() earlier_text = content[:start_pos] candidates = list(banner_pattern.finditer(earlier_text)) if candidates: last_match = candidates[-1] replacement = replace_banner_line(last_match) return ( content[:last_match.start()] + replacement + content[last_match.end():] ) raise ValueError("Header banner version line was not found.") def update_script_version_and_description_constants( self, content, new_version, new_description ): content, version_count = re.subn( r'^(\s*CURRENT_VER\s*=\s*[\'"])v?([^\'"]*)([\'"])', r'\g<1>%s\g<3>' % new_version, content, count=1, flags=re.MULTILINE ) if version_count == 0: raise ValueError("CURRENT_VER assignment was not found.") content, description_count = re.subn( r'^(\s*DESCRIPTION\s*=\s*[\'"])([^\'"]*)([\'"])', r'\g<1>%s\g<3>' % new_description.replace("\\", "\\\\"), content, count=1, flags=re.MULTILINE ) if description_count == 0: raise ValueError("DESCRIPTION assignment was not found.") return content def get_script_version_and_description(self, content): version_value = "" description_value = "" version_match = re.search( r'^\s*CURRENT_VER\s*=\s*[\'"]([^\'"]*)[\'"]', content, re.MULTILINE ) if version_match: version_value = normalize_version(version_match.group(1)) description_match = re.search( r'^\s*DESCRIPTION\s*=\s*[\'"]([^\'"]*)[\'"]', content, re.MULTILINE ) if description_match: description_value = description_match.group(1).strip() return version_value, description_value def write_file_bytes(self, path, content): handle = open(path, "wb") try: handle.write(content) finally: handle.close() def edit_active_plugin_locally(self): filename = self.get_active_plugin_name() if filename: pass else: tkMessageBox.showwarning( "No Plugin Selected", "Select or focus a plugin first." ) return local_path = self.get_local_path(filename) if os.path.isfile(local_path): pass else: tkMessageBox.showwarning( "Local File Missing", "This plugin is not installed locally:\n%s" % local_path ) return try: subprocess.Popen(["gedit", local_path]) self.status_text.set("Opened local file in gedit.") except Exception as exc: tkMessageBox.showerror( "Edit Locally Error", "Could not open gedit.\n\n%s" % str(exc) ) def download_plugin_text_from_server(self, filename): source_url = self.get_download_url(filename) response = urllib.urlopen(source_url) code = response.getcode() content = response.read() if code is not None and code != 200: raise IOError("server returned HTTP %s" % code) if content: pass else: raise IOError("downloaded file is empty") lowered = content.lstrip().lower() for marker in HTML_ERROR_MARKERS: if lowered.startswith(marker): raise IOError("server returned HTML instead of a plugin file") return content def inspect_active_plugin_from_server(self): filename = self.get_active_plugin_name() if filename: pass else: tkMessageBox.showwarning( "No Plugin Selected", "Select or focus a plugin first." ) return if self.is_premium_plugin(filename): tkMessageBox.showinfo( "View Server Copy Unavailable", "View Server Copy is not available for Premium plugins." ) return try: content = self.download_plugin_text_from_server(filename) temp_path = self.open_text_in_gedit( content, prefix="tormachtips_server_", suffix="_" + filename ) self.status_text.set("Opened server copy in gedit.") except Exception as exc: tkMessageBox.showerror( "Inspect From Server Error", "Could not open server copy.\n\n%s" % str(exc) ) def open_url_in_browser(self, url): browser_commands = [ ["google-chrome", url], ["google-chrome-stable", url], ["chromium-browser", url], ["chromium", url], ["xdg-open", url] ] for command in browser_commands: program = command[0] full_path = self.find_executable_in_path(program) if full_path: subprocess.Popen([full_path] + command[1:]) return True try: if webbrowser.open(url): return True except Exception: pass return False def rebuild_json(self): if os.path.isfile(self.ftp_config_path): pass else: tkMessageBox.showwarning( "FTP Config Missing", "Rebuild JSON is disabled because the FTP config file was not found:\n%s" % self.ftp_config_path ) self.refresh_upload_button_state() return try: config = self.load_ftp_config() except Exception as exc: tkMessageBox.showerror( "Rebuild JSON Error", str(exc) ) self.refresh_upload_button_state() return rebuild_json_url = str(config.get("rebuild_json_url", "")).strip() if rebuild_json_url: pass else: tkMessageBox.showerror( "Rebuild JSON Error", "FTP config is missing rebuild_json_url." ) return if self.open_url_in_browser(rebuild_json_url): self.status_text.set("Opened rebuild JSON URL.") else: tkMessageBox.showerror( "Browser Error", "Could not open a browser for:\n%s" % rebuild_json_url ) self.status_text.set("Could not open rebuild JSON URL.") def find_executable_in_path(self, program_name): if os.path.isabs(program_name): if os.path.isfile(program_name) and os.access(program_name, os.X_OK): return program_name return "" path_value = str(os.environ.get("PATH", "")).strip() if not path_value: return "" for path_dir in path_value.split(os.pathsep): path_dir = path_dir.strip() if not path_dir: continue candidate = os.path.join(path_dir, program_name) if os.path.isfile(candidate) and os.access(candidate, os.X_OK): return candidate return "" def run_startup_sequence(self): if self.remote_session: self.show_remote_session_warning() self.prompt_for_plugin_folder_setup() self.reload_plugins() def get_ui_hooks_path(self): return os.path.join(self.plugin_dir, UI_HOOKS_FILENAME) def has_plugin_dir(self): return os.path.isdir(self.plugin_dir) def has_ui_hooks(self): return os.path.isfile(self.get_ui_hooks_path()) def needs_plugin_folder_setup(self): if self.has_plugin_dir(): if self.has_ui_hooks(): return False return True return True def install_plugin_folder_setup(self): created_dir = False copied_ui_hooks = False if self.has_plugin_dir(): pass else: os.makedirs(self.plugin_dir) created_dir = True if self.has_ui_hooks(): pass else: response = urllib.urlopen(UI_HOOKS_URL) code = response.getcode() content = response.read() if code is not None and code != 200: raise IOError("server returned HTTP %s for %s" % (code, UI_HOOKS_URL)) if content: pass else: raise IOError("ui_hooks.py download is empty") lowered = content.lstrip().lower() for marker in HTML_ERROR_MARKERS: if lowered.startswith(marker): raise IOError("server returned HTML instead of ui_hooks.py") handle = open(self.get_ui_hooks_path(), "wb") try: handle.write(content) finally: handle.close() copied_ui_hooks = True return created_dir, copied_ui_hooks def prompt_for_plugin_folder_setup(self): if self.needs_plugin_folder_setup(): pass else: return dialog = tk.Toplevel(self.root) dialog.title("Plugin Setup Required") dialog.transient(self.root) dialog.grab_set() dialog.resizable(False, False) main_frame = tk.Frame(dialog, padx=14, pady=14) main_frame.pack(fill=tk.BOTH, expand=True) message = ( "Welcome to the TormachTips Download & Update Manager.\n\n" "Most PathPilot plugins need these two setup steps:\n\n" " Create %s\n" " Install %s\n\n" "Would you like me to do this now?" ) % (self.plugin_dir, UI_HOOKS_FILENAME) message_label = tk.Label( main_frame, text=message, justify=tk.LEFT, anchor="w" ) message_label.pack(fill=tk.X) done_font = tkFont.Font( family=DEFAULT_FONT_FAMILY, size=DEFAULT_FONT_SIZE, overstrike=1 ) created_dir_done = self.has_plugin_dir() ui_hooks_done = self.has_ui_hooks() if created_dir_done: dir_label_text = "Done: Create %s" % self.plugin_dir else: dir_label_text = "Pending: Create %s" % self.plugin_dir dir_label = tk.Label( main_frame, text=dir_label_text, justify=tk.LEFT, anchor="w" ) dir_label.pack(fill=tk.X, pady=(10, 0), padx=(16, 0)) if ui_hooks_done: hooks_label_text = "Done: Install %s" % UI_HOOKS_FILENAME else: hooks_label_text = "Pending: Install %s" % UI_HOOKS_FILENAME hooks_label = tk.Label( main_frame, text=hooks_label_text, justify=tk.LEFT, anchor="w" ) hooks_label.pack(fill=tk.X, pady=(4, 0), padx=(16, 0)) if created_dir_done: dir_label.config(fg="gray40", font=done_font) if ui_hooks_done: hooks_label.config(fg="gray40", font=done_font) buttons_frame = tk.Frame(main_frame) buttons_frame.pack(fill=tk.X, pady=(14, 0)) result = {"accepted": False} def on_yes(): result["accepted"] = True dialog.destroy() def on_no(): dialog.destroy() yes_button = tk.Button( buttons_frame, text="Yes", width=12, command=on_yes ) yes_button.pack(side=tk.LEFT) no_button = tk.Button( buttons_frame, text="No", width=12, command=on_no ) no_button.pack(side=tk.LEFT, padx=(6, 0)) dialog.protocol("WM_DELETE_WINDOW", on_no) self.root.wait_window(dialog) if result["accepted"]: pass else: self.status_text.set("Plugin prerequisite setup skipped.") return try: created_dir, copied_ui_hooks = self.install_plugin_folder_setup() except Exception as exc: tkMessageBox.showerror( "Setup Error", "Could not complete plugin setup.\n\n%s" % str(exc) ) self.status_text.set("Plugin prerequisite setup failed.") return result_lines = [] if created_dir: result_lines.append("%s created" % self.plugin_dir) if copied_ui_hooks: result_lines.append("%s installed" % UI_HOOKS_FILENAME) if result_lines: tkMessageBox.showinfo( "Setup Complete", "\n".join(result_lines) ) self.status_text.set("Plugin prerequisite setup completed.") else: self.status_text.set("Plugin prerequisite setup already complete.") def show_remote_session_warning(self): if self.remote_session: message = ( "Welcome to the TormachTips Download & Update Manager.\n\n" "This session was opened remotely from \"the cloud\".\n\n" "On the next page, you'll be presented with all my plugins " "(including this manager).\n\n" "You may continue using remote sessions, or download this " "script to run locally." ) tkMessageBox.showinfo("Remote Session Detected", message) def refresh_upload_button_state(self): show_ftp_buttons = os.path.isfile(self.ftp_config_path) if hasattr(self, "admin_buttons_row"): if show_ftp_buttons: if self.admin_buttons_row.winfo_manager(): pass else: self.admin_buttons_row.pack(fill=tk.X, pady=(6, 0)) else: if self.admin_buttons_row.winfo_manager(): self.admin_buttons_row.pack_forget() def prompt_for_ftp_password(self): if self.ftp_password: return True password = tkSimpleDialog.askstring( "FTP Password Required", "Enter FTP password:", show="" ) if password is None: return False password = password.strip() if password: self.ftp_password = password return True tkMessageBox.showwarning( "Missing Password", "A password is required for FTP upload." ) return False def update_status_text(self, prefix): self.status_text.set( "%s | Plugins: %d | Selected: %d" % ( prefix, len(self.plugins), len(self.selected_files) ) ) def is_load_meter_plugin(self, filename): return filename in PREMIUM_PLUGIN_FILES def is_pendant_plugin(self, filename): return filename in PENDANT_PLUGIN_FILES def is_premium_plugin(self, filename): if self.is_load_meter_plugin(filename): return True if self.is_pendant_plugin(filename): return True return False def get_upload_targets(self, filename): targets = [] targets.append({ "remote_filename": filename, "use_parent_of_remote_dir": False }) if filename == "download_manager.py": targets.append({ "remote_filename": "downloader.py", "use_parent_of_remote_dir": True }) return targets def get_download_url(self, filename): if self.is_load_meter_plugin(filename): return PREMIUM_BASE_URL + filename if self.is_pendant_plugin(filename): return PENDANT_BASE_URL + filename return BASE_URL + filename def find_curl_binary(self): for path in CURL_PATHS: if os.path.isfile(path) and os.access(path, os.X_OK): return path return None def prompt_for_premium_credentials(self): username = tkSimpleDialog.askstring( "Premium Download Login", "Enter username:") if username is None: return False password = tkSimpleDialog.askstring( "Premium Download Login", "Enter password:", show="") if password is None: return False username = username.strip() if username and password: self.download_username = username self.download_password = password return True tkMessageBox.showwarning( "Missing Credentials", "Both username and password are required for this plugin.") return False def validate_download_response( self, source_url, final_url, http_code, content_type, content ): if http_code is not None and http_code != 200: raise IOError( self.build_download_debug_message( source_url, final_url, http_code, content_type, content, "Server returned HTTP %s." % http_code ) ) if not content: raise IOError( self.build_download_debug_message( source_url, final_url, http_code, content_type, content, "Downloaded file is empty." ) ) lowered = content.lstrip().lower() content_type_text = str(content_type or "").lower() probe = lowered[:512] if "text/html" in content_type_text: raise IOError( self.build_download_debug_message( source_url, final_url, http_code, content_type, content, "Server returned HTML instead of a plugin file." ) ) for marker in HTML_ERROR_MARKERS: if probe.startswith(marker): raise IOError( self.build_download_debug_message( source_url, final_url, http_code, content_type, content, "Server returned HTML instead of a plugin file." ) ) def build_download_debug_message( self, source_url, final_url, http_code, content_type, content, prefix ): preview = "" if content: preview = content[:400] preview = preview.replace("\r", " ") preview = preview.replace("\n", " ") preview = preview.strip() details = [ prefix, "", "Requested URL: %s" % source_url, "Final URL: %s" % (final_url or "(unknown)"), "HTTP code: %s" % (http_code if http_code is not None else "(unknown)"), "Content-Type: %s" % (content_type or "(unknown)"), ] if preview: details.append("Response preview: %s" % preview) return "\n".join(details) def ensure_premium_credentials(self): if self.download_username and self.download_password: return True return self.prompt_for_premium_credentials() def download_plugin_with_curl(self, filename): curl_path = self.find_curl_binary() if not curl_path: raise IOError("curl was not found on this system") if not self.ensure_premium_credentials(): raise IOError("download cancelled: username/password required") source_url = self.get_download_url(filename) destination_path = self.get_local_path(filename) header_path = destination_path + ".headers.tmp" self.ensure_plugin_dir_exists() if not os.access(self.plugin_dir, os.W_OK): raise IOError("plugin folder is not writable: %s" % self.plugin_dir) command = [ curl_path, "--silent", "--show-error", "--location", "--user", "%s:%s" % (self.download_username, self.download_password), "--dump-header", header_path, "--output", destination_path, source_url, ] process = subprocess.Popen( command, stdout=subprocess.PIPE, stderr=subprocess.PIPE ) stdout_data, stderr_data = process.communicate() if process.returncode != 0: if os.path.isfile(destination_path): try: os.remove(destination_path) except Exception: pass if os.path.isfile(header_path): try: os.remove(header_path) except Exception: pass self.clear_premium_credentials() error_text = (stderr_data or "").strip() if not error_text: error_text = "curl failed with exit code %s" % process.returncode raise IOError(error_text) if not os.path.isfile(destination_path): if os.path.isfile(header_path): try: os.remove(header_path) except Exception: pass raise IOError("file was not created: %s" % destination_path) if os.path.getsize(destination_path) <= 0: if os.path.isfile(header_path): try: os.remove(header_path) except Exception: pass raise IOError("saved file is empty: %s" % destination_path) try: content = self.read_file_bytes(destination_path) except Exception as exc: if os.path.isfile(header_path): try: os.remove(header_path) except Exception: pass self.clear_premium_credentials() raise IOError("download completed but file could not be read: %s" % str(exc)) headers_text = "" if os.path.isfile(header_path): try: headers_text = self.read_file_bytes(header_path) except Exception: headers_text = "" final_url = source_url http_code = None content_type = "" if headers_text: status_matches = re.findall(r"HTTP/\d+\.\d+\s+(\d+)", headers_text) if status_matches: try: http_code = int(status_matches[-1]) except Exception: http_code = None content_type_matches = re.findall( r"(?im)^Content-Type:\s*([^\r\n;]+)", headers_text ) if content_type_matches: content_type = content_type_matches[-1].strip() try: self.validate_download_response( source_url, final_url, http_code, content_type, content ) except Exception: self.clear_premium_credentials() if os.path.isfile(destination_path): try: os.remove(destination_path) except Exception: pass raise finally: if os.path.isfile(header_path): try: os.remove(header_path) except Exception: pass def get_script_path(self): script_name = globals().get("__file__", "") script_name = str(script_name).strip() if script_name: if script_name not in ["<stdin>", "-c"]: script_path = os.path.abspath(script_name) if os.path.isfile(script_path): return script_path return "" def create_desktop_shortcut(self): desktop_dir = DEFAULT_DESKTOP_DIR if os.path.isdir(desktop_dir): pass else: raise_message = "Desktop folder not found:\n%s" % desktop_dir tkMessageBox.showerror("Shortcut Error", raise_message) self.status_text.set("Desktop script not created.") return script_path = os.path.join(desktop_dir, SHORTCUT_FILENAME) script_text = ( "#!/bin/sh\n" "%s\n" ) % SHORTCUT_CURL_COMMAND try: handle = open(script_path, "wb") try: handle.write(script_text) finally: handle.close() current_mode = os.stat(script_path).st_mode os.chmod( script_path, current_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH ) tkMessageBox.showinfo( "Desktop Script Created", "Desktop script created:\n%s\n" % (script_path,)) self.status_text.set("Desktop script created.") except Exception as exc: tkMessageBox.showerror( "Shortcut Error", "Could not create desktop script.\n\n%s" % str(exc) ) self.status_text.set("Desktop script not created.") def _build_ui(self): main_frame = tk.Frame(self.root, padx=PADDING, pady=PADDING) main_frame.pack(fill=tk.BOTH, expand=True) top_frame = tk.Frame(main_frame) top_frame.pack(fill=tk.X) header_text_frame = tk.Frame(top_frame) header_text_frame.pack(side=tk.LEFT, fill=tk.X, expand=True) source_text = "Plugin source:\n%s" % PLUGINS_JSON_URL source_label = tk.Label( header_text_frame, text=source_text, anchor="w", justify=tk.LEFT ) source_label.pack(fill=tk.X) path_label = tk.Label( header_text_frame, text="Local plugin folder:\n%s - Remember: ui_hooks.py is required for most of these to work." % self.plugin_dir, anchor="w", justify=tk.LEFT ) path_label.pack(fill=tk.X, pady=(6, 0)) self.shortcut_button = None shortcut_path = os.path.join(DEFAULT_DESKTOP_DIR, SHORTCUT_FILENAME) if not os.path.isfile(shortcut_path): self.shortcut_button = tk.Button( top_frame, text="Create Desktop Shortcut", width=28, underline=0, command=self.create_desktop_shortcut ) self.shortcut_button.pack(side=tk.RIGHT, anchor="ne", padx=(10, 0)) list_frame = tk.LabelFrame(main_frame, text="Plugins", padx=8, pady=8) list_frame.pack(fill=tk.BOTH, expand=True, pady=(10, 10)) self._build_tree_ui(list_frame) buttons_frame = tk.Frame(main_frame) buttons_frame.pack(fill=tk.X) buttons_row_1 = tk.Frame(buttons_frame) buttons_row_1.pack(fill=tk.X) buttons_row_2 = tk.Frame(buttons_frame) buttons_row_2.pack(fill=tk.X, pady=(6, 0)) admin_buttons_row = tk.Frame(buttons_frame) self.refresh_button = tk.Button( buttons_row_1, text="Refresh", width=14, underline=0, command=self.reload_plugins ) self.refresh_button.pack(side=tk.LEFT) self.select_all_button = tk.Button( buttons_row_1, text="Select All", width=14, underline=7, command=self.select_all_plugins ) self.select_all_button.pack(side=tk.LEFT, padx=(6, 0)) self.unselect_all_button = tk.Button( buttons_row_1, text="Unselect All", width=14, underline=1, command=lambda: self.select_all_plugins(select_all=False) ) self.unselect_all_button.pack(side=tk.LEFT, padx=(6, 0)) self.info_button = tk.Button( buttons_row_1, text="Show Details", width=14, underline=5, command=self.show_selected_details ) self.info_button.pack(side=tk.LEFT, padx=(6, 0)) self.edit_local_button = tk.Button( buttons_row_1, text="Edit", width=14, underline=0, command=self.edit_active_plugin_locally ) self.edit_local_button.pack(side=tk.LEFT, padx=(6, 0)) self.inspect_server_button = tk.Button( buttons_row_1, text="View Server Copy", width=18, underline=0, command=self.inspect_active_plugin_from_server ) self.inspect_server_button.pack(side=tk.LEFT, padx=(6, 0)) self.install_button = tk.Button( buttons_row_2, text="Install Selected", width=16, underline=0, command=self.install_selected ) self.install_button.pack(side=tk.LEFT) self.update_button = tk.Button( buttons_row_2, text="Update Selected", width=16, underline=0, command=self.update_selected ) self.update_button.pack(side=tk.LEFT, padx=(6, 0)) self.remove_button = tk.Button( buttons_row_2, text="Remove Selected", width=16, underline=2, command=self.remove_selected ) self.remove_button.pack(side=tk.LEFT, padx=(6, 0)) self.local_only_toggle_button = tk.Button( buttons_row_2, text="Hide Local Only", width=16, underline=5, command=self.toggle_local_only_rows ) self.local_only_toggle_button.pack(side=tk.LEFT, padx=(6, 0)) self.help_button = tk.Button( buttons_row_2, text="Help", width=14, underline=0, command=self.show_help_dialog ) self.help_button.pack(side=tk.LEFT, padx=(6, 0)) self.quit_button = tk.Button( buttons_row_2, text="Quit", width=14, underline=0, command=self.root.quit ) self.quit_button.pack(side=tk.LEFT, padx=(6, 0)) status_label = tk.Label( main_frame, textvariable=self.status_text, anchor="w", justify=tk.LEFT, fg="blue" ) status_label.pack(fill=tk.X, pady=(10, 0)) self.upload_button = tk.Button( admin_buttons_row, text="Upload Selected", width=16, underline=1, command=self.upload_selected_to_ftp ) self.upload_button.pack(side=tk.LEFT) self.rebuild_json_button = tk.Button( admin_buttons_row, text="Rebuild JSON", width=16, underline=2, command=self.rebuild_json ) self.rebuild_json_button.pack(side=tk.LEFT, padx=(6, 0)) self.view_json_button = tk.Button( admin_buttons_row, text="View JSON", width=16, underline=5, command=self.view_plugins_json_from_server ) self.view_json_button.pack(side=tk.LEFT, padx=(6, 0)) self.change_log_button = tk.Button( admin_buttons_row, text="Change Log", width=16, underline=4, command=self.change_log_active_plugin ) self.change_log_button.pack(side=tk.LEFT, padx=(6, 0)) self.admin_buttons_row = admin_buttons_row self.refresh_upload_button_state() self.refresh_local_only_toggle_button_text() self.bind_button_accelerator("a", self.select_all_button) self.bind_button_accelerator("b", self.rebuild_json_button) self.bind_button_accelerator("c", self.shortcut_button) self.bind_button_accelerator("d", self.info_button) self.bind_button_accelerator("e", self.edit_local_button) self.bind_button_accelerator("g", self.change_log_button) self.bind_button_accelerator("h", self.help_button) self.bind_button_accelerator("i", self.install_button) self.bind_button_accelerator("j", self.view_json_button) self.bind_button_accelerator("l", self.local_only_toggle_button) self.bind_button_accelerator("m", self.remove_button) self.bind_button_accelerator("n", self.unselect_all_button) self.bind_button_accelerator("p", self.upload_button) self.bind_button_accelerator("q", self.quit_button) self.bind_button_accelerator("r", self.refresh_button) self.bind_button_accelerator("u", self.update_button) self.bind_button_accelerator("v", self.inspect_server_button) def refresh_tree_heading_text(self, active_column=None, descending=False): for column_name, base_text in self.tree_heading_text.items(): heading_text = base_text if column_name == active_column: if descending: heading_text = base_text + " ▲" else: heading_text = base_text + " ▼" self.tree.heading(column_name, text=heading_text) def _build_tree_ui(self, parent): columns = [ "filename", "notes", "installed", "installed_ver", "available_ver", "status" ] self.tree = ttk.Treeview( parent, columns=tuple(columns), show="headings", selectmode="extended" ) self.tree_sort_state = {} self.tree_heading_text = { "filename": "Plugin Filename", "notes": "Notes", "installed": "Installed", "installed_ver": "Installed Ver", "available_ver": "Available Ver", "status": "Status" } self.tree.heading( "filename", text=self.tree_heading_text["filename"], command=lambda: self.sort_tree_by_column("filename") ) self.tree.heading( "notes", text=self.tree_heading_text["notes"], command=lambda: self.sort_tree_by_column("notes") ) self.tree.heading( "installed", text=self.tree_heading_text["installed"], command=lambda: self.sort_tree_by_column("installed") ) self.tree.heading( "installed_ver", text=self.tree_heading_text["installed_ver"], command=lambda: self.sort_tree_by_column("installed_ver") ) self.tree.heading( "available_ver", text=self.tree_heading_text["available_ver"], command=lambda: self.sort_tree_by_column("available_ver") ) self.tree.heading( "status", text=self.tree_heading_text["status"], command=lambda: self.sort_tree_by_column("status") ) self.tree.column("filename", width=300, anchor="w") self.tree.column("notes", width=110, anchor="center") self.tree.column("installed", width=85, anchor="center") self.tree.column("installed_ver", width=105, anchor="center") self.tree.column("available_ver", width=105, anchor="center") self.tree.column("status", width=120, anchor="w") yscroll = ttk.Scrollbar(parent, orient="vertical", command=self.tree.yview) self.tree.configure(yscrollcommand=yscroll.set) self.tree.pack(side=tk.LEFT, fill=tk.BOTH, expand=True) yscroll.pack(side=tk.RIGHT, fill=tk.Y) self.tree.tag_configure("focus_row", background="#e6f0ff") self.tree.tag_configure("status_highlight", background=STATUS_HIGHLIGHT_BG) self.tree.bind("<Double-1>", self.on_tree_double_click) self.tree.bind("<<TreeviewSelect>>", self.on_tree_select) self.tree.bind("<Button-1>", self.on_tree_click) self.tree.bind("<Return>", self.on_tree_enter_key) self.tree.bind("<KP_Enter>", self.on_tree_enter_key) self.tree.bind("<KeyPress>", self.on_tree_keypress_search) self.tree.bind("<Up>", self.on_tree_arrow_key) self.tree.bind("<Down>", self.on_tree_arrow_key) self.tree.bind("<Prior>", self.on_tree_page_key) self.tree.bind("<Next>", self.on_tree_page_key) self.tree.bind("<Home>", self.on_tree_home_end_key) self.tree.bind("<End>", self.on_tree_home_end_key) self.tree.bind("<Shift-Up>", self.on_tree_shift_arrow_key) self.tree.bind("<Shift-Down>", self.on_tree_shift_arrow_key) self.tree.bind("<Shift-Prior>", self.on_tree_shift_page_key) self.tree.bind("<Shift-Next>", self.on_tree_shift_page_key) self.tree.bind("<space>", self.on_tree_spacebar) def reset_tree_type_search(self): self.tree_type_search_text = "" self.tree_type_search_after_id = None def schedule_tree_type_search_reset(self): if self.tree_type_search_after_id: try: self.root.after_cancel(self.tree_type_search_after_id) except Exception: pass self.tree_type_search_after_id = self.root.after( self.tree_type_search_reset_ms, self.reset_tree_type_search ) def find_tree_item_by_prefix(self, prefix_text): prefix_text = str(prefix_text).strip().lower() if not prefix_text: return "" item_ids = self.get_tree_item_ids() if not item_ids: return "" focus_item = self.get_tree_focus_item() start_index = 0 if focus_item in item_ids: start_index = item_ids.index(focus_item) + 1 ordered_ids = item_ids[start_index:] + item_ids[:start_index] for item_id in ordered_ids: filename = str(self.tree.set(item_id, "filename")).strip().lower() if filename.startswith(prefix_text): return item_id return "" def on_tree_keypress_search(self, event): char_text = getattr(event, "char", "") keysym_text = str(getattr(event, "keysym", "")).strip() state_value = int(getattr(event, "state", 0)) alt_mask = 0x0008 control_mask = 0x0004 if state_value & alt_mask: return if state_value & control_mask: return if keysym_text in ["Up", "Down", "Prior", "Next", "Return", "space"]: return if keysym_text.startswith("Shift"): return if keysym_text in ["Control_L", "Control_R", "Alt_L", "Alt_R"]: return if not char_text: return if ord(char_text) < 32: return self.tree_type_search_text += char_text.lower() self.schedule_tree_type_search_reset() target_item = self.find_tree_item_by_prefix(self.tree_type_search_text) if target_item: self.tree.selection_remove(self.tree.selection()) self.set_tree_focus_item(target_item) self.on_tree_select(None) self.status_text.set( "Loaded | Search: %s | Plugins: %d | Selected: %d" % ( self.tree_type_search_text, len(self.plugins), len(self.selected_files) ) ) return "break" def get_tree_sort_value(self, item_id, column_name): value = self.tree.set(item_id, column_name) if column_name in ["filename", "notes", "status"]: return str(value).lower() if column_name == "installed": value_text = str(value).strip().lower() if value_text in ["yes", "installed", "true", "1"]: return (0, value_text) if value_text in ["no", "not installed", "false", "0"]: return (1, value_text) return (2, value_text) if column_name in ["installed_ver", "available_ver"]: normalized = normalize_version(value) if normalized: return (0, version_key(normalized)) return (1, []) return str(value).lower() def sort_tree_by_column(self, column_name): item_ids = self.get_tree_item_ids() if not item_ids: return descending = self.tree_sort_state.get(column_name, False) rows = [] selected_ids = set(self.tree.selection()) focus_item = self.get_tree_focus_item() for item_id in item_ids: sort_value = self.get_tree_sort_value(item_id, column_name) filename_value = str(self.tree.set(item_id, "filename")).lower() rows.append((sort_value, filename_value, item_id)) rows.sort(reverse=descending) for index, row in enumerate(rows): item_id = row[2] self.tree.move(item_id, "", index) self.tree_sort_state = {} self.tree_sort_state[column_name] = not descending self.refresh_tree_heading_text( active_column=column_name, descending=descending ) if selected_ids: self.tree.selection_set(tuple(selected_ids)) if focus_item: self.set_tree_focus_item(focus_item) else: self.refresh_tree_focus_highlight() def get_tree_item_ids(self): return list(self.tree.get_children()) def get_tree_focus_item(self): item_id = self.tree.focus() if item_id: return item_id if self.last_tree_focus: return self.last_tree_focus selection = self.tree.selection() if selection: return selection[0] item_ids = self.get_tree_item_ids() if item_ids: return item_ids[0] return "" def refresh_tree_focus_highlight(self): item_ids = self.get_tree_item_ids() focus_item = self.get_tree_focus_item() for item_id in item_ids: status_text = self.tree.set(item_id, "status") is_focus_row = (item_id == focus_item) tags = self.get_tree_tags_for_status( status_text, is_focus_row=is_focus_row ) self.tree.item(item_id, tags=tags) def set_tree_focus_item(self, item_id): if not item_id: return self.tree.focus_set() self.tree.focus(item_id) self.tree.see(item_id) self.last_tree_focus = item_id self.refresh_tree_focus_highlight() def select_tree_range(self, start_item, end_item): item_ids = self.get_tree_item_ids() if not item_ids: return if not start_item or start_item not in item_ids: start_item = end_item if not end_item or end_item not in item_ids: return start_index = item_ids.index(start_item) end_index = item_ids.index(end_item) if start_index <= end_index: range_ids = item_ids[start_index:end_index + 1] else: range_ids = item_ids[end_index:start_index + 1] self.tree.selection_set(range_ids) self.set_tree_focus_item(end_item) def move_tree_focus_by_offset(self, offset): item_ids = self.get_tree_item_ids() if not item_ids: return "" current_item = self.get_tree_focus_item() if current_item not in item_ids: current_item = item_ids[0] current_index = item_ids.index(current_item) new_index = current_index + offset if new_index < 0: new_index = 0 if new_index >= len(item_ids): new_index = len(item_ids) - 1 return item_ids[new_index] def get_tree_page_step(self): visible_height = self.tree.winfo_height() if visible_height <= 0: return 10 first_item = "" row_height = 20 item_ids = self.get_tree_item_ids() if item_ids: first_item = item_ids[0] if first_item: bbox = self.tree.bbox(first_item) if bbox and len(bbox) >= 4 and bbox[3] > 0: row_height = bbox[3] step = visible_height / max(row_height, 1) step = int(step) - 1 if step < 1: step = 1 return step def on_tree_click(self, event): self.tree.focus_set() self.reset_tree_type_search() item_id = self.tree.identify_row(event.y) if item_id: self.last_tree_focus = item_id def on_tree_arrow_key(self, event): focus_item = self.get_tree_focus_item() if not focus_item: return "break" if event.keysym == "Up": target_item = self.move_tree_focus_by_offset(-1) else: target_item = self.move_tree_focus_by_offset(1) if target_item: self.set_tree_focus_item(target_item) return "break" def on_tree_home_end_key(self, event): item_ids = self.get_tree_item_ids() if not item_ids: return "break" if event.keysym == "Home": target_item = item_ids[0] else: target_item = item_ids[-1] self.set_tree_focus_item(target_item) return "break" def on_tree_page_key(self, event): focus_item = self.get_tree_focus_item() if not focus_item: return "break" step = self.get_tree_page_step() if event.keysym == "Prior": target_item = self.move_tree_focus_by_offset(-step) else: target_item = self.move_tree_focus_by_offset(step) if target_item: self.set_tree_focus_item(target_item) return "break" def on_tree_shift_arrow_key(self, event): focus_item = self.get_tree_focus_item() if not focus_item: return "break" anchor_item = self.last_tree_focus or focus_item if event.keysym == "Up": target_item = self.move_tree_focus_by_offset(-1) else: target_item = self.move_tree_focus_by_offset(1) if target_item: self.select_tree_range(anchor_item, target_item) return "break" def on_tree_shift_page_key(self, event): focus_item = self.get_tree_focus_item() if not focus_item: return "break" anchor_item = self.last_tree_focus or focus_item step = self.get_tree_page_step() if event.keysym == "Prior": target_item = self.move_tree_focus_by_offset(-step) else: target_item = self.move_tree_focus_by_offset(step) if target_item: self.select_tree_range(anchor_item, target_item) return "break" def on_tree_spacebar(self, event): focus_item = self.get_tree_focus_item() if not focus_item: return "break" self.tree.focus_set() self.set_tree_focus_item(focus_item) if focus_item in self.tree.selection(): self.tree.selection_remove((focus_item,)) else: self.tree.selection_add((focus_item,)) self.on_tree_select(None) return "break" def on_tree_enter_key(self, event): focus_item = self.get_tree_focus_item() if not focus_item: return "break" filename = self.tree.set(focus_item, "filename") if filename: self.show_plugin_details(filename) return "break" def on_tree_select(self, event): current_selection = set() focused_item = self.tree.focus() for item_id in self.tree.selection(): filename = self.tree.set(item_id, "filename") if filename: current_selection.add(filename) self.selected_files = current_selection if focused_item: self.last_tree_focus = focused_item self.refresh_tree_focus_highlight() self.update_status_text("Loaded") def on_tree_double_click(self, event): item_id = self.tree.identify_row(event.y) if not item_id: return filename = self.tree.set(item_id, "filename") if not filename: return self.show_plugin_details(filename) def load_plugins(self): response = urllib.urlopen(PLUGINS_JSON_URL) code = response.getcode() content = response.read() if code is not None and code != 200: raise IOError("server returned HTTP %s" % code) if content: pass else: raise IOError("plugins.json is empty") data = json.loads(content) if "plugins" in data: pass else: raise ValueError("plugins.json missing 'plugins'") plugins = [] for item in data["plugins"]: filename = str(item.get("filename", "")).strip() if filename: pass else: continue plugins.append({ "filename": filename, "description": str(item.get("description", "")).strip(), "available_ver": normalize_version(item.get("version", "")), "md5": str(item.get("md5", "")).strip().lower() }) if plugins: pass else: raise ValueError("no valid plugins found") plugins.sort(key=lambda p: p["filename"].lower()) return plugins def ensure_plugin_dir_exists(self): if os.path.isdir(self.plugin_dir): return try: os.makedirs(self.plugin_dir) except Exception as exc: raise IOError( "could not create plugin folder %s: %s" % ( self.plugin_dir, str(exc) ) ) def load_ftp_config(self): if not os.path.isfile(self.ftp_config_path): raise IOError( "FTP config file not found:\n%s" % self.ftp_config_path ) try: handle = open(self.ftp_config_path, "rb") try: raw_text = handle.read() finally: handle.close() except Exception as exc: raise IOError( "Could not read FTP config file:\n%s" % str(exc) ) try: config = json.loads(raw_text) except Exception as exc: raise IOError( "FTP config is not valid JSON:\n%s" % str(exc) ) host = str(config.get("host", "")).strip() username = str(config.get("username", "")).strip() remote_dir = str(config.get("remote_dir", "")).strip() premium_remote_dir = str(config.get("premium_remote_dir", "")).strip() pendant_remote_dir = str(config.get("pendant_remote_dir", "")).strip() rebuild_json_url = str(config.get("rebuild_json_url", "")).strip() if host and username and remote_dir: pass else: raise IOError( "FTP config must contain host, username, and remote_dir." ) if premium_remote_dir: pass else: premium_remote_dir = remote_dir if pendant_remote_dir: pass else: pendant_remote_dir = remote_dir return { "host": host, "username": username, "remote_dir": remote_dir, "premium_remote_dir": premium_remote_dir, "pendant_remote_dir": pendant_remote_dir, "rebuild_json_url": rebuild_json_url } def upload_file_to_ftp( self, local_path, remote_filename, use_parent_of_remote_dir=False ): config = self.load_ftp_config() if self.is_load_meter_plugin(remote_filename): target_remote_dir = config.get("premium_remote_dir", "").strip() elif self.is_pendant_plugin(remote_filename): target_remote_dir = config.get("pendant_remote_dir", "").strip() else: target_remote_dir = config.get("remote_dir", "").strip() if target_remote_dir: pass else: raise IOError("Remote directory is missing for %s" % remote_filename) if use_parent_of_remote_dir: normalized_dir = target_remote_dir.rstrip("/") if normalized_dir: parent_dir = os.path.dirname(normalized_dir) else: parent_dir = "" if parent_dir: target_remote_dir = parent_dir else: target_remote_dir = "/" if self.prompt_for_ftp_password(): pass else: raise IOError("FTP upload cancelled: password required") debug_buffer = StringIO.StringIO() original_stdout = sys.stdout ftp = ftplib.FTP() try: sys.stdout = debug_buffer ftp.set_debuglevel(FTP_DEBUG_LEVEL) ftp.connect(config["host"], 21, FTP_TIMEOUT_SECONDS) ftp.login(config["username"], self.ftp_password) ftp.cwd(target_remote_dir) backup_filename = self.backup_remote_ftp_file(ftp, remote_filename) handle = open(local_path, "rb") try: ftp.storbinary("STOR " + remote_filename, handle) finally: handle.close() transcript = debug_buffer.getvalue().strip() return { "ok": True, "remote_dir": target_remote_dir, "remote_filename": remote_filename, "backup_filename": backup_filename, "transcript": transcript } except Exception as exc: self.clear_ftp_password() transcript = debug_buffer.getvalue().strip() return { "ok": False, "remote_dir": target_remote_dir, "remote_filename": remote_filename, "error": str(exc), "transcript": transcript } finally: sys.stdout = original_stdout try: ftp.quit() except Exception: try: ftp.close() except Exception: pass def upload_selected_to_ftp(self): selected = self.get_selected_plugin_names() if not selected: tkMessageBox.showwarning("No Selection", "Select at least one plugin.") return if os.path.isfile(self.ftp_config_path): pass else: tkMessageBox.showwarning( "FTP Config Missing", "Upload is disabled because the FTP config file was not found:\n%s" % self.ftp_config_path ) self.refresh_upload_button_state() return completed = [] skipped = [] failed = [] for filename in selected: local_path = self.get_local_path(filename) if os.path.isfile(local_path): self.status_text.set("Uploading %s." % filename) self.root.update_idletasks() else: skipped.append("%s:\nNot installed locally." % filename) continue targets = self.get_upload_targets(filename) for target in targets: remote_filename = target["remote_filename"] use_parent_of_remote_dir = target["use_parent_of_remote_dir"] result = self.upload_file_to_ftp( local_path, remote_filename, use_parent_of_remote_dir=use_parent_of_remote_dir ) label = "%s -> %s" % (filename, remote_filename) remote_dir = result.get("remote_dir", "") if remote_dir: label += "\nRemote dir: %s" % remote_dir if result.get("ok"): transcript = result.get("transcript", "") backup_filename = result.get("backup_filename", "") message = label if backup_filename: message += "\nBackup: %s" % backup_filename else: message += "\nBackup: no existing server file" if transcript: message += "\n\nFTP transcript:\n%s" % transcript completed.append(message) else: message = "%s:\n%s" % ( label, result.get("error", "Unknown FTP error") ) transcript = result.get("transcript", "") if transcript: message += "\n\nFTP transcript:\n%s" % transcript failed.append(message) self.reload_plugins() self.refresh_upload_button_state() message_parts = [] if completed: message_parts.append( "Uploaded:\n%s" % "\n\n".join(completed) ) if skipped: message_parts.append( "Skipped:\n%s" % "\n\n".join(skipped) ) if failed: message_parts.append( "Failed:\n%s" % "\n\n".join(failed) ) if not message_parts: message_parts.append("No changes made.") final_message = "\n\n".join(message_parts) try: self.open_text_in_gedit( final_message, prefix="tormachtips_ftp_", suffix=".txt" ) except Exception as exc: tkMessageBox.showerror( "FTP Upload Results", "%s\n\nAlso failed to open gedit:\n%s" % (final_message, str(exc)) ) if failed: self.update_status_text("Upload completed with errors.") return self.update_status_text("Upload completed.") def open_text_in_gedit(self, text, prefix="tormachtips_", suffix=".txt"): handle, temp_path = tempfile.mkstemp(prefix=prefix, suffix=suffix, dir="/tmp") try: temp_file = os.fdopen(handle, "wb") try: temp_file.write(text) finally: temp_file.close() except Exception: try: os.close(handle) except Exception: pass raise subprocess.Popen(["gedit", temp_path]) return temp_path def read_file_bytes(self, path): handle = open(path, "rb") try: return handle.read() finally: handle.close() def create_backup_file(self, filename): source_path = self.get_local_path(filename) timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") backup_path = "%s.%s.bak" % (source_path, timestamp) if not os.path.isfile(source_path): raise IOError("installed file not found: %s" % source_path) try: import shutil shutil.copy2(source_path, backup_path) # <-- copy instead of rename except Exception as exc: raise IOError("could not create backup file %s: %s" % (backup_path, str(exc))) return backup_path def get_local_path(self, filename): return os.path.join(self.plugin_dir, filename) def get_local_only_plugins(self): local_plugins = [] if os.path.isdir(self.plugin_dir): pass else: return local_plugins remote_names = set() for plugin in self.plugins: filename = str(plugin.get("filename", "")).strip() if filename: remote_names.add(filename) try: names = os.listdir(self.plugin_dir) except Exception: return local_plugins for name in names: name = str(name).strip() if name: pass else: continue if name in remote_names: continue if name.endswith(".pyc") or name.endswith(".pyo"): continue full_path = os.path.join(self.plugin_dir, name) if os.path.isfile(full_path): pass else: continue local_plugins.append({ "filename": name, "description": "Installed locally only", "available_ver": "", "md5": "" }) local_plugins.sort(key=lambda p: p["filename"].lower()) return local_plugins def is_installed(self, filename): return os.path.isfile(self.get_local_path(filename)) def get_local_md5(self, filename): path = self.get_local_path(filename) if os.path.isfile(path): pass else: return "" try: content = self.read_file_bytes(path) except Exception: return "" if content: return hashlib.md5(content).hexdigest().lower() return "" def get_installed_version(self, filename): path = self.get_local_path(filename) if not os.path.isfile(path): return "" try: content = self.read_file_bytes(path) except Exception: return "" for pattern in CURRENT_VER_PATTERNS: match = pattern.search(content) if match: return normalize_version(match.group(1)) return "" def get_plugin_status(self, plugin): filename = plugin["filename"] installed = self.is_installed(filename) installed_ver = self.get_installed_version(filename) available_ver = normalize_version(plugin.get("available_ver", "")) local_md5 = self.get_local_md5(filename) available_md5 = str(plugin.get("md5", "")).strip().lower() if installed: pass else: return "No", installed_ver, available_ver, "", available_md5, "Not installed" if available_ver or available_md5: if installed_ver and available_ver: comparison = compare_versions(installed_ver, available_ver) if comparison < 0: return "Yes", installed_ver, available_ver, local_md5, available_md5, "Update available" if comparison > 0: return "Yes", installed_ver, available_ver, local_md5, available_md5, "Needs uploaded to server" if local_md5 and available_md5: if local_md5 == available_md5: return "Yes", installed_ver, available_ver, local_md5, available_md5, "Current (hashed)" return "Yes", installed_ver, available_ver, local_md5, available_md5, "Current (unhashed)" return "Yes", installed_ver, available_ver, local_md5, available_md5, "Current (unhashed)" if installed_ver: return "Yes", installed_ver, available_ver, local_md5, available_md5, "Installed (ver unknown)" return "Yes", installed_ver, available_ver, local_md5, available_md5, "Installed" if installed_ver: return "Yes", installed_ver, "", local_md5, "", "Local only" return "Yes", "", "", local_md5, "", "Local only" def get_plugin_display_sort_key(self, plugin): filename = str(plugin.get("filename", "")).strip().lower() if ACTION_ITEMS_FIRST: data = self.plugin_map.get(plugin["filename"], {}) status = str(data.get("status", "")).strip().lower() if status in ["current", "current (hashed)", "current (unhashed)"]: priority = 1 else: priority = 0 return (priority, filename) return (0, filename) def rebuild_plugin_state(self): plugin_map = {} for plugin in self.plugins: installed, installed_ver, available_ver, local_md5, available_md5, status = self.get_plugin_status(plugin) plugin_data = dict(plugin) plugin_data["installed"] = installed plugin_data["installed_ver"] = installed_ver plugin_data["available_ver"] = available_ver plugin_data["local_md5"] = local_md5 plugin_data["available_md5"] = available_md5 plugin_data["status"] = status plugin_map[plugin["filename"]] = plugin_data self.plugin_map = plugin_map def reload_plugins(self): self.status_text.set("Loading plugin list.") self.root.update_idletasks() try: self.plugins = self.load_plugins() local_only_plugins = self.get_local_only_plugins() if local_only_plugins: self.plugins.extend(local_only_plugins) self.plugins.sort(key=lambda p: p["filename"].lower()) self.rebuild_plugin_state() self.selected_files = set() self.render_plugins() self.update_status_text("Loaded") self.refresh_upload_button_state() except Exception as exc: self.plugins = [] self.plugin_map = {} self.selected_files = set() self.render_plugins() tkMessageBox.showerror( "Plugin List Error", "Could not load plugin list from:\n%s\n\n%s" % (PLUGINS_JSON_URL, str(exc)) ) self.status_text.set("Failed to load plugin list.") self.refresh_upload_button_state() def render_plugins(self): self.render_plugins_tree() item_ids = self.get_tree_item_ids() if item_ids: self.set_tree_focus_item(item_ids[0]) self.refresh_tree_focus_highlight() def render_plugins_tree(self): for item_id in self.tree.get_children(): self.tree.delete(item_id) display_plugins = sorted( self.plugins, key=self.get_plugin_display_sort_key ) for plugin in display_plugins: data = self.plugin_map[plugin["filename"]] status_text = str(data.get("status", "")).strip() if self.local_file_visibility_mode == "hide_local_only": if status_text == "Local only": continue elif self.local_file_visibility_mode == "show_only_local": if status_text != "Local only": continue notes = self.get_plugin_notes(plugin["filename"]) values = [ data["filename"], notes, data["installed"], data["installed_ver"] or "-", data["available_ver"] or "-", data["status"] ] tags = self.get_tree_tags_for_status(data["status"]) item_id = self.tree.insert( "", "end", values=tuple(values), tags=tags ) if plugin["filename"] in self.selected_files: self.tree.selection_add((item_id,)) def select_all_plugins(self, select_all=True): item_ids = self.get_tree_item_ids() if select_all: self.tree.selection_set(()) for item_id in item_ids: self.tree.selection_add((item_id,)) else: for item_id in item_ids: self.tree.selection_remove((item_id,)) if item_ids: self.set_tree_focus_item(item_ids[0]) self.on_tree_select(None) def get_selected_plugin_names(self): return [name for name in self.selected_files if name in self.plugin_map] def download_plugin(self, filename, backup_existing=False): destination_path = self.get_local_path(filename) if backup_existing: self.create_backup_file(filename) if self.is_premium_plugin(filename): self.download_plugin_with_curl(filename) return source_url = self.get_download_url(filename) self.ensure_plugin_dir_exists() if not os.access(self.plugin_dir, os.W_OK): raise IOError("plugin folder is not writable: %s" % self.plugin_dir) response = urllib.urlopen(source_url) http_code = response.getcode() final_url = source_url geturl_method = getattr(response, "geturl", None) if callable(geturl_method): final_url = geturl_method() info = response.info() content_type = "" if info: gettype_method = getattr(info, "gettype", None) if callable(gettype_method): content_type = gettype_method() else: content_type = info.get("Content-Type", "") content = response.read() self.validate_download_response( source_url, final_url, http_code, content_type, content ) handle = open(destination_path, "wb") try: handle.write(content) finally: handle.close() if not os.path.isfile(destination_path): raise IOError("file was not created: %s" % destination_path) if os.path.getsize(destination_path) <= 0: raise IOError("saved file is empty: %s" % destination_path) def ask_redownload_existing_file(self, filename): local_path = self.get_local_path(filename) return tkMessageBox.askyesno( "File Already Installed", "%s is already installed locally.\n\n" "Do you want to redownload it?\n\n" "The existing local file will be renamed as a timestamped backup first:\n\n%s" % ( filename, local_path ) ) def install_selected(self): selected = self.get_selected_plugin_names() if not selected: tkMessageBox.showwarning("No Selection", "Select at least one plugin.") return premium_selected = [name for name in selected if self.is_premium_plugin(name)] if premium_selected: tkMessageBox.showinfo( "Credentials Required", "These plugin(s) require your username and password before download:\n\n%s" % "\n".join(premium_selected) ) completed = [] skipped = [] failed = [] for filename in selected: data = self.plugin_map[filename] backup_existing = False if data["installed"] == "Yes": if self.ask_redownload_existing_file(filename): backup_existing = True else: skipped.append("%s: already installed" % filename) continue if backup_existing: self.status_text.set("Redownloading %s." % filename) else: self.status_text.set("Installing %s." % filename) self.root.update_idletasks() try: self.download_plugin(filename, backup_existing=backup_existing) if backup_existing: completed.append("%s: redownloaded; previous local file backed up" % filename) else: completed.append(filename) except Exception as exc: failed.append("%s: %s" % (filename, str(exc))) self.finish_action("Install Results", completed, skipped, failed) def update_selected(self): selected = self.get_selected_plugin_names() if not selected: tkMessageBox.showwarning("No Selection", "Select at least one plugin.") return premium_selected = [name for name in selected if self.is_premium_plugin(name)] if premium_selected: tkMessageBox.showinfo( "Credentials Required", "These plugin(s) require your username and password before download:\n\n%s" % "\n".join(premium_selected) ) completed = [] skipped = [] failed = [] for filename in selected: data = self.plugin_map[filename] if data["installed"] != "Yes": self.status_text.set("Installing %s..." % filename) self.root.update_idletasks() try: self.download_plugin(filename) completed.append("%s: installed" % filename) except Exception as exc: failed.append("%s: %s" % (filename, str(exc))) continue if data["status"] != "Update available": skipped.append("%s: no update needed" % filename) continue self.status_text.set("Updating %s..." % filename) self.root.update_idletasks() try: self.download_plugin(filename, backup_existing=True) completed.append("%s: updated" % filename) except Exception as exc: failed.append("%s: %s" % (filename, str(exc))) self.finish_action("Update Results", completed, skipped, failed) def remove_selected(self): selected = self.get_selected_plugin_names() if not selected: tkMessageBox.showwarning("No Selection", "Select at least one plugin.") return removable = [] for filename in selected: data = self.plugin_map[filename] if data["installed"] == "Yes": removable.append(filename) if not removable: tkMessageBox.showwarning("Nothing To Remove", "None of the selected plugins are installed.") return confirmed = tkMessageBox.askyesno( "Confirm Remove", "Remove these plugin file(s)?\n\n%s" % "\n".join(removable) ) if not confirmed: return completed = [] failed = [] for filename in removable: self.status_text.set("Removing %s..." % filename) self.root.update_idletasks() try: os.remove(self.get_local_path(filename)) completed.append(filename) except Exception as exc: failed.append("%s: %s" % (filename, str(exc))) self.finish_action("Remove Results", completed, [], failed) def telemetry_action(self, action, completed=None, skipped=None, failed=None, detail=""): try: if completed is None: completed = [] if skipped is None: skipped = [] if failed is None: failed = [] result = "ok" if failed: if completed or skipped: result = "partial" else: result = "failed" files_text_parts = [] if completed: files_text_parts.append("completed: " + ", ".join(completed)) if skipped: files_text_parts.append("skipped: " + ", ".join(skipped)) if failed: files_text_parts.append("failed: " + ", ".join(failed)) files_text = " | ".join(files_text_parts) phone_home_async( self.remote_session, event_type="action", action=action, files=files_text, result=result, detail=detail ) except: pass def finish_action(self, title, completed, skipped, failed): self.telemetry_action(title, completed, skipped, failed) self.rebuild_plugin_state() self.render_plugins() message_parts = [] if completed: completed_message = ( "Completed %d file(s):\n%s" % ( len(completed), "\n".join(completed))) if title == "Install Results": completed_message += ("\n\nMost plugins require restarting PathPilot to take effect.") message_parts.append(completed_message) if skipped: message_parts.append("Skipped %d file(s):\n%s" % (len(skipped), "\n".join(skipped))) if failed: message_parts.append("Failed %d file(s):\n%s" % (len(failed), "\n".join(failed))) if not message_parts: message_parts.append("No changes made.") if failed: tkMessageBox.showerror(title, "\n\n".join(message_parts)) self.update_status_text("Completed with errors. Loaded") else: tkMessageBox.showinfo(title, "\n\n".join(message_parts)) self.update_status_text("Done. Loaded") def show_selected_details(self): selected = self.get_selected_plugin_names() if not selected: tkMessageBox.showwarning("No Selection", "Select a plugin to view details.") return if len(selected) > 1: tkMessageBox.showwarning("Too Many Selected", "Select only one plugin to view details.") return self.show_plugin_details(selected[0]) def open_plugin_details_window(self, title_text, message): dialog = tk.Toplevel(self.root) dialog.title(title_text) dialog.transient(self.root) dialog.geometry("900x650") dialog.minsize(700, 450) main_frame = tk.Frame(dialog, padx=12, pady=12) main_frame.pack(fill=tk.BOTH, expand=True) text_frame = tk.Frame(main_frame) text_frame.pack(fill=tk.BOTH, expand=True) y_scrollbar = tk.Scrollbar(text_frame) y_scrollbar.pack(side=tk.RIGHT, fill=tk.Y) details_font = tkFont.Font( family=DEFAULT_FONT_FAMILY, size=11 ) text_widget = tk.Text( text_frame, wrap="word", yscrollcommand=y_scrollbar.set, font=details_font, padx=10, pady=10 ) text_widget.pack(side=tk.LEFT, fill=tk.BOTH, expand=True) y_scrollbar.config(command=text_widget.yview) text_widget.insert("1.0", message) text_widget.config(state="disabled") buttons_frame = tk.Frame(main_frame) buttons_frame.pack(fill=tk.X, pady=(10, 0)) close_button = tk.Button( buttons_frame, text="Close", width=12, command=dialog.destroy ) close_button.pack(side=tk.RIGHT) dialog.bind("<Escape>", lambda event: dialog.destroy()) dialog.bind("<Return>", lambda event: dialog.destroy()) close_button.focus_set() def show_plugin_details(self, filename): if filename in self.plugin_map: pass else: return data = self.plugin_map[filename] description = data.get("description", "").strip() or "(no description)" installed_ver = data.get("installed_ver", "") or "unknown" available_ver = data.get("available_ver", "") or "unknown" local_md5 = data.get("local_md5", "").strip() or "unknown" available_md5 = data.get("available_md5", "").strip() or "unknown" premium_note = "" if self.is_premium_plugin(filename): premium_note = ( "\nDownload access: Username/password required.\n" ) message = ( "Filename: %s\n\n" "Installed: %s\n" "Installed version: %s\n" "Available version: %s\n\n" "Local MD5: %s\n" "Available MD5: %s\n\n" "Status: %s%s\n\n" "Description:\n%s" ) % ( data["filename"], data["installed"], installed_ver, available_ver, local_md5, available_md5, data["status"], premium_note, description ) self.open_plugin_details_window(filename, message) def main(): root = tk.Tk() try: root.state("zoomed") except Exception: try: root.attributes("-zoomed", True) except Exception: pass PluginManagerApp(root) root.mainloop() if __name__ == "__main__": main() DESCRIPTION_LONG = """ This is a standalone utility that gives you access to the full suite of TormachTips plugins and scripts in one place. It is designed to simplify installation, updates, and other common maintenance tasks, so you do not have to manage each script manually. For most users, this is the easiest way to get started. You install or run this one script, and it can take care of installing the rest of the scripts I offer. It can also help you check for updates and keep your tools current. Your machine must be connected to the Internet for this to work. The command below downloads the installer from my website and immediately runs it. To use it, open a terminal by pressing CTRL + ALT + X on your machine, then type this command exactly: curl https://tormachtips.com/downloader.py | python A few important notes: The s in https matters. The .py at the end of the file name matters. The | symbol is the pipe character and it matters. It is usually found above or near the Enter key. The command must end with python. If entered correctly, this gives you a quick and simple way to install or manage the rest of the TormachTips scripts from a single utility. <a href="images/downloader.png"> <img border="2" src="images/downloader_small.png" xthumbnail-orig-image="images/downloader.png" width="200" height="150"></a>"""

# coding: utf-8 # python 2 only # Copyright (c) 2026 TormachTips.com. All rights reserved. # Licensed under the TormachTips Personal Use License. # Permission is granted only for private personal use and private personal modification. # No sharing, publication, distribution, resale, sublicensing, screenshots, code excerpts, # benchmarks, or videos are permitted without prior written permission. # Requests: tormach.1100m@gmail.com # Information page: https://tormachtips.com/plugins.htm ############################################# ## ## ## Download & Update Manager 1.10 ## ## www.tormachtips.com ## ## ## ############################################# # 1.10 - added phone home telemetry - 5/12/2026 # 1.09 - added HOME and END functionality - 5/2/2026 # 1.08 - now asks if you wish to redownload pre existing local file - 5/1/2026 # 1.07 - added knobpendant premium files - 4/28/2026 # 1.06 - added feature to backup remote files when uploading new versions - 4/25/2026 # 1.05 - added feature to show only local files - 4/23/2026 # 1.04 - UI improvements. - 4/19/2026 # 1.03 - Added Show / Hide Local Files Only. - 4/17/2026 # 1.02 - Added logic for local-only scripts. - 4/17/2026 # 1.01 - Added pendants premium files and json directory. - 4/16/2026 # 1.00 - Added some more admin tools. - 4/15/2026 # 0.99 - Updated Help Section - 4/15/2026 # 0.98 - adds bolding, md5 comparison, and change log updating - 4/13/2026 # 0.97 - adds pre-install checks. - 4/13/2026 # 0.96 - adds upload to server logic. - 4/12/2026 import os import re import json import urllib import Tkinter as tk import tkMessageBox import tkFont import stat import subprocess import tkSimpleDialog import sys import ttk import datetime import ftplib import StringIO import tempfile import webbrowser import hashlib import threading import socket CURRENT_VER = "1.10" SCRIPT_NAME = "Download & Update Manager" DESCRIPTION = "Download and Update Manager for TormachTips scripts." WINDOW_TITLE = "TormachTips Plugin Manager " + CURRENT_VER BASE_URL = "https://tormachtips.com/plugins/" PREMIUM_BASE_URL = "https://tormachtips.com/load_meter/" PENDANT_BASE_URL = "https://tormachtips.com/pendants/" PLUGINS_JSON_FILE = "plugins.json" PLUGINS_JSON_URL = BASE_URL + PLUGINS_JSON_FILE UI_HOOKS_FILENAME = "ui_hooks.py" UI_HOOKS_URL = BASE_URL + UI_HOOKS_FILENAME DEFAULT_PLUGIN_DIR = "/home/operator/gcode/python/" DEFAULT_DESKTOP_DIR = os.path.expanduser("~/Desktop") DEFAULT_STATUS = "Loading plugins..." DEFAULT_FONT_FAMILY = "DejaVu Sans Mono" DEFAULT_FONT_SIZE = 10 PADDING = 10 STATUS_HIGHLIGHT_BG = "#F5DEB3" ACTION_ITEMS_FIRST = False SHORTCUT_FILENAME = "TormachTips Plugin Manager.sh" SHORTCUT_CURL_COMMAND = "curl https://tormachtips.com/downloader.py | python2" PHONE_HOME = 1 PHONE_HOME_URL = "https://tormachtips.com/cgi-bin/phone_home2.cgi" PHONE_HOME_TIMEOUT_SECONDS = 3 PHONE_HOME_ACTIONS = 1 PHONE_HOME_MAX_FIELD_CHARS = 1200 TMC_SYMLINK_PATH = "/home/operator/tmc" PATHPILOT_JSON_CANDIDATES = "/home/operator/pathpilot.json|/home/operator/tmc/configs/tormach_mill/pathpilot.json|/home/operator/tmc/configs/tormach_lathe/pathpilot.json".split("|") FTP_CONFIG_PATH = "/home/operator/gcode/python/download_manager_ftp.json" FTP_TIMEOUT_SECONDS = 20 FTP_DEBUG_LEVEL = 2 CURL_PATHS = "/usr/bin/curl|/bin/curl".split("|") HTML_ERROR_MARKERS = "|", "-c"]: if os.path.isfile(os.path.abspath(script_name)): return False stdin_obj = getattr(sys, "stdin", None) if stdin_obj: is_tty_method = getattr(stdin_obj, "isatty", None) if callable(is_tty_method): if is_tty_method(): return False return True def get_local_ip_for_telemetry(): try: sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) try: sock.connect(("8.8.8.8", 80)) return sock.getsockname()[0] finally: sock.close() except: return "" def read_json_file_safely(path): try: if not path: return {} if not os.path.isfile(path): return {} handle = open(path, "rb") try: content = handle.read() finally: handle.close() return json.loads(content) except: return {} def get_pathpilot_json_path(): for path in PATHPILOT_JSON_CANDIDATES: try: if os.path.isfile(path): return path except: pass try: if os.path.exists(TMC_SYMLINK_PATH): real_tmc_path = os.path.realpath(TMC_SYMLINK_PATH) for root, dirs, files in os.walk(real_tmc_path): if "pathpilot.json" in files: return os.path.join(root, "pathpilot.json") except: pass return "" def get_pathpilot_version_info(): result = { "pathpilot_version": "", "pathpilot_version_path": "", } try: if os.path.exists(TMC_SYMLINK_PATH): real_path = os.path.realpath(TMC_SYMLINK_PATH) result["pathpilot_version_path"] = real_path result["pathpilot_version"] = os.path.basename(real_path) except: pass return result def get_machine_telemetry(): result = { "machine_model": "", "machine_class": "", "machine_sim": "", "machine_cesim": "", "machine_rapidturn": "", "communication_method": "", "pathpilot_version": "", "pathpilot_version_path": "", } try: json_path = get_pathpilot_json_path() config = read_json_file_safely(json_path) machine_data = config.get("machine", {}) result["machine_model"] = str(machine_data.get("model", "")).strip() result["machine_class"] = str(machine_data.get("class", "")).strip() result["machine_sim"] = "1" if machine_data.get("sim", False) else "0" result["machine_cesim"] = "1" if machine_data.get("cesim", False) else "0" result["machine_rapidturn"] = "1" if machine_data.get("rapidturn", False) else "0" result["communication_method"] = str(machine_data.get("communication_method", "")).strip() except: pass try: version_info = get_pathpilot_version_info() result["pathpilot_version"] = version_info.get("pathpilot_version", "") result["pathpilot_version_path"] = version_info.get("pathpilot_version_path", "") except: pass return result def trim_telemetry_field(value): try: value = str(value) except: value = "" value = value.replace("\r", " ") value = value.replace("\n", " ") value = value.replace("\t", " ") if len(value) > PHONE_HOME_MAX_FIELD_CHARS: value = value[:PHONE_HOME_MAX_FIELD_CHARS] + "..." return value def phone_home_async(remote_session=False, event_type="startup", action="", files="", result="", detail=""): if not PHONE_HOME: return if event_type != "startup": if not PHONE_HOME_ACTIONS: return thread_obj = threading.Thread( target=phone_home_worker, args=(remote_session, event_type, action, files, result, detail) ) thread_obj.daemon = True thread_obj.start() def phone_home_worker(remote_session=False, event_type="startup", action="", files="", result="", detail=""): old_timeout = None try: machine_telemetry = get_machine_telemetry() data = { "script": SCRIPT_NAME, "version": CURRENT_VER, "local_ip": get_local_ip_for_telemetry(), "remote_session": "1" if remote_session else "0", "platform": sys.platform, "hostname": socket.gethostname(), "user": str(os.getenv("USER", "")), "machine_model": machine_telemetry.get("machine_model", ""), "machine_class": machine_telemetry.get("machine_class", ""), "machine_sim": machine_telemetry.get("machine_sim", ""), "machine_cesim": machine_telemetry.get("machine_cesim", ""), "machine_rapidturn": machine_telemetry.get("machine_rapidturn", ""), "communication_method": machine_telemetry.get("communication_method", ""), "pathpilot_version": machine_telemetry.get("pathpilot_version", ""), "pathpilot_version_path": machine_telemetry.get("pathpilot_version_path", ""), "event_type": trim_telemetry_field(event_type), "action": trim_telemetry_field(action), "files": trim_telemetry_field(files), "result": trim_telemetry_field(result), "detail": trim_telemetry_field(detail), } encoded_data = urllib.urlencode(data) try: old_timeout = socket.getdefaulttimeout() socket.setdefaulttimeout(PHONE_HOME_TIMEOUT_SECONDS) except: pass response = urllib.urlopen(PHONE_HOME_URL, encoded_data) response.read() except: pass try: socket.setdefaulttimeout(old_timeout) except: pass def compare_versions(left, right): left_key = version_key(left) right_key = version_key(right) if not left_key or not right_key: return 0 max_len = max(len(left_key), len(right_key)) for idx in range(max_len): left_part = left_key[idx] if idx < len(left_key) else 0 right_part = right_key[idx] if idx < len(right_key) else 0 if left_part == right_part: continue left_is_int = isinstance(left_part, int) right_is_int = isinstance(right_part, int) if left_is_int and right_is_int: if left_part < right_part: return -1 return 1 left_text = str(left_part) right_text = str(right_part) if left_text < right_text: return -1 return 1 return 0 class PluginManagerApp(object): def __init__(self, root): self.root = root self.root.title(WINDOW_TITLE) self.default_font = tkFont.Font( family=DEFAULT_FONT_FAMILY, size=DEFAULT_FONT_SIZE ) self.root.option_add("*Font", self.default_font) self.root.bind_class("Entry", "", lambda event: event.widget.event_generate("")) self.root.bind_class("TEntry", "", lambda event: event.widget.event_generate("")) self.ftp_config_path = FTP_CONFIG_PATH self.plugin_dir = DEFAULT_PLUGIN_DIR self.plugins = [] self.plugin_map = {} self.selected_files = set() self.last_tree_focus = "" self.tree_type_search_text = "" self.tree_type_search_after_id = None self.tree_type_search_reset_ms = 900 self.local_file_visibility_mode = "hide_local_only" self.download_username = "" self.download_password = "" self.ftp_password = "" self.remote_session = is_remote_session() phone_home_async(self.remote_session) self.status_text = tk.StringVar() self.status_text.set(DEFAULT_STATUS) self._build_ui() self.run_startup_sequence() def build_remote_backup_filename(self, remote_filename): timestamp_text = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") return "%s.%s" % (remote_filename, timestamp_text) def backup_remote_ftp_file(self, ftp, remote_filename): backup_filename = self.build_remote_backup_filename(remote_filename) try: ftp.rename(remote_filename, backup_filename) return backup_filename except Exception as exc: error_text = str(exc).strip() if "550" in error_text: return "" raise IOError("Could not backup existing server file %s:\n%s" % (remote_filename, error_text)) def get_tree_tags_for_status(self, status_text, is_focus_row=False): tags = [] status_text = str(status_text).strip().lower() if status_text in ["current (hashed)", "not installed"]: pass else: tags.append("status_highlight") if is_focus_row: tags.append("focus_row") return tuple(tags) def refresh_local_only_toggle_button_text(self): if self.local_file_visibility_mode == "hide_local_only": self.local_only_toggle_button.config(text="Show Local Files") return if self.local_file_visibility_mode == "show_all": self.local_only_toggle_button.config(text="Show Local Only") return self.local_only_toggle_button.config(text="Hide Local Files") def toggle_local_only_rows(self): if self.local_file_visibility_mode == "hide_local_only": self.local_file_visibility_mode = "show_all" elif self.local_file_visibility_mode == "show_all": self.local_file_visibility_mode = "show_only_local" else: self.local_file_visibility_mode = "hide_local_only" self.refresh_local_only_toggle_button_text() self.render_plugins() self.update_status_text("Loaded") def clear_premium_credentials(self): self.download_username = "" self.download_password = "" def clear_ftp_password(self): self.ftp_password = "" def view_plugins_json_from_server(self): try: response = urllib.urlopen(PLUGINS_JSON_URL) code = response.getcode() content = response.read() if code is not None and code != 200: raise IOError("server returned HTTP %s" % code) if content: pass else: raise IOError("plugins.json is empty") lowered = content.lstrip().lower() for marker in HTML_ERROR_MARKERS: if lowered.startswith(marker): raise IOError("server returned HTML instead of plugins.json") self.open_text_in_gedit( content, prefix="tormachtips_plugins_json_", suffix=".json" ) self.status_text.set("Opened server plugins.json in gedit.") except Exception as exc: tkMessageBox.showerror( "View JSON Error", "Could not open server plugins.json.\n\n%s" % str(exc) ) def show_help_dialog(self): help_text = ( "TormachTips Download & Update Manager\n" "\n" "This program serves as a real-time repository for my PathPilot plugins and\n" "standalone scripts.\n\n" "This script can either be downloaded and run locally like any other plugin,\n" "or it can be run directly from the cloud via a curl command or the Create\n" "Desktop Icon button.\n" "\n" "The advantage to that is you are always using the latest version of this script.\n" "The downside is that the machine must have internet access.\n" "\n" "Double-click any filename to view details, including a short description of\n" "what it does. For full descriptions, visit:\n\n" "https://tormachtips.com/plugins\n" "\n" "Most columns are self-explanatory, but the Status column includes a few that\n" "may not be obvious, especially Current (hashed) and Current (unhashed).\n" "\n" "If a plugin shows Current (hashed), that means your local copy is an exact\n" "match to the released version.\n" "\n" "If you edit the file, even by changing a single character, the MD5 checksum\n" "changes and the file is no longer an exact match. That usually only matters\n" "to developers, but it helps with version tracking and file verification.\n\ntormach.1100m@gmail.com" ) dialog = tk.Toplevel(self.root) dialog.title("Help") dialog.transient(self.root) screen_width = dialog.winfo_screenwidth() screen_height = dialog.winfo_screenheight() dialog_width = int(screen_width * 0.90) dialog_height = int(screen_height * 0.90) dialog_x = (screen_width - dialog_width) / 2 dialog_y = (screen_height - dialog_height) / 2 dialog.geometry("%dx%d+%d+%d" % ( dialog_width, dialog_height, dialog_x, dialog_y )) dialog.minsize(700, 500) main_frame = tk.Frame(dialog, padx=12, pady=12) main_frame.pack(fill=tk.BOTH, expand=True) text_frame = tk.Frame(main_frame) text_frame.pack(fill=tk.BOTH, expand=True) y_scrollbar = tk.Scrollbar(text_frame) y_scrollbar.pack(side=tk.RIGHT, fill=tk.Y) help_widget = tk.Text( text_frame, wrap="word", yscrollcommand=y_scrollbar.set ) help_widget.pack(side=tk.LEFT, fill=tk.BOTH, expand=True) y_scrollbar.config(command=help_widget.yview) help_widget.insert("1.0", help_text) help_widget.config(state="disabled") buttons_frame = tk.Frame(main_frame) buttons_frame.pack(fill=tk.X, pady=(10, 0)) close_button = tk.Button( buttons_frame, text="Close", width=12, command=dialog.destroy ) close_button.pack(side=tk.RIGHT) dialog.bind("", lambda event: dialog.destroy()) dialog.bind("", lambda event: dialog.destroy()) close_button.focus_set() def invoke_button_from_accelerator(self, button): if button: pass else: return try: state_value = str(button.cget("state")).strip().lower() except Exception: state_value = "" if state_value == "disabled": return try: button.invoke() except Exception: pass def bind_button_accelerator(self, key_text, button): sequence = "" % key_text.lower() self.root.bind_all( sequence, lambda event, target_button=button: self.invoke_button_from_accelerator(target_button) ) upper_sequence = "" % key_text.upper() self.root.bind_all( upper_sequence, lambda event, target_button=button: self.invoke_button_from_accelerator(target_button) ) def get_active_plugin_name(self): focus_item = self.get_tree_focus_item() if focus_item: filename = self.tree.set(focus_item, "filename") if filename: return filename selected = self.get_selected_plugin_names() if len(selected) == 1: return selected[0] return "" def get_plugin_notes(self, filename): notes = [] if filename == "download_manager.py": notes.append("this script") if filename == "ui_hooks.py": notes.append("required") if self.is_premium_plugin(filename): notes.append("premium") return ", ".join(notes) def change_log_active_plugin(self): filename = self.get_active_plugin_name() if filename: pass else: tkMessageBox.showwarning( "No Plugin Selected", "Select or focus a plugin first." ) return local_path = self.get_local_path(filename) if os.path.isfile(local_path): pass else: tkMessageBox.showwarning( "Local File Missing", "This plugin is not installed locally:\n%s" % local_path ) return try: original_content = self.read_file_bytes(local_path) except Exception as exc: tkMessageBox.showerror( "Read Error", "Could not read local file.\n\n%s" % str(exc) ) return current_version, current_description = self.get_script_version_and_description( original_content ) if current_version: pass else: tkMessageBox.showerror( "Version Not Found", "CURRENT_VER was not found in:\n%s" % local_path ) return if current_description: pass else: tkMessageBox.showerror( "Description Not Found", "DESCRIPTION was not found in:\n%s" % local_path ) return new_version = tkSimpleDialog.askstring( "Change CURRENT_VER", "Current CURRENT_VER: %s\n\nEnter new CURRENT_VER:" % current_version, initialvalue=current_version ) if new_version is None: return new_version = normalize_version(new_version) if new_version: pass else: tkMessageBox.showwarning( "Invalid Version", "CURRENT_VER cannot be blank." ) return change_text = tkSimpleDialog.askstring( "Change Log Text", "Enter changelog text for version %s:" % new_version ) if change_text is None: return change_text = change_text.strip() if change_text: pass else: tkMessageBox.showwarning( "Invalid Change Log Text", "Change log text cannot be blank." ) return new_description = tkSimpleDialog.askstring( "Change DESCRIPTION", "Current DESCRIPTION:\n%s\n\nEnter new DESCRIPTION:" % current_description, initialvalue=current_description ) if new_description is None: return new_description = new_description.strip() if new_description: pass else: tkMessageBox.showwarning( "Invalid Description", "DESCRIPTION cannot be blank." ) return try: self.create_backup_file(filename) updated_content = original_content updated_content = self.update_script_version_and_description_constants( updated_content, new_version, new_description ) updated_content = self.update_script_header_banner_version( updated_content, new_version ) updated_content = self.add_script_changelog_line( updated_content, new_version, change_text ) self.write_file_bytes(local_path, updated_content) except Exception as exc: tkMessageBox.showerror( "Change Log Error", "Could not update file.\n\n%s" % str(exc) ) return self.reload_plugins() try: subprocess.Popen(["gedit", local_path]) self.status_text.set("Updated and opened %s in gedit." % filename) except Exception as exc: self.status_text.set("Updated %s, but could not open gedit." % filename) tkMessageBox.showwarning( "Gedit Open Warning", "The file was updated, but gedit could not be opened.\n\n%s" % str(exc) ) tkMessageBox.showinfo( "Change Log Updated", "Updated:\n%s\n\nCURRENT_VER: %s\nDESCRIPTION: %s" % ( filename, new_version, new_description ) ) def add_script_changelog_line(self, content, new_version, change_text): today_now = datetime.datetime.now() today_text = "%d/%d/%d" % (today_now.month, today_now.day, today_now.year) change_text = str(change_text).strip() changelog_line = "# %s - %s - %s\n" % (new_version, change_text, today_text) banner_block_pattern = re.compile( r'(?ms)^(?P#{20,}\s*\n)' r'(?P(?:^\s*##.*##\s*\n)+)' r'(?P#{20,}\s*\n)' ) banner_match = banner_block_pattern.search(content) if banner_match: insert_at = banner_match.end() while insert_at < len(content) and content[insert_at] in "\r\n": insert_at += 1 return content[:banner_match.end()] + changelog_line + content[insert_at:] website_line_pattern = re.compile( r'^\s*##\s*(?:https?://)?(?:www\.)?tormachtips\.com\s*##\s*\n', re.MULTILINE | re.IGNORECASE ) website_match = website_line_pattern.search(content) if website_match: following_text = content[website_match.end():] next_border_match = re.search( r'^#{20,}\s*\n', following_text, re.MULTILINE ) if next_border_match: insert_at = website_match.end() + next_border_match.end() while insert_at < len(content) and content[insert_at] in "\r\n": insert_at += 1 return content[:website_match.end() + next_border_match.end()] + changelog_line + content[insert_at:] raise ValueError("Could not find where to insert changelog line.") def update_script_header_banner_version(self, content, new_version): banner_pattern = re.compile( r'^(?P\s*##\s*)(?P.*?)(?P<gap>\s+)(?P<old_version>v?[0-9]+(?:\.[0-9]+)*)(?P<trailing>\s*)(?P<suffix>##\s*)$', re.MULTILINE ) def replace_banner_line(match): title_text = match.group("title").strip() title_text = re.sub(r'\s+', ' ', title_text) old_version = match.group("old_version").strip() trailing_spaces = match.group("trailing") if old_version.lower().startswith("v"): trailing_spaces = trailing_spaces + " " return "%s%s%s%s%s%s" % ( match.group("prefix"), title_text, match.group("gap"), new_version, trailing_spaces, match.group("suffix") ) updated_content, count = banner_pattern.subn(replace_banner_line, content, 1) if count: return updated_content website_pattern = re.compile( r'^(?P<line>\s*##\s*www\.tormachtips\.com\s*##\s*)$', re.MULTILINE | re.IGNORECASE ) website_match = website_pattern.search(content) if website_match: start_pos = website_match.start() earlier_text = content[:start_pos] candidates = list(banner_pattern.finditer(earlier_text)) if candidates: last_match = candidates[-1] replacement = replace_banner_line(last_match) return ( content[:last_match.start()] + replacement + content[last_match.end():] ) raise ValueError("Header banner version line was not found.") def update_script_version_and_description_constants( self, content, new_version, new_description ): content, version_count = re.subn( r'^(\s*CURRENT_VER\s*=\s*[\'"])v?([^\'"]*)([\'"])', r'\g<1>%s\g<3>' % new_version, content, count=1, flags=re.MULTILINE ) if version_count == 0: raise ValueError("CURRENT_VER assignment was not found.") content, description_count = re.subn( r'^(\s*DESCRIPTION\s*=\s*[\'"])([^\'"]*)([\'"])', r'\g<1>%s\g<3>' % new_description.replace("\\", "\\\\"), content, count=1, flags=re.MULTILINE ) if description_count == 0: raise ValueError("DESCRIPTION assignment was not found.") return content def get_script_version_and_description(self, content): version_value = "" description_value = "" version_match = re.search( r'^\s*CURRENT_VER\s*=\s*[\'"]([^\'"]*)[\'"]', content, re.MULTILINE ) if version_match: version_value = normalize_version(version_match.group(1)) description_match = re.search( r'^\s*DESCRIPTION\s*=\s*[\'"]([^\'"]*)[\'"]', content, re.MULTILINE ) if description_match: description_value = description_match.group(1).strip() return version_value, description_value def write_file_bytes(self, path, content): handle = open(path, "wb") try: handle.write(content) finally: handle.close() def edit_active_plugin_locally(self): filename = self.get_active_plugin_name() if filename: pass else: tkMessageBox.showwarning( "No Plugin Selected", "Select or focus a plugin first." ) return local_path = self.get_local_path(filename) if os.path.isfile(local_path): pass else: tkMessageBox.showwarning( "Local File Missing", "This plugin is not installed locally:\n%s" % local_path ) return try: subprocess.Popen(["gedit", local_path]) self.status_text.set("Opened local file in gedit.") except Exception as exc: tkMessageBox.showerror( "Edit Locally Error", "Could not open gedit.\n\n%s" % str(exc) ) def download_plugin_text_from_server(self, filename): source_url = self.get_download_url(filename) response = urllib.urlopen(source_url) code = response.getcode() content = response.read() if code is not None and code != 200: raise IOError("server returned HTTP %s" % code) if content: pass else: raise IOError("downloaded file is empty") lowered = content.lstrip().lower() for marker in HTML_ERROR_MARKERS: if lowered.startswith(marker): raise IOError("server returned HTML instead of a plugin file") return content def inspect_active_plugin_from_server(self): filename = self.get_active_plugin_name() if filename: pass else: tkMessageBox.showwarning( "No Plugin Selected", "Select or focus a plugin first." ) return if self.is_premium_plugin(filename): tkMessageBox.showinfo( "View Server Copy Unavailable", "View Server Copy is not available for Premium plugins." ) return try: content = self.download_plugin_text_from_server(filename) temp_path = self.open_text_in_gedit( content, prefix="tormachtips_server_", suffix="_" + filename ) self.status_text.set("Opened server copy in gedit.") except Exception as exc: tkMessageBox.showerror( "Inspect From Server Error", "Could not open server copy.\n\n%s" % str(exc) ) def open_url_in_browser(self, url): browser_commands = [ ["google-chrome", url], ["google-chrome-stable", url], ["chromium-browser", url], ["chromium", url], ["xdg-open", url] ] for command in browser_commands: program = command[0] full_path = self.find_executable_in_path(program) if full_path: subprocess.Popen([full_path] + command[1:]) return True try: if webbrowser.open(url): return True except Exception: pass return False def rebuild_json(self): if os.path.isfile(self.ftp_config_path): pass else: tkMessageBox.showwarning( "FTP Config Missing", "Rebuild JSON is disabled because the FTP config file was not found:\n%s" % self.ftp_config_path ) self.refresh_upload_button_state() return try: config = self.load_ftp_config() except Exception as exc: tkMessageBox.showerror( "Rebuild JSON Error", str(exc) ) self.refresh_upload_button_state() return rebuild_json_url = str(config.get("rebuild_json_url", "")).strip() if rebuild_json_url: pass else: tkMessageBox.showerror( "Rebuild JSON Error", "FTP config is missing rebuild_json_url." ) return if self.open_url_in_browser(rebuild_json_url): self.status_text.set("Opened rebuild JSON URL.") else: tkMessageBox.showerror( "Browser Error", "Could not open a browser for:\n%s" % rebuild_json_url ) self.status_text.set("Could not open rebuild JSON URL.") def find_executable_in_path(self, program_name): if os.path.isabs(program_name): if os.path.isfile(program_name) and os.access(program_name, os.X_OK): return program_name return "" path_value = str(os.environ.get("PATH", "")).strip() if not path_value: return "" for path_dir in path_value.split(os.pathsep): path_dir = path_dir.strip() if not path_dir: continue candidate = os.path.join(path_dir, program_name) if os.path.isfile(candidate) and os.access(candidate, os.X_OK): return candidate return "" def run_startup_sequence(self): if self.remote_session: self.show_remote_session_warning() self.prompt_for_plugin_folder_setup() self.reload_plugins() def get_ui_hooks_path(self): return os.path.join(self.plugin_dir, UI_HOOKS_FILENAME) def has_plugin_dir(self): return os.path.isdir(self.plugin_dir) def has_ui_hooks(self): return os.path.isfile(self.get_ui_hooks_path()) def needs_plugin_folder_setup(self): if self.has_plugin_dir(): if self.has_ui_hooks(): return False return True return True def install_plugin_folder_setup(self): created_dir = False copied_ui_hooks = False if self.has_plugin_dir(): pass else: os.makedirs(self.plugin_dir) created_dir = True if self.has_ui_hooks(): pass else: response = urllib.urlopen(UI_HOOKS_URL) code = response.getcode() content = response.read() if code is not None and code != 200: raise IOError("server returned HTTP %s for %s" % (code, UI_HOOKS_URL)) if content: pass else: raise IOError("ui_hooks.py download is empty") lowered = content.lstrip().lower() for marker in HTML_ERROR_MARKERS: if lowered.startswith(marker): raise IOError("server returned HTML instead of ui_hooks.py") handle = open(self.get_ui_hooks_path(), "wb") try: handle.write(content) finally: handle.close() copied_ui_hooks = True return created_dir, copied_ui_hooks def prompt_for_plugin_folder_setup(self): if self.needs_plugin_folder_setup(): pass else: return dialog = tk.Toplevel(self.root) dialog.title("Plugin Setup Required") dialog.transient(self.root) dialog.grab_set() dialog.resizable(False, False) main_frame = tk.Frame(dialog, padx=14, pady=14) main_frame.pack(fill=tk.BOTH, expand=True) message = ( "Welcome to the TormachTips Download & Update Manager.\n\n" "Most PathPilot plugins need these two setup steps:\n\n" " Create %s\n" " Install %s\n\n" "Would you like me to do this now?" ) % (self.plugin_dir, UI_HOOKS_FILENAME) message_label = tk.Label( main_frame, text=message, justify=tk.LEFT, anchor="w" ) message_label.pack(fill=tk.X) done_font = tkFont.Font( family=DEFAULT_FONT_FAMILY, size=DEFAULT_FONT_SIZE, overstrike=1 ) created_dir_done = self.has_plugin_dir() ui_hooks_done = self.has_ui_hooks() if created_dir_done: dir_label_text = "Done: Create %s" % self.plugin_dir else: dir_label_text = "Pending: Create %s" % self.plugin_dir dir_label = tk.Label( main_frame, text=dir_label_text, justify=tk.LEFT, anchor="w" ) dir_label.pack(fill=tk.X, pady=(10, 0), padx=(16, 0)) if ui_hooks_done: hooks_label_text = "Done: Install %s" % UI_HOOKS_FILENAME else: hooks_label_text = "Pending: Install %s" % UI_HOOKS_FILENAME hooks_label = tk.Label( main_frame, text=hooks_label_text, justify=tk.LEFT, anchor="w" ) hooks_label.pack(fill=tk.X, pady=(4, 0), padx=(16, 0)) if created_dir_done: dir_label.config(fg="gray40", font=done_font) if ui_hooks_done: hooks_label.config(fg="gray40", font=done_font) buttons_frame = tk.Frame(main_frame) buttons_frame.pack(fill=tk.X, pady=(14, 0)) result = {"accepted": False} def on_yes(): result["accepted"] = True dialog.destroy() def on_no(): dialog.destroy() yes_button = tk.Button( buttons_frame, text="Yes", width=12, command=on_yes ) yes_button.pack(side=tk.LEFT) no_button = tk.Button( buttons_frame, text="No", width=12, command=on_no ) no_button.pack(side=tk.LEFT, padx=(6, 0)) dialog.protocol("WM_DELETE_WINDOW", on_no) self.root.wait_window(dialog) if result["accepted"]: pass else: self.status_text.set("Plugin prerequisite setup skipped.") return try: created_dir, copied_ui_hooks = self.install_plugin_folder_setup() except Exception as exc: tkMessageBox.showerror( "Setup Error", "Could not complete plugin setup.\n\n%s" % str(exc) ) self.status_text.set("Plugin prerequisite setup failed.") return result_lines = [] if created_dir: result_lines.append("%s created" % self.plugin_dir) if copied_ui_hooks: result_lines.append("%s installed" % UI_HOOKS_FILENAME) if result_lines: tkMessageBox.showinfo( "Setup Complete", "\n".join(result_lines) ) self.status_text.set("Plugin prerequisite setup completed.") else: self.status_text.set("Plugin prerequisite setup already complete.") def show_remote_session_warning(self): if self.remote_session: message = ( "Welcome to the TormachTips Download & Update Manager.\n\n" "This session was opened remotely from \"the cloud\".\n\n" "On the next page, you'll be presented with all my plugins " "(including this manager).\n\n" "You may continue using remote sessions, or download this " "script to run locally." ) tkMessageBox.showinfo("Remote Session Detected", message) def refresh_upload_button_state(self): show_ftp_buttons = os.path.isfile(self.ftp_config_path) if hasattr(self, "admin_buttons_row"): if show_ftp_buttons: if self.admin_buttons_row.winfo_manager(): pass else: self.admin_buttons_row.pack(fill=tk.X, pady=(6, 0)) else: if self.admin_buttons_row.winfo_manager(): self.admin_buttons_row.pack_forget() def prompt_for_ftp_password(self): if self.ftp_password: return True password = tkSimpleDialog.askstring( "FTP Password Required", "Enter FTP password:", show="" ) if password is None: return False password = password.strip() if password: self.ftp_password = password return True tkMessageBox.showwarning( "Missing Password", "A password is required for FTP upload." ) return False def update_status_text(self, prefix): self.status_text.set( "%s | Plugins: %d | Selected: %d" % ( prefix, len(self.plugins), len(self.selected_files) ) ) def is_load_meter_plugin(self, filename): return filename in PREMIUM_PLUGIN_FILES def is_pendant_plugin(self, filename): return filename in PENDANT_PLUGIN_FILES def is_premium_plugin(self, filename): if self.is_load_meter_plugin(filename): return True if self.is_pendant_plugin(filename): return True return False def get_upload_targets(self, filename): targets = [] targets.append({ "remote_filename": filename, "use_parent_of_remote_dir": False }) if filename == "download_manager.py": targets.append({ "remote_filename": "downloader.py", "use_parent_of_remote_dir": True }) return targets def get_download_url(self, filename): if self.is_load_meter_plugin(filename): return PREMIUM_BASE_URL + filename if self.is_pendant_plugin(filename): return PENDANT_BASE_URL + filename return BASE_URL + filename def find_curl_binary(self): for path in CURL_PATHS: if os.path.isfile(path) and os.access(path, os.X_OK): return path return None def prompt_for_premium_credentials(self): username = tkSimpleDialog.askstring( "Premium Download Login", "Enter username:") if username is None: return False password = tkSimpleDialog.askstring( "Premium Download Login", "Enter password:", show="") if password is None: return False username = username.strip() if username and password: self.download_username = username self.download_password = password return True tkMessageBox.showwarning( "Missing Credentials", "Both username and password are required for this plugin.") return False def validate_download_response( self, source_url, final_url, http_code, content_type, content ): if http_code is not None and http_code != 200: raise IOError( self.build_download_debug_message( source_url, final_url, http_code, content_type, content, "Server returned HTTP %s." % http_code ) ) if not content: raise IOError( self.build_download_debug_message( source_url, final_url, http_code, content_type, content, "Downloaded file is empty." ) ) lowered = content.lstrip().lower() content_type_text = str(content_type or "").lower() probe = lowered[:512] if "text/html" in content_type_text: raise IOError( self.build_download_debug_message( source_url, final_url, http_code, content_type, content, "Server returned HTML instead of a plugin file." ) ) for marker in HTML_ERROR_MARKERS: if probe.startswith(marker): raise IOError( self.build_download_debug_message( source_url, final_url, http_code, content_type, content, "Server returned HTML instead of a plugin file." ) ) def build_download_debug_message( self, source_url, final_url, http_code, content_type, content, prefix ): preview = "" if content: preview = content[:400] preview = preview.replace("\r", " ") preview = preview.replace("\n", " ") preview = preview.strip() details = [ prefix, "", "Requested URL: %s" % source_url, "Final URL: %s" % (final_url or "(unknown)"), "HTTP code: %s" % (http_code if http_code is not None else "(unknown)"), "Content-Type: %s" % (content_type or "(unknown)"), ] if preview: details.append("Response preview: %s" % preview) return "\n".join(details) def ensure_premium_credentials(self): if self.download_username and self.download_password: return True return self.prompt_for_premium_credentials() def download_plugin_with_curl(self, filename): curl_path = self.find_curl_binary() if not curl_path: raise IOError("curl was not found on this system") if not self.ensure_premium_credentials(): raise IOError("download cancelled: username/password required") source_url = self.get_download_url(filename) destination_path = self.get_local_path(filename) header_path = destination_path + ".headers.tmp" self.ensure_plugin_dir_exists() if not os.access(self.plugin_dir, os.W_OK): raise IOError("plugin folder is not writable: %s" % self.plugin_dir) command = [ curl_path, "--silent", "--show-error", "--location", "--user", "%s:%s" % (self.download_username, self.download_password), "--dump-header", header_path, "--output", destination_path, source_url, ] process = subprocess.Popen( command, stdout=subprocess.PIPE, stderr=subprocess.PIPE ) stdout_data, stderr_data = process.communicate() if process.returncode != 0: if os.path.isfile(destination_path): try: os.remove(destination_path) except Exception: pass if os.path.isfile(header_path): try: os.remove(header_path) except Exception: pass self.clear_premium_credentials() error_text = (stderr_data or "").strip() if not error_text: error_text = "curl failed with exit code %s" % process.returncode raise IOError(error_text) if not os.path.isfile(destination_path): if os.path.isfile(header_path): try: os.remove(header_path) except Exception: pass raise IOError("file was not created: %s" % destination_path) if os.path.getsize(destination_path) <= 0: if os.path.isfile(header_path): try: os.remove(header_path) except Exception: pass raise IOError("saved file is empty: %s" % destination_path) try: content = self.read_file_bytes(destination_path) except Exception as exc: if os.path.isfile(header_path): try: os.remove(header_path) except Exception: pass self.clear_premium_credentials() raise IOError("download completed but file could not be read: %s" % str(exc)) headers_text = "" if os.path.isfile(header_path): try: headers_text = self.read_file_bytes(header_path) except Exception: headers_text = "" final_url = source_url http_code = None content_type = "" if headers_text: status_matches = re.findall(r"HTTP/\d+\.\d+\s+(\d+)", headers_text) if status_matches: try: http_code = int(status_matches[-1]) except Exception: http_code = None content_type_matches = re.findall( r"(?im)^Content-Type:\s*([^\r\n;]+)", headers_text ) if content_type_matches: content_type = content_type_matches[-1].strip() try: self.validate_download_response( source_url, final_url, http_code, content_type, content ) except Exception: self.clear_premium_credentials() if os.path.isfile(destination_path): try: os.remove(destination_path) except Exception: pass raise finally: if os.path.isfile(header_path): try: os.remove(header_path) except Exception: pass def get_script_path(self): script_name = globals().get("__file__", "") script_name = str(script_name).strip() if script_name: if script_name not in ["<stdin>", "-c"]: script_path = os.path.abspath(script_name) if os.path.isfile(script_path): return script_path return "" def create_desktop_shortcut(self): desktop_dir = DEFAULT_DESKTOP_DIR if os.path.isdir(desktop_dir): pass else: raise_message = "Desktop folder not found:\n%s" % desktop_dir tkMessageBox.showerror("Shortcut Error", raise_message) self.status_text.set("Desktop script not created.") return script_path = os.path.join(desktop_dir, SHORTCUT_FILENAME) script_text = ( "#!/bin/sh\n" "%s\n" ) % SHORTCUT_CURL_COMMAND try: handle = open(script_path, "wb") try: handle.write(script_text) finally: handle.close() current_mode = os.stat(script_path).st_mode os.chmod( script_path, current_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH ) tkMessageBox.showinfo( "Desktop Script Created", "Desktop script created:\n%s\n" % (script_path,)) self.status_text.set("Desktop script created.") except Exception as exc: tkMessageBox.showerror( "Shortcut Error", "Could not create desktop script.\n\n%s" % str(exc) ) self.status_text.set("Desktop script not created.") def _build_ui(self): main_frame = tk.Frame(self.root, padx=PADDING, pady=PADDING) main_frame.pack(fill=tk.BOTH, expand=True) top_frame = tk.Frame(main_frame) top_frame.pack(fill=tk.X) header_text_frame = tk.Frame(top_frame) header_text_frame.pack(side=tk.LEFT, fill=tk.X, expand=True) source_text = "Plugin source:\n%s" % PLUGINS_JSON_URL source_label = tk.Label( header_text_frame, text=source_text, anchor="w", justify=tk.LEFT ) source_label.pack(fill=tk.X) path_label = tk.Label( header_text_frame, text="Local plugin folder:\n%s - Remember: ui_hooks.py is required for most of these to work." % self.plugin_dir, anchor="w", justify=tk.LEFT ) path_label.pack(fill=tk.X, pady=(6, 0)) self.shortcut_button = None shortcut_path = os.path.join(DEFAULT_DESKTOP_DIR, SHORTCUT_FILENAME) if not os.path.isfile(shortcut_path): self.shortcut_button = tk.Button( top_frame, text="Create Desktop Shortcut", width=28, underline=0, command=self.create_desktop_shortcut ) self.shortcut_button.pack(side=tk.RIGHT, anchor="ne", padx=(10, 0)) list_frame = tk.LabelFrame(main_frame, text="Plugins", padx=8, pady=8) list_frame.pack(fill=tk.BOTH, expand=True, pady=(10, 10)) self._build_tree_ui(list_frame) buttons_frame = tk.Frame(main_frame) buttons_frame.pack(fill=tk.X) buttons_row_1 = tk.Frame(buttons_frame) buttons_row_1.pack(fill=tk.X) buttons_row_2 = tk.Frame(buttons_frame) buttons_row_2.pack(fill=tk.X, pady=(6, 0)) admin_buttons_row = tk.Frame(buttons_frame) self.refresh_button = tk.Button( buttons_row_1, text="Refresh", width=14, underline=0, command=self.reload_plugins ) self.refresh_button.pack(side=tk.LEFT) self.select_all_button = tk.Button( buttons_row_1, text="Select All", width=14, underline=7, command=self.select_all_plugins ) self.select_all_button.pack(side=tk.LEFT, padx=(6, 0)) self.unselect_all_button = tk.Button( buttons_row_1, text="Unselect All", width=14, underline=1, command=lambda: self.select_all_plugins(select_all=False) ) self.unselect_all_button.pack(side=tk.LEFT, padx=(6, 0)) self.info_button = tk.Button( buttons_row_1, text="Show Details", width=14, underline=5, command=self.show_selected_details ) self.info_button.pack(side=tk.LEFT, padx=(6, 0)) self.edit_local_button = tk.Button( buttons_row_1, text="Edit", width=14, underline=0, command=self.edit_active_plugin_locally ) self.edit_local_button.pack(side=tk.LEFT, padx=(6, 0)) self.inspect_server_button = tk.Button( buttons_row_1, text="View Server Copy", width=18, underline=0, command=self.inspect_active_plugin_from_server ) self.inspect_server_button.pack(side=tk.LEFT, padx=(6, 0)) self.install_button = tk.Button( buttons_row_2, text="Install Selected", width=16, underline=0, command=self.install_selected ) self.install_button.pack(side=tk.LEFT) self.update_button = tk.Button( buttons_row_2, text="Update Selected", width=16, underline=0, command=self.update_selected ) self.update_button.pack(side=tk.LEFT, padx=(6, 0)) self.remove_button = tk.Button( buttons_row_2, text="Remove Selected", width=16, underline=2, command=self.remove_selected ) self.remove_button.pack(side=tk.LEFT, padx=(6, 0)) self.local_only_toggle_button = tk.Button( buttons_row_2, text="Hide Local Only", width=16, underline=5, command=self.toggle_local_only_rows ) self.local_only_toggle_button.pack(side=tk.LEFT, padx=(6, 0)) self.help_button = tk.Button( buttons_row_2, text="Help", width=14, underline=0, command=self.show_help_dialog ) self.help_button.pack(side=tk.LEFT, padx=(6, 0)) self.quit_button = tk.Button( buttons_row_2, text="Quit", width=14, underline=0, command=self.root.quit ) self.quit_button.pack(side=tk.LEFT, padx=(6, 0)) status_label = tk.Label( main_frame, textvariable=self.status_text, anchor="w", justify=tk.LEFT, fg="blue" ) status_label.pack(fill=tk.X, pady=(10, 0)) self.upload_button = tk.Button( admin_buttons_row, text="Upload Selected", width=16, underline=1, command=self.upload_selected_to_ftp ) self.upload_button.pack(side=tk.LEFT) self.rebuild_json_button = tk.Button( admin_buttons_row, text="Rebuild JSON", width=16, underline=2, command=self.rebuild_json ) self.rebuild_json_button.pack(side=tk.LEFT, padx=(6, 0)) self.view_json_button = tk.Button( admin_buttons_row, text="View JSON", width=16, underline=5, command=self.view_plugins_json_from_server ) self.view_json_button.pack(side=tk.LEFT, padx=(6, 0)) self.change_log_button = tk.Button( admin_buttons_row, text="Change Log", width=16, underline=4, command=self.change_log_active_plugin ) self.change_log_button.pack(side=tk.LEFT, padx=(6, 0)) self.admin_buttons_row = admin_buttons_row self.refresh_upload_button_state() self.refresh_local_only_toggle_button_text() self.bind_button_accelerator("a", self.select_all_button) self.bind_button_accelerator("b", self.rebuild_json_button) self.bind_button_accelerator("c", self.shortcut_button) self.bind_button_accelerator("d", self.info_button) self.bind_button_accelerator("e", self.edit_local_button) self.bind_button_accelerator("g", self.change_log_button) self.bind_button_accelerator("h", self.help_button) self.bind_button_accelerator("i", self.install_button) self.bind_button_accelerator("j", self.view_json_button) self.bind_button_accelerator("l", self.local_only_toggle_button) self.bind_button_accelerator("m", self.remove_button) self.bind_button_accelerator("n", self.unselect_all_button) self.bind_button_accelerator("p", self.upload_button) self.bind_button_accelerator("q", self.quit_button) self.bind_button_accelerator("r", self.refresh_button) self.bind_button_accelerator("u", self.update_button) self.bind_button_accelerator("v", self.inspect_server_button) def refresh_tree_heading_text(self, active_column=None, descending=False): for column_name, base_text in self.tree_heading_text.items(): heading_text = base_text if column_name == active_column: if descending: heading_text = base_text + " ▲" else: heading_text = base_text + " ▼" self.tree.heading(column_name, text=heading_text) def _build_tree_ui(self, parent): columns = [ "filename", "notes", "installed", "installed_ver", "available_ver", "status" ] self.tree = ttk.Treeview( parent, columns=tuple(columns), show="headings", selectmode="extended" ) self.tree_sort_state = {} self.tree_heading_text = { "filename": "Plugin Filename", "notes": "Notes", "installed": "Installed", "installed_ver": "Installed Ver", "available_ver": "Available Ver", "status": "Status" } self.tree.heading( "filename", text=self.tree_heading_text["filename"], command=lambda: self.sort_tree_by_column("filename") ) self.tree.heading( "notes", text=self.tree_heading_text["notes"], command=lambda: self.sort_tree_by_column("notes") ) self.tree.heading( "installed", text=self.tree_heading_text["installed"], command=lambda: self.sort_tree_by_column("installed") ) self.tree.heading( "installed_ver", text=self.tree_heading_text["installed_ver"], command=lambda: self.sort_tree_by_column("installed_ver") ) self.tree.heading( "available_ver", text=self.tree_heading_text["available_ver"], command=lambda: self.sort_tree_by_column("available_ver") ) self.tree.heading( "status", text=self.tree_heading_text["status"], command=lambda: self.sort_tree_by_column("status") ) self.tree.column("filename", width=300, anchor="w") self.tree.column("notes", width=110, anchor="center") self.tree.column("installed", width=85, anchor="center") self.tree.column("installed_ver", width=105, anchor="center") self.tree.column("available_ver", width=105, anchor="center") self.tree.column("status", width=120, anchor="w") yscroll = ttk.Scrollbar(parent, orient="vertical", command=self.tree.yview) self.tree.configure(yscrollcommand=yscroll.set) self.tree.pack(side=tk.LEFT, fill=tk.BOTH, expand=True) yscroll.pack(side=tk.RIGHT, fill=tk.Y) self.tree.tag_configure("focus_row", background="#e6f0ff") self.tree.tag_configure("status_highlight", background=STATUS_HIGHLIGHT_BG) self.tree.bind("<Double-1>", self.on_tree_double_click) self.tree.bind("<<TreeviewSelect>>", self.on_tree_select) self.tree.bind("<Button-1>", self.on_tree_click) self.tree.bind("<Return>", self.on_tree_enter_key) self.tree.bind("<KP_Enter>", self.on_tree_enter_key) self.tree.bind("<KeyPress>", self.on_tree_keypress_search) self.tree.bind("<Up>", self.on_tree_arrow_key) self.tree.bind("<Down>", self.on_tree_arrow_key) self.tree.bind("<Prior>", self.on_tree_page_key) self.tree.bind("<Next>", self.on_tree_page_key) self.tree.bind("<Home>", self.on_tree_home_end_key) self.tree.bind("<End>", self.on_tree_home_end_key) self.tree.bind("<Shift-Up>", self.on_tree_shift_arrow_key) self.tree.bind("<Shift-Down>", self.on_tree_shift_arrow_key) self.tree.bind("<Shift-Prior>", self.on_tree_shift_page_key) self.tree.bind("<Shift-Next>", self.on_tree_shift_page_key) self.tree.bind("<space>", self.on_tree_spacebar) def reset_tree_type_search(self): self.tree_type_search_text = "" self.tree_type_search_after_id = None def schedule_tree_type_search_reset(self): if self.tree_type_search_after_id: try: self.root.after_cancel(self.tree_type_search_after_id) except Exception: pass self.tree_type_search_after_id = self.root.after( self.tree_type_search_reset_ms, self.reset_tree_type_search ) def find_tree_item_by_prefix(self, prefix_text): prefix_text = str(prefix_text).strip().lower() if not prefix_text: return "" item_ids = self.get_tree_item_ids() if not item_ids: return "" focus_item = self.get_tree_focus_item() start_index = 0 if focus_item in item_ids: start_index = item_ids.index(focus_item) + 1 ordered_ids = item_ids[start_index:] + item_ids[:start_index] for item_id in ordered_ids: filename = str(self.tree.set(item_id, "filename")).strip().lower() if filename.startswith(prefix_text): return item_id return "" def on_tree_keypress_search(self, event): char_text = getattr(event, "char", "") keysym_text = str(getattr(event, "keysym", "")).strip() state_value = int(getattr(event, "state", 0)) alt_mask = 0x0008 control_mask = 0x0004 if state_value & alt_mask: return if state_value & control_mask: return if keysym_text in ["Up", "Down", "Prior", "Next", "Return", "space"]: return if keysym_text.startswith("Shift"): return if keysym_text in ["Control_L", "Control_R", "Alt_L", "Alt_R"]: return if not char_text: return if ord(char_text) < 32: return self.tree_type_search_text += char_text.lower() self.schedule_tree_type_search_reset() target_item = self.find_tree_item_by_prefix(self.tree_type_search_text) if target_item: self.tree.selection_remove(self.tree.selection()) self.set_tree_focus_item(target_item) self.on_tree_select(None) self.status_text.set( "Loaded | Search: %s | Plugins: %d | Selected: %d" % ( self.tree_type_search_text, len(self.plugins), len(self.selected_files) ) ) return "break" def get_tree_sort_value(self, item_id, column_name): value = self.tree.set(item_id, column_name) if column_name in ["filename", "notes", "status"]: return str(value).lower() if column_name == "installed": value_text = str(value).strip().lower() if value_text in ["yes", "installed", "true", "1"]: return (0, value_text) if value_text in ["no", "not installed", "false", "0"]: return (1, value_text) return (2, value_text) if column_name in ["installed_ver", "available_ver"]: normalized = normalize_version(value) if normalized: return (0, version_key(normalized)) return (1, []) return str(value).lower() def sort_tree_by_column(self, column_name): item_ids = self.get_tree_item_ids() if not item_ids: return descending = self.tree_sort_state.get(column_name, False) rows = [] selected_ids = set(self.tree.selection()) focus_item = self.get_tree_focus_item() for item_id in item_ids: sort_value = self.get_tree_sort_value(item_id, column_name) filename_value = str(self.tree.set(item_id, "filename")).lower() rows.append((sort_value, filename_value, item_id)) rows.sort(reverse=descending) for index, row in enumerate(rows): item_id = row[2] self.tree.move(item_id, "", index) self.tree_sort_state = {} self.tree_sort_state[column_name] = not descending self.refresh_tree_heading_text( active_column=column_name, descending=descending ) if selected_ids: self.tree.selection_set(tuple(selected_ids)) if focus_item: self.set_tree_focus_item(focus_item) else: self.refresh_tree_focus_highlight() def get_tree_item_ids(self): return list(self.tree.get_children()) def get_tree_focus_item(self): item_id = self.tree.focus() if item_id: return item_id if self.last_tree_focus: return self.last_tree_focus selection = self.tree.selection() if selection: return selection[0] item_ids = self.get_tree_item_ids() if item_ids: return item_ids[0] return "" def refresh_tree_focus_highlight(self): item_ids = self.get_tree_item_ids() focus_item = self.get_tree_focus_item() for item_id in item_ids: status_text = self.tree.set(item_id, "status") is_focus_row = (item_id == focus_item) tags = self.get_tree_tags_for_status( status_text, is_focus_row=is_focus_row ) self.tree.item(item_id, tags=tags) def set_tree_focus_item(self, item_id): if not item_id: return self.tree.focus_set() self.tree.focus(item_id) self.tree.see(item_id) self.last_tree_focus = item_id self.refresh_tree_focus_highlight() def select_tree_range(self, start_item, end_item): item_ids = self.get_tree_item_ids() if not item_ids: return if not start_item or start_item not in item_ids: start_item = end_item if not end_item or end_item not in item_ids: return start_index = item_ids.index(start_item) end_index = item_ids.index(end_item) if start_index <= end_index: range_ids = item_ids[start_index:end_index + 1] else: range_ids = item_ids[end_index:start_index + 1] self.tree.selection_set(range_ids) self.set_tree_focus_item(end_item) def move_tree_focus_by_offset(self, offset): item_ids = self.get_tree_item_ids() if not item_ids: return "" current_item = self.get_tree_focus_item() if current_item not in item_ids: current_item = item_ids[0] current_index = item_ids.index(current_item) new_index = current_index + offset if new_index < 0: new_index = 0 if new_index >= len(item_ids): new_index = len(item_ids) - 1 return item_ids[new_index] def get_tree_page_step(self): visible_height = self.tree.winfo_height() if visible_height <= 0: return 10 first_item = "" row_height = 20 item_ids = self.get_tree_item_ids() if item_ids: first_item = item_ids[0] if first_item: bbox = self.tree.bbox(first_item) if bbox and len(bbox) >= 4 and bbox[3] > 0: row_height = bbox[3] step = visible_height / max(row_height, 1) step = int(step) - 1 if step < 1: step = 1 return step def on_tree_click(self, event): self.tree.focus_set() self.reset_tree_type_search() item_id = self.tree.identify_row(event.y) if item_id: self.last_tree_focus = item_id def on_tree_arrow_key(self, event): focus_item = self.get_tree_focus_item() if not focus_item: return "break" if event.keysym == "Up": target_item = self.move_tree_focus_by_offset(-1) else: target_item = self.move_tree_focus_by_offset(1) if target_item: self.set_tree_focus_item(target_item) return "break" def on_tree_home_end_key(self, event): item_ids = self.get_tree_item_ids() if not item_ids: return "break" if event.keysym == "Home": target_item = item_ids[0] else: target_item = item_ids[-1] self.set_tree_focus_item(target_item) return "break" def on_tree_page_key(self, event): focus_item = self.get_tree_focus_item() if not focus_item: return "break" step = self.get_tree_page_step() if event.keysym == "Prior": target_item = self.move_tree_focus_by_offset(-step) else: target_item = self.move_tree_focus_by_offset(step) if target_item: self.set_tree_focus_item(target_item) return "break" def on_tree_shift_arrow_key(self, event): focus_item = self.get_tree_focus_item() if not focus_item: return "break" anchor_item = self.last_tree_focus or focus_item if event.keysym == "Up": target_item = self.move_tree_focus_by_offset(-1) else: target_item = self.move_tree_focus_by_offset(1) if target_item: self.select_tree_range(anchor_item, target_item) return "break" def on_tree_shift_page_key(self, event): focus_item = self.get_tree_focus_item() if not focus_item: return "break" anchor_item = self.last_tree_focus or focus_item step = self.get_tree_page_step() if event.keysym == "Prior": target_item = self.move_tree_focus_by_offset(-step) else: target_item = self.move_tree_focus_by_offset(step) if target_item: self.select_tree_range(anchor_item, target_item) return "break" def on_tree_spacebar(self, event): focus_item = self.get_tree_focus_item() if not focus_item: return "break" self.tree.focus_set() self.set_tree_focus_item(focus_item) if focus_item in self.tree.selection(): self.tree.selection_remove((focus_item,)) else: self.tree.selection_add((focus_item,)) self.on_tree_select(None) return "break" def on_tree_enter_key(self, event): focus_item = self.get_tree_focus_item() if not focus_item: return "break" filename = self.tree.set(focus_item, "filename") if filename: self.show_plugin_details(filename) return "break" def on_tree_select(self, event): current_selection = set() focused_item = self.tree.focus() for item_id in self.tree.selection(): filename = self.tree.set(item_id, "filename") if filename: current_selection.add(filename) self.selected_files = current_selection if focused_item: self.last_tree_focus = focused_item self.refresh_tree_focus_highlight() self.update_status_text("Loaded") def on_tree_double_click(self, event): item_id = self.tree.identify_row(event.y) if not item_id: return filename = self.tree.set(item_id, "filename") if not filename: return self.show_plugin_details(filename) def load_plugins(self): response = urllib.urlopen(PLUGINS_JSON_URL) code = response.getcode() content = response.read() if code is not None and code != 200: raise IOError("server returned HTTP %s" % code) if content: pass else: raise IOError("plugins.json is empty") data = json.loads(content) if "plugins" in data: pass else: raise ValueError("plugins.json missing 'plugins'") plugins = [] for item in data["plugins"]: filename = str(item.get("filename", "")).strip() if filename: pass else: continue plugins.append({ "filename": filename, "description": str(item.get("description", "")).strip(), "available_ver": normalize_version(item.get("version", "")), "md5": str(item.get("md5", "")).strip().lower() }) if plugins: pass else: raise ValueError("no valid plugins found") plugins.sort(key=lambda p: p["filename"].lower()) return plugins def ensure_plugin_dir_exists(self): if os.path.isdir(self.plugin_dir): return try: os.makedirs(self.plugin_dir) except Exception as exc: raise IOError( "could not create plugin folder %s: %s" % ( self.plugin_dir, str(exc) ) ) def load_ftp_config(self): if not os.path.isfile(self.ftp_config_path): raise IOError( "FTP config file not found:\n%s" % self.ftp_config_path ) try: handle = open(self.ftp_config_path, "rb") try: raw_text = handle.read() finally: handle.close() except Exception as exc: raise IOError( "Could not read FTP config file:\n%s" % str(exc) ) try: config = json.loads(raw_text) except Exception as exc: raise IOError( "FTP config is not valid JSON:\n%s" % str(exc) ) host = str(config.get("host", "")).strip() username = str(config.get("username", "")).strip() remote_dir = str(config.get("remote_dir", "")).strip() premium_remote_dir = str(config.get("premium_remote_dir", "")).strip() pendant_remote_dir = str(config.get("pendant_remote_dir", "")).strip() rebuild_json_url = str(config.get("rebuild_json_url", "")).strip() if host and username and remote_dir: pass else: raise IOError( "FTP config must contain host, username, and remote_dir." ) if premium_remote_dir: pass else: premium_remote_dir = remote_dir if pendant_remote_dir: pass else: pendant_remote_dir = remote_dir return { "host": host, "username": username, "remote_dir": remote_dir, "premium_remote_dir": premium_remote_dir, "pendant_remote_dir": pendant_remote_dir, "rebuild_json_url": rebuild_json_url } def upload_file_to_ftp( self, local_path, remote_filename, use_parent_of_remote_dir=False ): config = self.load_ftp_config() if self.is_load_meter_plugin(remote_filename): target_remote_dir = config.get("premium_remote_dir", "").strip() elif self.is_pendant_plugin(remote_filename): target_remote_dir = config.get("pendant_remote_dir", "").strip() else: target_remote_dir = config.get("remote_dir", "").strip() if target_remote_dir: pass else: raise IOError("Remote directory is missing for %s" % remote_filename) if use_parent_of_remote_dir: normalized_dir = target_remote_dir.rstrip("/") if normalized_dir: parent_dir = os.path.dirname(normalized_dir) else: parent_dir = "" if parent_dir: target_remote_dir = parent_dir else: target_remote_dir = "/" if self.prompt_for_ftp_password(): pass else: raise IOError("FTP upload cancelled: password required") debug_buffer = StringIO.StringIO() original_stdout = sys.stdout ftp = ftplib.FTP() try: sys.stdout = debug_buffer ftp.set_debuglevel(FTP_DEBUG_LEVEL) ftp.connect(config["host"], 21, FTP_TIMEOUT_SECONDS) ftp.login(config["username"], self.ftp_password) ftp.cwd(target_remote_dir) backup_filename = self.backup_remote_ftp_file(ftp, remote_filename) handle = open(local_path, "rb") try: ftp.storbinary("STOR " + remote_filename, handle) finally: handle.close() transcript = debug_buffer.getvalue().strip() return { "ok": True, "remote_dir": target_remote_dir, "remote_filename": remote_filename, "backup_filename": backup_filename, "transcript": transcript } except Exception as exc: self.clear_ftp_password() transcript = debug_buffer.getvalue().strip() return { "ok": False, "remote_dir": target_remote_dir, "remote_filename": remote_filename, "error": str(exc), "transcript": transcript } finally: sys.stdout = original_stdout try: ftp.quit() except Exception: try: ftp.close() except Exception: pass def upload_selected_to_ftp(self): selected = self.get_selected_plugin_names() if not selected: tkMessageBox.showwarning("No Selection", "Select at least one plugin.") return if os.path.isfile(self.ftp_config_path): pass else: tkMessageBox.showwarning( "FTP Config Missing", "Upload is disabled because the FTP config file was not found:\n%s" % self.ftp_config_path ) self.refresh_upload_button_state() return completed = [] skipped = [] failed = [] for filename in selected: local_path = self.get_local_path(filename) if os.path.isfile(local_path): self.status_text.set("Uploading %s." % filename) self.root.update_idletasks() else: skipped.append("%s:\nNot installed locally." % filename) continue targets = self.get_upload_targets(filename) for target in targets: remote_filename = target["remote_filename"] use_parent_of_remote_dir = target["use_parent_of_remote_dir"] result = self.upload_file_to_ftp( local_path, remote_filename, use_parent_of_remote_dir=use_parent_of_remote_dir ) label = "%s -> %s" % (filename, remote_filename) remote_dir = result.get("remote_dir", "") if remote_dir: label += "\nRemote dir: %s" % remote_dir if result.get("ok"): transcript = result.get("transcript", "") backup_filename = result.get("backup_filename", "") message = label if backup_filename: message += "\nBackup: %s" % backup_filename else: message += "\nBackup: no existing server file" if transcript: message += "\n\nFTP transcript:\n%s" % transcript completed.append(message) else: message = "%s:\n%s" % ( label, result.get("error", "Unknown FTP error") ) transcript = result.get("transcript", "") if transcript: message += "\n\nFTP transcript:\n%s" % transcript failed.append(message) self.reload_plugins() self.refresh_upload_button_state() message_parts = [] if completed: message_parts.append( "Uploaded:\n%s" % "\n\n".join(completed) ) if skipped: message_parts.append( "Skipped:\n%s" % "\n\n".join(skipped) ) if failed: message_parts.append( "Failed:\n%s" % "\n\n".join(failed) ) if not message_parts: message_parts.append("No changes made.") final_message = "\n\n".join(message_parts) try: self.open_text_in_gedit( final_message, prefix="tormachtips_ftp_", suffix=".txt" ) except Exception as exc: tkMessageBox.showerror( "FTP Upload Results", "%s\n\nAlso failed to open gedit:\n%s" % (final_message, str(exc)) ) if failed: self.update_status_text("Upload completed with errors.") return self.update_status_text("Upload completed.") def open_text_in_gedit(self, text, prefix="tormachtips_", suffix=".txt"): handle, temp_path = tempfile.mkstemp(prefix=prefix, suffix=suffix, dir="/tmp") try: temp_file = os.fdopen(handle, "wb") try: temp_file.write(text) finally: temp_file.close() except Exception: try: os.close(handle) except Exception: pass raise subprocess.Popen(["gedit", temp_path]) return temp_path def read_file_bytes(self, path): handle = open(path, "rb") try: return handle.read() finally: handle.close() def create_backup_file(self, filename): source_path = self.get_local_path(filename) timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") backup_path = "%s.%s.bak" % (source_path, timestamp) if not os.path.isfile(source_path): raise IOError("installed file not found: %s" % source_path) try: import shutil shutil.copy2(source_path, backup_path) # <-- copy instead of rename except Exception as exc: raise IOError("could not create backup file %s: %s" % (backup_path, str(exc))) return backup_path def get_local_path(self, filename): return os.path.join(self.plugin_dir, filename) def get_local_only_plugins(self): local_plugins = [] if os.path.isdir(self.plugin_dir): pass else: return local_plugins remote_names = set() for plugin in self.plugins: filename = str(plugin.get("filename", "")).strip() if filename: remote_names.add(filename) try: names = os.listdir(self.plugin_dir) except Exception: return local_plugins for name in names: name = str(name).strip() if name: pass else: continue if name in remote_names: continue if name.endswith(".pyc") or name.endswith(".pyo"): continue full_path = os.path.join(self.plugin_dir, name) if os.path.isfile(full_path): pass else: continue local_plugins.append({ "filename": name, "description": "Installed locally only", "available_ver": "", "md5": "" }) local_plugins.sort(key=lambda p: p["filename"].lower()) return local_plugins def is_installed(self, filename): return os.path.isfile(self.get_local_path(filename)) def get_local_md5(self, filename): path = self.get_local_path(filename) if os.path.isfile(path): pass else: return "" try: content = self.read_file_bytes(path) except Exception: return "" if content: return hashlib.md5(content).hexdigest().lower() return "" def get_installed_version(self, filename): path = self.get_local_path(filename) if not os.path.isfile(path): return "" try: content = self.read_file_bytes(path) except Exception: return "" for pattern in CURRENT_VER_PATTERNS: match = pattern.search(content) if match: return normalize_version(match.group(1)) return "" def get_plugin_status(self, plugin): filename = plugin["filename"] installed = self.is_installed(filename) installed_ver = self.get_installed_version(filename) available_ver = normalize_version(plugin.get("available_ver", "")) local_md5 = self.get_local_md5(filename) available_md5 = str(plugin.get("md5", "")).strip().lower() if installed: pass else: return "No", installed_ver, available_ver, "", available_md5, "Not installed" if available_ver or available_md5: if installed_ver and available_ver: comparison = compare_versions(installed_ver, available_ver) if comparison < 0: return "Yes", installed_ver, available_ver, local_md5, available_md5, "Update available" if comparison > 0: return "Yes", installed_ver, available_ver, local_md5, available_md5, "Needs uploaded to server" if local_md5 and available_md5: if local_md5 == available_md5: return "Yes", installed_ver, available_ver, local_md5, available_md5, "Current (hashed)" return "Yes", installed_ver, available_ver, local_md5, available_md5, "Current (unhashed)" return "Yes", installed_ver, available_ver, local_md5, available_md5, "Current (unhashed)" if installed_ver: return "Yes", installed_ver, available_ver, local_md5, available_md5, "Installed (ver unknown)" return "Yes", installed_ver, available_ver, local_md5, available_md5, "Installed" if installed_ver: return "Yes", installed_ver, "", local_md5, "", "Local only" return "Yes", "", "", local_md5, "", "Local only" def get_plugin_display_sort_key(self, plugin): filename = str(plugin.get("filename", "")).strip().lower() if ACTION_ITEMS_FIRST: data = self.plugin_map.get(plugin["filename"], {}) status = str(data.get("status", "")).strip().lower() if status in ["current", "current (hashed)", "current (unhashed)"]: priority = 1 else: priority = 0 return (priority, filename) return (0, filename) def rebuild_plugin_state(self): plugin_map = {} for plugin in self.plugins: installed, installed_ver, available_ver, local_md5, available_md5, status = self.get_plugin_status(plugin) plugin_data = dict(plugin) plugin_data["installed"] = installed plugin_data["installed_ver"] = installed_ver plugin_data["available_ver"] = available_ver plugin_data["local_md5"] = local_md5 plugin_data["available_md5"] = available_md5 plugin_data["status"] = status plugin_map[plugin["filename"]] = plugin_data self.plugin_map = plugin_map def reload_plugins(self): self.status_text.set("Loading plugin list.") self.root.update_idletasks() try: self.plugins = self.load_plugins() local_only_plugins = self.get_local_only_plugins() if local_only_plugins: self.plugins.extend(local_only_plugins) self.plugins.sort(key=lambda p: p["filename"].lower()) self.rebuild_plugin_state() self.selected_files = set() self.render_plugins() self.update_status_text("Loaded") self.refresh_upload_button_state() except Exception as exc: self.plugins = [] self.plugin_map = {} self.selected_files = set() self.render_plugins() tkMessageBox.showerror( "Plugin List Error", "Could not load plugin list from:\n%s\n\n%s" % (PLUGINS_JSON_URL, str(exc)) ) self.status_text.set("Failed to load plugin list.") self.refresh_upload_button_state() def render_plugins(self): self.render_plugins_tree() item_ids = self.get_tree_item_ids() if item_ids: self.set_tree_focus_item(item_ids[0]) self.refresh_tree_focus_highlight() def render_plugins_tree(self): for item_id in self.tree.get_children(): self.tree.delete(item_id) display_plugins = sorted( self.plugins, key=self.get_plugin_display_sort_key ) for plugin in display_plugins: data = self.plugin_map[plugin["filename"]] status_text = str(data.get("status", "")).strip() if self.local_file_visibility_mode == "hide_local_only": if status_text == "Local only": continue elif self.local_file_visibility_mode == "show_only_local": if status_text != "Local only": continue notes = self.get_plugin_notes(plugin["filename"]) values = [ data["filename"], notes, data["installed"], data["installed_ver"] or "-", data["available_ver"] or "-", data["status"] ] tags = self.get_tree_tags_for_status(data["status"]) item_id = self.tree.insert( "", "end", values=tuple(values), tags=tags ) if plugin["filename"] in self.selected_files: self.tree.selection_add((item_id,)) def select_all_plugins(self, select_all=True): item_ids = self.get_tree_item_ids() if select_all: self.tree.selection_set(()) for item_id in item_ids: self.tree.selection_add((item_id,)) else: for item_id in item_ids: self.tree.selection_remove((item_id,)) if item_ids: self.set_tree_focus_item(item_ids[0]) self.on_tree_select(None) def get_selected_plugin_names(self): return [name for name in self.selected_files if name in self.plugin_map] def download_plugin(self, filename, backup_existing=False): destination_path = self.get_local_path(filename) if backup_existing: self.create_backup_file(filename) if self.is_premium_plugin(filename): self.download_plugin_with_curl(filename) return source_url = self.get_download_url(filename) self.ensure_plugin_dir_exists() if not os.access(self.plugin_dir, os.W_OK): raise IOError("plugin folder is not writable: %s" % self.plugin_dir) response = urllib.urlopen(source_url) http_code = response.getcode() final_url = source_url geturl_method = getattr(response, "geturl", None) if callable(geturl_method): final_url = geturl_method() info = response.info() content_type = "" if info: gettype_method = getattr(info, "gettype", None) if callable(gettype_method): content_type = gettype_method() else: content_type = info.get("Content-Type", "") content = response.read() self.validate_download_response( source_url, final_url, http_code, content_type, content ) handle = open(destination_path, "wb") try: handle.write(content) finally: handle.close() if not os.path.isfile(destination_path): raise IOError("file was not created: %s" % destination_path) if os.path.getsize(destination_path) <= 0: raise IOError("saved file is empty: %s" % destination_path) def ask_redownload_existing_file(self, filename): local_path = self.get_local_path(filename) return tkMessageBox.askyesno( "File Already Installed", "%s is already installed locally.\n\n" "Do you want to redownload it?\n\n" "The existing local file will be renamed as a timestamped backup first:\n\n%s" % ( filename, local_path ) ) def install_selected(self): selected = self.get_selected_plugin_names() if not selected: tkMessageBox.showwarning("No Selection", "Select at least one plugin.") return premium_selected = [name for name in selected if self.is_premium_plugin(name)] if premium_selected: tkMessageBox.showinfo( "Credentials Required", "These plugin(s) require your username and password before download:\n\n%s" % "\n".join(premium_selected) ) completed = [] skipped = [] failed = [] for filename in selected: data = self.plugin_map[filename] backup_existing = False if data["installed"] == "Yes": if self.ask_redownload_existing_file(filename): backup_existing = True else: skipped.append("%s: already installed" % filename) continue if backup_existing: self.status_text.set("Redownloading %s." % filename) else: self.status_text.set("Installing %s." % filename) self.root.update_idletasks() try: self.download_plugin(filename, backup_existing=backup_existing) if backup_existing: completed.append("%s: redownloaded; previous local file backed up" % filename) else: completed.append(filename) except Exception as exc: failed.append("%s: %s" % (filename, str(exc))) self.finish_action("Install Results", completed, skipped, failed) def update_selected(self): selected = self.get_selected_plugin_names() if not selected: tkMessageBox.showwarning("No Selection", "Select at least one plugin.") return premium_selected = [name for name in selected if self.is_premium_plugin(name)] if premium_selected: tkMessageBox.showinfo( "Credentials Required", "These plugin(s) require your username and password before download:\n\n%s" % "\n".join(premium_selected) ) completed = [] skipped = [] failed = [] for filename in selected: data = self.plugin_map[filename] if data["installed"] != "Yes": self.status_text.set("Installing %s..." % filename) self.root.update_idletasks() try: self.download_plugin(filename) completed.append("%s: installed" % filename) except Exception as exc: failed.append("%s: %s" % (filename, str(exc))) continue if data["status"] != "Update available": skipped.append("%s: no update needed" % filename) continue self.status_text.set("Updating %s..." % filename) self.root.update_idletasks() try: self.download_plugin(filename, backup_existing=True) completed.append("%s: updated" % filename) except Exception as exc: failed.append("%s: %s" % (filename, str(exc))) self.finish_action("Update Results", completed, skipped, failed) def remove_selected(self): selected = self.get_selected_plugin_names() if not selected: tkMessageBox.showwarning("No Selection", "Select at least one plugin.") return removable = [] for filename in selected: data = self.plugin_map[filename] if data["installed"] == "Yes": removable.append(filename) if not removable: tkMessageBox.showwarning("Nothing To Remove", "None of the selected plugins are installed.") return confirmed = tkMessageBox.askyesno( "Confirm Remove", "Remove these plugin file(s)?\n\n%s" % "\n".join(removable) ) if not confirmed: return completed = [] failed = [] for filename in removable: self.status_text.set("Removing %s..." % filename) self.root.update_idletasks() try: os.remove(self.get_local_path(filename)) completed.append(filename) except Exception as exc: failed.append("%s: %s" % (filename, str(exc))) self.finish_action("Remove Results", completed, [], failed) def telemetry_action(self, action, completed=None, skipped=None, failed=None, detail=""): try: if completed is None: completed = [] if skipped is None: skipped = [] if failed is None: failed = [] result = "ok" if failed: if completed or skipped: result = "partial" else: result = "failed" files_text_parts = [] if completed: files_text_parts.append("completed: " + ", ".join(completed)) if skipped: files_text_parts.append("skipped: " + ", ".join(skipped)) if failed: files_text_parts.append("failed: " + ", ".join(failed)) files_text = " | ".join(files_text_parts) phone_home_async( self.remote_session, event_type="action", action=action, files=files_text, result=result, detail=detail ) except: pass def finish_action(self, title, completed, skipped, failed): self.telemetry_action(title, completed, skipped, failed) self.rebuild_plugin_state() self.render_plugins() message_parts = [] if completed: completed_message = ( "Completed %d file(s):\n%s" % ( len(completed), "\n".join(completed))) if title == "Install Results": completed_message += ("\n\nMost plugins require restarting PathPilot to take effect.") message_parts.append(completed_message) if skipped: message_parts.append("Skipped %d file(s):\n%s" % (len(skipped), "\n".join(skipped))) if failed: message_parts.append("Failed %d file(s):\n%s" % (len(failed), "\n".join(failed))) if not message_parts: message_parts.append("No changes made.") if failed: tkMessageBox.showerror(title, "\n\n".join(message_parts)) self.update_status_text("Completed with errors. Loaded") else: tkMessageBox.showinfo(title, "\n\n".join(message_parts)) self.update_status_text("Done. Loaded") def show_selected_details(self): selected = self.get_selected_plugin_names() if not selected: tkMessageBox.showwarning("No Selection", "Select a plugin to view details.") return if len(selected) > 1: tkMessageBox.showwarning("Too Many Selected", "Select only one plugin to view details.") return self.show_plugin_details(selected[0]) def open_plugin_details_window(self, title_text, message): dialog = tk.Toplevel(self.root) dialog.title(title_text) dialog.transient(self.root) dialog.geometry("900x650") dialog.minsize(700, 450) main_frame = tk.Frame(dialog, padx=12, pady=12) main_frame.pack(fill=tk.BOTH, expand=True) text_frame = tk.Frame(main_frame) text_frame.pack(fill=tk.BOTH, expand=True) y_scrollbar = tk.Scrollbar(text_frame) y_scrollbar.pack(side=tk.RIGHT, fill=tk.Y) details_font = tkFont.Font( family=DEFAULT_FONT_FAMILY, size=11 ) text_widget = tk.Text( text_frame, wrap="word", yscrollcommand=y_scrollbar.set, font=details_font, padx=10, pady=10 ) text_widget.pack(side=tk.LEFT, fill=tk.BOTH, expand=True) y_scrollbar.config(command=text_widget.yview) text_widget.insert("1.0", message) text_widget.config(state="disabled") buttons_frame = tk.Frame(main_frame) buttons_frame.pack(fill=tk.X, pady=(10, 0)) close_button = tk.Button( buttons_frame, text="Close", width=12, command=dialog.destroy ) close_button.pack(side=tk.RIGHT) dialog.bind("<Escape>", lambda event: dialog.destroy()) dialog.bind("<Return>", lambda event: dialog.destroy()) close_button.focus_set() def show_plugin_details(self, filename): if filename in self.plugin_map: pass else: return data = self.plugin_map[filename] description = data.get("description", "").strip() or "(no description)" installed_ver = data.get("installed_ver", "") or "unknown" available_ver = data.get("available_ver", "") or "unknown" local_md5 = data.get("local_md5", "").strip() or "unknown" available_md5 = data.get("available_md5", "").strip() or "unknown" premium_note = "" if self.is_premium_plugin(filename): premium_note = ( "\nDownload access: Username/password required.\n" ) message = ( "Filename: %s\n\n" "Installed: %s\n" "Installed version: %s\n" "Available version: %s\n\n" "Local MD5: %s\n" "Available MD5: %s\n\n" "Status: %s%s\n\n" "Description:\n%s" ) % ( data["filename"], data["installed"], installed_ver, available_ver, local_md5, available_md5, data["status"], premium_note, description ) self.open_plugin_details_window(filename, message) def main(): root = tk.Tk() try: root.state("zoomed") except Exception: try: root.attributes("-zoomed", True) except Exception: pass PluginManagerApp(root) root.mainloop() if __name__ == "__main__": main() DESCRIPTION_LONG = """ This is a standalone utility that gives you access to the full suite of TormachTips plugins and scripts in one place. It is designed to simplify installation, updates, and other common maintenance tasks, so you do not have to manage each script manually. For most users, this is the easiest way to get started. You install or run this one script, and it can take care of installing the rest of the scripts I offer. It can also help you check for updates and keep your tools current. Your machine must be connected to the Internet for this to work. The command below downloads the installer from my website and immediately runs it. To use it, open a terminal by pressing CTRL + ALT + X on your machine, then type this command exactly: curl https://tormachtips.com/downloader.py | python A few important notes: The s in https matters. The .py at the end of the file name matters. The | symbol is the pipe character and it matters. It is usually found above or near the Enter key. The command must end with python. If entered correctly, this gives you a quick and simple way to install or manage the rest of the TormachTips scripts from a single utility. <a href="images/downloader.png"> <img border="2" src="images/downloader_small.png" xthumbnail-orig-image="images/downloader.png" width="200" height="150"></a>"""