commit 874df3230e568c2446e8ae63a67eb0012ae9cb97 Author: Lenrui Date: Thu Nov 5 22:56:01 2026 -0800 upload commit diff --git a/README.md b/README.md new file mode 100644 index 0000000..bf9123f --- /dev/null +++ b/README.md @@ -0,0 +1,90 @@ +# gitdesk + +A GitHub Desktop-style git client that runs in your terminal. Use the mouse or the keyboard. + +## Setup + +1. Install Python 3.9+ and git (2.25 or newer). +2. Install the one dependency: `python -m pip install -r requirements.txt` + (or just run gitdesk once and say yes when it offers to install it). + +## Build a standalone executable (optional) + +This makes a single file that runs without Python or Textual installed. + +- **Windows:** double-click `build.bat`. You get `dist\gitdesk.exe`. +- **Linux / macOS:** run `./build.sh`. You get `dist/gitdesk`. + +Build on the system you want to run it on: a Windows exe has to be built on Windows. +The scripts use their own `.build-venv` folder and don't touch your Python install. +The exe takes a second or two to start because it unpacks itself first. + +## Run + +| | | +|---|---| +| **Windows** | `gitdesk.bat` inside a repo folder, or `gitdesk.bat C:\path\to\repo`. Windows Terminal gives the best mouse support and symbols. | +| **Linux / macOS** | `python3 gitdesk.py` inside a repo, or `python3 gitdesk.py ~/path/to/repo` | + +Tip: put this folder on your PATH (Windows), or symlink it on Linux/macOS +(`ln -s "$PWD/gitdesk.py" ~/.local/bin/gitdesk`). Then you can type `gitdesk` in any repo. + +## What's on screen + +- **Top bar:** + - **Current repository:** click it to open the picker. It lists your recent repos + and has **Add local repository…**, **New repository…** and **Clone repository…**. + - **Current branch:** click to switch branches or create one. + - **Fetch / Pull / Push / Publish:** one button that changes with the state of your + branch, like GitHub Desktop's. If the repo has no remote yet, it shows **No remote**: + click it to add the URL of an empty repo you created on GitHub or Gitea, then publish. + - **More:** lists every action. +- **Startup:** gitdesk reopens the repository you had open last. If you start it inside + a repo folder, or pass a path (`gitdesk C:\path\to\repo`), it opens that one instead. +- **Add local repository** (`o`): type a path, or click **Choose…** to pick the folder. + If the folder isn't a git repository yet, it offers **Create a repository here**: + the files already in it go into the first commit (pick the Unity .gitignore for a Unity + project so Library/ and Temp/ stay out). +- **Choose… buttons** (Add local, New repository, Clone) open the normal folder picker: + the Explorer "Select Folder" window on Windows, Finder on macOS, and zenity/kdialog on + Linux desktops. With no desktop (e.g. over SSH) you get a folder browser inside the terminal. + On Windows the Explorer picker needs Python's tkinter, which the python.org installer + includes by default. Without it you get the older Windows folder dialog instead. + The same goes for the exe: build it with a Python that has tkinter. +- **New repository:** choose a name, description and folder, a README, a .gitignore + template (Unity, Python, Node, Visual Studio), an MIT license, and an optional remote URL. + It makes the first commit for you. +- **Clone repository:** paste an HTTPS or SSH URL. The folder name fills in automatically + and progress shows in the top bar. If the server needs a password, use **Retry in terminal**. +- **Changes tab:** changed files with checkboxes. Ticked files go into the commit. + Click a file to see its diff, then **Discard** it if you don't want those changes. + Discard keeps a backup in Stashed changes, so it can be undone. More → Discard + ticked changes does several files at once. Below the list are Stash all / Stashed, and + the commit summary + description. +- **History tab:** recent commits (↑ = not pushed yet). Click one to see what changed, + or **View on GitHub** to open that commit's page. +- **View on GitHub** (`g`, More menu, or the card on the "No local changes" screen) opens + the repo's web page, worked out from the remote URL. It works for GitHub, Gitea, + GitLab and Bitbucket, over SSH or HTTPS, and any token in the URL is left out. + If the address can't be worked out (e.g. SSH goes through a separate hostname), set it once: + `git config gitdesk.webUrl https://git.example.com/owner/repo` +- **Log** (`l`): shows every git command gitdesk ran and what git printed. + +## Behaviour + +- It refreshes every 3 seconds and fetches in the background every 5 minutes. +- **Pull blocked by your local changes:** it offers *Stash, pull & re-apply*. + - If re-applying clashes, the file shows the `<<<<<<< ======= >>>>>>>` markers. + - Fix the file, then click **Mark as resolved**. The stash is kept until you drop it. +- **Pull hits a merge conflict:** choose *Resolve conflicts* (fix files, Mark as resolved, + Commit merge) or *Abort merge*. +- **Push when the remote has newer commits:** it offers *Pull, then push*. +- **Switching branch blocked by your changes:** it offers *Stash & switch*. +- **A network command needs a password, token, SSH passphrase or host-key confirmation:** + click **Retry in terminal**. gitdesk steps aside so git can ask you, then comes back. +- **Another program has the repository busy** (GitHub Desktop, VS Code, Unity, OneDrive, + antivirus): gitdesk waits and retries for a few seconds. If a `.git/index.lock` is left + behind by a crashed git, it offers **Remove lock file & retry**. + +Keys: `f` fetch · `p` pull · `u` push · `b` branch · `s` stash · `t` stashes · `d` discard · +`o` add local repo · `n` new repo · `c` clone · `g` view on web · `l` log · `r` refresh · `space` tick/untick file · `q` quit diff --git a/build.bat b/build.bat new file mode 100644 index 0000000..5824582 --- /dev/null +++ b/build.bat @@ -0,0 +1,48 @@ +@echo off +rem ========================================================================== +rem build.bat - builds gitdesk into a single Windows executable: dist\gitdesk.exe +rem +rem Needs Python 3.9+ installed. Double-click it or run it from this folder. +rem It makes its own throwaway environment in .build-venv, so it doesn't touch +rem your normal Python packages. +rem ========================================================================== +setlocal +cd /d "%~dp0" + +set "PY=" +where py >nul 2>&1 && set "PY=py -3" +if not defined PY where python >nul 2>&1 && set "PY=python" +if not defined PY ( + echo Python 3 was not found. Install it from https://www.python.org/downloads/ + goto :fail +) + +echo [1/3] Creating the build environment - .build-venv ... +if not exist ".build-venv\Scripts\python.exe" %PY% -m venv .build-venv +if not exist ".build-venv\Scripts\python.exe" goto :fail +set "VPY=.build-venv\Scripts\python.exe" + +echo [2/3] Installing PyInstaller and Textual... +"%VPY%" -m pip install --quiet --upgrade pip +"%VPY%" -m pip install --quiet pyinstaller -r requirements.txt +if errorlevel 1 goto :fail + +echo [3/3] Building gitdesk.exe... +rem Textual and Rich load some of their modules on demand, so bundle all of them. +"%VPY%" -m PyInstaller --noconfirm --clean --onefile --name gitdesk ^ + --collect-submodules textual --collect-data textual --collect-submodules rich ^ + --specpath build --workpath build --distpath dist ^ + gitdesk.py +if errorlevel 1 goto :fail + +echo. +echo Done: %CD%\dist\gitdesk.exe +echo Copy it anywhere - a folder on your PATH is handy - and run "gitdesk" inside a repo. +pause +exit /b 0 + +:fail +echo. +echo Build failed - see the messages above. +pause +exit /b 1 diff --git a/build.sh b/build.sh new file mode 100644 index 0000000..e555e47 --- /dev/null +++ b/build.sh @@ -0,0 +1,36 @@ +#!/usr/bin/env bash +# +# build.sh - builds gitdesk into a single executable: dist/gitdesk +# +# Run it on the system you want the executable for (Linux builds a Linux +# binary, macOS builds a macOS one). For Windows use build.bat on Windows. +# It makes its own throwaway environment in .build-venv, so it doesn't touch +# your system Python packages. +set -euo pipefail +cd "$(dirname "$0")" + +PY=$(command -v python3 || command -v python || true) +[[ -n $PY ]] || { echo "Python 3 wasn't found. Install it first."; exit 1; } + +echo "[1/3] Creating the build environment (.build-venv)..." +if [[ ! -x .build-venv/bin/python ]]; then + "$PY" -m venv .build-venv || { + echo "Couldn't create a virtualenv. On Debian/Ubuntu: sudo apt install python3-venv"; exit 1; } +fi +VPY=.build-venv/bin/python + +echo "[2/3] Installing PyInstaller and Textual..." +"$VPY" -m pip install --quiet --upgrade pip +"$VPY" -m pip install --quiet pyinstaller -r requirements.txt + +echo "[3/3] Building..." +# Textual and Rich load some of their modules on demand, so tell PyInstaller to +# bundle all of them. +"$VPY" -m PyInstaller --noconfirm --clean --onefile --name gitdesk \ + --collect-submodules textual --collect-data textual --collect-submodules rich \ + --specpath build --workpath build --distpath dist \ + gitdesk.py + +echo +echo "Done: $PWD/dist/gitdesk" +echo "Copy it somewhere on your PATH (e.g. ~/.local/bin) and run 'gitdesk' inside a repo." diff --git a/gitdesk-preview.png b/gitdesk-preview.png new file mode 100644 index 0000000..e65c64a Binary files /dev/null and b/gitdesk-preview.png differ diff --git a/gitdesk.bat b/gitdesk.bat new file mode 100644 index 0000000..8445ddb --- /dev/null +++ b/gitdesk.bat @@ -0,0 +1,15 @@ +@echo off +rem gitdesk launcher for Windows. +rem Run it from inside a repo folder (or pass a path): gitdesk / gitdesk C:\path\to\repo +rem Needs Python 3.9+ and git. On first run it offers to install the "textual" package. +setlocal +set "PY=" +where py >nul 2>&1 && set "PY=py -3" +if not defined PY where python >nul 2>&1 && set "PY=python" +if not defined PY ( + echo Python 3 was not found. Install it from https://www.python.org/downloads/ + pause + exit /b 1 +) +%PY% "%~dp0gitdesk.py" %* +if errorlevel 1 pause diff --git a/gitdesk.py b/gitdesk.py new file mode 100644 index 0000000..205eb99 --- /dev/null +++ b/gitdesk.py @@ -0,0 +1,3536 @@ +#!/usr/bin/env python3 +""" +gitdesk - a GitHub Desktop-style git client that runs in your terminal. + +Works with the mouse and the keyboard: click the repository / branch / +Fetch-Pull-Push buttons along the top, tick the files you want in a commit, +click a file to see its diff, and use the History tab to browse commits. + + python gitdesk.py open the repository in the current folder, or the one + you had open last time + python gitdesk.py PATH open the repository at PATH + +New / Add local / Clone repository are in the repository picker (top-left). + +Needs Python 3.9+, git 2.25+ and the "textual" package (pip install textual). +Runs in Windows Terminal / PowerShell / cmd and in Linux / macOS terminals. +""" +from __future__ import annotations + +import asyncio +import json +import os +import queue +import re +import shutil +import subprocess +import sys +import threading +import time +from dataclasses import dataclass, field +from typing import Callable, List, Optional, Set, Tuple + +IS_WIN = os.name == "nt" + + +def _pick_folder_helper(args: List[str]) -> int: + """`gitdesk --pick-folder TITLE START`: show the system folder picker, print the choice. + + Runs as its own short-lived process so the GUI dialog can't interfere with the + terminal UI. Exit codes: 0 = chosen (path on stdout), 1 = cancelled, 3 = unavailable. + """ + title = args[0] if args else "Choose a folder" + start = args[1] if len(args) > 1 else "" + try: + import tkinter as tk + from tkinter import filedialog + except ImportError: + return 3 + try: + if IS_WIN: + try: # crisp (not blurry) dialog on high-DPI screens + import ctypes + ctypes.windll.shcore.SetProcessDpiAwareness(2) + except Exception: + pass + root = tk.Tk() + root.withdraw() + try: + root.attributes("-topmost", True) # don't open behind the terminal window + except Exception: + pass + root.update() + path = filedialog.askdirectory(parent=root, title=title, initialdir=start or None, mustexist=True) + root.destroy() + except Exception: + return 3 + if not path: + return 1 + sys.stdout.buffer.write(os.path.normpath(path).encode("utf-8")) + sys.stdout.flush() + return 0 + + +if __name__ == "__main__" and len(sys.argv) > 1 and sys.argv[1] == "--pick-folder": + sys.exit(_pick_folder_helper(sys.argv[2:])) + + +def _ensure_textual() -> None: + """Offer to install Textual if it's missing, then restart.""" + try: + import textual # noqa: F401 + return + except ImportError: + if getattr(sys, "frozen", False): # a built executable can't pip-install anything + raise + print("gitdesk needs the Python package 'textual'.") + try: + answer = input("Install it now with pip? [Y/n] ").strip().lower() + except EOFError: + answer = "n" + if answer not in ("", "y", "yes"): + print("Install it with: python -m pip install textual") + sys.exit(1) + cmd = [sys.executable, "-m", "pip", "install", "textual"] + if sys.prefix == sys.base_prefix: # not inside a virtualenv + cmd.insert(4, "--user") + if subprocess.call(cmd) != 0: + print("\npip couldn't install it. Try: python -m pip install textual") + print("(Some Linux distros need a virtualenv or pipx for this.)") + sys.exit(1) + sys.exit(subprocess.call([sys.executable, *sys.argv])) # restart with textual available + + +_ensure_textual() + +from rich.console import Group # noqa: E402 +from rich.table import Table # noqa: E402 +from rich.text import Text # noqa: E402 +from textual import events, on # noqa: E402 +from textual.app import App, ComposeResult, SuspendNotSupported # noqa: E402 +from textual.binding import Binding # noqa: E402 +from textual.css.query import NoMatches # noqa: E402 +from textual.containers import Horizontal, Vertical, VerticalScroll # noqa: E402 +from textual.message import Message # noqa: E402 +from textual.reactive import reactive # noqa: E402 +from textual.screen import ModalScreen # noqa: E402 +from textual.theme import Theme # noqa: E402 +from textual.widgets import ( # noqa: E402 + Button, Checkbox, ContentSwitcher, DirectoryTree, Footer, Input, Label, ListItem, ListView, + RichLog, Select, Static, TabbedContent, TabPane, TextArea, Tree, +) + +# -------------------------------------------------------------------------- +# Colours (GitHub Desktop dark-ish) +# -------------------------------------------------------------------------- +C_MUTED = "#959da5" +C_DIM = "#6a737d" +C_TEXT = "#e1e4e8" +C_BLUE = "#79b8ff" +C_GREEN = "#34d058" +C_RED = "#ea4a5a" +C_YELLOW = "#dbab09" +C_ORANGE = "#f66a0a" + +KIND_ICON = { + "M": ("●", C_YELLOW, "modified"), + "A": ("+", C_GREEN, "added"), + "?": ("+", C_GREEN, "new file"), + "D": ("−", C_RED, "deleted"), + "R": ("→", C_BLUE, "renamed"), + "U": ("!", C_ORANGE, "conflict"), +} + +MAX_FILES_SHOWN = 1000 +MAX_DIFF_ROWS = 4000 +EMPTY_TREE = "4b825dc642cb6eb9a060e54bf8d69288fbee4904" + +# Background git commands must never sit waiting for a password prompt; +# if one is needed the user gets a "Retry in terminal" button instead. +GIT_ENV = dict(os.environ) +GIT_ENV.update({ + "GIT_TERMINAL_PROMPT": "0", + "GIT_MERGE_AUTOEDIT": "no", + "GIT_OPTIONAL_LOCKS": "0", # status refreshes don't grab the index lock +}) + + +# -------------------------------------------------------------------------- +# Git layer +# -------------------------------------------------------------------------- +@dataclass +class Result: + code: int + out: str = "" + err: str = "" + lockish: bool = False # failed because another program had the repository locked + handled: bool = False # the user has already been shown this failure + + @property + def ok(self) -> bool: + return self.code == 0 + + @property + def message(self) -> str: + return self.err.strip() or self.out.strip() or f"git exited with code {self.code}" + + +# Failures that mean "the repository was busy" rather than "this can't be done". +LOCKISH_RE = re.compile( + r"index\.lock|could not write index|unable to write (new )?index|unable to create '[^']*\.lock'" + r"|cannot lock ref|resource busy|permission denied|file exists", + re.I) +# git's normal "no, because..." answers - never retried as if the repo were busy. +NOT_LOCK = ("CONFLICT", "would be overwritten", "already exists", "needs merge", "did not match", + "not a valid", "unknown revision", "nothing to commit", "no changes added", + "Please tell me who you are", "hook", "Filename too long") +STASH_KEPT_RE = re.compile(r"The stash entry is kept in case you need it again\.?") + + +def looks_lockish(r: Result) -> bool: + text = f"{r.err}\n{r.out}" + if any(s in text for s in NOT_LOCK): + return False + residue = STASH_KEPT_RE.sub("", text).strip() + return not residue or bool(LOCKISH_RE.search(residue)) + + +def hint_for(message: str) -> str: + """Plain-English next step for common git errors.""" + if "Filename too long" in message: + return ("Windows limits paths to 260 characters unless git is told otherwise. Run this once, " + "then try again:\n git config --global core.longpaths true") + if "Please tell me who you are" in message: + return ("git needs your name and email for commits. Run these once:\n" + ' git config --global user.name "Your Name"\n' + " git config --global user.email you@example.com") + if "Permission denied" in message or "Invalid argument" in message: + return ("A file is probably open in another program (Unity, an editor, a sync tool like OneDrive). " + "Close it and try again.") + return "" + + +def git_version() -> Tuple[int, ...]: + r = run_git(os.getcwd(), ["--version"]) + m = re.search(r"(\d+)\.(\d+)", r.out) + return (int(m.group(1)), int(m.group(2))) if m else (0, 0) + + +@dataclass +class FileChange: + path: str + kind: str # M A D R U ? (see KIND_ICON) + orig: Optional[str] = None # old path for renames + + @property + def is_conflict(self) -> bool: + return self.kind == "U" + + +@dataclass +class Commit: + sha: str + short: str + subject: str + author: str + when: str + + +@dataclass +class RepoState: + branch: Optional[str] = None # None = detached HEAD + oid: Optional[str] = None # None = no commits yet + upstream: Optional[str] = None # e.g. "origin/main" + upstream_gone: bool = False + ahead: int = 0 + behind: int = 0 + files: List[FileChange] = field(default_factory=list) + merging: bool = False + rebasing: bool = False + stashes: int = 0 + remotes: List[str] = field(default_factory=list) + remote_urls: dict = field(default_factory=dict) # name -> fetch URL + + @property + def conflicts(self) -> List[FileChange]: + return [f for f in self.files if f.is_conflict] + + @property + def has_upstream(self) -> bool: + return bool(self.upstream) and not self.upstream_gone + + +class GitError(Exception): + pass + + +def run_git(cwd: str, args, input: Optional[str] = None, env=None) -> Result: + cmd = ["git", "-c", "core.quotepath=false", "-c", "color.ui=false", *args] + kw = dict(cwd=cwd, capture_output=True, text=True, encoding="utf-8", + errors="replace", env=env or GIT_ENV) + if input is None: + kw["stdin"] = subprocess.DEVNULL + else: + kw["input"] = input + # Detach from the terminal so ssh/git can't grab the keyboard behind the UI. + if IS_WIN: + kw["creationflags"] = subprocess.CREATE_NO_WINDOW + else: + kw["start_new_session"] = True + try: + p = subprocess.run(cmd, **kw) + except FileNotFoundError: + return Result(127, "", "git was not found - is it installed and in your PATH?") + except OSError as e: + return Result(1, "", str(e)) + return Result(p.returncode, p.stdout or "", p.stderr or "") + + +def _kind(xy: str, entry: str) -> str: + if entry == "2": + return "R" + if "D" in xy: + return "D" + if "A" in xy: + return "A" + return "M" + + +def parse_status(out: str, st: RepoState) -> None: + """Parse `git status --porcelain=v2 --branch -z` output into st.""" + fields = out.split("\0") + i = 0 + while i < len(fields): + f = fields[i] + i += 1 + if not f: + continue + t = f[0] + if t == "#": + key, _, val = f[2:].partition(" ") + if key == "branch.oid": + st.oid = None if val == "(initial)" else val + elif key == "branch.head": + st.branch = None if val == "(detached)" else val + elif key == "branch.upstream": + st.upstream = val + st.upstream_gone = True # until branch.ab shows up + elif key == "branch.ab": + st.upstream_gone = False + a, b = val.split() + st.ahead, st.behind = int(a.lstrip("+")), int(b.lstrip("-")) + elif t == "1": + p = f.split(" ", 8) + st.files.append(FileChange(p[8], _kind(p[1], "1"))) + elif t == "2": + p = f.split(" ", 9) + orig = fields[i] if i < len(fields) else None + i += 1 + st.files.append(FileChange(p[9], "R", orig)) + elif t == "u": + p = f.split(" ", 10) + st.files.append(FileChange(p[10], "U")) + elif t == "?": + st.files.append(FileChange(f[2:], "?")) + st.files.sort(key=lambda fc: fc.path.lower()) + + +class Git: + def __init__(self, path: str, git_dir: str, common_dir: str, log: Callable): + self.path = path + self.git_dir = git_dir + self.common_dir = common_dir + self._log = log + # Only one git process at a time from gitdesk, so the background refresh can + # never be holding the repository while an action (stash, commit...) runs. + self.lock = threading.RLock() + + @classmethod + def open(cls, path: str, log: Callable) -> Optional["Git"]: + path = os.path.abspath(os.path.expanduser(path)) + if not os.path.isdir(path): + return None + r = run_git(path, ["rev-parse", "--show-toplevel", "--absolute-git-dir", "--git-common-dir"]) + lines = r.out.splitlines() + if not r.ok or len(lines) < 3: + return None + top, git_dir, common = lines[0], lines[1], lines[2] + if not os.path.isabs(common): + common = os.path.normpath(os.path.join(path, common)) + return cls(os.path.normpath(top), git_dir, common, log) + + def run(self, *args, input: Optional[str] = None, log: bool = True, serialize: bool = True) -> Result: + if serialize: + with self.lock: + r = run_git(self.path, list(args), input=input) + else: + r = run_git(self.path, list(args), input=input) + if log: + self._log(list(args), r) + return r + + def snapshot(self) -> tuple: + """Everything a failed command could have changed - used to decide if a retry is safe.""" + r = self.run("status", "--porcelain=v2", "--branch", "-z", "-uall", log=False) + return (r.out, self.stash_count(), os.path.exists(os.path.join(self.git_dir, "MERGE_HEAD"))) + + def index_lock_path(self) -> str: + return os.path.join(self.git_dir, "index.lock") + + # ---- state ------------------------------------------------------------- + def state(self) -> RepoState: + r = self.run("status", "--porcelain=v2", "--branch", "-z", "-uall", log=False) + if not r.ok: + raise GitError(r.message) + st = RepoState() + parse_status(r.out, st) + st.merging = os.path.exists(os.path.join(self.git_dir, "MERGE_HEAD")) + st.rebasing = any(os.path.isdir(os.path.join(self.git_dir, d)) + for d in ("rebase-merge", "rebase-apply")) + st.stashes = self.stash_count() + for line in self.run("remote", "-v", log=False).out.splitlines(): + parts = line.split() + if len(parts) >= 2 and parts[0] not in st.remote_urls: + st.remotes.append(parts[0]) + st.remote_urls[parts[0]] = parts[1] + return st + + def stash_count(self) -> int: + try: + with open(os.path.join(self.common_dir, "logs", "refs", "stash"), "rb") as fh: + return sum(1 for line in fh if line.strip()) + except OSError: + return 0 + + def last_fetch_time(self) -> Optional[float]: + for d in (self.git_dir, self.common_dir): + try: + return os.path.getmtime(os.path.join(d, "FETCH_HEAD")) + except OSError: + pass + return None + + def merge_message(self) -> str: + try: + with open(os.path.join(self.git_dir, "MERGE_MSG"), encoding="utf-8", errors="replace") as fh: + return fh.readline().strip() + except OSError: + return "" + + # ---- diffs ------------------------------------------------------------- + def file_diff(self, fc: FileChange, has_head: bool) -> str: + if fc.kind == "?": + return self._untracked_diff(fc.path) + args = ["diff", "HEAD" if has_head else EMPTY_TREE, "--no-ext-diff", "-M", "--"] + if fc.orig: + args.append(fc.orig) + args.append(fc.path) + r = self.run(*args, log=False) + return r.out if r.ok else r.message + + def _untracked_diff(self, path: str) -> str: + try: + with open(os.path.join(self.path, path), "rb") as fh: + data = fh.read(512_000) + except OSError as e: + return f"Can't read this file: {e}" + if b"\0" in data[:8000]: + return f"diff --git a/{path} b/{path}\nBinary files /dev/null and b/{path} differ" + text = data.decode("utf-8", errors="replace") + lines = text.splitlines() + out = [f"diff --git a/{path} b/{path}", f"@@ -0,0 +1,{len(lines)} @@"] + out += ["+" + line for line in lines] + if text and not text.endswith(("\n", "\r")): + out.append("\\ No newline at end of file") + return "\n".join(out) + + def changed_between(self, a: str, b: str) -> Set[str]: + r = self.run("diff", "--name-only", "-z", f"{a}...{b}", log=False) + return {p for p in r.out.split("\0") if p} + + def commit_header(self, sha: str) -> str: + return self.run("show", "-s", "--date=format:%Y-%m-%d %H:%M", + "--format=%H%n%an <%ae>%n%ad%n%B", sha, log=False).out + + def commit_diff(self, sha: str) -> str: + return self.run("show", "--format=", "-p", "-M", "--first-parent", + "--no-ext-diff", sha, log=False).out + + # ---- history / branches / stashes ------------------------------------- + def history(self, has_upstream: bool, limit: int = 300) -> Tuple[List[Commit], Set[str]]: + r = self.run("log", f"-n{limit}", "--format=%H%x1f%h%x1f%s%x1f%an%x1f%ar", log=False) + commits = [] + for line in r.out.splitlines(): + parts = line.split("\x1f") + if len(parts) == 5: + commits.append(Commit(*parts)) + unpushed: Set[str] = set() + if has_upstream: + unpushed = set(self.run("rev-list", "@{u}..HEAD", log=False).out.split()) + return commits, unpushed + + def local_branches(self) -> List[Tuple[str, str]]: + r = self.run("for-each-ref", "--sort=-committerdate", + "--format=%(refname:short)%1f%(committerdate:relative)", "refs/heads", log=False) + return [tuple(line.split("\x1f", 1)) for line in r.out.splitlines() if "\x1f" in line] + + def remote_branches(self, remote: str) -> List[Tuple[str, str]]: + r = self.run("for-each-ref", "--sort=-committerdate", + "--format=%(refname:short)%1f%(committerdate:relative)", + f"refs/remotes/{remote}", log=False) + out = [] + for line in r.out.splitlines(): + if "\x1f" not in line: + continue + name, when = line.split("\x1f", 1) + if name in (remote, f"{remote}/HEAD"): + continue + out.append((name, when)) + return out + + def stash_list(self) -> List[Tuple[str, str, str]]: + r = self.run("stash", "list", "--format=%gd%x1f%gs%x1f%cr", log=False) + return [tuple(line.split("\x1f", 2)) for line in r.out.splitlines() if line.count("\x1f") >= 2] + + def stash_preview(self, ref: str) -> str: + r = self.run("stash", "show", "--stat", "--include-untracked", ref, log=False) + if not r.ok: + r = self.run("stash", "show", "--stat", ref, log=False) + return r.out.rstrip() or r.message + + +# -------------------------------------------------------------------------- +# Helpers +# -------------------------------------------------------------------------- +def config_dir() -> str: + if IS_WIN: + base = os.environ.get("APPDATA") or os.path.expanduser("~") + else: + base = os.environ.get("XDG_CONFIG_HOME") or os.path.expanduser("~/.config") + return os.path.join(base, "gitdesk") + + +def _settings_file() -> str: + return os.path.join(config_dir(), "recent.json") + + +def load_settings() -> dict: + try: + with open(_settings_file(), encoding="utf-8") as fh: + data = json.load(fh) + return data if isinstance(data, dict) else {} + except (OSError, ValueError): + return {} + + +def save_settings(data: dict) -> None: + try: + os.makedirs(config_dir(), exist_ok=True) + tmp = _settings_file() + ".tmp" + with open(tmp, "w", encoding="utf-8") as fh: + json.dump(data, fh, indent=1) + os.replace(tmp, _settings_file()) + except OSError: + pass + + +def load_recent() -> List[str]: + return [p for p in load_settings().get("recent", []) if isinstance(p, str)] + + +def save_recent(path: str) -> None: + """Remember a repository; the first one in the list is reopened next launch.""" + norm = os.path.normcase(path) + data = load_settings() + data["recent"] = ([path] + [p for p in load_recent() if os.path.normcase(p) != norm])[:15] + save_settings(data) + + +def default_parent_folder() -> str: + """Where new / cloned repositories go by default (like GitHub Desktop's Documents\\GitHub).""" + saved = load_settings().get("parent") + if isinstance(saved, str) and os.path.isdir(saved): + return saved + home = os.path.expanduser("~") + for cand in (os.path.join(home, "Documents", "GitHub"), os.path.join(home, "Documents")): + if os.path.isdir(cand): + return cand + return home + + +def remember_parent_folder(folder: str) -> None: + data = load_settings() + data["parent"] = folder + save_settings(data) + + +def repo_name_from_url(url: str) -> str: + """'git@host:owner/My-Repo.git' -> 'My-Repo'.""" + tail = re.split(r"[/:\\]", url.strip().rstrip("/\\"))[-1] if url.strip() else "" + return re.sub(r"\.git$", "", tail) + + +BAD_NAME_RE = re.compile(r'[\\/:*?"<>|]') + + +GITIGNORE_OS = """ +# OS / editor clutter +.DS_Store +Thumbs.db +desktop.ini +""" + +GITIGNORE_TEMPLATES = { + "Unity": """# Unity generated folders +/[Ll]ibrary/ +/[Tt]emp/ +/[Oo]bj/ +/[Bb]uild/ +/[Bb]uilds/ +/[Ll]ogs/ +/[Uu]ser[Ss]ettings/ +/[Mm]emoryCaptures/ +/[Rr]ecordings/ + +# Asset meta data should only be ignored when the corresponding asset is also ignored +!/[Aa]ssets/**/*.meta + +# Uncomment this line if you wish to ignore the asset store tools plugin +# /[Aa]ssets/AssetStoreTools* + +# Autogenerated Jetbrains Rider plugin +/[Aa]ssets/Plugins/Editor/JetBrains* + +# Visual Studio / Rider cache directories +.vs/ +.idea/ +.gradle/ + +# Autogenerated VS/MD/Consulo solution and project files +ExportedObj/ +.consulo/ +*.csproj +*.unityproj +*.sln +*.suo +*.tmp +*.user +*.userprefs +*.pidb +*.booproj +*.svd +*.pdb +*.mdb +*.opendb +*.VC.db + +# Unity3D generated meta files +*.pidb.meta +*.pdb.meta +*.mdb.meta + +# Unity3D generated file on crash reports +sysinfo.txt + +# Builds +*.apk +*.aab +*.unitypackage +*.unitypackage.meta +*.app + +# Crashlytics generated file +crashlytics-build.properties + +# Packed Addressables +/[Aa]ssets/[Aa]ddressable[Aa]ssets[Dd]ata/*/*.bin* + +# Temporary auto-generated Android Assets +/[Aa]ssets/[Ss]treamingAssets/aa.meta +/[Aa]ssets/[Ss]treamingAssets/aa/* +""", + "Python": """# Python +__pycache__/ +*.py[cod] +*.egg-info/ +.eggs/ +build/ +dist/ +.venv/ +venv/ +env/ +.env +.pytest_cache/ +.mypy_cache/ +.ruff_cache/ +.coverage +htmlcov/ +""", + "Node": """# Node +node_modules/ +npm-debug.log* +yarn-debug.log* +yarn-error.log* +pnpm-debug.log* +dist/ +build/ +coverage/ +.env +.env.local +.cache/ +""", + "Visual Studio / .NET": """# Visual Studio / .NET +.vs/ +[Bb]in/ +[Oo]bj/ +[Dd]ebug/ +[Rr]elease/ +x64/ +x86/ +*.user +*.suo +*.userosscache +*.sln.docstates +packages/ +TestResults/ +*.log +""", +} + +MIT_LICENSE = """MIT License + +Copyright (c) {year} {name} + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. +""" + +PROGRESS_RE = re.compile(r"^(?:remote: )?(Enumerating|Counting|Compressing|Receiving|Resolving|" + r"Updating files|Filtering content|Checking out|Total|Downloading)", re.I) + + +def run_git_streaming(cwd: str, args: List[str], on_progress: Callable[[str], None]) -> Result: + """Run a long git command (clone), passing each progress line to on_progress.""" + kw = {"creationflags": subprocess.CREATE_NO_WINDOW} if IS_WIN else {"start_new_session": True} + try: + p = subprocess.Popen(["git", *args], cwd=cwd, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, + stderr=subprocess.PIPE, env=GIT_ENV, **kw) + except OSError as e: + return Result(1, "", str(e)) + out: List[bytes] = [] + reader = threading.Thread(target=lambda: out.append(p.stdout.read()), daemon=True) + reader.start() + messages: List[str] = [] + buf = b"" + while True: + chunk = p.stderr.read1(4096) + if not chunk: + break + buf += chunk + *lines, buf = re.split(rb"[\r\n]", buf) + for raw in lines: + line = raw.decode("utf-8", "replace").strip() + if not line: + continue + if PROGRESS_RE.match(line): + on_progress(line) + elif line not in messages: + messages.append(line) + if buf.strip(): + messages.append(buf.decode("utf-8", "replace").strip()) + p.wait() + reader.join(timeout=5) + return Result(p.returncode, (out[0] if out else b"").decode("utf-8", "replace"), "\n".join(messages)) + + +def ago(ts: Optional[float]) -> str: + if not ts: + return "never" + s = max(0, time.time() - ts) + if s < 60: + return "just now" + for unit, size in (("day", 86400), ("hour", 3600), ("minute", 60)): + if s >= size: + n = int(s // size) + return f"{n} {unit}{'s' if n != 1 else ''} ago" + return "just now" + + +def plural(n: int, word: str) -> str: + return f"{n} {word}{'' if n == 1 else 's'}" + + +MARKER_RE = re.compile(r"^(<{7}|>{7})( |$)", re.M) + + +def file_has_markers(path: str) -> bool: + try: + with open(path, encoding="utf-8", errors="replace") as fh: + return bool(MARKER_RE.search(fh.read())) + except OSError: + return False + + +def editor_name() -> str: + if shutil.which("code"): + return "VS Code" + ed = os.environ.get("VISUAL") or os.environ.get("EDITOR") + return os.path.basename(ed.split()[0]) if ed else "your editor" + + +_SSH_HOSTS: dict = {} + + +def resolve_ssh_host(host: str) -> str: + """Turn a Host alias from ~/.ssh/config into the real hostname (cached).""" + if host not in _SSH_HOSTS: + real = "" + if shutil.which("ssh"): + try: + kw = {"creationflags": subprocess.CREATE_NO_WINDOW} if IS_WIN else {"start_new_session": True} + p = subprocess.run(["ssh", "-G", host], capture_output=True, text=True, timeout=3, + stdin=subprocess.DEVNULL, **kw) + for line in p.stdout.splitlines(): + if line.lower().startswith("hostname "): + real = line.split(None, 1)[1].strip() + break + except (OSError, subprocess.SubprocessError): + pass + _SSH_HOSTS[host] = real or host + return _SSH_HOSTS[host] + + +def web_url_for(remote_url: str) -> Optional[str]: + """The repository's web page for a remote URL (GitHub, Gitea, GitLab...), or None.""" + u = remote_url.strip() + m = re.match(r"^(https?)://(?:[^@/]+@)?([^/]+)/(.+?)(?:\.git)?/*$", u) + if m: # https://[user:token@]host[:port]/owner/repo.git + return f"{m.group(1)}://{m.group(2)}/{m.group(3)}" + m = re.match(r"^(?:ssh|git\+ssh|ssh\+git|git)://(?:[^@/]+@)?([^/:]+)(?::\d+)?/(.+?)(?:\.git)?/*$", u) + if not m and "://" not in u: # git@host:owner/repo.git + m = re.match(r"^(?:[^@/:]+@)?([^/:]{2,}):/*(.+?)(?:\.git)?/*$", u) + if not m: + return None # a local path - no web page + return f"https://{resolve_ssh_host(m.group(1))}/{m.group(2).lstrip('/')}" + + +def web_site_name(web_url: str) -> str: + host = re.sub(r"^https?://", "", web_url).split("/")[0].split(":")[0].lower() + if host == "github.com" or host.endswith(".github.com"): + return "GitHub" + if "gitlab" in host: + return "GitLab" + if "bitbucket" in host: + return "Bitbucket" + return host + + +def commit_web_url(web_url: str, sha: str) -> str: + host = web_url.lower() + if "gitlab" in host: + return f"{web_url}/-/commit/{sha}" + if "bitbucket" in host: + return f"{web_url}/commits/{sha}" + return f"{web_url}/commit/{sha}" # GitHub, Gitea, Forgejo, Gogs + + +def _helper_command(title: str, start: str) -> List[str]: + if getattr(sys, "frozen", False): + return [sys.executable, "--pick-folder", title, start] + return [sys.executable, os.path.abspath(__file__), "--pick-folder", title, start] + + +def _powershell_pick(title: str, start: str) -> Tuple[str, Optional[str]]: + def q(text: str) -> str: + return "'" + text.replace("'", "''") + "'" + ps = ("Add-Type -AssemblyName System.Windows.Forms;" + "$f = New-Object System.Windows.Forms.FolderBrowserDialog;" + f"$f.Description = {q(title)}; $f.SelectedPath = {q(start)}; $f.ShowNewFolderButton = $true;" + "$o = New-Object System.Windows.Forms.Form -Property @{TopMost = $true};" + "if ($f.ShowDialog($o) -eq [System.Windows.Forms.DialogResult]::OK) {" + " try { [Console]::OutputEncoding = [Text.Encoding]::UTF8 } catch {}; $f.SelectedPath }") + p = subprocess.run(["powershell", "-NoProfile", "-STA", "-Command", ps], capture_output=True, + stdin=subprocess.DEVNULL, creationflags=subprocess.CREATE_NO_WINDOW) + out = p.stdout.decode("utf-8", "replace").strip() + if p.returncode == 0: + return ("ok", out) if out else ("cancel", None) + return "unavailable", None + + +def pick_folder_native(title: str, start: str) -> Tuple[str, Optional[str]]: + """Show the operating system's folder picker (blocks until it closes). + + Returns ("ok", path), ("cancel", None), or ("unavailable", None) when there's no + desktop to show one on - then gitdesk shows its own in-terminal folder browser. + """ + try: + if sys.platform == "darwin": + def aq(text: str) -> str: + return '"' + text.replace("\\", "\\\\").replace('"', '\\"') + '"' + script = f"POSIX path of (choose folder with prompt {aq(title)} default location (POSIX file {aq(start)}))" + p = subprocess.run(["osascript", "-e", script], capture_output=True, text=True) + if p.returncode == 0 and p.stdout.strip(): + return "ok", p.stdout.strip().rstrip("/") or "/" + return ("cancel", None) if "-128" in p.stderr else ("unavailable", None) + if not IS_WIN: + if not (os.environ.get("DISPLAY") or os.environ.get("WAYLAND_DISPLAY")): + return "unavailable", None + for cmd in (["zenity", "--file-selection", "--directory", f"--title={title}", + f"--filename={start.rstrip('/')}/"], + ["kdialog", "--getexistingdirectory", start, "--title", title]): + if shutil.which(cmd[0]): + p = subprocess.run(cmd, capture_output=True, text=True, stdin=subprocess.DEVNULL, + start_new_session=True) + if p.returncode == 0 and p.stdout.strip(): + return "ok", p.stdout.strip() + if p.returncode == 1: + return "cancel", None + # Windows (and Linux without zenity/kdialog): a helper copy of gitdesk shows the + # native picker through tkinter - on Windows that's the Explorer "Select Folder" window. + kw = {"creationflags": subprocess.CREATE_NO_WINDOW} if IS_WIN else {"start_new_session": True} + p = subprocess.run(_helper_command(title, start), capture_output=True, stdin=subprocess.DEVNULL, **kw) + if p.returncode == 0 and p.stdout.strip(): + return "ok", p.stdout.decode("utf-8", "replace").strip() + if p.returncode == 1: + return "cancel", None + if IS_WIN: + return _powershell_pick(title, start) + except (OSError, subprocess.SubprocessError): + pass + return "unavailable", None + + +def existing_folder(path: str) -> str: + """The nearest folder that exists, walking up from path (falls back to home).""" + path = os.path.abspath(os.path.expanduser(path)) if path.strip() else "" + while path and not os.path.isdir(path): + parent = os.path.dirname(path) + if parent == path: + break + path = parent + return path if path and os.path.isdir(path) else os.path.expanduser("~") + + +def file_manager_name() -> str: + return "Explorer" if IS_WIN else ("Finder" if sys.platform == "darwin" else "file manager") + + +def set_label(button, label: str) -> None: + """Change a button's label and let it resize to fit.""" + if str(button.label) != label: + button.label = label + button.refresh(layout=True) + + +def path_text(fc: FileChange) -> Text: + t = Text(no_wrap=True, overflow="ellipsis") + d, _, name = fc.path.rpartition("/") + if fc.orig: + t.append(fc.orig + " → ", style=C_DIM) + if d: + t.append(d + "/", style=C_MUTED) + t.append(name) + return t + + +def render_diff(text: str, multi_file: bool = False): + """Turn unified diff text into a table with old/new line numbers.""" + t = Table.grid(expand=True, padding=(0, 1)) + t.add_column(justify="right", width=5, style=C_DIM, no_wrap=True) + t.add_column(justify="right", width=5, style=C_DIM, no_wrap=True) + t.add_column(width=1, no_wrap=True) + t.add_column(ratio=1, overflow="fold") + old = new = 0 + in_hunk = False + rows = 0 + for raw in text.split("\n"): + line = raw.rstrip("\r") + if rows >= MAX_DIFF_ROWS: + t.add_row("", "", "", Text(f"… diff truncated after {MAX_DIFF_ROWS} lines", style=f"italic {C_MUTED}")) + break + if line.startswith("diff --git "): + in_hunk = False + if multi_file: + name = line.split(" b/", 1)[-1] if " b/" in line else line[11:] + t.add_row("", "", "", Text(name, style=f"bold {C_TEXT}"), style="on #2f363d") + rows += 1 + continue + if line.startswith("@@"): + m = re.match(r"@@ -(\d+)(?:,\d+)? \+(\d+)(?:,\d+)? @@", line) + if m: + old, new = int(m.group(1)), int(m.group(2)) + in_hunk = True + t.add_row("", "", "", Text(line, style=C_BLUE), style="on #1c2938") + rows += 1 + continue + if not in_hunk: + if line.startswith("Binary files"): + t.add_row("", "", "", Text("Binary file - there's no text diff to show.", style=f"italic {C_MUTED}")) + rows += 1 + elif multi_file and line.startswith(("rename from", "rename to", "new file mode", "deleted file mode")): + t.add_row("", "", "", Text(line, style=C_DIM)) + rows += 1 + continue + if not line: + continue + body = Text(line[1:].expandtabs(4)) + if line[0] == "+": + if MARKER_RE.match(line[1:]) or line[1:].startswith("======="): + body.stylize(f"bold {C_ORANGE}") + t.add_row("", str(new), Text("+", style=C_GREEN), body, style="on #12261e") + new += 1 + elif line[0] == "-": + t.add_row(str(old), "", Text("-", style=C_RED), body, style="on #3a1d22") + old += 1 + elif line[0] == "\\": + t.add_row("", "", "", Text(line[2:], style=f"italic {C_MUTED}")) + else: + t.add_row(str(old), str(new), " ", body) + old += 1 + new += 1 + rows += 1 + if rows == 0: + return Text("No differences to show.", style=f"italic {C_MUTED}") + return t + + +def create_repository(o: dict) -> Tuple[bool, str]: + """Create a repo from the New repository dialog's choices. + + Returns (created, problem). created=False means nothing usable was made. + """ + path, name, desc = o["path"], o["name"], o.get("description", "") + try: + os.makedirs(path, exist_ok=True) + except OSError as e: + return False, f"Couldn't create the folder: {e}" + branch = run_git(path, ["config", "--get", "init.defaultBranch"]).out.strip() or "main" + r = run_git(path, ["init"]) + if not r.ok: + return False, r.message + run_git(path, ["symbolic-ref", "HEAD", f"refs/heads/{branch}"]) + wrote = [] + try: + def write(fname: str, text: str) -> None: + with open(os.path.join(path, fname), "w", encoding="utf-8", newline="\n") as fh: + fh.write(text) + wrote.append(fname) + if o.get("readme") and not os.path.exists(os.path.join(path, "README.md")): + write("README.md", f"# {name}\n" + (f"\n{desc}\n" if desc else "")) + if o.get("gitignore") in GITIGNORE_TEMPLATES: + ignore_path = os.path.join(path, ".gitignore") + template = GITIGNORE_TEMPLATES[o["gitignore"]] + GITIGNORE_OS + if os.path.exists(ignore_path): # existing folder: add to what's there + with open(ignore_path, "a", encoding="utf-8", newline="\n") as fh: + fh.write("\n" + template) + wrote.append(".gitignore") + else: + write(".gitignore", template) + if o.get("license") == "MIT" and not os.path.exists(os.path.join(path, "LICENSE")): + author = run_git(path, ["config", "--get", "user.name"]).out.strip() or "" + write("LICENSE", MIT_LICENSE.format(year=time.localtime().tm_year, name=author)) + except OSError as e: + return True, f"Couldn't write {e.filename}: {e.strerror}" + if o.get("remote"): + r = run_git(path, ["remote", "add", "origin", o["remote"]]) + if not r.ok: + return True, "Couldn't add the remote: " + r.message + if wrote: + r = run_git(path, ["add", "-A"]) + if r.ok: + r = run_git(path, ["commit", "-q", "-m", "Initial commit"]) + if not r.ok: + return True, r.message + return True, "" + + +# -------------------------------------------------------------------------- +# Small widgets +# -------------------------------------------------------------------------- +class ToolbarButton(Static, can_focus=True): + """Two-line clickable block in the top bar (like GitHub Desktop's).""" + + class Pressed(Message): + def __init__(self, button: "ToolbarButton") -> None: + super().__init__() + self.button = button + + def __init__(self, icon: str, top: str, bottom: str, *, id: str, + emphasis: str = "bottom", dropdown: bool = False) -> None: + super().__init__(id=id, classes="tb") + self.icon, self.top, self.bottom = icon, top, bottom + self.badge = "" + self.emphasis = emphasis + self.dropdown = dropdown + + def set(self, *, icon=None, top=None, bottom=None, badge=None) -> None: + if icon is not None: + self.icon = icon + if top is not None: + self.top = top + if bottom is not None: + self.bottom = bottom + if badge is not None: + self.badge = badge + self.refresh() + + def render(self) -> Text: + strong, weak = f"bold {C_TEXT}", C_MUTED + top_style, bottom_style = (weak, strong) if self.emphasis == "bottom" else (strong, weak) + t = Text(no_wrap=True, overflow="ellipsis") + t.append(f"{self.icon} ", style=f"bold {C_TEXT}") + t.append(self.top, style=top_style) + if self.badge: + t.append(f" {self.badge}", style=f"bold {C_BLUE}") + if self.dropdown: + t.append(" ▾", style=C_MUTED) + t.append("\n ") + t.append(self.bottom, style=bottom_style) + return t + + def on_click(self, event: events.Click) -> None: + event.stop() + self.post_message(self.Pressed(self)) + + def key_enter(self) -> None: + self.post_message(self.Pressed(self)) + + def key_space(self) -> None: + self.post_message(self.Pressed(self)) + + +class Tick(Static): + """A small clickable checkbox. Clicking it doesn't select the row.""" + + class Toggled(Message): + def __init__(self, tick: "Tick") -> None: + super().__init__() + self.tick = tick + + checked = reactive(True) + + def __init__(self, checked: bool = True, key: Optional[str] = None, **kw) -> None: + super().__init__(**kw) + self.set_reactive(Tick.checked, checked) + self.key = key + + def render(self) -> Text: + return Text("[x]" if self.checked else "[ ]", + style=f"bold {C_BLUE}" if self.checked else C_MUTED) + + def on_click(self, event: events.Click) -> None: + event.stop() + self.checked = not self.checked + self.post_message(self.Toggled(self)) + + +class FileRow(ListItem): + def __init__(self, fc: FileChange, checked: bool) -> None: + super().__init__(classes="filerow") + self.fc = fc + self._checked = checked + + def compose(self) -> ComposeResult: + icon, color, _ = KIND_ICON.get(self.fc.kind, ("●", C_YELLOW, "")) + with Horizontal(): + yield Tick(self._checked, key=self.fc.path, classes="tick") + yield Label(path_text(self.fc), classes="fpath") + yield Label(Text(icon, style=f"bold {color}"), classes="fkind") + + +class CommitRow(ListItem): + def __init__(self, c: Commit, unpushed: bool) -> None: + super().__init__(classes="commitrow") + self.commit = c + self.unpushed = unpushed + + def compose(self) -> ComposeResult: + t = Text(no_wrap=True, overflow="ellipsis") + if self.unpushed: + t.append("↑ ", style=f"bold {C_BLUE}") + t.append(self.commit.subject) + yield Label(t) + yield Label(Text(f"{self.commit.author} • {self.commit.when}", style=C_MUTED, + no_wrap=True, overflow="ellipsis")) + + +class Card(Horizontal): + def __init__(self, title: str, desc: str, button: str, action: str) -> None: + super().__init__(classes="card") + self._title, self._desc, self._button, self._action = title, desc, button, action + + def compose(self) -> ComposeResult: + with Vertical(classes="card-text"): + yield Label(Text(self._title, style=f"bold {C_TEXT}")) + yield Label(Text(self._desc, style=C_MUTED)) + yield Button(self._button, name=self._action, classes="card-btn", compact=True) + + +# -------------------------------------------------------------------------- +# Dialogs +# -------------------------------------------------------------------------- +class ChoiceDialog(ModalScreen): + """Message + a row of buttons. Returns the chosen button's value.""" + BINDINGS = [Binding("escape", "cancel", "Close")] + + def __init__(self, title: str, body, buttons: List[Tuple[str, str, str]], + cancel: Optional[str] = None) -> None: + super().__init__() + self._title = title + self._body = body if isinstance(body, Text) else Text(str(body)) + self._buttons = buttons + self._cancel = cancel if cancel is not None else buttons[-1][0] + + def compose(self) -> ComposeResult: + with Vertical(classes="dialog"): + yield Label(Text(self._title, style="bold"), classes="dialog-title") + with VerticalScroll(classes="dialog-body"): + yield Static(self._body) + with Horizontal(classes="dialog-buttons"): + for value, label, variant in self._buttons: + yield Button(label, variant=variant, name=value, compact=True) + + def on_mount(self) -> None: + self.query(Button).first().focus() + + def on_button_pressed(self, event: Button.Pressed) -> None: + event.stop() + self.dismiss(event.button.name) + + def action_cancel(self) -> None: + self.dismiss(self._cancel) + + +class InputDialog(ModalScreen): + BINDINGS = [Binding("escape", "cancel", "Cancel")] + + def __init__(self, title: str, prompt: str, ok: str, placeholder: str = "", value: str = "") -> None: + super().__init__() + self._title, self._prompt, self._ok = title, prompt, ok + self._placeholder, self._value = placeholder, value + + def compose(self) -> ComposeResult: + with Vertical(classes="dialog"): + yield Label(Text(self._title, style="bold"), classes="dialog-title") + yield Static(Text(self._prompt)) + yield Input(value=self._value, placeholder=self._placeholder, id="dlg-input") + with Horizontal(classes="dialog-buttons"): + yield Button(self._ok, variant="primary", name="ok", compact=True) + yield Button("Cancel", name="cancel", compact=True) + + def on_mount(self) -> None: + self.query_one(Input).focus() + + def on_input_submitted(self, event: Input.Submitted) -> None: + event.stop() + self.dismiss(event.value) + + def on_button_pressed(self, event: Button.Pressed) -> None: + event.stop() + self.dismiss(self.query_one(Input).value if event.button.name == "ok" else None) + + def action_cancel(self) -> None: + self.dismiss(None) + + +class FolderTree(DirectoryTree): + def filter_paths(self, paths): + return [p for p in paths if p.is_dir() and p.name != ".git"] + + +class FolderBrowser(ModalScreen): + """In-terminal folder picker, used when there's no desktop folder picker available.""" + BINDINGS = [Binding("escape", "cancel", "Cancel"), Binding("backspace", "up", "Up a folder")] + + def __init__(self, title: str, start: str) -> None: + super().__init__() + self._title = title + self.current = existing_folder(start) + + def compose(self) -> ComposeResult: + with Vertical(classes="dialog wide"): + yield Label(Text(self._title, style="bold"), classes="dialog-title") + yield Static(id="fb-current") + yield FolderTree(self.current, id="fb-tree", classes="folder-tree") + with Horizontal(classes="dialog-buttons"): + yield Button("Up a folder", name="up", compact=True) + yield Button("Select this folder", variant="primary", name="select", compact=True) + yield Button("Cancel", name="cancel", compact=True) + + def on_mount(self) -> None: + self._show() + self.query_one("#fb-tree").focus() + + def _show(self) -> None: + self.query_one("#fb-current", Static).update(Text.assemble(("Selected: ", C_MUTED), (self.current, "bold"))) + + @on(Tree.NodeHighlighted, "#fb-tree") + def _highlighted(self, event: Tree.NodeHighlighted) -> None: + data = event.node.data + if data is not None and getattr(data, "path", None) is not None: + self.current = str(data.path) + self._show() + + def action_up(self) -> None: + parent = os.path.dirname(self.current.rstrip("\\/")) or self.current + if os.path.isdir(parent) and parent != self.current: + self.current = parent + self.query_one("#fb-tree", FolderTree).path = parent + self._show() + + def on_button_pressed(self, event: Button.Pressed) -> None: + event.stop() + if event.button.name == "up": + self.action_up() + elif event.button.name == "select": + self.dismiss(self.current) + else: + self.dismiss(None) + + def action_cancel(self) -> None: + self.dismiss(None) + + +class ChooseFolderMixin: + """For dialogs with a path box + "Choose…" button.""" + + async def choose_into(self, input_id: str, title: str, button_id: str, + start: Optional[str] = None) -> Optional[str]: + inp = self.query_one(f"#{input_id}", Input) + btn = self.query_one(f"#{button_id}", Button) + btn.disabled = True + try: + chosen = await self.app.choose_folder(title, start if start is not None else inp.value) + finally: + btn.disabled = False + if chosen: + inp.value = chosen + inp.cursor_position = len(chosen) + return chosen + + +class AddLocalDialog(ChooseFolderMixin, ModalScreen): + """Returns ("open", path), ("create_here", path) or None.""" + BINDINGS = [Binding("escape", "cancel", "Cancel")] + + def compose(self) -> ComposeResult: + with Vertical(classes="dialog"): + yield Label(Text("Add local repository", style="bold"), classes="dialog-title") + yield Label("Local path", classes="field-label") + with Horizontal(classes="path-row"): + yield Input(placeholder="repository path", id="al-path", compact=True) + yield Button("Choose…", name="choose", id="al-choose", compact=True) + yield Static(id="al-error", classes="error") + with Horizontal(classes="dialog-buttons"): + yield Button("Create a repository here", name="create_here", id="al-create", + variant="success", compact=True, classes="hidden") + yield Button("Add repository", variant="primary", name="add", compact=True) + yield Button("Cancel", name="cancel", compact=True) + + def on_mount(self) -> None: + self.query_one("#al-path", Input).focus() + + _checked_value: Optional[str] = None + + def on_input_changed(self, event: Input.Changed) -> None: + # Only clear the message when the path really changed since it was checked + # (filling the box from Choose… also sends a Changed event, a moment later). + if event.value != self._checked_value: + self.query_one("#al-error", Static).update("") + self.query_one("#al-create").add_class("hidden") + + def on_input_submitted(self, event: Input.Submitted) -> None: + event.stop() + self._submit() + + def _path(self) -> str: + raw = self.query_one("#al-path", Input).value.strip() + return os.path.abspath(os.path.expanduser(raw)) if raw else "" + + def _submit(self) -> None: + self._checked_value = self.query_one("#al-path", Input).value + path = self._path() + err = self.query_one("#al-error", Static) + if not path: + err.update(Text("Type a folder path or click Choose….", style=C_RED)) + elif not os.path.isdir(path): + err.update(Text("That folder doesn't exist.", style=C_RED)) + elif run_git(path, ["rev-parse", "--show-toplevel"]).ok: + self.dismiss(("open", path)) + else: + err.update(Text("This folder doesn't appear to be a git repository. Would you like to " + "create a repository here instead?", style=C_ORANGE)) + self.query_one("#al-create").remove_class("hidden") + + async def on_button_pressed(self, event: Button.Pressed) -> None: + event.stop() + name = event.button.name + if name == "choose": + self.run_worker(self._choose(), exit_on_error=False) + elif name == "add": + self._submit() + elif name == "create_here": + self.dismiss(("create_here", self._path())) + else: + self.dismiss(None) + + async def _choose(self) -> None: + if await self.choose_into("al-path", "Choose a repository folder", "al-choose"): + self._submit() + + def action_cancel(self) -> None: + self.dismiss(None) + + +class SimpleRow(ListItem): + def __init__(self, value, *lines: Text) -> None: + super().__init__() + self.value = value + self._lines = lines + + def compose(self) -> ComposeResult: + for line in self._lines: + yield Label(line) + + +class StashDialog(ModalScreen): + BINDINGS = [Binding("escape", "cancel", "Close")] + + def __init__(self, git: Git) -> None: + super().__init__() + self.git = git + self.stashes = git.stash_list() + + def compose(self) -> ComposeResult: + with Vertical(classes="dialog wide"): + yield Label(Text("Stashed changes", style="bold"), classes="dialog-title") + if not self.stashes: + yield Static(Text("You don't have any stashed changes.", style=C_MUTED)) + else: + rows = [SimpleRow(ref, Text(msg, no_wrap=True, overflow="ellipsis"), + Text(f"{ref} • {when}", style=C_MUTED)) + for ref, msg, when in self.stashes] + yield ListView(*rows, id="stash-lv", classes="dialog-list") + yield Static(id="stash-preview", classes="preview") + with Horizontal(classes="dialog-buttons"): + if self.stashes: + yield Button("Pop (restore + remove)", variant="primary", name="pop", compact=True) + yield Button("Apply (keep stash)", name="apply", compact=True) + yield Button("Drop", variant="error", name="drop", compact=True) + yield Button("Close", name="close", compact=True) + + def on_mount(self) -> None: + if self.stashes: + lv = self.query_one("#stash-lv", ListView) + lv.index = 0 + lv.focus() + + @on(ListView.Highlighted, "#stash-lv") + def _preview(self, event: ListView.Highlighted) -> None: + if isinstance(event.item, SimpleRow): + self.query_one("#stash-preview", Static).update( + Text(self.git.stash_preview(event.item.value), style=C_MUTED)) + + def _current(self) -> Optional[str]: + lv = self.query_one("#stash-lv", ListView) + item = lv.highlighted_child + return item.value if isinstance(item, SimpleRow) else None + + def on_button_pressed(self, event: Button.Pressed) -> None: + event.stop() + if event.button.name == "close": + self.dismiss(None) + return + ref = self._current() + self.dismiss((event.button.name, ref) if ref else None) + + def action_cancel(self) -> None: + self.dismiss(None) + + +class BranchDialog(ModalScreen): + BINDINGS = [Binding("escape", "cancel", "Close")] + + def __init__(self, git: Git, current: Optional[str], remote: Optional[str]) -> None: + super().__init__() + self.current = current + self.local = git.local_branches() + local_names = {n for n, _ in self.local} + self.remote_only = [] + if remote: + for name, when in git.remote_branches(remote): + short = name[len(remote) + 1:] + if short not in local_names: + self.remote_only.append((short, name, when)) + + def compose(self) -> ComposeResult: + with Vertical(classes="dialog wide"): + yield Label(Text("Branches", style="bold"), classes="dialog-title") + yield Input(placeholder="Filter, or type a new branch name", id="branch-filter") + yield ListView(id="branch-lv", classes="dialog-list") + with Horizontal(classes="dialog-buttons"): + yield Button("New branch", variant="success", name="create", id="branch-create", + compact=True, disabled=True) + yield Button("Close", name="close", compact=True) + + async def on_mount(self) -> None: + await self._fill("") + self.query_one("#branch-filter", Input).focus() + + async def _fill(self, text: str) -> None: + lv = self.query_one("#branch-lv", ListView) + await lv.clear() + needle = text.lower() + rows = [] + for name, when in self.local: + if needle and needle not in name.lower(): + continue + t = Text(no_wrap=True, overflow="ellipsis") + if name == self.current: + t.append("✓ " + name, style=f"bold {C_GREEN}") + else: + t.append(" " + name) + t.append(f" {when}", style=C_DIM) + rows.append(SimpleRow(("switch", name, None), t)) + for short, full, when in self.remote_only: + if needle and needle not in short.lower(): + continue + t = Text(no_wrap=True, overflow="ellipsis") + t.append(" " + short) + t.append(f" on {full.split('/')[0]} • {when}", style=C_DIM) + rows.append(SimpleRow(("switch", short, full), t)) + await lv.extend(rows) + btn = self.query_one("#branch-create", Button) + names = {n for n, _ in self.local} + btn.disabled = not text.strip() or text.strip() in names + set_label(btn, f'Create branch "{text.strip()}"' if text.strip() else "New branch") + + async def on_input_changed(self, event: Input.Changed) -> None: + await self._fill(event.value) + + def on_input_submitted(self, event: Input.Submitted) -> None: + event.stop() + lv = self.query_one("#branch-lv", ListView) + rows = list(lv.query(SimpleRow)) + if len(rows) == 1: + self._pick(rows[0].value) + elif not self.query_one("#branch-create", Button).disabled: + self.dismiss(("create", event.value.strip(), None)) + + def _pick(self, value) -> None: + if value[1] == self.current and value[2] is None: + self.dismiss(None) + else: + self.dismiss(value) + + @on(ListView.Selected, "#branch-lv") + def _selected(self, event: ListView.Selected) -> None: + if isinstance(event.item, SimpleRow): + self._pick(event.item.value) + + def on_button_pressed(self, event: Button.Pressed) -> None: + event.stop() + if event.button.name == "create": + self.dismiss(("create", self.query_one("#branch-filter", Input).value.strip(), None)) + else: + self.dismiss(None) + + def action_cancel(self) -> None: + self.dismiss(None) + + +class RepoDialog(ModalScreen): + """Repository picker: recent repos, plus Add local / New / Clone. + + Returns a path, "add_local", "new", "clone", or None. + """ + BINDINGS = [Binding("escape", "cancel", "Cancel")] + + def __init__(self, current: str, first: bool, error: str = "") -> None: + super().__init__() + self.current, self.first, self.error = current, first, error + self.recent = [p for p in load_recent() if os.path.isdir(p)] + + def compose(self) -> ComposeResult: + with Vertical(classes="dialog wide"): + yield Label(Text("Repositories", style="bold"), classes="dialog-title") + if self.error: + yield Static(Text(self.error, style=C_RED)) + if self.recent: + yield Static(Text("Recent", style=f"bold {C_MUTED}")) + rows = [SimpleRow(p, Text(os.path.basename(p) or p, style="bold"), + Text(p, style=C_MUTED, no_wrap=True, overflow="ellipsis")) + for p in self.recent] + yield ListView(*rows, id="repo-lv", classes="dialog-list") + else: + yield Static(Text("No recent repositories yet. Add one you already have, create a new " + "one, or clone one from GitHub, Gitea, etc.", style=C_MUTED)) + with Horizontal(classes="dialog-buttons"): + yield Button("Add local repository…", name="add_local", variant="success", compact=True) + yield Button("New repository…", name="new", variant="success", compact=True) + yield Button("Clone repository…", name="clone", variant="success", compact=True) + yield Button("Quit" if self.first else "Cancel", name="cancel", compact=True) + + def on_mount(self) -> None: + if self.recent: + self.query_one("#repo-lv", ListView).focus() + else: + self.query(Button).first().focus() + + @on(ListView.Selected, "#repo-lv") + def _selected(self, event: ListView.Selected) -> None: + if isinstance(event.item, SimpleRow): + self.dismiss(event.item.value) + + def on_button_pressed(self, event: Button.Pressed) -> None: + event.stop() + name = event.button.name + if name in ("new", "clone", "add_local"): + self.dismiss(name) + else: + self.dismiss(None) + + def action_cancel(self) -> None: + self.dismiss(None) + + +class NewRepoDialog(ChooseFolderMixin, ModalScreen): + """Create a new repository. Returns a dict of the choices, or None.""" + BINDINGS = [Binding("escape", "cancel", "Cancel")] + + def __init__(self, parent: str, name: str = "") -> None: + super().__init__() + self.parent_folder = parent + self.initial_name = name + + def compose(self) -> ComposeResult: + ignore_opts = [("None", "")] + [(k, k) for k in GITIGNORE_TEMPLATES] + with Vertical(classes="dialog wide"): + yield Label(Text("Create a new repository", style="bold"), classes="dialog-title") + yield Static(id="nr-error", classes="error pinned") + with VerticalScroll(classes="form"): + yield Label("Name", classes="field-label") + yield Input(value=self.initial_name, placeholder="my-project", id="nr-name", compact=True) + yield Label("Description (optional)", classes="field-label") + yield Input(id="nr-desc", compact=True) + yield Label("Local path - the new folder is created inside this one", classes="field-label") + with Horizontal(classes="path-row"): + yield Input(value=self.parent_folder, id="nr-parent", compact=True) + yield Button("Choose…", name="choose", id="nr-choose", compact=True) + yield Static(id="nr-where", classes="hint") + yield Checkbox("Initialize this repository with a README", value=True, id="nr-readme", + compact=True) + yield Label("Git ignore", classes="field-label") + yield Select(ignore_opts, value="", allow_blank=False, id="nr-ignore", compact=True) + yield Label("License", classes="field-label") + yield Select([("None", ""), ("MIT License", "MIT")], value="", allow_blank=False, id="nr-license", + compact=True) + yield Label("Remote URL (optional) - an empty repository you've already created on " + "GitHub, Gitea, etc.", classes="field-label") + yield Input(placeholder="e.g. git@github.com:you/my-project.git", id="nr-remote", compact=True) + with Horizontal(classes="dialog-buttons pinned"): + yield Button("Create repository", variant="primary", name="create", compact=True) + yield Button("Cancel", name="cancel", compact=True) + + def on_mount(self) -> None: + self.query_one("#nr-name", Input).focus() + self._update_where() + + def _target(self) -> str: + name = self.query_one("#nr-name", Input).value.strip() + parent = os.path.expanduser(self.query_one("#nr-parent", Input).value.strip()) + return os.path.join(parent, name) if name else parent + + def _update_where(self) -> None: + name = self.query_one("#nr-name", Input).value.strip() + target = self._target() + text = Text(f"Will be created at: {target}" if name else "", style=C_MUTED) + if name and os.path.isdir(os.path.join(target, ".git")): + text = Text(f"{target} is already a repository - you'll be offered to open it.", style=C_ORANGE) + elif name and os.path.isdir(target) and os.listdir(target): + text = Text(f"{target} already exists - it will become a repository, and the files already " + "in it will be included in the first commit.", style=C_ORANGE) + self.query_one("#nr-where", Static).update(text) + + def on_input_changed(self, event: Input.Changed) -> None: + if event.input.id in ("nr-name", "nr-parent"): + self._update_where() + + def on_input_submitted(self, event: Input.Submitted) -> None: + event.stop() + self._submit() + + def _error(self, msg: str) -> None: + self.query_one("#nr-error", Static).update(Text(msg, style=C_RED)) + + def _submit(self) -> None: + name = self.query_one("#nr-name", Input).value.strip() + parent = os.path.expanduser(self.query_one("#nr-parent", Input).value.strip()) + if not name: + return self._error("Give the repository a name.") + if BAD_NAME_RE.search(name) or name in (".", ".."): + return self._error('The name can\'t contain \\ / : * ? " < > |') + if not parent: + return self._error("Choose the folder to create it in.") + target = os.path.join(parent, name) + if os.path.exists(target) and not os.path.isdir(target): + return self._error(f"{target} is a file, not a folder.") + self.dismiss({ + "name": name, + "parent": parent, + "path": target, + "description": self.query_one("#nr-desc", Input).value.strip(), + "readme": self.query_one("#nr-readme", Checkbox).value, + "gitignore": self.query_one("#nr-ignore", Select).value or "", + "license": self.query_one("#nr-license", Select).value or "", + "remote": self.query_one("#nr-remote", Input).value.strip(), + }) + + def on_button_pressed(self, event: Button.Pressed) -> None: + event.stop() + if event.button.name == "choose": + self.run_worker(self.choose_into("nr-parent", "Choose where to create the repository", "nr-choose"), + exit_on_error=False) + elif event.button.name == "create": + self._submit() + else: + self.dismiss(None) + + def action_cancel(self) -> None: + self.dismiss(None) + + +class CloneDialog(ChooseFolderMixin, ModalScreen): + """Clone a repository. Returns (url, path) or None.""" + BINDINGS = [Binding("escape", "cancel", "Cancel")] + + def __init__(self, parent: str) -> None: + super().__init__() + self.parent_folder = parent + self._auto_path = parent + + def compose(self) -> ComposeResult: + with Vertical(classes="dialog wide"): + yield Label(Text("Clone a repository", style="bold"), classes="dialog-title") + yield Label("Repository URL - HTTPS or SSH", classes="field-label") + yield Input(placeholder="https://github.com/owner/repo.git or git@host:owner/repo.git", + id="cl-url", compact=True) + yield Label("Local path", classes="field-label") + with Horizontal(classes="path-row"): + yield Input(value=self.parent_folder, id="cl-path", compact=True) + yield Button("Choose…", name="choose", id="cl-choose", compact=True) + yield Static(id="cl-error", classes="error") + with Horizontal(classes="dialog-buttons"): + yield Button("Clone", variant="primary", name="clone", compact=True) + yield Button("Cancel", name="cancel", compact=True) + + def on_mount(self) -> None: + self.query_one("#cl-url", Input).focus() + + def on_input_changed(self, event: Input.Changed) -> None: + if event.input.id != "cl-url": + return + path_input = self.query_one("#cl-path", Input) + if path_input.value == self._auto_path: # user hasn't typed their own path + name = repo_name_from_url(event.value) + self._auto_path = os.path.join(self.parent_folder, name) if name else self.parent_folder + path_input.value = self._auto_path + + def on_input_submitted(self, event: Input.Submitted) -> None: + event.stop() + self._submit() + + def _submit(self) -> None: + url = self.query_one("#cl-url", Input).value.strip() + path = os.path.expanduser(self.query_one("#cl-path", Input).value.strip()) + err = self.query_one("#cl-error", Static) + if not url: + err.update(Text("Paste the repository's URL.", style=C_RED)) + elif not path: + err.update(Text("Choose where to put it.", style=C_RED)) + elif os.path.isdir(path) and os.listdir(path): + err.update(Text(f"{path} already exists and isn't empty - pick another folder.", style=C_RED)) + else: + self.dismiss((url, path)) + + def on_button_pressed(self, event: Button.Pressed) -> None: + event.stop() + if event.button.name == "choose": + self.run_worker(self._choose(), exit_on_error=False) + elif event.button.name == "clone": + self._submit() + else: + self.dismiss(None) + + async def _choose(self) -> None: + chosen = await self.choose_into("cl-path", "Choose where to put the cloned repository", "cl-choose", + start=self.parent_folder) + if chosen: + self.parent_folder = chosen + name = repo_name_from_url(self.query_one("#cl-url", Input).value) + self._auto_path = os.path.join(chosen, name) if name else chosen + self.query_one("#cl-path", Input).value = self._auto_path + + def action_cancel(self) -> None: + self.dismiss(None) + + +ACTIONS = [ + ("fetch", "Fetch"), + ("pull", "Pull"), + ("push", "Push"), + ("sync", "Sync (pull, then push)"), + ("stash", "Stash all changes…"), + ("stashes", "Stashed changes…"), + ("discard_ticked", "Discard ticked changes…"), + ("branches", "Switch / create branch…"), + ("new_repo", "New repository…"), + ("repositories", "Repositories…"), + ("open_repo", "Add local repository…"), + ("clone_repo", "Clone repository…"), + ("add_remote", "Add / change remote URL…"), + ("open_editor", "Open in editor"), + ("open_folder", "Show in file manager"), + ("view_remote", "View on GitHub / remote site"), + ("toggle_log", "Show / hide git output"), + ("refresh", "Refresh"), +] + + +class ActionsDialog(ModalScreen): + BINDINGS = [Binding("escape", "cancel", "Close")] + + def compose(self) -> ComposeResult: + with Vertical(classes="dialog narrow"): + yield Label(Text("Actions", style="bold"), classes="dialog-title") + for action, label in ACTIONS: + yield Button(label, name=action, classes="action-btn", compact=True) + with Horizontal(classes="dialog-buttons"): + yield Button("Close", name="", compact=True) + + def on_mount(self) -> None: + self.query(Button).first().focus() + + def on_button_pressed(self, event: Button.Pressed) -> None: + event.stop() + self.dismiss(event.button.name or None) + + def action_cancel(self) -> None: + self.dismiss(None) + + +# -------------------------------------------------------------------------- +# The app +# -------------------------------------------------------------------------- +GITDESK_THEME = Theme( + name="gitdesk", + primary="#2188ff", + secondary="#79b8ff", + accent="#2188ff", + warning="#f66a0a", + error="#d73a49", + success="#28a745", + foreground=C_TEXT, + background="#24292e", + surface="#2f363d", + panel="#1f2428", + dark=True, +) + +CSS = """ +Screen { background: #24292e; } +.hidden { display: none; } + +#toolbar { height: 3; background: #1f2428; } +.tb { height: 3; width: 32; padding: 0 1; content-align: left middle; + border-right: vkey #444d56; background: #1f2428; } +.tb:hover { background: #2f363d; } +.tb:focus { background: #2f363d; text-style: none; } +.tb:disabled { color: #6a737d; } +#tb-remote { width: 34; } +#tb-more { width: 14; } + +#main { height: 1fr; } +#left { width: 42; min-width: 30; border-right: vkey #444d56; } +#right { width: 1fr; } + +#tabs { height: 1fr; } +#tabs > ContentSwitcher { height: 1fr; } +TabPane { padding: 0; height: 1fr; } +#tab-changes > Vertical, #tab-history > Vertical { height: 1fr; } + +#banner { height: auto; background: #3b2a12; padding: 0 1; border-bottom: solid #f66a0a; } +#banner-buttons { height: auto; margin-top: 1; } +#changes-header { height: 1; padding: 0 1; background: #2b3036; } +#changes-header .tick { width: 4; } +#changes-count { width: 1fr; } + +#files { height: 1fr; background: #24292e; } +.filerow { padding: 0 1; height: 1; } +.filerow > Horizontal { height: 1; } +.filerow .tick { width: 4; } +.filerow .fpath { width: 1fr; } +.filerow .fkind { width: 2; } + +#stash-bar { height: 1; padding: 0 1; margin-top: 1; } +#stash-bar Button { margin-right: 1; } + +#commit-box { height: auto; padding: 1 1 0 1; border-top: solid #444d56; } +#summary { margin-bottom: 1; } +#description { height: 5; } +#commit { width: 100%; margin: 1 0; } + +#history { height: 1fr; } +.commitrow { height: auto; padding: 0 1; border-bottom: solid #30363d; } + +#placeholder { padding: 1 4; } +#ph-title { text-style: bold; margin-bottom: 1; } +#ph-cards { height: auto; margin-top: 1; } +.card { height: auto; border: round #444d56; padding: 0 1; margin-bottom: 1; } +.card-text { width: 1fr; height: auto; margin-right: 2; } +.card-text > Label { width: 100%; } +.card-btn { margin-top: 1; } + +#diffview { height: 1fr; } +#diff-header { height: 1; background: #1f2428; padding: 0 1; } +#diff-title { width: 1fr; } +#diff-header Button { margin-left: 1; } +#diff-scroll { height: 1fr; } + +#gitlog { height: 12; border-top: solid #444d56; background: #1f2428; } + +ModalScreen { align: center middle; background: rgba(0, 0, 0, 0.55); } +.dialog { width: 76; max-width: 95%; height: auto; max-height: 90%; + background: #2f363d; border: round #586069; padding: 1 2; } +.dialog.wide { width: 90; } +.dialog.narrow { width: 44; } +.dialog-title { margin-bottom: 1; } +.dialog-body { height: auto; max-height: 22; } +.dialog-buttons { height: auto; align-horizontal: right; margin-top: 1; } +.dialog-buttons Button { margin-left: 2; } +.dialog-list { height: auto; max-height: 16; margin-top: 1; background: #24292e; } +.dialog-list > ListItem { padding: 0 1; } +.preview { height: auto; max-height: 8; margin-top: 1; } +.action-btn { width: 100%; margin-bottom: 0; } +#dlg-input { margin-top: 1; } +.form { height: auto; } +.pinned { dock: bottom; } +.dialog-buttons.pinned { margin-top: 0; padding-top: 1; } +.error.pinned { margin-top: 0; } +.form Input, .dialog Input.-textual-compact { background: #24292e; } +.form Select, .form Checkbox { width: 100%; } +.field-label { margin-top: 1; color: #959da5; } +.hint { height: auto; color: #959da5; } +.error { height: auto; margin-top: 1; } +#nr-readme { margin-top: 1; } +.path-row { height: auto; } +.path-row Input { width: 1fr; } +.path-row Button { margin-left: 1; min-width: 12; } +.folder-tree { height: 18; background: #24292e; margin-top: 1; } +""" + + +class GitDesk(App): + CSS = CSS + TITLE = "gitdesk" + BINDINGS = [ + Binding("f", "fetch", "Fetch"), + Binding("p", "pull", "Pull"), + Binding("u", "push", "Push"), + Binding("b", "branches", "Branch"), + Binding("s", "stash", "Stash"), + Binding("t", "stashes", "Stashes"), + Binding("d", "discard", "Discard"), + Binding("o", "open_repo", "Add local"), + Binding("n", "new_repo", "New"), + Binding("c", "clone_repo", "Clone"), + Binding("g", "view_remote", "Web"), + Binding("l", "toggle_log", "Log"), + Binding("r", "refresh", "Refresh"), + Binding("space", "toggle_file", "Include file", show=False), + Binding("q", "quit", "Quit"), + ] + + def __init__(self, start_path: Optional[str] = None) -> None: + super().__init__() + self.start_path = start_path + self.busy_detail = "" + self.git: Optional[Git] = None + self.state: Optional[RepoState] = None + self.excluded: Set[str] = set() + self.selected: Optional[str] = None + self.mode = "changes" # or "history" + self.busy = False + self.busy_label = "" + self.last_fetch: Optional[float] = None + self.log_q: "queue.Queue" = queue.Queue() + self._files_key = None + self._ph_key = None + self._diff_key = None + self._history_key = None + self._merge_prefilled = False + self._lock: Optional[asyncio.Lock] = None + self._net_lock: Optional[asyncio.Lock] = None + self.fetching = False + + # ---- layout ------------------------------------------------------------ + def compose(self) -> ComposeResult: + with Horizontal(id="toolbar"): + yield ToolbarButton("▤", "Current repository", "—", id="tb-repo", dropdown=True) + yield ToolbarButton("⎇", "Current branch", "—", id="tb-branch", dropdown=True) + yield ToolbarButton("↻", "Fetch origin", "", id="tb-remote", emphasis="top") + yield ToolbarButton("☰", "More", "Actions", id="tb-more") + with Horizontal(id="main"): + with Vertical(id="left"): + with TabbedContent(id="tabs", initial="tab-changes"): + with TabPane("Changes", id="tab-changes"): + with Vertical(): + with Vertical(id="banner", classes="hidden"): + yield Static(id="banner-text") + with Horizontal(id="banner-buttons"): + yield Button("Abort merge", id="abort-merge", variant="error", compact=True) + with Horizontal(id="changes-header"): + yield Tick(True, id="tick-all", classes="tick") + yield Label("0 changed files", id="changes-count") + yield ListView(id="files") + with Horizontal(id="stash-bar"): + yield Button("Stash all", id="stash-all", compact=True) + yield Button("Stashed (0)", id="stash-list", compact=True) + with Vertical(id="commit-box"): + yield Input(placeholder="Summary (required)", id="summary") + yield TextArea(id="description", placeholder="Description", + show_line_numbers=False, soft_wrap=True) + yield Button("Commit", id="commit", variant="primary", compact=True) + with TabPane("History", id="tab-history"): + with Vertical(): + yield ListView(id="history") + with ContentSwitcher(id="right", initial="placeholder"): + with VerticalScroll(id="placeholder"): + yield Static(id="ph-title") + yield Static(id="ph-sub") + yield Vertical(id="ph-cards") + with Vertical(id="diffview"): + with Horizontal(id="diff-header"): + yield Label("", id="diff-title") + yield Button("Open in editor", id="open-file", compact=True) + yield Button("Discard", id="discard-file", variant="error", compact=True) + yield Button("View on web", id="view-commit", compact=True) + yield Button("Mark as resolved", id="mark-resolved", variant="warning", compact=True) + with VerticalScroll(id="diff-scroll"): + yield Static(id="diff") + yield RichLog(id="gitlog", classes="hidden", markup=False, wrap=True, max_lines=3000) + yield Footer() + + async def on_mount(self) -> None: + self.register_theme(GITDESK_THEME) + self.theme = "gitdesk" + self._lock = asyncio.Lock() + self._net_lock = asyncio.Lock() + self.set_interval(3, self.poll) + self.set_interval(0.3, self.drain_log) + self.set_interval(30, self.update_remote_button) + self.set_interval(300, self.auto_fetch) + if await self.open_start_repo(): + self.run_worker(self.auto_fetch(), exit_on_error=False) + else: + self.run_worker(self.open_repo_flow(first=True), exit_on_error=False) + + async def open_start_repo(self) -> bool: + """A path on the command line wins; otherwise the folder you launched from if it's a + repository (but not gitdesk's own folder); otherwise the repository you had open last.""" + if self.start_path: + return await self.load_repo(self.start_path) + candidates = [] + here = await asyncio.to_thread(Git.open, os.getcwd(), self._log_cmd) + if here is not None: + app_dir = os.path.dirname(os.path.abspath( + sys.executable if getattr(sys, "frozen", False) else __file__)) + own = await asyncio.to_thread(Git.open, app_dir, self._log_cmd) + if own is None or os.path.normcase(own.path) != os.path.normcase(here.path): + candidates.append(here.path) + candidates += [p for p in load_recent() if os.path.isdir(p)] + for path in candidates: + if await self.load_repo(path): + return True + return False + + APP_KEYS = {"fetch", "pull", "push", "branches", "stash", "stashes", "open_repo", "discard", "view_remote", + "new_repo", "clone_repo", "add_remote", "repositories", + "toggle_log", "refresh", "toggle_file", "quit"} + + def check_action(self, action: str, parameters) -> Optional[bool]: + # Single-letter shortcuts are switched off while a popup is open. + if action in self.APP_KEYS and isinstance(self.screen, ModalScreen): + return False + return True + + # ---- small accessors --------------------------------------------------- + def remote_name(self) -> str: + st = self.state + if st and st.upstream and st.remotes: + for r in sorted(st.remotes, key=len, reverse=True): + if st.upstream.startswith(r + "/"): + return r + if st and st.remotes: + return "origin" if "origin" in st.remotes else st.remotes[0] + return "origin" + + def web_url(self) -> Optional[str]: + """Web page of the current remote. Override per repo: git config gitdesk.webUrl """ + st = self.state + if st is None or self.git is None: + return None + url = st.remote_urls.get(self.remote_name(), "") + key = (self.git.path, url) + if getattr(self, "_web_cache_key", None) != key: + override = self.git.run("config", "--get", "gitdesk.webUrl", log=False).out.strip() + self._web_cache_key = key + self._web_cache = override.rstrip("/") or (web_url_for(url) if url else None) + return self._web_cache + + def open_url(self, url: str) -> None: + """Open a link in the browser, without letting a text browser take over this terminal.""" + try: + if IS_WIN: + os.startfile(url) # type: ignore[attr-defined] + elif sys.platform == "darwin": + subprocess.Popen(["open", url]) + elif (os.environ.get("DISPLAY") or os.environ.get("WAYLAND_DISPLAY")) and shutil.which("xdg-open"): + subprocess.Popen(["xdg-open", url], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + stdin=subprocess.DEVNULL, start_new_session=True) + elif shutil.which("wslview"): + subprocess.Popen(["wslview", url], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + start_new_session=True) + else: + self.copy_to_clipboard(url) + self.notify(f"No browser found here - the link is copied to your clipboard:\n{url}", timeout=20) + return + self.notify(f"Opening {url}") + except OSError as e: + self.notify(f"Couldn't open the browser ({e}). The link is:\n{url}", severity="error", timeout=20) + + def action_view_remote(self) -> None: + if self.git is None or self.state is None: + return + url = self.web_url() + if not url: + remote = self.state.remote_urls.get(self.remote_name()) + if not remote: + self.notify("This repository has no remote, so there's no web page to open.", severity="warning") + else: + self.notify(f"Couldn't work out a web page from {remote}. Set one with:\n" + "git config gitdesk.webUrl https://host/owner/repo", severity="warning", timeout=12) + return + self.open_url(url) + + def included_files(self) -> List[FileChange]: + return [f for f in (self.state.files if self.state else []) if f.path not in self.excluded] + + async def bg(self, *args, input: Optional[str] = None, serialize: bool = True) -> Result: + return await asyncio.to_thread(self.git.run, *args, input=input, serialize=serialize) + + async def load_state(self) -> RepoState: + st = await asyncio.to_thread(self.git.state) + self.state = st + return st + + def _log_cmd(self, args, res: Result) -> None: + self.log_q.put((args, res)) # called from worker threads + + def drain_log(self) -> None: + try: + log = self.query_one("#gitlog", RichLog) + except NoMatches: # app is starting up or shutting down + return + while True: + try: + args, res = self.log_q.get_nowait() + except queue.Empty: + break + log.write(Text("$ git " + " ".join(args), style=f"bold {C_BLUE}")) + out = (res.out + ("\n" if res.out and res.err else "") + res.err).strip() + if out: + lines = out.splitlines() + if len(lines) > 40: + lines = lines[:40] + [f"… ({len(lines) - 40} more lines)"] + log.write(Text("\n".join(lines), style=C_TEXT if res.ok else "#f97583")) + + # ---- loading / refreshing ---------------------------------------------- + async def load_repo(self, path: str) -> bool: + g = await asyncio.to_thread(Git.open, path, self._log_cmd) + if g is None: + return False + self.git = g + self.state = None + self.excluded.clear() + self.selected = None + self._files_key = self._ph_key = self._diff_key = self._history_key = None + self.last_fetch = g.last_fetch_time() + save_recent(g.path) + self.sub_title = g.path + await self.query_one("#history", ListView).clear() + await self.refresh_state(force=True) + return True + + async def poll(self) -> None: + if not self.busy and self.git is not None: + await self.refresh_state() + + async def refresh_state(self, force: bool = False) -> None: + if self.git is None or self._lock is None: + return + if self._lock.locked() and not force: + return + async with self._lock: + g = self.git + try: + st = await asyncio.to_thread(g.state) + except GitError as e: + self.notify(str(e), title="git status failed", severity="error") + return + if g is not self.git: + return + self.state = st + try: + await self.apply_state(st) + except NoMatches: # app is shutting down + pass + + async def apply_state(self, st: RepoState) -> None: + g = self.git + self.query_one("#tb-repo", ToolbarButton).set(bottom=os.path.basename(g.path) or g.path) + branch = st.branch or (f"detached at {st.oid[:7]}" if st.oid else "detached") + self.query_one("#tb-branch", ToolbarButton).set(bottom=branch) + self.update_remote_button() + + key = tuple((f.path, f.kind, f.orig) for f in st.files) + if key != self._files_key: + self._files_key = key + await self.rebuild_files(st) + self.update_counts() + self.update_banner(st) + set_label(self.query_one("#stash-list", Button), f"Stashed ({st.stashes})") + self.query_one("#stash-list", Button).disabled = self.busy or st.stashes == 0 + self.query_one("#stash-all", Button).disabled = self.busy or not st.files + + if st.merging and not self._merge_prefilled: + self._merge_prefilled = True + summary = self.query_one("#summary", Input) + if not summary.value: + summary.value = g.merge_message() + elif not st.merging: + self._merge_prefilled = False + self.update_commit_button() + + if self.mode == "changes": + await self.show_changes_side(st) + else: + await self.load_history() + + async def rebuild_files(self, st: RepoState) -> None: + lv = self.query_one("#files", ListView) + paths = {f.path for f in st.files} + self.excluded &= paths + await lv.clear() + shown = st.files[:MAX_FILES_SHOWN] + rows: List[ListItem] = [FileRow(f, f.path not in self.excluded) for f in shown] + if len(st.files) > len(shown): + rows.append(ListItem(Label(Text(f"… and {len(st.files) - len(shown)} more files", + style=C_MUTED)), disabled=True)) + await lv.extend(rows) + if self.selected not in paths: + self.selected = shown[0].path if shown else None + for i, f in enumerate(shown): + if f.path == self.selected: + lv.index = i + break + + def update_counts(self) -> None: + st = self.state + n = len(st.files) if st else 0 + inc = len(self.included_files()) + self.query_one("#changes-count", Label).update( + Text(plural(n, "changed file"), style="bold" if n else C_MUTED)) + self.query_one("#tick-all", Tick).checked = n > 0 and inc == n + self.query_one("#tabs", TabbedContent).get_tab("tab-changes").label = ( + f"Changes ({n})" if n else "Changes") + + def update_banner(self, st: RepoState) -> None: + banner = self.query_one("#banner") + abort = self.query_one("#abort-merge", Button) + text = None + if st.rebasing: + text = Text.assemble(("Rebase in progress. ", "bold"), + "Finish it in a terminal with git rebase --continue (or --abort).") + abort.display = False + elif st.merging: + if st.conflicts: + text = Text.assemble(("Merge conflicts: ", "bold"), + f"{plural(len(st.conflicts), 'file')} need fixing (marked ", (" ! ", f"bold {C_ORANGE}"), + "). Open each one, keep the parts you want, remove the <<<<<<< ======= >>>>>>> " + "lines, click Mark as resolved, then commit the merge.") + else: + text = Text.assemble(("All conflicts resolved. ", "bold"), "Commit the merge to finish it.") + abort.display = True + elif st.conflicts: + text = Text.assemble(("Conflicts: ", "bold"), + f"{plural(len(st.conflicts), 'file')} couldn't be merged with your stashed changes. " + "Fix the <<<<<<< ======= >>>>>>> parts, then click Mark as resolved.") + abort.display = False + banner.set_class(text is None, "hidden") + if text is not None: + self.query_one("#banner-text", Static).update(text) + + def update_commit_button(self) -> None: + st = self.state + btn = self.query_one("#commit", Button) + if st is None: + btn.disabled = True + return + summary = self.query_one("#summary", Input).value.strip() + if st.merging: + btn.label = "Commit merge" + btn.disabled = self.busy or bool(st.conflicts) + else: + n = len(self.included_files()) + btn.label = f"Commit {plural(n, 'file')} to {st.branch or 'HEAD'}" + btn.disabled = self.busy or n == 0 or not summary or bool(st.conflicts) + + def remote_mode(self) -> str: + st = self.state + if st is None or not st.remotes: + return "none" + if st.branch is None: + return "fetch" + if not st.has_upstream: + return "publish" + if st.behind: + return "pull" + if st.ahead: + return "push" + return "fetch" + + def update_remote_button(self) -> None: + try: + b = self.query_one("#tb-remote", ToolbarButton) + except NoMatches: # app is shutting down + return + st = self.state + if self.busy: + b.set(icon="…", top=self.busy_label + "…", bottom=self.busy_detail or "Working", badge="") + b.disabled = True + return + b.disabled = False + remote = self.remote_name() + fetched = f"Last fetched {ago(self.last_fetch)}" + if self.fetching: + fetched = "Checking for changes…" + mode = self.remote_mode() + if mode == "none": + b.set(icon="+", top="No remote", bottom="Click to add one", badge="") + elif mode == "publish": + b.set(icon="↑", top="Publish branch", bottom=f"Push it to {remote}", badge="") + elif mode == "pull": + badge = f"↓{st.behind}" + (f" ↑{st.ahead}" if st.ahead else "") + b.set(icon="↓", top=f"Pull {remote}", bottom=fetched, badge=badge) + elif mode == "push": + b.set(icon="↑", top=f"Push {remote}", bottom=fetched, badge=f"↑{st.ahead}") + else: + b.set(icon="↻", top=f"Fetch {remote}", bottom=fetched, badge="") + + def set_busy(self, busy: bool, label: str = "") -> None: + self.busy, self.busy_label = busy, label + for sel in ("#tb-branch", "#tb-repo", "#stash-all", "#stash-list", "#abort-merge", "#mark-resolved", "#discard-file"): + self.query_one(sel).disabled = busy + self.update_remote_button() + self.update_commit_button() + + # ---- right-hand side --------------------------------------------------- + async def show_changes_side(self, st: RepoState) -> None: + switcher = self.query_one("#right", ContentSwitcher) + if not st.files: + self._diff_key = None + await self.rebuild_placeholder(st) + switcher.current = "placeholder" + return + fc = next((f for f in st.files if f.path == self.selected), None) + if fc is None: + fc = st.files[0] + self.selected = fc.path + switcher.current = "diffview" + self.run_worker(self.show_file_diff(fc), group="diff", exclusive=True, exit_on_error=False) + + async def show_file_diff(self, fc: FileChange) -> None: + g, st = self.git, self.state + text = await asyncio.to_thread(g.file_diff, fc, st.oid is not None) + if self.mode != "changes" or self.selected != fc.path: + return + key = ("file", fc.path, fc.kind, text) + if key == self._diff_key: + return + new_file = not self._diff_key or self._diff_key[1] != fc.path + self._diff_key = key + icon, color, label = KIND_ICON.get(fc.kind, ("●", C_YELLOW, "")) + title = Text(no_wrap=True, overflow="ellipsis") + title.append(f"{icon} ", style=f"bold {color}") + title.append(fc.path, style="bold") + title.append(f" {label}", style=C_MUTED) + self.query_one("#diff-title", Label).update(title) + self.query_one("#open-file").display = True + self.query_one("#discard-file").display = not fc.is_conflict + self.query_one("#view-commit").display = False + self.query_one("#mark-resolved").display = fc.is_conflict + self.query_one("#diff", Static).update(render_diff(text)) + self.query_one("#right", ContentSwitcher).current = "diffview" + if new_file: + self.query_one("#diff-scroll", VerticalScroll).scroll_home(animate=False) + + async def rebuild_placeholder(self, st: RepoState) -> None: + key = (st.behind, st.ahead, st.has_upstream, bool(st.remotes), st.branch, st.stashes, self.web_url()) + if key == self._ph_key: + return + self._ph_key = key + self.query_one("#ph-title", Static).update(Text("No local changes", style=f"bold {C_TEXT}")) + self.query_one("#ph-sub", Static).update(Text( + "There are no uncommitted changes in this repository. Here are some friendly " + "suggestions for what to do next.", style=C_MUTED)) + remote = self.remote_name() + cards = [] + if not st.remotes: + cards.append(Card("Publish your repository", + "It isn't on GitHub, Gitea or another server yet. Create an empty repository " + "there, then add its URL here.", "Add remote URL", "add_remote")) + if st.remotes and st.branch: + if not st.has_upstream and st.oid: + cards.append(Card("Publish your branch", + f"'{st.branch}' isn't on {remote} yet. Publish it so it's backed up and shareable.", + "Publish branch", "push")) + elif st.behind: + cards.append(Card(f"Pull {plural(st.behind, 'commit')} from {remote}", + "The remote has changes you don't have yet.", f"Pull {remote}", "pull")) + elif st.ahead: + cards.append(Card(f"Push {plural(st.ahead, 'commit')} to {remote}", + "You have local commits that aren't on the remote yet.", f"Push {remote}", "push")) + if st.stashes: + cards.append(Card(f"You have {plural(st.stashes, 'stash')}", + "Changes you put aside earlier. Bring them back any time.", + "View stashed changes", "stashes")) + cards.append(Card("Open the repository in your external editor", + f"Opens it in {editor_name()}.", "Open in editor", "open_editor")) + cards.append(Card("View the files of your repository", + f"Opens the folder in {file_manager_name()}.", + f"Show in {file_manager_name()}", "open_folder")) + web = self.web_url() + if web: + site = web_site_name(web) + cards.append(Card(f"Open the repository page on {site} in your browser", web, + f"View on {site}", "view_remote")) + box = self.query_one("#ph-cards", Vertical) + await box.remove_children() + await box.mount_all(cards) + + async def load_history(self) -> None: + st, g = self.state, self.git + if st is None: + return + key = (st.oid, st.upstream, st.ahead) + if key == self._history_key: + return + self._history_key = key + commits, unpushed = await asyncio.to_thread(g.history, st.has_upstream) if st.oid else ([], set()) + lv = self.query_one("#history", ListView) + await lv.clear() + await lv.extend([CommitRow(c, c.sha in unpushed) for c in commits]) + if not commits: + self._show_message("History", "No commits yet.") + + def _show_message(self, title: str, message: str) -> None: + self._diff_key = None + self.query_one("#diff-title", Label).update(Text(title, style="bold")) + self.query_one("#open-file").display = False + self.query_one("#discard-file").display = False + self.query_one("#mark-resolved").display = False + self.query_one("#view-commit").display = False + self.query_one("#diff", Static).update(Text(message, style=f"italic {C_MUTED}")) + self.query_one("#right", ContentSwitcher).current = "diffview" + + async def show_commit(self, c: Commit, unpushed: bool = False) -> None: + g = self.git + header = await asyncio.to_thread(g.commit_header, c.sha) + diff = await asyncio.to_thread(g.commit_diff, c.sha) + if self.mode != "history": + return + lines = header.splitlines() + head = Text() + if len(lines) >= 3: + body = "\n".join(lines[3:]).strip() + _, _, rest = body.partition("\n") # subject is already in the title bar + head.append(f" {lines[1]} • {lines[2]} • {lines[0][:10]}\n", style=C_MUTED) + if rest.strip(): + head.append("\n" + "\n".join(" " + ln for ln in rest.strip().splitlines()) + "\n", + style=C_TEXT) + head.append("\n") + self._diff_key = ("commit", c.sha) + title = Text(no_wrap=True, overflow="ellipsis") + title.append(c.short + " ", style=C_BLUE) + title.append(c.subject, style="bold") + self.query_one("#diff-title", Label).update(title) + self.query_one("#open-file").display = False + self.query_one("#discard-file").display = False + self.query_one("#mark-resolved").display = False + web = self.web_url() + view = self.query_one("#view-commit", Button) + view.display = bool(web) and not unpushed # unpushed commits aren't on the site yet + if web: + set_label(view, f"View on {web_site_name(web)}") + self._shown_commit = c.sha + self.query_one("#diff", Static).update(Group(head, render_diff(diff, multi_file=True))) + self.query_one("#right", ContentSwitcher).current = "diffview" + self.query_one("#diff-scroll", VerticalScroll).scroll_home(animate=False) + + # ---- UI events --------------------------------------------------------- + @on(TabbedContent.TabActivated, "#tabs") + async def _tab_changed(self, event: TabbedContent.TabActivated) -> None: + self.mode = "history" if event.pane.id == "tab-history" else "changes" + self._diff_key = None + if self.state is None: + return + if self.mode == "history": + await self.load_history() + lv = self.query_one("#history", ListView) + if isinstance(lv.highlighted_child, CommitRow): + self.run_worker(self.show_commit(lv.highlighted_child.commit, lv.highlighted_child.unpushed), + group="diff", + exclusive=True, exit_on_error=False) + elif lv.children: + self._show_message("History", "Click a commit to see what changed.") + else: + await self.show_changes_side(self.state) + + @on(ListView.Highlighted, "#files") + def _file_highlighted(self, event: ListView.Highlighted) -> None: + if isinstance(event.item, FileRow) and self.mode == "changes": + self.selected = event.item.fc.path + self.run_worker(self.show_file_diff(event.item.fc), group="diff", exclusive=True, + exit_on_error=False) + + @on(ListView.Highlighted, "#history") + def _commit_highlighted(self, event: ListView.Highlighted) -> None: + if isinstance(event.item, CommitRow) and self.mode == "history": + self.run_worker(self.show_commit(event.item.commit, event.item.unpushed), group="diff", exclusive=True, + exit_on_error=False) + + @on(Tick.Toggled) + def _tick_toggled(self, event: Tick.Toggled) -> None: + st = self.state + if st is None: + return + tick = event.tick + if tick.id == "tick-all": + self.excluded = set() if tick.checked else {f.path for f in st.files} + for t in self.query("#files Tick"): + t.checked = tick.checked + elif tick.key: + (self.excluded.discard if tick.checked else self.excluded.add)(tick.key) + self.update_counts() + self.update_commit_button() + + def action_toggle_file(self) -> None: + lv = self.query_one("#files", ListView) + if self.mode != "changes" or not lv.has_focus or not isinstance(lv.highlighted_child, FileRow): + return + tick = lv.highlighted_child.query_one(Tick) + tick.checked = not tick.checked + tick.post_message(Tick.Toggled(tick)) + + @on(Input.Changed, "#summary") + def _summary_changed(self) -> None: + self.update_commit_button() + + @on(Input.Submitted, "#summary") + def _summary_submitted(self) -> None: + if not self.query_one("#commit", Button).disabled: + self.start_op("Committing", self.commit_flow) + + @on(ToolbarButton.Pressed) + def _toolbar(self, event: ToolbarButton.Pressed) -> None: + bid = event.button.id + if bid == "tb-repo": + self.action_repositories() + elif bid == "tb-branch": + self.action_branches() + elif bid == "tb-more": + self.run_worker(self.actions_flow(), exit_on_error=False) + elif bid == "tb-remote": + mode = self.remote_mode() + if mode == "none": + self.action_add_remote() + elif mode == "pull": + self.action_pull() + elif mode in ("push", "publish"): + self.action_push() + else: + self.action_fetch() + + async def on_button_pressed(self, event: Button.Pressed) -> None: + b = event.button + if b.has_class("card-btn") and b.name: + await self.run_action(b.name) + elif b.id == "commit": + self.start_op("Committing", self.commit_flow) + elif b.id == "stash-all": + self.action_stash() + elif b.id == "stash-list": + self.action_stashes() + elif b.id == "abort-merge": + self.start_op("Aborting merge", self.abort_merge_flow) + elif b.id == "mark-resolved": + self.start_op("Marking resolved", self.mark_resolved_flow) + elif b.id == "view-commit" and self.web_url() and getattr(self, "_shown_commit", None): + self.open_url(commit_web_url(self.web_url(), self._shown_commit)) + elif b.id == "discard-file": + self.action_discard() + elif b.id == "open-file" and self.selected and self.git: + self.action_open_editor(os.path.join(self.git.path, self.selected)) + + # ---- actions (keyboard + buttons) -------------------------------------- + def action_fetch(self) -> None: + self.start_op("Fetching", self.fetch_flow) + + def action_pull(self) -> None: + self.start_op("Pulling", self.pull_flow) + + def action_push(self) -> None: + self.start_op("Pushing", self.push_flow) + + def action_sync(self) -> None: + self.start_op("Syncing", lambda: self.push_flow(auto_pull=True)) + + def action_stash(self) -> None: + self.run_worker(self.stash_all_flow(), exit_on_error=False) + + def action_stashes(self) -> None: + self.run_worker(self.stashes_flow(), exit_on_error=False) + + def action_discard(self) -> None: + st = self.state + fc = next((f for f in (st.files if st else []) if f.path == self.selected), None) + if self.mode != "changes" or fc is None: + self.notify("Select a changed file first.", severity="warning") + return + self.run_worker(self.discard_flow([fc]), exit_on_error=False) + + def action_discard_ticked(self) -> None: + self.run_worker(self.discard_flow(self.included_files()), exit_on_error=False) + + def action_branches(self) -> None: + self.run_worker(self.branches_flow(), exit_on_error=False) + + def action_open_repo(self) -> None: + self.run_worker(self.add_local_flow(), exit_on_error=False) + + def action_toggle_log(self) -> None: + self.query_one("#gitlog").toggle_class("hidden") + + async def action_refresh(self) -> None: + self._files_key = self._ph_key = self._diff_key = self._history_key = None + await self.refresh_state(force=True) + + def action_open_editor(self, path: Optional[str] = None) -> None: + if self.git is None: + return + target = path or self.git.path + code = shutil.which("code") + try: + if code: + if IS_WIN: + subprocess.Popen(f'"{code}" "{target}"', shell=True, stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + creationflags=subprocess.CREATE_NO_WINDOW) + else: + subprocess.Popen([code, target], stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, start_new_session=True) + self.notify(f"Opening in VS Code: {os.path.basename(target)}") + return + editor = os.environ.get("VISUAL") or os.environ.get("EDITOR") + if editor: + with self.suspend(): + subprocess.call(f'{editor} "{target}"', shell=True) + return + except (OSError, SuspendNotSupported) as e: + self.notify(f"Couldn't open the editor: {e}", severity="error") + return + self.notify("Couldn't find VS Code ('code' command) or an EDITOR environment variable.", + severity="warning") + + def action_open_folder(self) -> None: + if self.git is None: + return + try: + if IS_WIN: + os.startfile(self.git.path) # type: ignore[attr-defined] + elif sys.platform == "darwin": + subprocess.Popen(["open", self.git.path]) + elif shutil.which("xdg-open"): + subprocess.Popen(["xdg-open", self.git.path], stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, start_new_session=True) + else: + self.notify("No file manager found (xdg-open isn't installed).", severity="warning") + except OSError as e: + self.notify(f"Couldn't open the folder: {e}", severity="error") + + # ---- running operations ------------------------------------------------ + def idle_or_warn(self) -> bool: + if self.git is None or self.state is None: + self.notify("Open a repository first.", severity="warning") + return False + if self.busy: + self.notify(f"Still busy ({self.busy_label.lower()}) - try again in a moment.", severity="warning") + return False + return True + + def start_op(self, label: str, flow: Callable) -> None: + if self.git is None or self.state is None: + self.notify("Open a repository first.", severity="warning") + return + if self.busy: + self.notify("Still working on the last action…", severity="warning") + return + self.set_busy(True, label) + self.run_worker(self._run_op(flow), exit_on_error=False) + + async def _run_op(self, flow: Callable) -> None: + try: + await flow() + except Exception as e: # keep the app alive whatever happens + await self.error_dialog("Something went wrong", f"{type(e).__name__}: {e}") + finally: + self.set_busy(False) + await self.refresh_state(force=True) + self.update_remote_button() + + async def ask(self, title: str, body, buttons, cancel=None) -> str: + return await self.push_screen_wait(ChoiceDialog(title, body, buttons, cancel)) + + async def error_dialog(self, title: str, message: str) -> None: + await self.ask(title, Text(message.strip(), style="#f97583"), [("ok", "OK", "primary")]) + + async def fail(self, title: str, r: Result, hint: str = "") -> None: + """Show a git failure, unless the user has already seen it (e.g. the lock popup).""" + if r.handled: + return + hint = hint or hint_for(r.message) + await self.error_dialog(title, r.message + (f"\n\n{hint}" if hint else "")) + + async def gitop(self, *args, input: Optional[str] = None) -> Result: + """Run a git command that changes the repository. + + If it fails because another program (GitHub Desktop, VS Code, Unity, a + terminal...) has the repository locked, and the failed attempt didn't + change anything, wait a moment and try again. If the lock doesn't go + away, ask the user what to do. + """ + while True: + r = await self._gitop_retrying(args, input) + if r.ok or not r.lockish: + return r + choice = await self.lock_dialog(r) + if choice == "remove": + try: + os.remove(self.git.index_lock_path()) + self._log_cmd(["(gitdesk removed .git/index.lock)"], Result(0)) + except OSError as e: + await self.error_dialog("Couldn't remove the lock file", str(e)) + r.handled = True + return r + elif choice != "retry": + r.handled = True + return r + + async def _gitop_retrying(self, args, input) -> Result: + before = await asyncio.to_thread(self.git.snapshot) + for delay in (0.25, 0.5, 0.75, 1.0, 1.5, None): + r = await self.bg(*args, input=input) + if r.ok or not looks_lockish(r): + return r + after = await asyncio.to_thread(self.git.snapshot) + if after != before: + return r # it changed something - never repeat blindly + r.lockish = True + if delay is None: + return r + await asyncio.sleep(delay) + return r + + async def lock_dialog(self, r: Result) -> str: + lock = self.git.index_lock_path() + body = Text() + detail = STASH_KEPT_RE.sub("", f"{r.err}\n{r.out}").strip() + if detail: + body.append(detail + "\n\n", style="#f97583") + if os.path.exists(lock): + try: + age = ago(os.path.getmtime(lock)) + except OSError: + age = "a moment ago" + body.append("Another git program is using this repository right now, so git couldn't " + "update it. The lock file ") + body.append(".git/index.lock", style="bold") + body.append(f" was created {age}.\n\n") + body.append("If GitHub Desktop, VS Code, Unity or another terminal is busy with this repository, " + "wait for it to finish and try again. If nothing else is running, the lock was " + "left behind by a git program that crashed, and it's safe to remove it.", style=C_MUTED) + buttons = [("retry", "Try again", "primary"), ("remove", "Remove lock file & retry", "warning"), + ("close", "Close", "default")] + else: + body.append("Another program had the repository busy, so git couldn't update it. This is usually " + "GitHub Desktop, VS Code, Unity, an antivirus scan, or a sync tool (OneDrive, Dropbox) " + "briefly holding the repository's files. Wait a moment and try again.", style=C_MUTED) + buttons = [("retry", "Try again", "primary"), ("close", "Close", "default")] + return await self.ask("The repository is busy", body, buttons, cancel="close") + + async def net(self, args: List[str], what: str) -> bool: + """Run a network git command in the background; on failure offer to retry.""" + while True: + async with self._net_lock: + r = await self.bg(*args, serialize=False) + if r.ok: + return True + tail = "\n".join(r.message.splitlines()[-10:]) + body = Text.assemble((tail, "#f97583"), "\n\n", + ("If git needs a password or token, an SSH passphrase, or wants you to " + "confirm the server's key, 'Retry in terminal' lets it ask you.", C_MUTED)) + choice = await self.ask(f"Couldn't {what}", body, + [("retry", "Try again", "default"), + ("terminal", "Retry in terminal", "primary"), + ("close", "Close", "default")], cancel="close") + if choice == "terminal": + return self.run_in_terminal(args) + if choice != "retry": + return False + + def run_in_terminal(self, args: List[str], cwd: Optional[str] = None) -> bool: + rc = 1 + try: + with self.suspend(): + print(f"\n[gitdesk] git {' '.join(args)}\n", flush=True) + rc = subprocess.call(["git", *args], cwd=cwd or self.git.path) + print("\n[gitdesk] " + ("Done." if rc == 0 else f"git exited with code {rc}.")) + try: + input("Press Enter to go back to gitdesk...") + except EOFError: + pass + except SuspendNotSupported: + self.notify("This terminal can't hand over to git for a prompt.", severity="error") + return False + self._log_cmd(args, Result(rc, "(ran in the terminal)")) + return rc == 0 + + async def auto_fetch(self) -> None: + """Quiet background fetch. Never blocks other actions and never pops up dialogs.""" + if self.busy or self.git is None or self.state is None or not self.state.remotes: + return + if self._net_lock.locked(): + return + async with self._net_lock: + self.fetching = True + self.update_remote_button() + try: + r = await self.bg("fetch", "--prune", self.remote_name(), serialize=False) + if r.ok: + self.last_fetch = time.time() + finally: + self.fetching = False + await self.refresh_state(force=True) + self.update_remote_button() + + async def do_fetch(self) -> bool: + remote = self.remote_name() + ok = await self.net(["fetch", "--prune", remote], f"fetch from {remote}") + if ok: + self.last_fetch = time.time() + return ok + + def blocked_by_merge(self, st: RepoState) -> bool: + if st.merging or st.conflicts or st.rebasing: + self.notify("Finish resolving the conflicts (or abort the merge) first.", severity="warning") + return True + return False + + # ---- flows ------------------------------------------------------------- + async def fetch_flow(self) -> None: + if not self.state.remotes: + self.notify("This repository has no remote to fetch from.", severity="warning") + return + if await self.do_fetch(): + self.notify(f"Fetched {self.remote_name()}.") + + async def pull_flow(self) -> None: + st = self.state + if self.blocked_by_merge(st): + return + if st.branch is None: + self.notify("You're not on a branch, so there's nothing to pull.", severity="warning") + return + if not st.has_upstream: + self.notify(f"'{st.branch}' isn't on {self.remote_name()} yet. Publish it first.", severity="warning") + return + if await self.do_fetch(): + await self.merge_upstream() + + async def merge_upstream(self) -> bool: + """Merge the upstream branch in, offering to stash if local changes are in the way.""" + st = await self.load_state() + if st.behind == 0: + self.notify("Already up to date.") + return True + n = st.behind + remote = self.remote_name() + r = await self.gitop("merge", "--no-edit", st.upstream) + if r.ok: + self.notify(f"Pulled {plural(n, 'commit')} from {remote}.") + return True + st = await self.load_state() + if st.merging: + await self.conflict_dialog(auto_stashed=False) + return False + if not st.files: + await self.fail("Pull failed", r) + return False + + # git refused because uncommitted changes are in the way + incoming = await asyncio.to_thread(self.git.changed_between, "HEAD", st.upstream) + local = {f.path for f in st.files} + blocking = sorted(local & incoming) or sorted(local) + body = Text("Unable to pull when changes are present on your branch. " + "These files would be overwritten:\n\n") + for p in blocking[:12]: + body.append(f" {p}\n", style="bold") + if len(blocking) > 12: + body.append(f" … and {len(blocking) - 12} more\n", style=C_MUTED) + body.append("\nYou can stash your changes now, pull, and then they'll be put back automatically.") + choice = await self.ask("Your changes are in the way", body, + [("stash", "Stash, pull & re-apply", "primary"), ("close", "Close", "default")]) + if choice != "stash": + return False + if not await self.stash_push("gitdesk: auto-stash before pull"): + return False + r = await self.gitop("merge", "--no-edit", st.upstream) + if not r.ok: + st = await self.load_state() + if st.merging: + await self.conflict_dialog(auto_stashed=True) + return False + await self.fail("Pull failed", r) + await self.stash_pop("stash@{0}") + return False + self.notify(f"Pulled {plural(n, 'commit')} from {remote}.") + await self.stash_pop("stash@{0}") + return True + + async def conflict_dialog(self, auto_stashed: bool) -> None: + st = await self.load_state() + body = Text("Your commits and the commits you pulled changed the same lines in:\n\n") + for f in st.conflicts[:12]: + body.append(f" {f.path}\n", style="bold") + body.append("\nTo fix it, click each file on the left, keep the parts you want and delete the " + "<<<<<<< ======= >>>>>>> marker lines (Open in editor helps). Then click " + "'Mark as resolved' and commit the merge.\n\nOr abort to put everything back the " + "way it was before the pull.") + if auto_stashed: + body.append("\n\nYour own uncommitted changes are safe in Stashed changes.", style=C_MUTED) + choice = await self.ask("Merge conflicts", body, + [("resolve", "Resolve conflicts", "primary"), ("abort", "Abort merge", "error")], + cancel="resolve") + if choice == "abort": + r = await self.gitop("merge", "--abort") + if not r.ok: + await self.fail("Couldn't abort the merge", r) + return + self.notify("Merge aborted - nothing was changed.") + if auto_stashed: + await self.stash_pop("stash@{0}") + elif auto_stashed: + self.notify("After you commit the merge, bring your changes back from Stashed changes.", + timeout=10) + + async def stash_push(self, message: Optional[str], paths: Optional[List[str]] = None) -> bool: + """Stash changes (including new files). Returns True if a stash entry was created.""" + st = await self.load_state() + if st.oid is None: + await self.error_dialog("Can't stash yet", + "git can only stash once the repository has at least one commit. " + "Make a first commit, then stash.") + return False + if st.conflicts: + self.notify("Resolve the conflicts first - git can't stash files that are mid-merge.", + severity="warning") + return False + before = self.git.stash_count() + args = ["stash", "push", "-u"] + (["-m", message] if message else []) + if paths: + args = ["--literal-pathspecs", *args, "--", *paths] + r = await self.gitop(*args) + created = self.git.stash_count() > before + if created and not r.ok: + # The stash was saved but git hit a problem afterwards (often a file that + # another program - Unity, an editor - has open on Windows). + await self.error_dialog( + "Stashed, but git reported a problem", + r.message + "\n\nYour changes ARE saved in Stashed changes. Some files may not have been " + "removed from your folder - close any program using them and check the Changes list.") + return True + if not r.ok: + await self.fail("Couldn't stash your changes", r) + return False + if not created: + self.notify("There was nothing to stash.", severity="warning") + return False + return True + + async def stash_pop(self, ref: str, keep: bool = False) -> bool: + st = await self.load_state() + if st.conflicts or st.merging: + self.notify("Resolve the current conflicts (or abort the merge) first.", severity="warning") + return False + r = await self.gitop("stash", "apply" if keep else "pop", ref) + if r.ok: + self.notify("Your stashed changes are back." if not keep else "Applied - the stash is still saved too.") + return True + if r.handled: + return False + st = await self.load_state() + if st.conflicts: + body = Text("Your stashed changes clash with the current version of:\n\n") + for f in st.conflicts[:12]: + body.append(f" {f.path}\n", style="bold") + body.append("\nThe clashing parts are marked inside those files with <<<<<<< / ======= / >>>>>>>. " + "Edit each file to keep what you want, then click 'Mark as resolved'.\n\n" + "The stash itself was kept, so nothing is lost - drop it from Stashed changes " + "once you're happy.") + await self.ask("Stashed changes conflict", body, [("ok", "OK", "primary")]) + else: + msg = r.message + if "would be overwritten" in msg: + hint = ("Some of the stashed files also have changes in your folder right now. Commit or " + "stash your current changes first, then bring this stash back.") + elif "already exists" in msg: + hint = ("Some new files in the stash already exist in your folder. Move or delete those " + "files, then try again.") + else: + hint = "" + await self.error_dialog("Couldn't bring back your stashed changes", + msg + (f"\n\n{hint}" if hint else "") + "\n\nThe stash was kept, so nothing is lost.") + return False + + async def push_flow(self, auto_pull: bool = False) -> None: + st = self.state + if self.blocked_by_merge(st): + return + if st.branch is None: + self.notify("You're not on a branch, so there's nothing to push.", severity="warning") + return + if not st.remotes: + self.notify("This repository has no remote. Add one with: git remote add origin ", + severity="warning") + return + if st.oid is None: + self.notify("Make a commit first - there's nothing to push yet.", severity="warning") + return + remote = self.remote_name() + if not st.has_upstream: + if await self.net(["push", "-u", remote, st.branch], f"publish to {remote}"): + self.notify(f"Published '{st.branch}' to {remote}.") + return + if not await self.do_fetch(): + return + st = await self.load_state() + if st.behind: + if not auto_pull: + choice = await self.ask( + "Newer commits on the remote", + f"{remote} has {plural(st.behind, 'commit')} on '{st.branch}' that you don't have yet, " + "so a push would be rejected.\n\nPull them first, then push?", + [("pull", "Pull, then push", "primary"), ("close", "Cancel", "default")]) + if choice != "pull": + return + if not await self.merge_upstream(): + return + st = await self.load_state() + if st.ahead == 0: + self.notify("Nothing to push - you're up to date.") + return + target = st.upstream[len(remote) + 1:] + n = st.ahead + if await self.net(["push", remote, f"HEAD:refs/heads/{target}"], f"push to {remote}"): + self.notify(f"Pushed {plural(n, 'commit')} to {remote}.") + + async def commit_flow(self) -> None: + st = await self.load_state() + summary = self.query_one("#summary", Input).value.strip() + desc = self.query_one("#description", TextArea).text.strip() + if st.conflicts: + self.notify("Resolve the conflicts first (click each ! file and Mark as resolved).", + severity="warning") + return + message = summary + (f"\n\n{desc}" if desc else "") + if st.merging: + r = await self.gitop("add", "-A") + if r.ok: + if summary: + r = await self.gitop("commit", "--cleanup=whitespace", "-F", "-", input=message) + else: + r = await self.gitop("commit", "--no-edit") + else: + files = [f for f in st.files if f.path not in self.excluded] + if not files: + self.notify("Tick at least one file to commit.", severity="warning") + return + if not summary: + self.notify("Write a summary for the commit.", severity="warning") + return + if len(files) == len(st.files): + r = await self.gitop("add", "-A") + else: + # Commit only the ticked files: reset the index, then add just those. + if st.oid: + r = await self.gitop("reset", "-q") + else: + r = await self.gitop("rm", "-r", "-q", "--cached", "--ignore-unmatch", "--", ".") + if r.ok: + now = await self.load_state() + present = {f.path for f in now.files} + specs = [] + for f in files: + specs += [p for p in (f.path, f.orig) if p and p in present] + if not specs: + self.notify("The ticked files have no changes left to commit.", severity="warning") + return + r = await self.gitop("--literal-pathspecs", "add", "-A", "--pathspec-from-file=-", + "--pathspec-file-nul", input="\0".join(specs)) + if r.ok: + r = await self.gitop("commit", "--cleanup=whitespace", "-F", "-", input=message) + if not r.ok: + await self.fail("Commit failed", r) + return + self.query_one("#summary", Input).value = "" + self.query_one("#description", TextArea).text = "" + self.excluded.clear() + self.notify(f"Committed: {summary or 'merge'}") + + async def mark_resolved_flow(self) -> None: + st = self.state + fc = next((f for f in st.files if f.path == self.selected), None) + if fc is None or not fc.is_conflict: + return + if await asyncio.to_thread(file_has_markers, os.path.join(self.git.path, fc.path)): + choice = await self.ask( + "Conflict markers are still in this file", + f"{fc.path} still has <<<<<<< / >>>>>>> lines in it. Mark it as resolved anyway?", + [("yes", "Mark as resolved", "warning"), ("no", "Cancel", "default")]) + if choice != "yes": + return + r = await self.gitop("--literal-pathspecs", "add", "--", fc.path) + if r.ok and not st.merging and st.oid: + # a stash conflict: leave the file as a normal (unstaged) change + r = await self.gitop("--literal-pathspecs", "restore", "--staged", "--", fc.path) + if r.ok: + self.notify(f"Marked {fc.path} as resolved.") + else: + await self.fail("Couldn't mark it as resolved", r) + + async def abort_merge_flow(self) -> None: + choice = await self.ask( + "Abort the merge?", + "This throws away the merge and any conflict fixes you've made, and puts the branch back " + "the way it was before the pull.", + [("abort", "Abort merge", "error"), ("cancel", "Cancel", "default")]) + if choice != "abort": + return + r = await self.gitop("merge", "--abort") + if r.ok: + self.notify("Merge aborted.") + else: + await self.fail("Couldn't abort the merge", r) + + async def stash_all_flow(self) -> None: + if not self.idle_or_warn(): + return + if not self.state.files: + self.notify("There's nothing to stash.", severity="warning") + return + name = await self.push_screen_wait(InputDialog( + "Stash all changes", + "Puts all your uncommitted changes (including new files) aside so you have a clean " + "working folder. Give the stash a name if you like:", + "Stash changes", placeholder="e.g. half-done README edit")) + if name is None: + return + + async def flow(): + if await self.stash_push(name.strip() or None): + self.notify("Changes stashed. Bring them back from Stashed changes.") + self.start_op("Stashing", flow) + + async def discard_flow(self, files: List[FileChange]) -> None: + st = self.state + if not self.idle_or_warn(): + return + if not files: + self.notify("Tick the files you want to discard first.", severity="warning") + return + if st.merging or st.conflicts: + self.notify("Finish resolving the conflicts (or abort the merge) first.", severity="warning") + return + if st.oid is None: + self.notify("Discard needs at least one commit in the repository.", severity="warning") + return + n = len(files) + body = Text("These changes will be removed from your folder:\n\n") + for f in files[:12]: + body.append(f" {f.path}\n", style="bold") + if n > 12: + body.append(f" … and {n - 12} more\n", style=C_MUTED) + body.append("\ngitdesk keeps a backup as a stash named 'gitdesk: discarded …', so you can get " + "them back from Stashed changes if you change your mind.", style=C_MUTED) + choice = await self.ask(f"Discard changes to {plural(n, 'file')}?", body, + [("discard", "Discard changes", "error"), ("cancel", "Cancel", "default")], + cancel="cancel") + if choice != "discard": + return + paths: List[str] = [] + for f in files: + paths.append(f.path) + if f.orig: + paths.append(f.orig) + everything = n == len(st.files) + if not everything and sum(len(p) + 3 for p in paths) > 20000: + self.notify("That's too many files to discard in one go - tick fewer.", severity="warning") + return + label = files[0].path if n == 1 else plural(n, "file") + + async def flow(): + limit = None + if not everything: + # A path-limited stash records the whole index, so unstage first (contents are + # untouched); then only the chosen files end up in the backup stash. This also + # turns staged renames/additions into plain deletions/new files that stash handles. + r = await self.gitop("reset", "-q") + if not r.ok: + await self.fail("Couldn't discard the changes", r) + return + now = await self.load_state() + present = {f.path for f in now.files} + limit = [p for p in paths if p in present] + if not limit: + self.notify("Those files have no changes left to discard.") + return + if await self.stash_push(f"gitdesk: discarded {label}", paths=limit): + self.notify("Discarded. A backup is in Stashed changes.") + self.start_op("Discarding", flow) + + async def stashes_flow(self) -> None: + if not self.idle_or_warn(): + return + res = await self.push_screen_wait(StashDialog(self.git)) + if not res: + return + action, ref = res + if action == "drop": + choice = await self.ask("Drop this stash?", + f"{ref} will be deleted for good. This can't be undone.", + [("drop", "Drop stash", "error"), ("cancel", "Cancel", "default")]) + if choice != "drop": + return + + async def drop(): + r = await self.gitop("stash", "drop", ref) + if r.ok: + self.notify(f"Dropped {ref}.") + else: + await self.fail("Couldn't drop the stash", r) + self.start_op("Dropping stash", drop) + return + + async def restore(): + st = await self.load_state() + if st.conflicts: + self.notify("Resolve the current conflicts first.", severity="warning") + return + await self.stash_pop(ref, keep=(action == "apply")) + self.start_op("Restoring stash", restore) + + async def branches_flow(self) -> None: + if not self.idle_or_warn(): + return + st = self.state + remote = self.remote_name() if st.remotes else None + res = await self.push_screen_wait(BranchDialog(self.git, st.branch, remote)) + if not res: + return + action, name, remote_ref = res + if action == "create": + self.start_op("Creating branch", lambda: self.create_branch(name)) + else: + self.start_op("Switching branch", lambda: self.switch_branch(name, remote_ref)) + + async def create_branch(self, name: str) -> None: + r = await self.bg("check-ref-format", "--branch", name) + if not r.ok: + self.notify(f"'{name}' isn't a valid branch name.", severity="error") + return + r = await self.gitop("switch", "-c", name) + if r.ok: + self.notify(f"Created and switched to '{name}'.") + else: + await self.fail("Couldn't create the branch", r) + + async def switch_branch(self, name: str, remote_ref: Optional[str]) -> None: + st = self.state + if self.blocked_by_merge(st): + return + args = ["switch", "--track", remote_ref] if remote_ref else ["switch", name] + r = await self.gitop(*args) + if r.ok: + self.notify(f"Switched to '{name}'.") + return + if not st.files: + await self.fail("Couldn't switch branch", r) + return + choice = await self.ask( + "Your changes are in the way", + f"Switching to '{name}' would overwrite some of your uncommitted changes.\n\n" + "Stash them first? They'll be saved under Stashed changes so you can bring them back " + "later, on any branch.", + [("stash", "Stash & switch", "primary"), ("close", "Cancel", "default")]) + if choice != "stash" or not await self.stash_push(f"gitdesk: changes from {st.branch}"): + return + r = await self.gitop(*args) + if r.ok: + self.notify(f"Switched to '{name}'. Your changes are saved in Stashed changes.", timeout=8) + else: + await self.fail("Couldn't switch branch", r) + + async def open_repo_flow(self, first: bool = False) -> None: + if self.busy: + self.notify("Still busy - try again in a moment.", severity="warning") + return + current = self.git.path if self.git else os.path.abspath(self.start_path or os.getcwd()) + error = f"{current} isn't a git repository." if first and self.start_path and self.git is None else "" + while True: + choice = await self.push_screen_wait(RepoDialog(current, first and self.git is None, error)) + if choice == "add_local": + if await self.add_local_flow(): + return + continue + if choice == "new": + if await self.new_repo_flow(): + return + continue + if choice == "clone": + if await self.clone_flow(): + return + continue + if not choice: + if first and self.git is None: + self.exit() + return + if await self.load_repo(choice): + self.run_worker(self.auto_fetch(), exit_on_error=False) + return + current, error = choice, f"{choice} isn't a git repository." + + # ---- new / clone / remote --------------------------------------------- + async def choose_folder(self, title: str, start: str) -> Optional[str]: + """System folder picker (Explorer on Windows); the in-terminal browser if there isn't one.""" + start = existing_folder(start or "") + status, path = await asyncio.to_thread(pick_folder_native, title, start) + if status == "ok": + return path + if status == "cancel": + return None + return await self.push_screen_wait(FolderBrowser(title, start)) + + def action_repositories(self) -> None: + self.run_worker(self.open_repo_flow(), exit_on_error=False) + + async def add_local_flow(self) -> bool: + if self.busy: + self.notify("Still busy - try again in a moment.", severity="warning") + return False + res = await self.push_screen_wait(AddLocalDialog()) + if not res: + return False + action, path = res + if action == "create_here": + return await self.new_repo_flow(parent=os.path.dirname(path), name=os.path.basename(path)) + if await self.load_repo(path): + self.run_worker(self.auto_fetch(), exit_on_error=False) + self.notify(f"Added {os.path.basename(self.git.path)}.") + return True + await self.error_dialog("Couldn't open the repository", path) + return False + + def action_new_repo(self) -> None: + self.run_worker(self.new_repo_flow(), exit_on_error=False) + + def action_clone_repo(self) -> None: + self.run_worker(self.clone_flow(), exit_on_error=False) + + def action_add_remote(self) -> None: + self.run_worker(self.add_remote_flow(), exit_on_error=False) + + async def new_repo_flow(self, parent: Optional[str] = None, name: str = "") -> bool: + if self.busy: + self.notify("Still busy - try again in a moment.", severity="warning") + return False + o = await self.push_screen_wait(NewRepoDialog(parent or default_parent_folder(), name)) + if not o: + return False + remember_parent_folder(o["parent"]) + path = o["path"] + if os.path.isdir(os.path.join(path, ".git")): + choice = await self.ask("That's already a repository", f"{path} is already a git repository. Open it?", + [("open", "Open it", "primary"), ("cancel", "Cancel", "default")]) + return choice == "open" and await self.load_repo(path) + self.set_busy(True, "Creating repository") + try: + created, problem = await asyncio.to_thread(create_repository, o) + finally: + self.set_busy(False) + if not created: + await self.error_dialog("Couldn't create the repository", problem) + return False + if not await self.load_repo(path): + await self.error_dialog("Couldn't open the new repository", path) + return False + if problem: + hint = hint_for(problem) + await self.error_dialog("The repository was created, but there was a problem", + problem + (f"\n\n{hint}" if hint else "")) + else: + self.notify(f"Created {o['name']}.") + return True + + def _progress_from_thread(self, line: str) -> None: + self.call_from_thread(self._show_progress, line) + + def _show_progress(self, line: str) -> None: + m = re.search(r"([A-Za-z][A-Za-z ]+?):\s+(\d+)%", line) + self.busy_detail = f"{m.group(1)} {m.group(2)}%" if m else line[:40] + self.update_remote_button() + + async def clone_flow(self) -> bool: + if self.busy: + self.notify("Still busy - try again in a moment.", severity="warning") + return False + res = await self.push_screen_wait(CloneDialog(default_parent_folder())) + if not res: + return False + url, path = res + path = os.path.abspath(path) + parent = os.path.dirname(path) + try: + os.makedirs(parent, exist_ok=True) + except OSError as e: + await self.error_dialog("Couldn't create the folder", str(e)) + return False + remember_parent_folder(parent) + args = ["clone", "--progress", url, path] + while True: + self.busy_detail = "Connecting…" + self.set_busy(True, "Cloning") + try: + r = await asyncio.to_thread(run_git_streaming, parent, args, self._progress_from_thread) + finally: + self.busy_detail = "" + self.set_busy(False) + self._log_cmd(args, r) + if r.ok: + break + tail = "\n".join(r.message.splitlines()[-10:]) + body = Text.assemble((tail, "#f97583"), "\n\n", + ("If the server needs a password or token, an SSH passphrase, or wants you to " + "confirm its key, 'Retry in terminal' lets git ask you.", C_MUTED)) + choice = await self.ask("Couldn't clone the repository", body, + [("retry", "Try again", "default"), ("terminal", "Retry in terminal", "primary"), + ("close", "Close", "default")], cancel="close") + if choice == "terminal": + if self.run_in_terminal(["clone", url, path], cwd=parent): + break + return False + if choice != "retry": + return False + if not await self.load_repo(path): + await self.error_dialog("Cloned, but couldn't open it", path) + return False + self.last_fetch = time.time() + self.update_remote_button() + self.notify(f"Cloned {os.path.basename(path)}.") + return True + + async def add_remote_flow(self) -> None: + if not self.idle_or_warn(): + return + st = self.state + name = self.remote_name() if st.remotes else "origin" + current = st.remote_urls.get(name, "") + url = await self.push_screen_wait(InputDialog( + "Remote URL", + f"Paste the address of this repository on GitHub, Gitea, etc. It's saved as the '{name}' remote.\n" + "If it doesn't exist there yet, create an empty repository on the website first " + "(without a README), then paste its URL here.", + "Save", placeholder="e.g. git@github.com:you/project.git", value=current)) + if url is None or not url.strip() or url.strip() == current: + return + url = url.strip() + + async def flow(): + args = ["remote", "set-url", name, url] if name in st.remotes else ["remote", "add", name, url] + r = await self.gitop(*args) + if r.ok: + self._web_cache_key = None + self.notify(f"Saved the {name} remote." + ("" if st.has_upstream else + " Click Publish branch to push your commits.")) + else: + await self.fail("Couldn't save the remote", r) + self.start_op("Saving remote", flow) + + async def actions_flow(self) -> None: + action = await self.push_screen_wait(ActionsDialog()) + if action: + await self.run_action(action) + + +def main() -> None: + if shutil.which("git") is None: + print("gitdesk: git isn't installed or isn't in your PATH.") + sys.exit(1) + ver = git_version() + if ver < (2, 25): + print(f"gitdesk needs git 2.25 or newer (found {ver[0]}.{ver[1]}). Please update git.") + sys.exit(1) + start = sys.argv[1] if len(sys.argv) > 1 else None + if start in ("-h", "--help"): + print(__doc__) + return + GitDesk(start).run() + + +if __name__ == "__main__": + main() diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..3ecd026 --- /dev/null +++ b/requirements.txt @@ -0,0 +1 @@ +textual>=8.2