223 lines
8.0 KiB
Python
223 lines
8.0 KiB
Python
"""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())
|