730 lines
28 KiB
Python
730 lines
28 KiB
Python
"""
|
|
Git Update Checker - GUI multi-repo.
|
|
Vérifie les mises à jour de plusieurs dépôts Git et propose de les télécharger.
|
|
Accès lecture seule uniquement (fetch/pull/checkout, jamais de push).
|
|
Tous les chemins sont relatifs à l'emplacement de l'exécutable.
|
|
"""
|
|
|
|
import subprocess
|
|
import sys
|
|
import os
|
|
import configparser
|
|
import logging
|
|
import threading
|
|
import tkinter as tk
|
|
from tkinter import ttk, messagebox
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
|
|
# Forcer UTF-8 sur Windows
|
|
if sys.platform == "win32":
|
|
os.system("chcp 65001 >nul 2>&1")
|
|
try:
|
|
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
|
|
sys.stderr.reconfigure(encoding="utf-8", errors="replace")
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
# ── Utilitaires ──────────────────────────────────────────────────────────────
|
|
|
|
def get_exe_dir():
|
|
if getattr(sys, "frozen", False):
|
|
return Path(sys.executable).parent
|
|
return Path(__file__).parent
|
|
|
|
|
|
def resolve_relative(path_str):
|
|
"""Résout un chemin relatif par rapport au dossier de l'exe."""
|
|
p = Path(path_str)
|
|
if not p.is_absolute():
|
|
p = get_exe_dir() / p
|
|
return p.resolve()
|
|
|
|
|
|
# ── Logging ──────────────────────────────────────────────────────────────────
|
|
|
|
def setup_logging():
|
|
"""Configure le logging dans un dossier log/ à côté de l'exe."""
|
|
log_dir = get_exe_dir() / "log"
|
|
log_dir.mkdir(exist_ok=True)
|
|
|
|
log_file = log_dir / f"{datetime.now().strftime('%Y-%m-%d')}.log"
|
|
|
|
logger = logging.getLogger("GitUpdateChecker")
|
|
logger.setLevel(logging.DEBUG)
|
|
|
|
# Handler fichier
|
|
fh = logging.FileHandler(log_file, encoding="utf-8")
|
|
fh.setLevel(logging.DEBUG)
|
|
fh.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(message)s", datefmt="%H:%M:%S"))
|
|
logger.addHandler(fh)
|
|
|
|
# Nettoyage des vieux logs (garder 30 jours)
|
|
for old_log in sorted(log_dir.glob("*.log"))[:-30]:
|
|
try:
|
|
old_log.unlink()
|
|
except OSError:
|
|
pass
|
|
|
|
return logger
|
|
|
|
|
|
log = setup_logging()
|
|
|
|
|
|
def run_git(args, cwd=None):
|
|
# -c safe.directory=* : évite l'erreur "dubious ownership" sur clé USB
|
|
cmd = ["git", "-c", "safe.directory=*"] + args
|
|
log.debug(f"git {' '.join(args)} (cwd={cwd})")
|
|
try:
|
|
result = subprocess.run(
|
|
cmd, cwd=cwd, capture_output=True, text=True, timeout=30,
|
|
creationflags=subprocess.CREATE_NO_WINDOW if sys.platform == "win32" else 0,
|
|
)
|
|
if result.returncode != 0 and result.stderr.strip():
|
|
log.warning(f"git {args[0]} erreur: {result.stderr.strip()}")
|
|
return result.returncode, result.stdout.strip(), result.stderr.strip()
|
|
except FileNotFoundError:
|
|
log.error("Git non trouve dans le PATH")
|
|
return -1, "", "Git n'est pas installe ou pas dans le PATH."
|
|
except subprocess.TimeoutExpired:
|
|
log.error(f"Timeout: git {' '.join(args)}")
|
|
return 1, "", "Timeout"
|
|
|
|
|
|
# ── Auto-update du programme ─────────────────────────────────────────────────
|
|
|
|
def check_self_update():
|
|
"""
|
|
Vérifie si le dossier de l'exe est un dépôt git avec des MAJ disponibles.
|
|
Retourne (needs_update: bool, info: str).
|
|
"""
|
|
exe_dir = str(get_exe_dir())
|
|
|
|
if not os.path.isdir(os.path.join(exe_dir, ".git")):
|
|
log.info("Auto-update: pas de .git dans le dossier de l'exe, skip")
|
|
return False, ""
|
|
|
|
log.info("Auto-update: verification...")
|
|
|
|
code, branch, _ = run_git(["rev-parse", "--abbrev-ref", "HEAD"], cwd=exe_dir)
|
|
if code != 0:
|
|
return False, "Impossible de lire la branche"
|
|
|
|
code, _, err = run_git(["fetch", "origin"], cwd=exe_dir)
|
|
if code != 0:
|
|
log.warning(f"Auto-update: fetch echoue: {err}")
|
|
return False, f"Fetch echoue: {err}"
|
|
|
|
code, local_hash, _ = run_git(["rev-parse", "HEAD"], cwd=exe_dir)
|
|
code2, remote_hash, _ = run_git(["rev-parse", f"origin/{branch}"], cwd=exe_dir)
|
|
|
|
if code != 0 or code2 != 0:
|
|
return False, "Impossible de comparer les commits"
|
|
|
|
if local_hash == remote_hash:
|
|
log.info("Auto-update: programme a jour")
|
|
return False, ""
|
|
|
|
# Compter les commits en retard
|
|
code, log_out, _ = run_git(
|
|
["log", "--oneline", f"HEAD..origin/{branch}"],
|
|
cwd=exe_dir,
|
|
)
|
|
count = len(log_out.splitlines()) if code == 0 and log_out else 0
|
|
info = f"{count} commit(s) en retard sur origin/{branch}"
|
|
log.info(f"Auto-update: MAJ disponible - {info}")
|
|
return True, info
|
|
|
|
|
|
def do_self_update():
|
|
"""Pull les mises à jour du programme lui-même. Retourne (ok, message)."""
|
|
exe_dir = str(get_exe_dir())
|
|
|
|
code, branch, _ = run_git(["rev-parse", "--abbrev-ref", "HEAD"], cwd=exe_dir)
|
|
if code != 0:
|
|
return False, "Impossible de lire la branche"
|
|
|
|
# Restaurer les fichiers locaux modifiés d'abord
|
|
run_git(["checkout", "--", "."], cwd=exe_dir)
|
|
|
|
code, out, err = run_git(["pull", "origin", branch], cwd=exe_dir)
|
|
if code == 0:
|
|
log.info(f"Auto-update: pull OK")
|
|
return True, "Mise a jour du programme reussie !\nRedemarre le programme pour appliquer."
|
|
else:
|
|
log.error(f"Auto-update: pull echoue: {err}")
|
|
return False, f"Erreur pull: {err}"
|
|
|
|
|
|
# ── Configuration ────────────────────────────────────────────────────────────
|
|
|
|
def get_config_path():
|
|
return get_exe_dir() / "config.ini"
|
|
|
|
|
|
def load_repos():
|
|
"""Charge la liste des dépôts depuis config.ini."""
|
|
config_path = get_config_path()
|
|
config = configparser.ConfigParser()
|
|
repos = []
|
|
|
|
if not config_path.exists():
|
|
# Créer un config.ini exemple
|
|
config["repo:Exemple"] = {
|
|
"url": "http://192.168.1.235:3125/user/repo",
|
|
"path": "../MonRepo",
|
|
}
|
|
with open(config_path, "w", encoding="utf-8") as f:
|
|
f.write("; Configuration des depots Git a surveiller\n")
|
|
f.write("; Les chemins (path) sont relatifs a l'emplacement de l'exe\n")
|
|
f.write("; Ajouter autant de sections [repo:NomDuRepo] que necessaire\n\n")
|
|
config.write(f)
|
|
return repos
|
|
|
|
config.read(config_path, encoding="utf-8")
|
|
|
|
for section in config.sections():
|
|
if section.startswith("repo:"):
|
|
name = section[5:]
|
|
url = config.get(section, "url", fallback="").strip()
|
|
path = config.get(section, "path", fallback="").strip()
|
|
if url and path:
|
|
repos.append({"name": name, "url": url, "path": path})
|
|
|
|
return repos
|
|
|
|
|
|
# ── Logique Git ──────────────────────────────────────────────────────────────
|
|
|
|
def check_repo(repo):
|
|
"""Vérifie un dépôt et retourne son état."""
|
|
name = repo["name"]
|
|
url = repo["url"]
|
|
local_path = str(resolve_relative(repo["path"]))
|
|
log.info(f"[{name}] Verification - {url} -> {local_path}")
|
|
|
|
result = {
|
|
"name": name,
|
|
"url": url,
|
|
"local_path": local_path,
|
|
"relative_path": repo["path"],
|
|
"exists": False,
|
|
"up_to_date": False,
|
|
"error": None,
|
|
"commits": [],
|
|
"files": [],
|
|
"local_changes": [],
|
|
"local_only": False,
|
|
"branch": "",
|
|
"local_hash": "",
|
|
"remote_hash": "",
|
|
"needs_clone": False,
|
|
}
|
|
|
|
# Vérifier si le dossier existe et est un repo git
|
|
if not os.path.isdir(os.path.join(local_path, ".git")):
|
|
if os.path.isdir(local_path) and os.listdir(local_path):
|
|
result["error"] = "Le dossier existe mais n'est pas un depot git."
|
|
return result
|
|
result["needs_clone"] = True
|
|
return result
|
|
|
|
result["exists"] = True
|
|
|
|
# Branche courante
|
|
code, branch, _ = run_git(["rev-parse", "--abbrev-ref", "HEAD"], cwd=local_path)
|
|
if code != 0:
|
|
result["error"] = "Impossible de lire la branche courante."
|
|
return result
|
|
result["branch"] = branch
|
|
|
|
# Fetch
|
|
code, _, err = run_git(["fetch", "origin"], cwd=local_path)
|
|
if code != 0:
|
|
result["error"] = f"Erreur fetch : {err}"
|
|
return result
|
|
|
|
# Comparer HEAD local vs origin
|
|
code, local_hash, _ = run_git(["rev-parse", "HEAD"], cwd=local_path)
|
|
if code != 0:
|
|
result["error"] = "Impossible de lire le commit local."
|
|
return result
|
|
|
|
code, remote_hash, _ = run_git(["rev-parse", f"origin/{branch}"], cwd=local_path)
|
|
if code != 0:
|
|
result["error"] = f"Impossible de trouver origin/{branch}."
|
|
return result
|
|
|
|
result["local_hash"] = local_hash[:8]
|
|
result["remote_hash"] = remote_hash[:8]
|
|
|
|
# Modifications locales
|
|
code, local_diff, _ = run_git(["diff", "--name-status", "HEAD"], cwd=local_path)
|
|
if code == 0 and local_diff:
|
|
for line in local_diff.splitlines():
|
|
parts = line.split("\t", 1)
|
|
if len(parts) == 2:
|
|
sc = parts[0].strip()
|
|
fn = parts[1].strip()
|
|
status_map = {"D": "Supprime localement", "M": "Modifie localement"}
|
|
result["local_changes"].append({
|
|
"status": status_map.get(sc[0], sc),
|
|
"status_char": sc[0],
|
|
"file": fn,
|
|
})
|
|
|
|
if local_hash == remote_hash and not result["local_changes"]:
|
|
log.info(f"[{name}] A jour ({local_hash[:8]})")
|
|
result["up_to_date"] = True
|
|
return result
|
|
|
|
if local_hash == remote_hash:
|
|
log.info(f"[{name}] {len(result['local_changes'])} fichier(s) locaux modifies/supprimes")
|
|
result["local_only"] = True
|
|
return result
|
|
|
|
# Nouveaux commits distants
|
|
code, log_out, _ = run_git(
|
|
["log", "--oneline", "--format=%h|%an|%ar|%s", f"HEAD..origin/{branch}"],
|
|
cwd=local_path,
|
|
)
|
|
if code == 0 and log_out:
|
|
for line in log_out.splitlines():
|
|
parts = line.split("|", 3)
|
|
if len(parts) == 4:
|
|
result["commits"].append({
|
|
"hash": parts[0], "author": parts[1],
|
|
"date": parts[2], "message": parts[3],
|
|
})
|
|
|
|
# Fichiers distants modifiés
|
|
code, diff_out, _ = run_git(
|
|
["diff", "--name-status", f"HEAD..origin/{branch}"],
|
|
cwd=local_path,
|
|
)
|
|
if code == 0 and diff_out:
|
|
for line in diff_out.splitlines():
|
|
parts = line.split("\t", 1)
|
|
if len(parts) == 2:
|
|
sc = parts[0].strip()
|
|
fn = parts[1].strip()
|
|
status_map = {"A": "Ajoute", "M": "Modifie", "D": "Supprime", "R": "Renomme"}
|
|
result["files"].append({
|
|
"status": status_map.get(sc[0], sc),
|
|
"status_char": sc[0],
|
|
"file": fn,
|
|
})
|
|
|
|
return result
|
|
|
|
|
|
def do_clone(repo):
|
|
"""Clone un dépôt."""
|
|
local_path = str(resolve_relative(repo["path"]))
|
|
log.info(f"Clonage: {repo['url']} -> {local_path}")
|
|
code, _, err = run_git(["clone", repo["url"], local_path])
|
|
if code == 0:
|
|
log.info(f"Clonage reussi: {local_path}")
|
|
else:
|
|
log.error(f"Clonage echoue: {err}")
|
|
return code == 0, err
|
|
|
|
|
|
def do_pull(local_path, branch):
|
|
"""Pull les mises à jour (lecture seule)."""
|
|
log.info(f"Pull: {local_path} (branche {branch})")
|
|
code, out, err = run_git(["pull", "origin", branch], cwd=local_path)
|
|
if code == 0:
|
|
log.info(f"Pull reussi: {local_path}")
|
|
else:
|
|
log.error(f"Pull echoue: {err}")
|
|
return code == 0, out, err
|
|
|
|
|
|
def do_restore(local_path):
|
|
"""Restaure les fichiers locaux modifiés/supprimés."""
|
|
log.info(f"Restauration: {local_path}")
|
|
code, _, err = run_git(["checkout", "--", "."], cwd=local_path)
|
|
if code == 0:
|
|
log.info(f"Restauration reussie: {local_path}")
|
|
else:
|
|
log.error(f"Restauration echouee: {err}")
|
|
return code == 0, err
|
|
|
|
|
|
# ── Interface graphique ──────────────────────────────────────────────────────
|
|
|
|
class App(tk.Tk):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.title("Git Update Checker")
|
|
self.geometry("820x600")
|
|
self.minsize(700, 450)
|
|
self.configure(bg="#1e1e2e")
|
|
|
|
self.repos_config = load_repos()
|
|
self.repo_results = []
|
|
|
|
log.info("=== Demarrage Git Update Checker ===")
|
|
self._build_ui()
|
|
self.after(100, self._check_self_update_then_repos)
|
|
|
|
def _build_ui(self):
|
|
style = ttk.Style(self)
|
|
style.theme_use("clam")
|
|
|
|
# Couleurs sombres
|
|
bg = "#1e1e2e"
|
|
fg = "#cdd6f4"
|
|
accent = "#89b4fa"
|
|
green = "#a6e3a1"
|
|
yellow = "#f9e2af"
|
|
red = "#f38ba8"
|
|
|
|
style.configure("TFrame", background=bg)
|
|
style.configure("TLabel", background=bg, foreground=fg, font=("Segoe UI", 10))
|
|
style.configure("Title.TLabel", background=bg, foreground=fg, font=("Segoe UI", 14, "bold"))
|
|
style.configure("Status.TLabel", background=bg, foreground=accent, font=("Segoe UI", 9))
|
|
style.configure("TButton", font=("Segoe UI", 10))
|
|
style.configure("Green.TLabel", background=bg, foreground=green, font=("Segoe UI", 10, "bold"))
|
|
style.configure("Yellow.TLabel", background=bg, foreground=yellow, font=("Segoe UI", 10, "bold"))
|
|
style.configure("Red.TLabel", background=bg, foreground=red, font=("Segoe UI", 10, "bold"))
|
|
|
|
# Header
|
|
header = ttk.Frame(self)
|
|
header.pack(fill="x", padx=15, pady=(15, 5))
|
|
ttk.Label(header, text="Git Update Checker", style="Title.TLabel").pack(side="left")
|
|
self.status_label = ttk.Label(header, text="Verification en cours...", style="Status.TLabel")
|
|
self.status_label.pack(side="right")
|
|
|
|
# Date
|
|
date_label = ttk.Label(self, text=datetime.now().strftime(" %d/%m/%Y %H:%M:%S"), style="Status.TLabel")
|
|
date_label.pack(anchor="w", padx=15)
|
|
|
|
# Zone scrollable pour les repos
|
|
container = ttk.Frame(self)
|
|
container.pack(fill="both", expand=True, padx=15, pady=10)
|
|
|
|
self.canvas = tk.Canvas(container, bg=bg, highlightthickness=0)
|
|
scrollbar = ttk.Scrollbar(container, orient="vertical", command=self.canvas.yview)
|
|
self.scroll_frame = ttk.Frame(self.canvas)
|
|
|
|
self.scroll_frame.bind("<Configure>", lambda e: self.canvas.configure(scrollregion=self.canvas.bbox("all")))
|
|
self.canvas.create_window((0, 0), window=self.scroll_frame, anchor="nw")
|
|
self.canvas.configure(yscrollcommand=scrollbar.set)
|
|
|
|
self.canvas.pack(side="left", fill="both", expand=True)
|
|
scrollbar.pack(side="right", fill="y")
|
|
|
|
# Scroll avec la molette
|
|
self.canvas.bind_all("<MouseWheel>", lambda e: self.canvas.yview_scroll(int(-1 * (e.delta / 120)), "units"))
|
|
|
|
# Boutons en bas
|
|
btn_frame = ttk.Frame(self)
|
|
btn_frame.pack(fill="x", padx=15, pady=(0, 15))
|
|
|
|
self.btn_refresh = ttk.Button(btn_frame, text="Rafraichir", command=self._start_check)
|
|
self.btn_refresh.pack(side="left", padx=(0, 10))
|
|
|
|
self.btn_update_all = ttk.Button(btn_frame, text="Tout mettre a jour", command=self._update_all)
|
|
self.btn_update_all.pack(side="left")
|
|
self.btn_update_all.state(["disabled"])
|
|
|
|
ttk.Button(btn_frame, text="Ouvrir config.ini", command=self._open_config).pack(side="right")
|
|
|
|
self.btn_open_log = ttk.Button(btn_frame, text="Ouvrir les logs", command=self._open_log)
|
|
self.btn_open_log.pack(side="right", padx=(0, 10))
|
|
|
|
def _check_self_update_then_repos(self):
|
|
"""Vérifie d'abord la MAJ du programme, puis les repos."""
|
|
self.status_label.configure(text="Verification auto-update...")
|
|
|
|
def work():
|
|
needs, info = check_self_update()
|
|
self.after(0, lambda: self._handle_self_update(needs, info))
|
|
|
|
threading.Thread(target=work, daemon=True).start()
|
|
|
|
def _handle_self_update(self, needs_update, info):
|
|
"""Gère le résultat de l'auto-update."""
|
|
if needs_update:
|
|
answer = messagebox.askyesno(
|
|
"Mise a jour du programme",
|
|
f"Une mise a jour du programme est disponible !\n\n{info}\n\nMettre a jour maintenant ?",
|
|
)
|
|
if answer:
|
|
self.status_label.configure(text="Mise a jour du programme...")
|
|
|
|
def work():
|
|
ok, msg = do_self_update()
|
|
self.after(0, lambda: self._self_update_done(ok, msg))
|
|
|
|
threading.Thread(target=work, daemon=True).start()
|
|
return
|
|
|
|
self._start_check()
|
|
|
|
def _self_update_done(self, ok, msg):
|
|
"""Callback après auto-update."""
|
|
if ok:
|
|
messagebox.showinfo("Auto-update", msg)
|
|
log.info("Auto-update appliquee, fermeture")
|
|
self.destroy()
|
|
return
|
|
else:
|
|
messagebox.showwarning("Auto-update", msg)
|
|
self._start_check()
|
|
|
|
def _start_check(self):
|
|
"""Lance la vérification dans un thread."""
|
|
self.btn_refresh.state(["disabled"])
|
|
self.btn_update_all.state(["disabled"])
|
|
self.status_label.configure(text="Verification en cours...")
|
|
self.repos_config = load_repos()
|
|
|
|
if not self.repos_config:
|
|
self._clear_cards()
|
|
self.status_label.configure(text="Aucun depot configure")
|
|
self._show_no_repos()
|
|
self.btn_refresh.state(["!disabled"])
|
|
return
|
|
|
|
threading.Thread(target=self._check_all, daemon=True).start()
|
|
|
|
def _check_all(self):
|
|
"""Vérifie tous les repos (dans un thread)."""
|
|
results = []
|
|
for repo in self.repos_config:
|
|
results.append(check_repo(repo))
|
|
self.repo_results = results
|
|
self.after(0, self._display_results)
|
|
|
|
def _display_results(self):
|
|
"""Affiche les résultats dans la GUI."""
|
|
self._clear_cards()
|
|
|
|
has_any_updates = False
|
|
for res in self.repo_results:
|
|
card = self._create_card(res)
|
|
card.pack(fill="x", pady=5)
|
|
if not res["up_to_date"] and not res.get("error"):
|
|
has_any_updates = True
|
|
|
|
total = len(self.repo_results)
|
|
up = sum(1 for r in self.repo_results if r["up_to_date"])
|
|
log.info(f"Resultat: {up}/{total} depots a jour")
|
|
self.status_label.configure(text=f"{up}/{total} depots a jour")
|
|
self.btn_refresh.state(["!disabled"])
|
|
|
|
if has_any_updates:
|
|
self.btn_update_all.state(["!disabled"])
|
|
|
|
def _clear_cards(self):
|
|
for w in self.scroll_frame.winfo_children():
|
|
w.destroy()
|
|
|
|
def _create_card(self, res):
|
|
"""Crée une carte pour un dépôt."""
|
|
bg_card = "#313244"
|
|
fg = "#cdd6f4"
|
|
green = "#a6e3a1"
|
|
yellow = "#f9e2af"
|
|
red = "#f38ba8"
|
|
cyan = "#89dceb"
|
|
dim = "#6c7086"
|
|
|
|
card = tk.Frame(self.scroll_frame, bg=bg_card, bd=0, highlightthickness=1, highlightbackground="#45475a")
|
|
|
|
# Header de la carte
|
|
top = tk.Frame(card, bg=bg_card)
|
|
top.pack(fill="x", padx=12, pady=(10, 5))
|
|
|
|
tk.Label(top, text=res["name"], bg=bg_card, fg=fg, font=("Segoe UI", 11, "bold")).pack(side="left")
|
|
|
|
if res.get("error"):
|
|
tk.Label(top, text="ERREUR", bg=bg_card, fg=red, font=("Segoe UI", 9, "bold")).pack(side="right")
|
|
elif res.get("needs_clone"):
|
|
tk.Label(top, text="A CLONER", bg=bg_card, fg=yellow, font=("Segoe UI", 9, "bold")).pack(side="right")
|
|
elif res["up_to_date"]:
|
|
tk.Label(top, text="A JOUR", bg=bg_card, fg=green, font=("Segoe UI", 9, "bold")).pack(side="right")
|
|
else:
|
|
count = len(res["commits"]) + len(res["local_changes"])
|
|
tk.Label(top, text=f"MAJ DISPO ({count})", bg=bg_card, fg=yellow, font=("Segoe UI", 9, "bold")).pack(side="right")
|
|
|
|
# Infos
|
|
info = tk.Frame(card, bg=bg_card)
|
|
info.pack(fill="x", padx=12)
|
|
tk.Label(info, text=f"Chemin : {res['relative_path']}", bg=bg_card, fg=dim, font=("Segoe UI", 8)).pack(anchor="w")
|
|
if res["branch"]:
|
|
tk.Label(info, text=f"Branche : {res['branch']} | {res['local_hash']}", bg=bg_card, fg=dim, font=("Segoe UI", 8)).pack(anchor="w")
|
|
|
|
# Erreur
|
|
if res.get("error"):
|
|
tk.Label(card, text=f" {res['error']}", bg=bg_card, fg=red, font=("Segoe UI", 9), anchor="w").pack(fill="x", padx=12, pady=(5, 0))
|
|
|
|
# Commits distants
|
|
if res["commits"]:
|
|
tk.Label(card, text=f"Nouveaux commits ({len(res['commits'])}) :", bg=bg_card, fg=fg, font=("Segoe UI", 9, "bold"), anchor="w").pack(fill="x", padx=12, pady=(8, 2))
|
|
for c in res["commits"][:5]:
|
|
line_frame = tk.Frame(card, bg=bg_card)
|
|
line_frame.pack(fill="x", padx=20)
|
|
tk.Label(line_frame, text=c["hash"], bg=bg_card, fg=cyan, font=("Consolas", 9)).pack(side="left")
|
|
tk.Label(line_frame, text=f" {c['message']}", bg=bg_card, fg=fg, font=("Segoe UI", 9), anchor="w").pack(side="left", fill="x")
|
|
if len(res["commits"]) > 5:
|
|
tk.Label(card, text=f" ... et {len(res['commits']) - 5} autres commits", bg=bg_card, fg=dim, font=("Segoe UI", 8)).pack(anchor="w", padx=20)
|
|
|
|
# Fichiers distants modifiés
|
|
if res["files"]:
|
|
tk.Label(card, text=f"Fichiers distants modifies ({len(res['files'])}) :", bg=bg_card, fg=fg, font=("Segoe UI", 9, "bold"), anchor="w").pack(fill="x", padx=12, pady=(5, 2))
|
|
color_map = {"A": green, "M": yellow, "D": red, "R": cyan}
|
|
for f in res["files"][:8]:
|
|
line_frame = tk.Frame(card, bg=bg_card)
|
|
line_frame.pack(fill="x", padx=20)
|
|
c = color_map.get(f["status_char"], fg)
|
|
tk.Label(line_frame, text=f"[{f['status']:>9}]", bg=bg_card, fg=c, font=("Consolas", 9)).pack(side="left")
|
|
tk.Label(line_frame, text=f" {f['file']}", bg=bg_card, fg=fg, font=("Segoe UI", 9)).pack(side="left")
|
|
if len(res["files"]) > 8:
|
|
tk.Label(card, text=f" ... et {len(res['files']) - 8} autres fichiers", bg=bg_card, fg=dim, font=("Segoe UI", 8)).pack(anchor="w", padx=20)
|
|
|
|
# Fichiers locaux manquants
|
|
if res["local_changes"]:
|
|
tk.Label(card, text=f"Fichiers locaux a restaurer ({len(res['local_changes'])}) :", bg=bg_card, fg=fg, font=("Segoe UI", 9, "bold"), anchor="w").pack(fill="x", padx=12, pady=(5, 2))
|
|
for f in res["local_changes"][:8]:
|
|
line_frame = tk.Frame(card, bg=bg_card)
|
|
line_frame.pack(fill="x", padx=20)
|
|
c = red if f["status_char"] == "D" else yellow
|
|
tk.Label(line_frame, text=f"[{f['status']:>20}]", bg=bg_card, fg=c, font=("Consolas", 8)).pack(side="left")
|
|
tk.Label(line_frame, text=f" {f['file']}", bg=bg_card, fg=fg, font=("Segoe UI", 9)).pack(side="left")
|
|
if len(res["local_changes"]) > 8:
|
|
tk.Label(card, text=f" ... et {len(res['local_changes']) - 8} autres fichiers", bg=bg_card, fg=dim, font=("Segoe UI", 8)).pack(anchor="w", padx=20)
|
|
|
|
# Bouton d'action
|
|
if not res["up_to_date"] and not res.get("error"):
|
|
btn_frame = tk.Frame(card, bg=bg_card)
|
|
btn_frame.pack(fill="x", padx=12, pady=(8, 0))
|
|
|
|
if res.get("needs_clone"):
|
|
btn = ttk.Button(btn_frame, text="Cloner le depot", command=lambda r=res: self._do_clone(r))
|
|
else:
|
|
btn = ttk.Button(btn_frame, text="Mettre a jour", command=lambda r=res: self._do_update(r))
|
|
btn.pack(side="left")
|
|
# Stocker ref du bouton pour le désactiver après
|
|
res["_btn"] = btn
|
|
|
|
# Padding bas
|
|
tk.Frame(card, bg=bg_card, height=10).pack()
|
|
|
|
return card
|
|
|
|
def _do_update(self, res):
|
|
"""Met à jour un dépôt (pull + restore)."""
|
|
log.info(f"[{res['name']}] MAJ unitaire demandee")
|
|
if "_btn" in res:
|
|
res["_btn"].state(["disabled"])
|
|
|
|
def work():
|
|
messages = []
|
|
success = True
|
|
local_path = res["local_path"]
|
|
branch = res["branch"]
|
|
|
|
if res["commits"]:
|
|
log.info(f"[{res['name']}] Pull de {len(res['commits'])} commit(s)...")
|
|
ok, out, err = do_pull(local_path, branch)
|
|
if ok:
|
|
msg = f"Pull OK : {len(res['commits'])} commits telecharges."
|
|
log.info(f"[{res['name']}] {msg}")
|
|
messages.append(msg)
|
|
else:
|
|
msg = f"Erreur pull : {err}"
|
|
log.error(f"[{res['name']}] {msg}")
|
|
messages.append(msg)
|
|
success = False
|
|
|
|
if res["local_changes"]:
|
|
log.info(f"[{res['name']}] Restauration de {len(res['local_changes'])} fichier(s)...")
|
|
ok, err = do_restore(local_path)
|
|
if ok:
|
|
msg = f"Restauration OK : {len(res['local_changes'])} fichiers restaures."
|
|
log.info(f"[{res['name']}] {msg}")
|
|
messages.append(msg)
|
|
else:
|
|
msg = f"Erreur restauration : {err}"
|
|
log.error(f"[{res['name']}] {msg}")
|
|
messages.append(msg)
|
|
success = False
|
|
|
|
status = "SUCCES" if success else "ECHEC"
|
|
log.info(f"[{res['name']}] MAJ unitaire terminee - {status}")
|
|
self.after(0, lambda: self._show_update_result(res, messages, success))
|
|
|
|
threading.Thread(target=work, daemon=True).start()
|
|
|
|
def _do_clone(self, res):
|
|
"""Clone un dépôt."""
|
|
log.info(f"[{res['name']}] Clonage demande - {res['url']}")
|
|
if "_btn" in res:
|
|
res["_btn"].state(["disabled"])
|
|
|
|
repo = {"url": res["url"], "path": res["relative_path"]}
|
|
|
|
def work():
|
|
ok, err = do_clone(repo)
|
|
if ok:
|
|
msg = f"Depot '{res['name']}' clone avec succes !"
|
|
log.info(f"[{res['name']}] {msg}")
|
|
else:
|
|
msg = f"Erreur de clonage : {err}"
|
|
log.error(f"[{res['name']}] {msg}")
|
|
self.after(0, lambda: messagebox.showinfo("Clonage", msg))
|
|
self.after(100, self._start_check)
|
|
|
|
threading.Thread(target=work, daemon=True).start()
|
|
|
|
def _show_update_result(self, res, messages, success):
|
|
title = "Mise a jour" if success else "Erreur"
|
|
messagebox.showinfo(title, f"{res['name']}\n\n" + "\n".join(messages))
|
|
self._start_check()
|
|
|
|
def _update_all(self):
|
|
"""Met à jour tous les dépôts qui ont des MAJ."""
|
|
to_update = [r for r in self.repo_results if not r["up_to_date"] and not r.get("error") and not r.get("needs_clone")]
|
|
log.info(f"MAJ globale demandee - {len(to_update)} depot(s) a mettre a jour")
|
|
for res in to_update:
|
|
self._do_update(res)
|
|
|
|
def _show_no_repos(self):
|
|
bg_card = "#313244"
|
|
fg = "#cdd6f4"
|
|
dim = "#6c7086"
|
|
|
|
card = tk.Frame(self.scroll_frame, bg=bg_card, bd=0, highlightthickness=1, highlightbackground="#45475a")
|
|
card.pack(fill="x", pady=5)
|
|
|
|
tk.Label(card, text="Aucun depot configure", bg=bg_card, fg=fg, font=("Segoe UI", 11, "bold")).pack(padx=12, pady=(12, 5))
|
|
tk.Label(card, text="Edite config.ini pour ajouter des depots :", bg=bg_card, fg=dim, font=("Segoe UI", 9)).pack(padx=12)
|
|
example = (
|
|
"[repo:MonRepo]\n"
|
|
"url = http://192.168.1.235:3125/user/repo\n"
|
|
"path = ../MonRepo"
|
|
)
|
|
tk.Label(card, text=example, bg="#1e1e2e", fg="#a6e3a1", font=("Consolas", 9), justify="left", padx=10, pady=8).pack(padx=12, pady=8, fill="x")
|
|
tk.Frame(card, bg=bg_card, height=10).pack()
|
|
|
|
def _open_config(self):
|
|
config_path = str(get_config_path())
|
|
if sys.platform == "win32":
|
|
os.startfile(config_path)
|
|
|
|
def _open_log(self):
|
|
log_dir = str(get_exe_dir() / "log")
|
|
if sys.platform == "win32":
|
|
os.startfile(log_dir)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
app = App()
|
|
app.mainloop()
|