# --- FILE: gitsync_tool/core/wiki_updater.py --- import os import shutil import tempfile import re import time from typing import Optional from urllib.parse import urlparse, urlunparse from enum import Enum, auto from dataclasses import dataclass from ..commands.git_commands import GitCommands, GitCommandError from ..logging_setup import log_handler class WikiUpdateStatus(Enum): """Represents the result of the wiki update operation.""" SUCCESS = auto() NO_CHANGES = auto() URL_DERIVATION_FAILED = auto() CLONE_FAILED = auto() PUSH_FAILED = auto() COMMIT_FAILED = auto() DOC_NOT_FOUND = auto() BRANCH_NOT_FOUND = auto() GENERIC_ERROR = auto() @dataclass class WikiUpdateResult: """Holds the result of the wiki update operation.""" status: WikiUpdateStatus message: str class WikiUpdater: DEFAULT_WIKI_EN_FILENAME = "English-Manual.md" DEFAULT_WIKI_IT_FILENAME = "Italian-Manual.md" def __init__(self, git_commands: GitCommands): if not isinstance(git_commands, GitCommands): raise TypeError("WikiUpdater requires a GitCommands instance.") self.git_commands: GitCommands = git_commands log_handler.log_debug("WikiUpdater initialized.", func_name="__init__") def _get_wiki_repo_url(self, main_repo_url: str) -> Optional[str]: if not main_repo_url: return None try: parsed_url = urlparse(main_repo_url) path = parsed_url.path if path.lower().endswith(".git"): new_path = path[:-4] + ".wiki.git" else: new_path = path.rstrip("/") + ".wiki.git" wiki_url_parts = parsed_url._replace(path=new_path) return urlunparse(wiki_url_parts) except Exception as e: log_handler.log_error( f"Could not derive Wiki URL from '{main_repo_url}': {e}", func_name="_get_wiki_repo_url", ) return None def update_wiki_from_docs( self, main_repo_path: str, main_repo_remote_url: str, doc_dir_name: str = "doc", en_manual_filename: str = "English-Manual.md", it_manual_filename: str = "Italian-Manual.md", wiki_en_target_filename: Optional[str] = None, wiki_it_target_filename: Optional[str] = None, commit_message: str = "Update Wiki documentation from local files", ) -> WikiUpdateResult: """ Clones the Gitea wiki repo, updates pages from local doc files, and pushes. """ func_name = "update_wiki_from_docs" log_handler.log_debug("Starting wiki update process.", func_name=func_name) wiki_url = self._get_wiki_repo_url(main_repo_remote_url) if not wiki_url: msg = "Could not derive Wiki repository URL from main remote URL." log_handler.log_error(msg, func_name=func_name) return WikiUpdateResult( status=WikiUpdateStatus.URL_DERIVATION_FAILED, message=msg ) log_handler.log_debug(f"Derived wiki URL: {wiki_url}", func_name=func_name) doc_path = os.path.join(main_repo_path, doc_dir_name) en_doc_file = os.path.join(doc_path, en_manual_filename) it_doc_file = os.path.join(doc_path, it_manual_filename) target_en_wiki_file = wiki_en_target_filename or self.DEFAULT_WIKI_EN_FILENAME target_it_wiki_file = wiki_it_target_filename or self.DEFAULT_WIKI_IT_FILENAME en_exists = os.path.isfile(en_doc_file) it_exists = os.path.isfile(it_doc_file) if not en_exists and not it_exists: msg = f"Neither '{en_manual_filename}' nor '{it_manual_filename}' found in '{doc_path}'. Cannot update Wiki." log_handler.log_warning(msg, func_name=func_name) return WikiUpdateResult(status=WikiUpdateStatus.DOC_NOT_FOUND, message=msg) temp_dir: Optional[str] = None try: temp_dir = tempfile.mkdtemp(prefix="gitea_wiki_update_") log_handler.log_info( f"Cloning Wiki repo into temp dir: {temp_dir}", func_name=func_name ) clone_result = self.git_commands.git_clone(wiki_url, temp_dir) if clone_result.returncode != 0: stderr_msg = clone_result.stderr or "Unknown clone error" if ( "repository not found" in stderr_msg.lower() or "does not appear to be a git repository" in stderr_msg.lower() ): msg = f"Failed to clone Wiki: Repository at '{wiki_url}' not found. Please create the first page manually." else: msg = f"Failed to clone Wiki repository '{wiki_url}'. Error: {stderr_msg.strip()}" log_handler.log_error(msg, func_name=func_name) return WikiUpdateResult( status=WikiUpdateStatus.CLONE_FAILED, message=f"{msg} (Check authentication?)", ) log_handler.log_info( "Wiki repository cloned successfully.", func_name=func_name ) if en_exists: with open(en_doc_file, "r", encoding="utf-8") as f_en: en_content = f_en.read() with open( os.path.join(temp_dir, target_en_wiki_file), "w", encoding="utf-8", newline="\n", ) as f_wiki_en: f_wiki_en.write(en_content) if it_exists: with open(it_doc_file, "r", encoding="utf-8") as f_it: it_content = f_it.read() with open( os.path.join(temp_dir, target_it_wiki_file), "w", encoding="utf-8", newline="\n", ) as f_wiki_it: f_wiki_it.write(it_content) if not self.git_commands.git_status_has_changes(temp_dir): msg = "Wiki content is already up-to-date with local doc files." log_handler.log_info(msg, func_name=func_name) return WikiUpdateResult(status=WikiUpdateStatus.NO_CHANGES, message=msg) log_handler.log_info( "Changes detected. Staging with --renormalize to handle line endings.", func_name=func_name, ) self.git_commands.add_file(temp_dir, ".", renormalize=True) commit_success = self.git_commands.git_commit( temp_dir, commit_message, stage_all_first=False ) if not commit_success: msg = "Staged changes, but commit reported no changes to commit. Push will be skipped." log_handler.log_warning(msg, func_name=func_name) return WikiUpdateResult(status=WikiUpdateStatus.NO_CHANGES, message=msg) log_handler.log_info("Wiki changes committed locally.", func_name=func_name) current_wiki_branch = self.git_commands.get_current_branch_name(temp_dir) if not current_wiki_branch: msg = "Could not determine the current branch in the cloned wiki repository." log_handler.log_error(msg, func_name=func_name) return WikiUpdateResult( status=WikiUpdateStatus.BRANCH_NOT_FOUND, message=msg ) log_handler.log_info( "Pushing wiki changes to remote...", func_name=func_name ) push_result = self.git_commands.git_push( working_directory=temp_dir, remote_name="origin", branch_name=current_wiki_branch, force=False, ) if push_result.returncode == 0: msg = "Wiki update pushed successfully to Gitea." log_handler.log_info(msg, func_name=func_name) return WikiUpdateResult(status=WikiUpdateStatus.SUCCESS, message=msg) else: stderr_msg = push_result.stderr or "Unknown push error" msg = f"Failed to push Wiki updates. Error: {stderr_msg.strip()}" log_handler.log_error(msg, func_name=func_name) return WikiUpdateResult( status=WikiUpdateStatus.PUSH_FAILED, message=msg ) except (GitCommandError, ValueError, IOError, Exception) as e: log_handler.log_exception( f"Error during wiki update process: {e}", func_name=func_name ) return WikiUpdateResult( status=WikiUpdateStatus.GENERIC_ERROR, message=f"Wiki update failed: {e}", ) finally: if temp_dir and os.path.isdir(temp_dir): log_handler.log_debug( f"Attempting to clean up temporary directory: {temp_dir}", func_name=func_name, ) for attempt in range(3): try: shutil.rmtree(temp_dir) log_handler.log_info( f"Cleanup of temporary directory successful.", func_name=func_name, ) break except Exception as clean_e: log_handler.log_warning( f"Cleanup attempt {attempt + 1} failed for '{temp_dir}': {clean_e}", func_name=func_name, ) if attempt < 2: time.sleep(0.5) else: log_handler.log_error( f"Final cleanup attempt failed for '{temp_dir}'.", func_name=func_name, )