513 lines
27 KiB
Python
513 lines
27 KiB
Python
# --- FILE: gitsync_tool/async_tasks/async_result_handler.py ---
|
|
|
|
import tkinter as tk
|
|
from tkinter import messagebox
|
|
import queue
|
|
import os
|
|
import logging
|
|
from typing import (
|
|
TYPE_CHECKING,
|
|
Dict,
|
|
Any,
|
|
List,
|
|
Tuple,
|
|
Callable,
|
|
Optional,
|
|
Set,
|
|
)
|
|
|
|
from gitutility.logging_setup import log_handler
|
|
from gitutility.commands.git_commands import GitCommandError
|
|
from gitutility.async_tasks import async_workers
|
|
from gitutility.config.config_manager import DEFAULT_REMOTE_NAME
|
|
|
|
if TYPE_CHECKING:
|
|
from ..app import GitSvnSyncApp
|
|
from ..gui.main_frame import MainFrame
|
|
|
|
class AsyncResultHandler:
|
|
"""
|
|
Handles processing the results received from asynchronous worker functions.
|
|
Updates the GUI based on the outcome of background tasks and triggers
|
|
necessary follow-up actions like GUI refreshes or subsequent tasks.
|
|
"""
|
|
|
|
def __init__(self, app: "GitSvnSyncApp"):
|
|
if app is None or not hasattr(app, "main_frame") or not app.main_frame:
|
|
raise ValueError("AsyncResultHandler requires a valid app instance with an initialized main_frame.")
|
|
self.app: "GitSvnSyncApp" = app
|
|
self.main_frame: "MainFrame" = app.main_frame
|
|
log_handler.log_debug("AsyncResultHandler initialized.", func_name="__init__")
|
|
|
|
def process(self, result_data: Dict[str, Any], context: Dict[str, Any]) -> None:
|
|
task_context = context.get("context", "unknown")
|
|
status = result_data.get("status", "error")
|
|
func_name = f"process (ctx: {task_context})"
|
|
log_handler.log_debug(f"Processing async result for '{task_context}' with status '{status}'.", func_name=func_name)
|
|
|
|
should_trigger_refreshes = False
|
|
post_action_sync_refresh_needed = False
|
|
|
|
handler_map: Dict[str, Callable] = {
|
|
"compare_branches": self._handle_compare_branches_result,
|
|
"refresh_tags": self._handle_refresh_tags_result,
|
|
"refresh_branches": self._handle_refresh_local_branches_result,
|
|
"refresh_history": self._handle_refresh_history_result,
|
|
"refresh_changes": self._handle_refresh_changes_result,
|
|
"refresh_remote_branches": self._handle_refresh_remote_branches_result,
|
|
"get_ahead_behind": self._handle_get_ahead_behind_result,
|
|
"prepare_repo": self._handle_prepare_repo_result,
|
|
"create_bundle": self._handle_create_bundle_result,
|
|
"fetch_bundle": self._handle_fetch_bundle_result,
|
|
"manual_backup": self._handle_manual_backup_result,
|
|
"commit": self._handle_commit_result,
|
|
"_handle_gitignore_save": self._handle_untrack_result,
|
|
"add_file": self._handle_add_file_result,
|
|
"create_tag": self._handle_create_tag_result,
|
|
"checkout_tag": self._handle_checkout_tag_result,
|
|
"create_branch": self._handle_create_branch_result,
|
|
"checkout_branch": self._handle_checkout_branch_result,
|
|
"delete_local_branch": self._handle_delete_local_branch_result,
|
|
"merge_local_branch": self._handle_merge_local_branch_result,
|
|
"check_connection": self._handle_check_connection_result,
|
|
"interactive_auth": self._handle_interactive_auth_result,
|
|
"apply_remote_config": self._handle_apply_remote_config_result,
|
|
"fetch_remote": self._handle_fetch_remote_result,
|
|
"pull_remote": self._handle_pull_remote_result,
|
|
"push_remote": self._handle_push_remote_result,
|
|
"push_tags_remote": self._handle_push_tags_remote_result,
|
|
"clone_remote": self._handle_clone_remote_result,
|
|
"checkout_tracking_branch": self._handle_checkout_tracking_branch_result,
|
|
"get_commit_details": self._handle_get_commit_details_result,
|
|
"update_wiki": self._handle_generic_result,
|
|
"revert_to_tag": self._handle_revert_to_tag_result,
|
|
"analyze_history": self._handle_analyze_history_result,
|
|
"purge_history": self._handle_purge_history_result,
|
|
"refresh_submodules": self._handle_refresh_submodules_result,
|
|
"add_submodule": self._handle_add_submodule_result,
|
|
"sync_submodules": self._handle_sync_all_submodules_result,
|
|
"remove_submodule": self._handle_remove_submodule_result,
|
|
"init_submodules": self._handle_init_submodules_result,
|
|
}
|
|
|
|
handler_method = handler_map.get(task_context)
|
|
|
|
if handler_method:
|
|
refresh_flags = handler_method(result_data, context)
|
|
if isinstance(refresh_flags, tuple) and len(refresh_flags) == 2:
|
|
should_trigger_refreshes, post_action_sync_refresh_needed = refresh_flags
|
|
else:
|
|
log_handler.log_warning(f"No handler for context '{task_context}'.", func_name=func_name)
|
|
should_trigger_refreshes, post_action_sync_refresh_needed = self._handle_generic_result(result_data, context)
|
|
|
|
if should_trigger_refreshes:
|
|
self._trigger_post_action_refreshes(post_action_sync_refresh_needed)
|
|
|
|
def _handle_refresh_tags_result(self, result_data: Dict[str, Any], context: Dict[str, Any]) -> Tuple[bool, bool]:
|
|
if result_data.get("status") == "success":
|
|
self.main_frame.tags_tab.update_tag_list(result_data.get("result", []))
|
|
else:
|
|
self.main_frame.tags_tab.update_tag_list([("(Error)", "")])
|
|
return False, False
|
|
|
|
def _handle_refresh_local_branches_result(self, result_data: Dict[str, Any], context: Dict[str, Any]) -> Tuple[bool, bool]:
|
|
branches, current = ([], None)
|
|
if result_data.get("status") == "success":
|
|
result_tuple = result_data.get("result")
|
|
if isinstance(result_tuple, tuple) and len(result_tuple) == 2:
|
|
branches, current = result_tuple
|
|
self.app.current_local_branch = current
|
|
|
|
self.main_frame.update_branch_list(branches, current)
|
|
self.main_frame.history_tab.update_history_branch_filter(branches)
|
|
|
|
# Don't trigger remote status refresh from here anymore, it's handled by check_connection result
|
|
return False, False
|
|
|
|
def _handle_refresh_history_result(self, result_data: Dict[str, Any], context: Dict[str, Any]) -> Tuple[bool, bool]:
|
|
if result_data.get("status") == "success":
|
|
self.main_frame.history_tab.update_history_display(result_data.get("result", []))
|
|
else:
|
|
self.main_frame.history_tab.update_history_display(["(Error retrieving history)"])
|
|
return False, False
|
|
|
|
def _handle_refresh_changes_result(self, result_data: Dict[str, Any], context: Dict[str, Any]) -> Tuple[bool, bool]:
|
|
if result_data.get("status") == "success":
|
|
self.main_frame.commit_tab.update_changed_files_list(result_data.get("result", []))
|
|
else:
|
|
self.main_frame.commit_tab.update_changed_files_list(["(Error refreshing changes)"])
|
|
return False, False
|
|
|
|
def _handle_refresh_remote_branches_result(self, result_data: Dict[str, Any], context: Dict[str, Any]) -> Tuple[bool, bool]:
|
|
if result_data.get("status") == "success":
|
|
self.main_frame.remote_tab.update_remote_branches_list(result_data.get("result", []))
|
|
else:
|
|
self.main_frame.remote_tab.update_remote_branches_list(["(Error)"])
|
|
return False, False
|
|
|
|
def _handle_get_ahead_behind_result(self, result_data: Dict[str, Any], context: Dict[str, Any]) -> Tuple[bool, bool]:
|
|
func_name = "_handle_get_ahead_behind_result"
|
|
remote_tab = getattr(self.main_frame, "remote_tab", None)
|
|
if not remote_tab:
|
|
log_handler.log_error("RemoteTab is missing.", func_name=func_name)
|
|
return False, False
|
|
|
|
status, result, message = result_data.get("status"), result_data.get("result"), result_data.get("message")
|
|
local_branch_ctx = context.get("local_branch")
|
|
|
|
if status == "success":
|
|
ahead, behind = result if isinstance(result, tuple) else (None, None)
|
|
remote_tab.update_ahead_behind_status(current_branch=local_branch_ctx, ahead=ahead, behind=behind)
|
|
else:
|
|
error_text = f"Sync Status: Error ({message})" if message else "Sync Status: Error"
|
|
remote_tab.update_ahead_behind_status(current_branch=local_branch_ctx, status_text=error_text)
|
|
return False, False
|
|
|
|
def _handle_prepare_repo_result(self, result_data: Dict[str, Any], context: Dict[str, Any]) -> Tuple[bool, bool]:
|
|
status, message = result_data.get("status"), result_data.get("message")
|
|
if status == "success":
|
|
return True, True
|
|
elif status == "warning":
|
|
self.main_frame.show_info("Prepare Info", message)
|
|
return True, True
|
|
elif status == "error":
|
|
self.main_frame.show_error("Prepare Error", f"Failed:\n{message}")
|
|
|
|
self.app.master.after(50, lambda: self.app.update_svn_status_indicator(self.app.main_frame.repo_tab.svn_path_entry.get()))
|
|
return False, False
|
|
|
|
def _handle_create_bundle_result(self, result_data: Dict[str, Any], context: Dict[str, Any]) -> Tuple[bool, bool]:
|
|
status, message, committed = result_data.get("status"), result_data.get("message"), result_data.get("committed", False)
|
|
if status == "success":
|
|
if committed:
|
|
return True, True
|
|
elif status == "error":
|
|
self.main_frame.show_error("Create Bundle Error", f"Failed:\n{message}")
|
|
return False, False
|
|
|
|
def _handle_fetch_bundle_result(self, result_data: Dict[str, Any], context: Dict[str, Any]) -> Tuple[bool, bool]:
|
|
status, message = result_data.get("status"), result_data.get("message")
|
|
if status == "success":
|
|
return True, True
|
|
elif status == "error":
|
|
if result_data.get("conflict"):
|
|
self.main_frame.show_error("Merge Conflict", f"Conflict during bundle fetch. Resolve in:\n{result_data.get('repo_path')}")
|
|
self._reenable_widgets_after_modal()
|
|
return True, False
|
|
else:
|
|
self.main_frame.show_error("Fetch Bundle Error", f"Failed:\n{message}")
|
|
|
|
self.app.master.after(50, lambda: self.app.update_svn_status_indicator(self.app.main_frame.repo_tab.svn_path_entry.get()))
|
|
return False, False
|
|
|
|
def _handle_manual_backup_result(self, result_data: Dict[str, Any], context: Dict[str, Any]) -> Tuple[bool, bool]:
|
|
if result_data.get("status") == "error":
|
|
self.main_frame.show_error("Manual Backup Error", f"Failed:\n{result_data.get('message')}")
|
|
return False, False
|
|
|
|
def _handle_commit_result(self, result_data: Dict[str, Any], context: Dict[str, Any]) -> Tuple[bool, bool]:
|
|
if result_data.get("status") == "success" and result_data.get("committed"):
|
|
self.main_frame.commit_tab.clear_commit_message()
|
|
return True, True
|
|
elif result_data.get("status") == "error":
|
|
self.main_frame.show_error("Commit Error", f"Failed:\n{result_data.get('message')}")
|
|
return False, False
|
|
|
|
def _handle_untrack_result(self, result_data: Dict[str, Any], context: Dict[str, Any]) -> Tuple[bool, bool]:
|
|
if result_data.get("status") == "success" and result_data.get("committed"):
|
|
return True, True
|
|
elif result_data.get("status") == "error":
|
|
self.main_frame.show_error("Untrack Error", f"Failed:\n{result_data.get('message')}")
|
|
return False, False
|
|
|
|
def _handle_add_file_result(self, result_data: Dict[str, Any], context: Dict[str, Any]) -> Tuple[bool, bool]:
|
|
if result_data.get("status") == "success":
|
|
return True, False
|
|
elif result_data.get("status") == "error":
|
|
self.main_frame.show_error("Add File Error", f"Failed:\n{result_data.get('message')}")
|
|
return False, False
|
|
|
|
def _handle_create_tag_result(self, result_data: Dict[str, Any], context: Dict[str, Any]) -> Tuple[bool, bool]:
|
|
if result_data.get("status") == "success":
|
|
return True, False
|
|
elif result_data.get("status") == "error":
|
|
self.main_frame.show_error("Create Tag Error", f"Failed:\n{result_data.get('message')}")
|
|
return False, False
|
|
|
|
def _handle_checkout_tag_result(self, result_data: Dict[str, Any], context: Dict[str, Any]) -> Tuple[bool, bool]:
|
|
if result_data.get("status") == "success":
|
|
return True, True
|
|
elif result_data.get("status") == "error":
|
|
self.main_frame.show_error("Checkout Tag Error", f"Failed:\n{result_data.get('message')}")
|
|
self._reenable_widgets_after_modal()
|
|
return False, False
|
|
|
|
def _handle_create_branch_result(self, result_data: Dict[str, Any], context: Dict[str, Any]) -> Tuple[bool, bool]:
|
|
if result_data.get("status") == "success":
|
|
new_branch = context.get("new_branch_name")
|
|
if new_branch and self.main_frame.ask_yes_no("Checkout?", f"Switch to new branch '{new_branch}'?"):
|
|
self.app.repository_handler.checkout_branch(branch_to_checkout=new_branch)
|
|
return False, False # Checkout will trigger its own refresh
|
|
return True, False
|
|
elif result_data.get("status") == "error":
|
|
self.main_frame.show_error("Create Branch Error", f"Failed:\n{result_data.get('message')}")
|
|
return False, False
|
|
|
|
def _handle_checkout_branch_result(self, result_data: Dict[str, Any], context: Dict[str, Any]) -> Tuple[bool, bool]:
|
|
if result_data.get("status") == "success":
|
|
return True, True
|
|
elif result_data.get("status") == "error":
|
|
self.main_frame.show_error("Checkout Branch Error", f"Failed:\n{result_data.get('message')}")
|
|
self._reenable_widgets_after_modal()
|
|
return False, False
|
|
|
|
def _handle_delete_local_branch_result(self, result_data: Dict[str, Any], context: Dict[str, Any]) -> Tuple[bool, bool]:
|
|
if result_data.get("status") == "success":
|
|
return True, False
|
|
elif result_data.get("status") == "error":
|
|
self.main_frame.show_error("Delete Error", result_data.get("message"))
|
|
return False, False
|
|
|
|
def _handle_merge_local_branch_result(self, result_data: Dict[str, Any], context: Dict[str, Any]) -> Tuple[bool, bool]:
|
|
status = result_data.get("status")
|
|
if status == "success":
|
|
return True, True
|
|
elif status == "conflict":
|
|
self.main_frame.show_error("Merge Conflict", result_data.get("message"))
|
|
self._reenable_widgets_after_modal()
|
|
return True, False
|
|
elif status == "error":
|
|
self.main_frame.show_error("Merge Error", result_data.get("message"))
|
|
self._reenable_widgets_after_modal()
|
|
return False, False
|
|
|
|
def _handle_check_connection_result(self, result_data: Dict[str, Any], context: Dict[str, Any]) -> Tuple[bool, bool]:
|
|
func_name = "_handle_check_connection_result"
|
|
status = result_data.get("status")
|
|
message = result_data.get("message")
|
|
remote_name = context.get("remote_name_checked", "unknown remote")
|
|
|
|
if status == "success":
|
|
log_handler.log_info(f"Connection check successful for '{remote_name}'.", func_name=func_name)
|
|
self.app._update_gui_auth_status("ok")
|
|
log_handler.log_info("Connection OK. Triggering remote-dependent refreshes.", func_name=func_name)
|
|
if hasattr(self.app, "refresh_remote_status"): self.app.master.after(50, self.app.refresh_remote_status)
|
|
if hasattr(self.app, "refresh_remote_branches"): self.app.master.after(100, self.app.refresh_remote_branches)
|
|
elif status == "auth_required":
|
|
self.app._update_gui_auth_status("required")
|
|
if self.main_frame.ask_yes_no("Authentication Required", f"Authentication is required for remote '{remote_name}'.\n\nAttempt authentication now?"):
|
|
args = (self.app.git_commands, context.get("repo_path_checked"), remote_name)
|
|
self.app._start_async_operation(
|
|
async_workers.run_interactive_auth_attempt_async,
|
|
args,
|
|
{"context": "interactive_auth", "original_context": context}
|
|
)
|
|
else:
|
|
self._reenable_widgets_after_modal()
|
|
elif status == "error":
|
|
error_type = result_data.get("result", "unknown_error")
|
|
self.app._update_gui_auth_status(error_type)
|
|
self.main_frame.show_error("Connection Error", message)
|
|
|
|
return False, False
|
|
|
|
def _handle_interactive_auth_result(self, result_data: Dict[str, Any], context: Dict[str, Any]) -> Tuple[bool, bool]:
|
|
if result_data.get("status") == "success":
|
|
self.app.master.after(100, self.app.remote_handler.check_connection_auth)
|
|
else:
|
|
self.app._update_gui_auth_status("failed")
|
|
self.main_frame.show_warning("Authentication Failed", result_data.get("message"))
|
|
self._reenable_widgets_after_modal()
|
|
return False, False
|
|
|
|
def _handle_apply_remote_config_result(self, result_data: Dict[str, Any], context: Dict[str, Any]) -> Tuple[bool, bool]:
|
|
if result_data.get("status") == "success":
|
|
self.app.master.after(50, self.app.remote_handler.check_connection_auth)
|
|
else:
|
|
self.main_frame.show_error("Apply Config Error", f"Failed:\n{result_data.get('message')}")
|
|
self.app._update_gui_auth_status("unknown")
|
|
return False, False
|
|
|
|
def _handle_fetch_remote_result(self, result_data: Dict[str, Any], context: Dict[str, Any]) -> Tuple[bool, bool]:
|
|
if result_data.get("status") == "success":
|
|
return True, True
|
|
else:
|
|
self.main_frame.show_error("Fetch Error", f"Failed:\n{result_data.get('message')}")
|
|
auth_status = self._determine_auth_status_from_error(result_data.get("exception"))
|
|
self.app._update_gui_auth_status(auth_status)
|
|
return False, False
|
|
|
|
def _handle_pull_remote_result(self, result_data: Dict[str, Any], context: Dict[str, Any]) -> Tuple[bool, bool]:
|
|
status = result_data.get("status")
|
|
if status == "success":
|
|
return True, True
|
|
elif status == "conflict":
|
|
self.main_frame.show_error("Merge Conflict", result_data.get("message"))
|
|
self._reenable_widgets_after_modal()
|
|
return True, False
|
|
else:
|
|
self.main_frame.show_error("Pull Error", f"Failed:\n{result_data.get('message')}")
|
|
auth_status = self._determine_auth_status_from_error(result_data.get("exception"))
|
|
self.app._update_gui_auth_status(auth_status)
|
|
return False, False
|
|
|
|
def _handle_push_remote_result(self, result_data: Dict[str, Any], context: Dict[str, Any]) -> Tuple[bool, bool]:
|
|
status = result_data.get("status")
|
|
if status == "success":
|
|
return False, True
|
|
elif status == "rejected":
|
|
self.main_frame.show_warning("Push Rejected", result_data.get("message"))
|
|
self.app.master.after(100, self.app.remote_handler.fetch_remote)
|
|
else:
|
|
self.main_frame.show_error("Push Error", f"Failed:\n{result_data.get('message')}")
|
|
auth_status = self._determine_auth_status_from_error(result_data.get("exception"))
|
|
self.app._update_gui_auth_status(auth_status)
|
|
return False, False
|
|
|
|
def _handle_push_tags_remote_result(self, result_data: Dict[str, Any], context: Dict[str, Any]) -> Tuple[bool, bool]:
|
|
if result_data.get("status") == "success":
|
|
return True, False
|
|
else:
|
|
self.main_frame.show_error("Push Tags Error", f"Failed:\n{result_data.get('message')}")
|
|
auth_status = self._determine_auth_status_from_error(result_data.get("exception"))
|
|
self.app._update_gui_auth_status(auth_status)
|
|
return False, False
|
|
|
|
def _handle_clone_remote_result(self, result_data: Dict[str, Any], context: Dict[str, Any]) -> Tuple[bool, bool]:
|
|
if result_data.get("status") == "success":
|
|
self.app.profile_handler.create_profile_after_clone(context.get("clone_success_data"))
|
|
else:
|
|
self.main_frame.show_error("Clone Error", result_data.get("message"))
|
|
self._reenable_widgets_after_modal()
|
|
return False, False
|
|
|
|
def _handle_checkout_tracking_branch_result(self, result_data: Dict[str, Any], context: Dict[str, Any]) -> Tuple[bool, bool]:
|
|
if result_data.get("status") == "success":
|
|
return True, True
|
|
else:
|
|
self.main_frame.show_error("Checkout Error", result_data.get("message"))
|
|
self._reenable_widgets_after_modal()
|
|
return False, False
|
|
|
|
def _handle_compare_branches_result(self, result_data: Dict[str, Any], context: Dict[str, Any]) -> Tuple[bool, bool]:
|
|
if result_data.get("status") == "success":
|
|
self.app.window_handler.handle_show_comparison_summary(
|
|
context.get("ref1"), context.get("ref2"), context.get("repo_path"), result_data.get("result", [])
|
|
)
|
|
else:
|
|
self.main_frame.show_error("Comparison Error", f"Could not compare branches:\n{result_data.get('message')}")
|
|
return False, False
|
|
|
|
def _handle_get_commit_details_result(self, result_data: Dict[str, Any], context: Dict[str, Any]) -> Tuple[bool, bool]:
|
|
if result_data.get("status") == "success":
|
|
self.app.window_handler.handle_show_commit_details(result_data.get("result"))
|
|
else:
|
|
self.main_frame.show_error("Commit Details Error", f"Could not get details:\n{result_data.get('message')}")
|
|
return False, False
|
|
|
|
def _handle_revert_to_tag_result(self, result_data: Dict[str, Any], context: Dict[str, Any]) -> Tuple[bool, bool]:
|
|
if result_data.get("status") == "success":
|
|
self.main_frame.show_info("Operation Successful", result_data.get("message"))
|
|
return True, True
|
|
else:
|
|
self.main_frame.show_error("Revert to Tag Error", f"Failed:\n{result_data.get('message')}")
|
|
return False, False
|
|
|
|
def _handle_analyze_history_result(self, result_data: Dict[str, Any], context: Dict[str, Any]) -> Tuple[bool, bool]:
|
|
if result_data.get("status") == "success":
|
|
purgeable_files = result_data.get("result", [])
|
|
if purgeable_files:
|
|
self.app.window_handler.handle_purge_confirmation(context.get("repo_path"), purgeable_files)
|
|
else:
|
|
self.main_frame.show_info("Analysis Complete", result_data.get("message"))
|
|
self._reenable_widgets_after_modal()
|
|
else:
|
|
self.main_frame.show_error("Analysis Error", f"Failed to analyze repository history:\n{result_data.get('message')}")
|
|
self._reenable_widgets_after_modal()
|
|
return False, False
|
|
|
|
def _handle_purge_history_result(self, result_data: Dict[str, Any], context: Dict[str, Any]) -> Tuple[bool, bool]:
|
|
if result_data.get("status") == "success":
|
|
self.main_frame.show_info("Purge Successful", result_data.get("message"))
|
|
return True, True
|
|
else:
|
|
self.main_frame.show_error("Purge Failed", f"The history cleaning process failed:\n\n{result_data.get('message')}")
|
|
self._reenable_widgets_after_modal()
|
|
return False, False
|
|
|
|
def _handle_refresh_submodules_result(self, result_data: Dict[str, Any], context: Dict[str, Any]) -> Tuple[bool, bool]:
|
|
if result_data.get("status") == "success":
|
|
self.main_frame.submodules_tab.update_submodules_list(result_data.get("result", []))
|
|
else:
|
|
self.main_frame.submodules_tab.update_submodules_list(None)
|
|
return False, False
|
|
|
|
def _handle_add_submodule_result(self, result_data: Dict[str, Any], context: Dict[str, Any]) -> Tuple[bool, bool]:
|
|
if result_data.get("status") == "success":
|
|
return True, True
|
|
else:
|
|
self.main_frame.show_error("Add Submodule Error", f"Failed:\n{result_data.get('message')}")
|
|
return False, False
|
|
|
|
def _handle_sync_all_submodules_result(self, result_data: Dict[str, Any], context: Dict[str, Any]) -> Tuple[bool, bool]:
|
|
if result_data.get("status") == "success" and result_data.get("committed"):
|
|
return True, True
|
|
elif result_data.get("status") == "error":
|
|
self.main_frame.show_error("Sync Submodules Error", f"Failed:\n{result_data.get('message')}")
|
|
return False, False
|
|
|
|
def _handle_remove_submodule_result(self, result_data: Dict[str, Any], context: Dict[str, Any]) -> Tuple[bool, bool]:
|
|
if result_data.get("status") == "success":
|
|
return True, True
|
|
else:
|
|
self.main_frame.show_error("Remove Submodule Error", f"Failed:\n{result_data.get('message')}")
|
|
return False, False
|
|
|
|
def _handle_init_submodules_result(self, result_data: Dict[str, Any], context: Dict[str, Any]) -> Tuple[bool, bool]:
|
|
if result_data.get("status") == "success":
|
|
return True, False
|
|
else:
|
|
self.main_frame.show_error("Initialize Submodules Error", f"Failed:\n{result_data.get('message')}")
|
|
return False, False
|
|
|
|
def _handle_generic_result(self, result_data: Dict[str, Any], context: Dict[str, Any]) -> Tuple[bool, bool]:
|
|
status, message = result_data.get("status"), result_data.get("message", "Operation finished.")
|
|
if status == "error":
|
|
self.main_frame.show_error(f"Error: {context.get('context', 'Task')}", message)
|
|
elif status == "warning":
|
|
self.main_frame.show_warning("Operation Info", message)
|
|
return False, False
|
|
|
|
def _determine_auth_status_from_error(self, exception: Optional[Exception]) -> str:
|
|
if isinstance(exception, GitCommandError) and exception.stderr:
|
|
stderr_low = exception.stderr.lower()
|
|
if any(e in stderr_low for e in ["authentication failed", "permission denied"]): return "failed"
|
|
if any(e in stderr_low for e in ["repository not found", "could not resolve host"]): return "connection_failed"
|
|
return "unknown_error"
|
|
|
|
def _reenable_widgets_after_modal(self):
|
|
if hasattr(self.app, "master") and self.app.master.winfo_exists():
|
|
self.app.master.after(50, self.app._reenable_widgets_if_ready)
|
|
|
|
def _trigger_post_action_refreshes(self, sync_refresh_needed: bool):
|
|
repo_path = self.app._get_and_validate_svn_path("Post-Action Refresh")
|
|
if not repo_path or not self.app._is_repo_ready(repo_path):
|
|
self._reenable_widgets_after_modal()
|
|
return
|
|
|
|
refresh_funcs = [
|
|
self.app.refresh_commit_history, self.app.refresh_branch_list,
|
|
self.app.refresh_tag_list, self.app.refresh_changed_files_list,
|
|
self.app.refresh_remote_branches, self.app.submodule_logic_handler.refresh_submodules
|
|
]
|
|
|
|
delay = 50
|
|
for func in refresh_funcs:
|
|
self.app.master.after(delay, func)
|
|
delay += 75
|
|
|
|
if sync_refresh_needed:
|
|
self.app.master.after(delay, self.app.refresh_remote_status)
|
|
delay += 75
|
|
|
|
self.app.master.after(delay + 50, self._reenable_widgets_if_ready) |