347 lines
14 KiB
Python
347 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""Small desktop front door for Labyricorn repository sync and editors."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
from typing import Callable
|
|
|
|
from git_repo import (
|
|
GitAuthenticationError,
|
|
GitRepository,
|
|
GitRepositoryError,
|
|
RepositoryState,
|
|
)
|
|
|
|
|
|
SCRIPT_DIR = Path(__file__).resolve().parent
|
|
SITE_ROOT = SCRIPT_DIR.parent
|
|
EDITOR_SCRIPTS = (
|
|
("Blog Editor", "blog_editor.py", True),
|
|
("Article Editor", "article_editor.py", True),
|
|
("Tag Editor", "tag_editor.py", True),
|
|
("Project Editor", "project_editor.py", True),
|
|
)
|
|
|
|
|
|
def run_gui() -> None:
|
|
import tkinter as tk
|
|
from tkinter import messagebox, simpledialog, ttk
|
|
|
|
class ManagementLauncher:
|
|
def __init__(self) -> None:
|
|
self.root = tk.Tk()
|
|
self.root.title("Labyricorn Management")
|
|
self.root.minsize(620, 430)
|
|
self.repository: GitRepository | None = None
|
|
self.state: RepositoryState | None = None
|
|
self.busy = False
|
|
|
|
outer = ttk.Frame(self.root, padding=14)
|
|
outer.grid(row=0, column=0, sticky="nsew")
|
|
self.root.columnconfigure(0, weight=1)
|
|
self.root.rowconfigure(0, weight=1)
|
|
outer.columnconfigure(0, weight=1)
|
|
|
|
ttk.Label(
|
|
outer,
|
|
text="Labyricorn Management",
|
|
font=("TkDefaultFont", 15, "bold"),
|
|
).grid(row=0, column=0, sticky="w", pady=(0, 12))
|
|
|
|
repository_frame = ttk.LabelFrame(outer, text="Repository", padding=10)
|
|
repository_frame.grid(row=1, column=0, sticky="ew")
|
|
repository_frame.columnconfigure(1, weight=1)
|
|
self.values: dict[str, tk.StringVar] = {}
|
|
fields = (
|
|
("Repository root", "root"),
|
|
("Repository", "repository"),
|
|
("Branch", "branch"),
|
|
("Remote", "remote"),
|
|
("Upstream", "upstream"),
|
|
("Status", "status"),
|
|
("Working tree", "working_tree"),
|
|
)
|
|
for row, (label, key) in enumerate(fields):
|
|
ttk.Label(repository_frame, text=f"{label}:").grid(
|
|
row=row, column=0, sticky="nw", padx=(0, 10), pady=2
|
|
)
|
|
value = tk.StringVar(value="Checking…" if key == "status" else "—")
|
|
self.values[key] = value
|
|
ttk.Label(repository_frame, textvariable=value, wraplength=460).grid(
|
|
row=row, column=1, sticky="w", pady=2
|
|
)
|
|
|
|
self.details_var = tk.StringVar(value="Fetching remote metadata…")
|
|
ttk.Label(
|
|
repository_frame,
|
|
textvariable=self.details_var,
|
|
wraplength=560,
|
|
foreground="#7a5200",
|
|
).grid(row=len(fields), column=0, columnspan=2, sticky="w", pady=(8, 2))
|
|
|
|
controls = ttk.Frame(repository_frame)
|
|
controls.grid(
|
|
row=len(fields) + 1,
|
|
column=0,
|
|
columnspan=2,
|
|
sticky="w",
|
|
pady=(10, 0),
|
|
)
|
|
self.refresh_button = ttk.Button(controls, text="Refresh", command=self.refresh)
|
|
self.refresh_button.grid(row=0, column=0)
|
|
self.pull_button = ttk.Button(
|
|
controls, text="Pull", command=self.pull, state="disabled"
|
|
)
|
|
self.pull_button.grid(row=0, column=1, padx=(8, 0))
|
|
self.push_button = ttk.Button(
|
|
controls, text="Commit & Push", command=self.push, state="disabled"
|
|
)
|
|
self.push_button.grid(row=0, column=2, padx=(8, 0))
|
|
|
|
editors_frame = ttk.LabelFrame(outer, text="Editors", padding=10)
|
|
editors_frame.grid(row=2, column=0, sticky="ew", pady=(14, 0))
|
|
editors_frame.columnconfigure(0, weight=1)
|
|
editor_row = 0
|
|
for label, filename, required in EDITOR_SCRIPTS:
|
|
path = SCRIPT_DIR / filename
|
|
if not required and not path.is_file():
|
|
continue
|
|
button = ttk.Button(
|
|
editors_frame,
|
|
text=label,
|
|
command=lambda selected=path, name=label: self.launch_editor(
|
|
selected, name
|
|
),
|
|
)
|
|
button.grid(row=editor_row, column=0, sticky="ew", pady=3)
|
|
if not path.is_file():
|
|
button.configure(state="disabled")
|
|
ttk.Label(editors_frame, text=f"{filename} is missing").grid(
|
|
row=editor_row, column=1, sticky="w", padx=(10, 0)
|
|
)
|
|
editor_row += 1
|
|
|
|
self.root.after(50, self.refresh)
|
|
|
|
def refresh(self) -> None:
|
|
def operation() -> RepositoryState:
|
|
repository = GitRepository.locate(SCRIPT_DIR)
|
|
self.repository = repository
|
|
return repository.refresh()
|
|
|
|
self._start_git_task("Refreshing repository status", operation)
|
|
|
|
def pull(self) -> None:
|
|
if self.repository is None or self.state is None:
|
|
return
|
|
repository = self.repository
|
|
expected = self.state
|
|
self._start_git_task(
|
|
"Pulling with fast-forward only", lambda: repository.pull(expected)
|
|
)
|
|
|
|
def push(self) -> None:
|
|
if self.repository is None or self.state is None:
|
|
return
|
|
repository = self.repository
|
|
expected = self.state
|
|
try:
|
|
changes = repository.publishing_changes()
|
|
except GitRepositoryError as exc:
|
|
messagebox.showerror(
|
|
"Cannot prepare publication", str(exc), parent=self.root
|
|
)
|
|
return
|
|
|
|
commit_message: str | None = None
|
|
if changes:
|
|
commit_message = simpledialog.askstring(
|
|
"Commit Labyricorn Changes",
|
|
"Enter the Git commit message for these reviewed changes:",
|
|
parent=self.root,
|
|
)
|
|
if commit_message is None:
|
|
return
|
|
commit_message = " ".join(commit_message.splitlines()).strip()
|
|
if not commit_message:
|
|
messagebox.showerror(
|
|
"Commit message required",
|
|
"Enter a non-empty commit message before publishing.",
|
|
parent=self.root,
|
|
)
|
|
return
|
|
|
|
lines = changes.splitlines()
|
|
preview_limit = 40
|
|
preview = "\n".join(lines[:preview_limit])
|
|
if len(lines) > preview_limit:
|
|
preview += f"\n… and {len(lines) - preview_limit} more paths"
|
|
if not preview:
|
|
preview = "No uncommitted paths; existing local commits will be pushed."
|
|
branch_warning = (
|
|
"\n\nWARNING: Pushing main automatically deploys production."
|
|
if expected.branch == "main"
|
|
else ""
|
|
)
|
|
confirmed = messagebox.askyesno(
|
|
"Confirm Commit and Push",
|
|
f"Repository changes to publish:\n\n{preview}"
|
|
f"\n\nCommit message: {commit_message or '(existing commits only)'}"
|
|
f"{branch_warning}\n\nContinue?",
|
|
parent=self.root,
|
|
)
|
|
if not confirmed:
|
|
return
|
|
self._start_git_task(
|
|
"Committing and pushing reviewed changes",
|
|
lambda: repository.publish(expected, commit_message, changes),
|
|
)
|
|
|
|
def _start_git_task(
|
|
self, label: str, operation: Callable[[], RepositoryState]
|
|
) -> None:
|
|
if self.busy:
|
|
return
|
|
self.busy = True
|
|
self.details_var.set(f"{label}…")
|
|
self._update_buttons()
|
|
|
|
def worker() -> None:
|
|
try:
|
|
state = operation()
|
|
except GitAuthenticationError as exc:
|
|
self.root.after(0, self._finish_authentication_error, label, str(exc))
|
|
except (GitRepositoryError, OSError) as exc:
|
|
self.root.after(0, self._finish_error, label, str(exc))
|
|
else:
|
|
self.root.after(0, self._finish_success, state)
|
|
|
|
threading.Thread(target=worker, daemon=True).start()
|
|
|
|
def _finish_authentication_error(self, label: str, detail: str) -> None:
|
|
self.busy = False
|
|
self.state = None
|
|
self.details_var.set(detail)
|
|
self.values["status"].set("Authentication required")
|
|
self._update_buttons()
|
|
if self.repository is None:
|
|
messagebox.showerror(label, detail, parent=self.root)
|
|
return
|
|
retry = messagebox.askyesno(
|
|
"Git authentication failed",
|
|
f"{detail}\n\n"
|
|
"Clear only the saved credential for this repository's HTTPS host "
|
|
"and open the Git Credential Manager sign-in flow?\n\n"
|
|
"Working files and local commits will not be changed.",
|
|
parent=self.root,
|
|
)
|
|
if not retry:
|
|
return
|
|
repository = self.repository
|
|
|
|
def clear_and_refresh() -> RepositoryState:
|
|
repository.clear_https_credential()
|
|
return repository.refresh(allow_prompt=True)
|
|
|
|
self._start_git_task("Clearing rejected credential and signing in", clear_and_refresh)
|
|
|
|
def _finish_success(self, state: RepositoryState) -> None:
|
|
self.busy = False
|
|
self.state = state
|
|
self.values["root"].set(str(state.root))
|
|
self.values["repository"].set(state.repository_name)
|
|
self.values["branch"].set(state.branch)
|
|
if not state.origin_url:
|
|
remote_text = "Not configured"
|
|
elif state.origin_push_url and state.origin_push_url != state.origin_url:
|
|
remote_text = (
|
|
f"origin — {state.origin_url}\nPush target — {state.origin_push_url}"
|
|
)
|
|
else:
|
|
remote_text = f"origin — {state.origin_url}"
|
|
self.values["remote"].set(remote_text)
|
|
self.values["upstream"].set(state.upstream or "Not configured")
|
|
self.values["status"].set(state.status_text)
|
|
self.values["working_tree"].set(state.working_tree_text)
|
|
if state.problem:
|
|
self.details_var.set(state.problem)
|
|
elif state.sync_state == "diverged":
|
|
self.details_var.set(
|
|
f"Local is {state.ahead} ahead and {state.behind} behind. "
|
|
"Resolve the histories manually; no automatic action is offered."
|
|
)
|
|
elif state.dirty_count and state.sync_state == "remote-ahead":
|
|
self.details_var.set(
|
|
"Pull is unavailable until the working-tree changes are "
|
|
"reviewed or committed."
|
|
)
|
|
elif state.dirty_count and state.can_publish:
|
|
self.details_var.set(
|
|
"Working-tree changes are ready for review with Commit & Push."
|
|
)
|
|
else:
|
|
self.details_var.set("Remote metadata fetched successfully.")
|
|
self._update_buttons()
|
|
|
|
def _finish_error(self, label: str, detail: str) -> None:
|
|
self.busy = False
|
|
self.state = None
|
|
self.details_var.set(detail)
|
|
self.values["status"].set("Unavailable")
|
|
self._update_buttons()
|
|
messagebox.showerror(label, detail, parent=self.root)
|
|
|
|
def _update_buttons(self) -> None:
|
|
self.refresh_button.configure(state="disabled" if self.busy else "normal")
|
|
pull_enabled = (
|
|
not self.busy and self.state is not None and self.state.can_pull
|
|
)
|
|
push_enabled = (
|
|
not self.busy and self.state is not None and self.state.can_publish
|
|
)
|
|
self.pull_button.configure(state="normal" if pull_enabled else "disabled")
|
|
self.push_button.configure(state="normal" if push_enabled else "disabled")
|
|
|
|
def launch_editor(self, path: Path, label: str) -> None:
|
|
if not path.is_file():
|
|
messagebox.showerror(
|
|
f"Cannot launch {label}",
|
|
f"The editor script is missing: {path}",
|
|
parent=self.root,
|
|
)
|
|
return
|
|
try:
|
|
subprocess.Popen(
|
|
[sys.executable, str(path), "--site-root", str(SITE_ROOT)],
|
|
cwd=SITE_ROOT,
|
|
)
|
|
except OSError as exc:
|
|
messagebox.showerror(
|
|
f"Cannot launch {label}",
|
|
f"Could not start {path.name}: {exc}",
|
|
parent=self.root,
|
|
)
|
|
|
|
try:
|
|
app = ManagementLauncher()
|
|
except tk.TclError as exc:
|
|
raise GitRepositoryError(f"Cannot open the Tkinter window: {exc}") from exc
|
|
app.root.mainloop()
|
|
|
|
|
|
def main() -> int:
|
|
try:
|
|
run_gui()
|
|
except GitRepositoryError as exc:
|
|
print(f"management-launcher: ERROR: {exc}", file=sys.stderr)
|
|
return 1
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|