commit 5823ea71113deea163fbaff099653aa96c65a409 Author: jzzliu Date: Sat Nov 7 20:06:45 2026 -0700 Build Git Scout repository monitor diff --git a/README.md b/README.md new file mode 100644 index 0000000..49d31ab --- /dev/null +++ b/README.md @@ -0,0 +1,34 @@ +# Git Scout + +Git Scout monitors local Git working trees and remote default branches. It uses +Python 3.9+ and Git, with no additional packages. + +```sh +python3 git_scout.py ~/Projects +python3 git_scout.py ~/Projects --needs-attention +python3 git_scout.py ~/Projects --gitea-imac --record +``` + +Local scans include linked worktrees and skip hidden subdirectories. The +attention view shows modified or untracked paths, ahead or behind counts, and +errors. Ahead and behind counts use locally recorded upstream refs. + +Use `--remote URL` repeatedly to include other remotes. Remote checks read the +default branch and commit without fetching objects. The `--gitea-imac` shortcut +uses the iMac's existing SSH alias to check `personal-site` and `python-project`. + +## Compare scans + +Run the same scan twice with `--record`, then use: + +```sh +python3 git_scout.py ~/Projects --history +python3 git_scout.py ~/Projects --compare +``` + +Snapshots are grouped by absolute scan directory. Keep the same remote options +across recordings to compare the same set of targets. SQLite history is stored +at `~/.local/state/git-scout/history.sqlite3` by default. Use `--db PATH` to +choose a different location. Snapshots record Git status and commit IDs, not +file contents. A scan with an error returns status 1 and still saves the +results when `--record` is set. diff --git a/git_scout.py b/git_scout.py new file mode 100644 index 0000000..cbc90ba --- /dev/null +++ b/git_scout.py @@ -0,0 +1,222 @@ +"""Scan Git working trees and compare recorded snapshots.""" + +import argparse +from concurrent.futures import ThreadPoolExecutor +from datetime import datetime, timezone +import json +import os +from pathlib import Path +import sqlite3 +import subprocess +import sys + + +def working_trees(root): + """Find working trees, including linked worktrees with a .git file.""" + for directory, subdirs, files in os.walk(root): + if ".git" in subdirs or ".git" in files: + yield Path(directory) + subdirs[:] = sorted(name for name in subdirs if not name.startswith(".")) + + +def git(*args): + env = dict(os.environ, GIT_TERMINAL_PROMPT="0") + env.setdefault("GIT_SSH_COMMAND", "ssh -o BatchMode=yes") + try: + result = subprocess.run( + ["git", "--no-optional-locks", *args], + capture_output=True, + text=True, + errors="replace", + timeout=15, + env=env, + ) + except subprocess.TimeoutExpired: + return "", "Git command timed out after 15 seconds" + except FileNotFoundError: + return "", "Git executable not found" + if result.returncode: + return "", result.stderr.strip() or f"Git exited with status {result.returncode}" + return result.stdout, None + + +def inspect_tree(repo): + output, error = git("-C", str(repo), "status", "--porcelain=v2", "--branch") + if error: + return {"error": error} + headers = {} + changes = [] + for line in output.splitlines(): + if line.startswith("# "): + name, _, value = line[2:].partition(" ") + headers[name] = value + else: + changes.append(line) + + ahead, behind = map(int, headers.get("branch.ab", "+0 -0").split()) + return { + "branch": headers.get("branch.head"), + "head": headers.get("branch.oid"), + "upstream": headers.get("branch.upstream"), + "ahead": ahead, + "behind": abs(behind), + "changes": changes, + } + + +def inspect_remote(url): + output, error = git("ls-remote", "--symref", "--", url, "HEAD") + if error: + return {"error": error} + branch = None + head = None + for line in output.splitlines(): + ref, _, name = line.partition("\t") + if name == "HEAD": + if ref.startswith("ref: "): + branch = ref.removeprefix("ref: refs/heads/") + else: + head = ref + return {"branch": branch, "head": head} + + +def scan(root, remotes): + tasks = [ + (f"local {repo.relative_to(root)}", inspect_tree, repo) + for repo in working_trees(root) + ] + tasks.extend((f"remote {url}", inspect_remote, url) for url in dict.fromkeys(remotes)) + with ThreadPoolExecutor(max_workers=8) as pool: + pending = [(label, pool.submit(inspect, value)) for label, inspect, value in tasks] + return {label: future.result() for label, future in pending} + + +def summarize(state): + if "error" in state: + return f"error: {state['error']}" + branch = state["branch"] or "no default branch" + head = state["head"] + summary = f"{branch} @ {head[:10]}" if head and head != "(initial)" else f"{branch} (no commits)" + if "changes" in state: + changes = state["changes"] + summary += f" | {len(changes)} changed paths" if changes else " | clean" + if state["upstream"]: + summary += f" | {state['upstream']} +{state['ahead']}/-{state['behind']}" + return summary + + +def report(snapshot, attention_only=False): + for label, state in snapshot.items(): + if attention_only and not any( + state.get(field) for field in ("error", "changes", "ahead", "behind") + ): + continue + print(f"{label}: {summarize(state)}") + + +def compare(before, after): + changed = False + for label in sorted(before.keys() | after.keys()): + old = before.get(label) + new = after.get(label) + if old == new: + continue + changed = True + if old is None: + print(f"+ {label}: {summarize(new)}") + elif new is None: + print(f"- {label}: {summarize(old)}") + else: + fields = ", ".join( + field for field in sorted(old.keys() | new.keys()) + if old.get(field) != new.get(field) + ) + print(f"~ {label} [{fields}]") + print(f" before: {summarize(old)}") + print(f" after: {summarize(new)}") + if not changed: + print("No changes between the last two snapshots.") + + +def snapshot_history(path, root, snapshot=None): + if snapshot is not None: + path.parent.mkdir(parents=True, exist_ok=True) + elif not path.is_file(): + return [] + with sqlite3.connect(path) as db: + db.execute(""" + CREATE TABLE IF NOT EXISTS snapshots ( + id INTEGER PRIMARY KEY, + root TEXT NOT NULL, + recorded_at TEXT NOT NULL, + data TEXT NOT NULL + ) + """) + db.execute("CREATE INDEX IF NOT EXISTS snapshots_root ON snapshots(root, id)") + if snapshot is not None: + recorded_at = datetime.now(timezone.utc).isoformat(timespec="seconds") + db.execute( + "INSERT INTO snapshots (root, recorded_at, data) VALUES (?, ?, ?)", + (str(root), recorded_at, json.dumps(snapshot)), + ) + return [] + return db.execute( + "SELECT id, recorded_at, data FROM snapshots WHERE root = ? ORDER BY id DESC LIMIT 20", + (str(root),), + ).fetchall() + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("directory", nargs="?", type=Path, default=Path.cwd()) + parser.add_argument("--db", type=Path, default=Path.home() / ".local/state/git-scout/history.sqlite3") + mode = parser.add_mutually_exclusive_group() + mode.add_argument("--record", action="store_true", help="save the current scan") + mode.add_argument("--history", action="store_true", help="list the latest 20 saved scans") + mode.add_argument("--compare", action="store_true", help="compare the latest two saved scans") + parser.add_argument("--remote", action="append", default=[], metavar="URL", help="check a remote HEAD (repeatable)") + parser.add_argument("--gitea-imac", action="store_true", help="check personal-site and python-project via SSH") + parser.add_argument("--needs-attention", action="store_true", help="show changes, divergence, and errors") + args = parser.parse_args() + if (args.history or args.compare) and (args.remote or args.gitea_imac or args.needs_attention): + parser.error("remote and attention options apply only to live scans") + + root = args.directory.expanduser().resolve() + db_path = args.db.expanduser() + try: + if args.history or args.compare: + rows = snapshot_history(db_path, root) + if args.compare: + if len(rows) < 2: + parser.error("record two snapshots of this directory first") + compare(json.loads(rows[1][2]), json.loads(rows[0][2])) + elif rows: + for row_id, recorded_at, data in rows: + print(f"{row_id} {recorded_at} {len(json.loads(data))} targets") + else: + print(f"No snapshots recorded for {root}") + return 0 + + if not root.is_dir(): + parser.error(f"directory does not exist: {root}") + remotes = args.remote + if args.gitea_imac: + remotes.extend( + f"gitea-imac:jzzliu/{name}.git" for name in ("personal-site", "python-project") + ) + snapshot = scan(root, remotes) + if snapshot: + report(snapshot, args.needs_attention) + else: + print(f"No Git working trees found under {root}") + if args.record: + snapshot_history(db_path, root, snapshot) + print(f"Snapshot saved to {db_path}") + return int(any("error" in state for state in snapshot.values())) + except (OSError, sqlite3.Error) as error: + print(f"Git Scout: {error}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + sys.exit(main())