238 lines
12 KiB
Python
238 lines
12 KiB
Python
# --- FILE: gitsync_tool/core/wiki_updater.py ---
|
|
|
|
import os
|
|
import shutil
|
|
import tempfile
|
|
import re
|
|
import time
|
|
from typing import Optional, Tuple
|
|
|
|
from ..commands.git_commands import GitCommands, GitCommandError
|
|
from ..logging_setup import log_handler
|
|
|
|
class WikiUpdater:
|
|
"""Handles updating the Gitea Wiki repository from local Markdown files."""
|
|
|
|
# Nomi file di default nella Wiki (potrebbero diventare configurabili)
|
|
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]:
|
|
"""
|
|
Attempts to derive the .wiki.git URL from the main repository URL.
|
|
|
|
Args:
|
|
main_repo_url (str): The URL of the main Gitea repository.
|
|
|
|
Returns:
|
|
Optional[str]: The derived Wiki repository URL (ending in .wiki.git)
|
|
or None if derivation fails.
|
|
"""
|
|
if not main_repo_url:
|
|
return None
|
|
# Rimuovi eventuale / finale
|
|
url = main_repo_url.rstrip('/')
|
|
# Gestisci common suffixes .git
|
|
if url.lower().endswith(".git"):
|
|
base_url = url[:-4]
|
|
else:
|
|
base_url = url
|
|
# Aggiungi .wiki.git
|
|
if base_url:
|
|
return f"{base_url}.wiki.git"
|
|
return None # Fallback
|
|
|
|
def update_wiki_from_docs(
|
|
self,
|
|
main_repo_path: str, # Path del repo principale per trovare /doc
|
|
main_repo_remote_url: str, # URL del repo principale per derivare wiki URL
|
|
doc_dir_name: str = "doc", # Nome cartella documenti
|
|
en_manual_filename: str = "English-Manual.md",
|
|
it_manual_filename: str = "Italian-Manual.md",
|
|
wiki_en_target_filename: Optional[str] = None, # Override per nome file wiki EN
|
|
wiki_it_target_filename: Optional[str] = None, # Override per nome file wiki IT
|
|
commit_message: str = "Update Wiki documentation from local files"
|
|
) -> Tuple[bool, str]: # Ritorna (successo, messaggio)
|
|
"""
|
|
Clones the Gitea wiki repo, updates pages from local doc files, and pushes.
|
|
|
|
Returns:
|
|
Tuple[bool, str]: (True, success_message) on success,
|
|
(False, error_message) on failure.
|
|
"""
|
|
func_name = "update_wiki_from_docs"
|
|
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 False, msg
|
|
|
|
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)
|
|
|
|
# Determina nomi file di destinazione nella wiki
|
|
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
|
|
|
|
log_handler.log_info(f"Starting Wiki update for Gitea repo: {wiki_url}", func_name=func_name)
|
|
log_handler.log_debug(f" Source EN doc: {en_doc_file}", func_name=func_name)
|
|
log_handler.log_debug(f" Source IT doc: {it_doc_file}", func_name=func_name)
|
|
log_handler.log_debug(f" Target EN wiki file: {target_en_wiki_file}", func_name=func_name)
|
|
log_handler.log_debug(f" Target IT wiki file: {target_it_wiki_file}", func_name=func_name)
|
|
|
|
# Verifica esistenza file sorgente
|
|
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 False, msg # Considera questo un "successo" vuoto o errore? Per ora errore.
|
|
|
|
temp_dir: Optional[str] = None
|
|
try:
|
|
# 1. Crea directory temporanea sicura
|
|
temp_dir = tempfile.mkdtemp(prefix="gitea_wiki_update_")
|
|
log_handler.log_info(f"Cloning Wiki repo '{wiki_url}' into temp dir: {temp_dir}", func_name=func_name)
|
|
|
|
# 2. Clona il repository Wiki
|
|
# Usa GitCommands.clone (non solleva eccezioni di default, controlla RC)
|
|
clone_result = self.git_commands.git_clone(wiki_url, temp_dir)
|
|
if clone_result.returncode != 0:
|
|
# Gestisci errore clone (es. repo non esiste, auth fallita)
|
|
stderr_msg = clone_result.stderr or "Unknown clone error"
|
|
# Controlla se il repo non esiste (tipico per wiki non inizializzate)
|
|
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 or not initialized. "
|
|
f"Please create the first page in the Gitea Wiki manually first.")
|
|
log_handler.log_error(msg, func_name=func_name)
|
|
return False, msg
|
|
else:
|
|
msg = f"Failed to clone Wiki repository '{wiki_url}'. Error: {stderr_msg.strip()}"
|
|
log_handler.log_error(msg, func_name=func_name)
|
|
# Potrebbe essere un problema di autenticazione
|
|
return False, f"{msg} (Check authentication?)"
|
|
|
|
log_handler.log_info("Wiki repository cloned successfully.", func_name=func_name)
|
|
|
|
# 3. Leggi i file locali e Scrivi/Aggiorna i file nella Wiki clonata
|
|
files_updated = False
|
|
if en_exists:
|
|
try:
|
|
with open(en_doc_file, 'r', encoding='utf-8') as f_en:
|
|
en_content = f_en.read()
|
|
wiki_en_path = os.path.join(temp_dir, target_en_wiki_file)
|
|
# Sovrascrivi o crea il file nella wiki
|
|
with open(wiki_en_path, 'w', encoding='utf-8', newline='\n') as f_wiki_en:
|
|
f_wiki_en.write(en_content)
|
|
log_handler.log_info(f"Updated/Created '{target_en_wiki_file}' in local wiki clone.", func_name=func_name)
|
|
files_updated = True
|
|
except Exception as e:
|
|
msg = f"Error reading local file '{en_doc_file}' or writing to wiki file '{target_en_wiki_file}': {e}"
|
|
log_handler.log_exception(msg, func_name=func_name)
|
|
return False, msg # Errore critico
|
|
|
|
if it_exists:
|
|
try:
|
|
with open(it_doc_file, 'r', encoding='utf-8') as f_it:
|
|
it_content = f_it.read()
|
|
wiki_it_path = os.path.join(temp_dir, target_it_wiki_file)
|
|
with open(wiki_it_path, 'w', encoding='utf-8', newline='\n') as f_wiki_it:
|
|
f_wiki_it.write(it_content)
|
|
log_handler.log_info(f"Updated/Created '{target_it_wiki_file}' in local wiki clone.", func_name=func_name)
|
|
files_updated = True
|
|
except Exception as e:
|
|
msg = f"Error reading local file '{it_doc_file}' or writing to wiki file '{target_it_wiki_file}': {e}"
|
|
log_handler.log_exception(msg, func_name=func_name)
|
|
return False, msg
|
|
|
|
if not files_updated:
|
|
msg = "Source files found but failed to write to local wiki clone (check logs)."
|
|
log_handler.log_error(msg, func_name=func_name)
|
|
return False, msg
|
|
|
|
# 4. Controlla se ci sono cambiamenti effettivi da committare
|
|
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 True, msg # Successo, nessuna azione necessaria
|
|
|
|
log_handler.log_info("Changes detected in local wiki clone. Proceeding with commit and push.", func_name=func_name)
|
|
|
|
# 5. Commit
|
|
commit_success = self.git_commands.git_commit(temp_dir, commit_message)
|
|
if not commit_success:
|
|
# Questo non dovrebbe accadere se status_has_changes era True, ma controlliamo
|
|
msg = "Failed to commit changes to local wiki clone (git commit reported no changes unexpectedly)."
|
|
log_handler.log_warning(msg, func_name=func_name)
|
|
# Consideriamo questo un errore lieve? O un fallimento?
|
|
# Proseguiamo con il push per sicurezza, ma segnaliamo
|
|
# return False, msg # Potremmo fermarci qui
|
|
log_handler.log_info("Wiki changes committed locally.", func_name=func_name)
|
|
|
|
# 6. Push
|
|
log_handler.log_info("Pushing wiki changes to remote...", func_name=func_name)
|
|
# Assumiamo che il push vada al branch di default ('master' o 'main') su 'origin' (il remote della wiki)
|
|
# Nota: git_push richiede il nome del branch; otteniamolo dal clone
|
|
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 False, msg
|
|
|
|
push_result = self.git_commands.git_push(
|
|
working_directory=temp_dir,
|
|
remote_name='origin', # 'origin' è il default per un clone
|
|
branch_name=current_wiki_branch,
|
|
set_upstream=False, # Non necessario qui
|
|
force=False
|
|
)
|
|
|
|
if push_result.returncode == 0:
|
|
msg = "Wiki update pushed successfully to Gitea."
|
|
log_handler.log_info(msg, func_name=func_name)
|
|
return True, msg
|
|
else:
|
|
# Push fallito (conflitto? auth?)
|
|
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)
|
|
# Distingui tra rejected e altri errori
|
|
if "rejected" in stderr_msg.lower():
|
|
return False, f"{msg} (Remote may have changes, try updating wiki manually first?)"
|
|
else:
|
|
return False, f"{msg} (Check authentication?)"
|
|
|
|
except (GitCommandError, ValueError, IOError, Exception) as e:
|
|
# Cattura eccezioni generali dal processo
|
|
log_handler.log_exception(f"Error during wiki update process: {e}", func_name=func_name)
|
|
return False, f"Wiki update failed: {e}"
|
|
|
|
finally:
|
|
# 7. Pulizia directory temporanea (SEMPRE)
|
|
if temp_dir and os.path.isdir(temp_dir):
|
|
try:
|
|
shutil.rmtree(temp_dir)
|
|
log_handler.log_info(f"Cleaned up temporary wiki directory: {temp_dir}", func_name=func_name)
|
|
except Exception as clean_e:
|
|
log_handler.log_error(f"Failed to clean up temporary directory '{temp_dir}': {clean_e}", func_name=func_name)
|
|
try:
|
|
time.sleep(0.5) # Aspetta 500 millisecondi
|
|
shutil.rmtree(temp_dir)
|
|
log_handler.log_info(f"Successfully cleaned up temporary wiki directory on retry: {temp_dir}", func_name=func_name)
|
|
except Exception as clean_e2:
|
|
log_handler.log_error(
|
|
f"Failed to clean up temporary directory '{temp_dir}' even after retry: {clean_e2}",
|
|
func_name=func_name
|
|
)
|
|
# Nota: L'operazione principale potrebbe essere comunque riuscita.
|
|
# Non cambiamo success_flag qui, ma l'errore è loggato.
|
|
|