951 lines
39 KiB
Python
951 lines
39 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,
|
|
)
|
|
import re
|
|
|
|
from gitutility.logging_setup import log_handler
|
|
from gitutility.commands.git_commands import GitCommandError
|
|
from gitutility.async_tasks import async_workers
|
|
from ..core.wiki_updater import WikiUpdateStatus
|
|
|
|
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,
|
|
"promote_branch": self._handle_promote_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_update_wiki_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,
|
|
"check_submodule_updates": self._handle_check_submodule_updates_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,
|
|
"force_clean_submodule": self._handle_force_clean_submodule_result,
|
|
"clean_submodule": self._handle_clean_submodule_result,
|
|
"reset_to_commit": self._handle_reset_to_commit_result,
|
|
"repository_transformation": self._handle_repository_transformation_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.app._trigger_post_action_refreshes(post_action_sync_refresh_needed)
|
|
|
|
def _handle_reset_to_commit_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(
|
|
"Reset to Commit Error", f"Failed:\n{result_data.get('message')}"
|
|
)
|
|
return False, False
|
|
|
|
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.app._reenable_widgets_if_ready()
|
|
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.app._reenable_widgets_if_ready()
|
|
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.app._reenable_widgets_if_ready()
|
|
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.app._reenable_widgets_if_ready()
|
|
return True, False
|
|
elif status == "error":
|
|
self.main_frame.show_error("Merge Error", result_data.get("message"))
|
|
self.app._reenable_widgets_if_ready()
|
|
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.app._reenable_widgets_if_ready()
|
|
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.app._reenable_widgets_if_ready()
|
|
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.app._reenable_widgets_if_ready()
|
|
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":
|
|
|
|
# --- MODIFICA CHIAVE ---
|
|
log_handler.log_info(
|
|
"Push Tags successful. Triggering automatic 'Refresh Sync Status'.",
|
|
func_name="_handle_push_tags_remote_result",
|
|
)
|
|
self.app.master.after(100, self.app.refresh_remote_status)
|
|
# ---------------------
|
|
|
|
# Push tags non cambia la storia dei branch, quindi non serve un refresh completo.
|
|
# Ma un refresh dei tag è utile.
|
|
self.app.master.after(200, self.app.refresh_tag_list)
|
|
return (
|
|
False,
|
|
False,
|
|
) # Non triggeriamo il refresh completo, ma solo quelli mirati.
|
|
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.app._reenable_widgets_if_ready()
|
|
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.app._reenable_widgets_if_ready()
|
|
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_promote_branch_result(
|
|
self, result_data: Dict[str, Any], context: Dict[str, Any]
|
|
) -> Tuple[bool, bool]:
|
|
"""Handle result of promoting a branch to main."""
|
|
status = result_data.get("status")
|
|
if status == "success":
|
|
message = result_data.get("message", "Promotion completed.")
|
|
backup = None
|
|
try:
|
|
backup = result_data.get("result", {}).get("backup_branch")
|
|
except Exception:
|
|
backup = None
|
|
info_msg = message
|
|
if backup:
|
|
info_msg += f"\nBackup created: {backup}"
|
|
self.main_frame.show_info("Promotion Successful", info_msg)
|
|
return True, True
|
|
else:
|
|
self.main_frame.show_error(
|
|
"Promotion Failed", 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.app._reenable_widgets_if_ready()
|
|
else:
|
|
self.main_frame.show_error(
|
|
"Analysis Error",
|
|
f"Failed to analyze repository history:\n{result_data.get('message')}",
|
|
)
|
|
self.app._reenable_widgets_if_ready()
|
|
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.app._reenable_widgets_if_ready()
|
|
return False, False
|
|
|
|
def _handle_refresh_submodules_result(
|
|
self, result_data: Dict[str, Any], context: Dict[str, Any]
|
|
) -> Tuple[bool, bool]:
|
|
submodules_list = result_data.get("result", [])
|
|
if result_data.get("status") == "success":
|
|
self.main_frame.submodules_tab.update_submodules_list(submodules_list)
|
|
else:
|
|
self.main_frame.submodules_tab.update_submodules_list(None)
|
|
|
|
on_complete_cb = context.get("on_complete")
|
|
if callable(on_complete_cb):
|
|
on_complete_cb(submodules_list)
|
|
|
|
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]:
|
|
status = result_data.get("status")
|
|
message = result_data.get("message")
|
|
exception = result_data.get("exception")
|
|
|
|
# --- NUOVA LOGICA DI RECUPERO (CORRETTA) ---
|
|
# Controlliamo se è un GitCommandError e se il suo stderr contiene il messaggio specifico
|
|
if (
|
|
isinstance(exception, GitCommandError)
|
|
and "untracked working tree" in (exception.stderr or "").lower()
|
|
):
|
|
|
|
# Applichiamo la regex a exception.stderr, non a str(exception)
|
|
stderr_text = exception.stderr or ""
|
|
submodule_path_match = re.search(r"in submodule path '(.*?)'", stderr_text)
|
|
|
|
submodule_path = (
|
|
submodule_path_match.group(1)
|
|
if submodule_path_match
|
|
else "an unknown submodule"
|
|
)
|
|
|
|
# Chiedi all'utente se vuole pulire
|
|
if self.main_frame.ask_yes_no(
|
|
"Sync Failed: Untracked Files",
|
|
f"The update failed because untracked files in submodule '{submodule_path}' would be overwritten.\n\n"
|
|
"Do you want to automatically clean these files (git clean -fd) and retry the sync?",
|
|
):
|
|
# L'utente ha detto sì, avvia l'operazione di pulizia
|
|
repo_path = self.app._get_and_validate_svn_path("Submodule Clean")
|
|
if repo_path:
|
|
# Se il path è 'unknown', non procediamo per sicurezza
|
|
if "unknown" in submodule_path:
|
|
self.main_frame.show_error(
|
|
"Error",
|
|
"Could not automatically determine the submodule path from the error message. Cannot proceed with automatic clean.",
|
|
)
|
|
return False, False
|
|
|
|
args = (self.app.submodule_handler, repo_path, submodule_path)
|
|
self.app._start_async_operation(
|
|
async_workers.run_clean_submodule_async,
|
|
args,
|
|
{
|
|
"context": "clean_submodule",
|
|
"status_msg": f"Cleaning submodule '{submodule_path}'...",
|
|
"on_success_retry": "sync_all",
|
|
},
|
|
)
|
|
return False, False
|
|
else:
|
|
# L'utente ha detto no, mostra l'errore originale
|
|
self.main_frame.show_error(
|
|
"Sync Submodules Error", f"Failed:\n{message}"
|
|
)
|
|
return False, False
|
|
|
|
# --- Logica esistente per altri casi ---
|
|
if status == "success":
|
|
log_handler.log_info(
|
|
"Sync/Update successful. Triggering automatic 'Check for Updates' to refresh remote status.",
|
|
func_name="_handle_sync_all_submodules_result",
|
|
)
|
|
# Usiamo 'after' per assicurarci che questa azione parta dopo che il ciclo di eventi corrente è terminato.
|
|
self.app.master.after(
|
|
100, self.app.main_frame.submodules_tab.check_for_updates_callback
|
|
)
|
|
|
|
# Se è stato fatto un commit, facciamo anche un refresh completo.
|
|
if result_data.get("committed"):
|
|
return True, True
|
|
elif status == "error":
|
|
self.main_frame.show_error("Sync Submodules Error", f"Failed:\n{message}")
|
|
|
|
return False, False
|
|
|
|
def _handle_clean_submodule_result(
|
|
self, result_data: Dict[str, Any], context: Dict[str, Any]
|
|
) -> Tuple[bool, bool]:
|
|
"""Handles the result of the submodule cleaning operation."""
|
|
if result_data.get("status") == "success":
|
|
self.main_frame.show_info("Submodule Cleaned", result_data.get("message"))
|
|
|
|
# Controlla se dobbiamo ritentare un'azione
|
|
retry_action = context.get("on_success_retry")
|
|
if retry_action == "sync_all":
|
|
log_handler.log_info(
|
|
"Clean successful, retrying 'Sync/Update All' operation.",
|
|
func_name="_handle_clean_submodule_result",
|
|
)
|
|
# Lancia di nuovo l'operazione di sync
|
|
self.app.submodule_logic_handler.sync_all_submodules()
|
|
else:
|
|
self.main_frame.show_error(
|
|
"Clean Failed",
|
|
f"The submodule cleaning process failed:\n\n{result_data.get('message')}",
|
|
)
|
|
|
|
return (
|
|
False,
|
|
False,
|
|
) # Il refresh verrà triggerato dall'azione successiva (o nessuno se fallisce)
|
|
|
|
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_force_clean_submodule_result(
|
|
self, result_data: Dict[str, Any], context: Dict[str, Any]
|
|
) -> Tuple[bool, bool]:
|
|
"""Handles the result of the submodule cleaning operation."""
|
|
if result_data.get("status") == "success":
|
|
self.main_frame.show_info(
|
|
"Operation Successful", result_data.get("message")
|
|
)
|
|
# Trigger a full refresh because the index and history have changed
|
|
return True, True
|
|
else:
|
|
self.main_frame.show_error(
|
|
"Clean Failed",
|
|
f"The submodule cleaning process failed:\n\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 _handle_check_submodule_updates_result(
|
|
self, result_data: Dict[str, Any], context: Dict[str, Any]
|
|
) -> Tuple[bool, bool]:
|
|
if result_data.get("status") == "success":
|
|
statuses = result_data.get("result", {})
|
|
self.main_frame.submodules_tab.update_remote_statuses(statuses)
|
|
|
|
return False, False
|
|
|
|
def _handle_update_wiki_result(
|
|
self, result_data: Dict[str, Any], context: Dict[str, Any]
|
|
) -> Tuple[bool, bool]:
|
|
"""Handles the result of the update_wiki_from_docs async task."""
|
|
status = result_data.get("status")
|
|
message = result_data.get("message", "Wiki operation finished.")
|
|
|
|
if status == "success":
|
|
# In our new implementation, 'success' is for both actual updates and no-changes scenarios
|
|
self.main_frame.show_info("Wiki Update", message)
|
|
# We only need to trigger a full refresh if something actually changed.
|
|
# The result object from the worker now contains the detailed status.
|
|
update_result = result_data.get("result")
|
|
if update_result and update_result.status == WikiUpdateStatus.SUCCESS:
|
|
return True, True
|
|
elif status == "warning":
|
|
self.main_frame.show_warning("Wiki Update Info", message)
|
|
elif status == "error":
|
|
self.main_frame.show_error("Wiki Update Error", f"Failed:\n{message}")
|
|
|
|
return False, False
|
|
|
|
def _handle_repository_transformation_result(
|
|
self, result_data: Dict[str, Any], context: Dict[str, Any]
|
|
) -> Tuple[bool, bool]:
|
|
"""Handles the result of the repository transformation worker."""
|
|
status = result_data.get("status")
|
|
message = result_data.get("message")
|
|
|
|
if status == "success":
|
|
self.main_frame.show_info("Transformation Successful", message)
|
|
|
|
# Aggiorna la lista dei profili nella GUI
|
|
new_profile_name = result_data.get("result", {}).get("new_profile_name")
|
|
sections = self.app.config_manager.get_profile_sections()
|
|
self.main_frame.update_profile_dropdown(sections)
|
|
|
|
# Seleziona automaticamente il nuovo profilo
|
|
if new_profile_name and new_profile_name in sections:
|
|
self.main_frame.profile_var.set(new_profile_name)
|
|
|
|
# Non serve un refresh completo, perché il cambio di profilo lo triggera già
|
|
return False, False
|
|
else: # status == "error"
|
|
self.main_frame.show_error("Transformation Failed", message)
|
|
# Un refresh completo è utile per ricaricare lo stato del repo originale
|
|
return True, True
|