-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtracker.py
More file actions
108 lines (87 loc) · 4.05 KB
/
tracker.py
File metadata and controls
108 lines (87 loc) · 4.05 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
import os
import subprocess
import concurrent.futures # For speed with 100s of repos
from rich.console import Console
from rich.table import Table
from rich.progress import Progress
# Search depth: 1 = same folder only, 2 = one folder deep, etc.
MAX_DEPTH = 3
# Folder to search (default is where the script is)
ROOT_DIR = os.path.dirname(os.path.abspath(__file__))
console = Console()
def is_git_repo(path):
"""Checks if a path is a git repo (handles folders, files, and bare repos)."""
# Standard or Worktree/Submodule
if os.path.exists(os.path.join(path, ".git")):
return True
# Bare repository (contains HEAD and config but no .git folder)
if os.path.exists(os.path.join(path, "HEAD")) and os.path.exists(os.path.join(path, "config")):
return True
return False
def run_git_cmd(repo_path, args):
try:
result = subprocess.run(
["git", "-C", repo_path] + args,
capture_output=True, text=True, check=True, timeout=5
)
return result.stdout.strip()
except Exception:
return None
def get_single_repo_data(path):
"""Gathers data for one specific repository."""
name = os.path.basename(path)
# Check if we are in a subfolder, show relative path if so
rel_path = os.path.relpath(path, ROOT_DIR)
branch = run_git_cmd(path, ["rev-parse", "--abbrev-ref", "HEAD"])
status = run_git_cmd(path, ["status", "--porcelain"])
unpushed = run_git_cmd(path, ["log", "@{u}..HEAD", "--oneline"])
stashes = run_git_cmd(path, ["stash", "list"])
return {
"name": rel_path if rel_path != "." else name,
"branch": branch or "N/A",
"changes": len(status.splitlines()) if status else 0,
"unpushed": len(unpushed.splitlines()) if unpushed else 0,
"stashes": len(stashes.splitlines()) if stashes else 0
}
def find_repos(root, max_depth):
"""Finds all git repositories up to a certain depth."""
repos = []
root = os.path.abspath(root)
base_depth = root.count(os.sep)
for dirpath, dirnames, filenames in os.walk(root):
current_depth = dirpath.count(os.sep) - base_depth
if current_depth >= max_depth:
del dirnames[:] # Don't go deeper
continue
if is_git_repo(dirpath):
repos.append(dirpath)
del dirnames[:] # Don't search inside a repo for other repos
return repos
def main():
console.print(f"[bold blue]Scanning for Repositories in:[/bold blue] {ROOT_DIR}\n")
repo_paths = find_repos(ROOT_DIR, MAX_DEPTH)
if not repo_paths:
console.print("[bold red]No repositories found![/bold red]")
return
table = Table(title=f"Tracker: {len(repo_paths)} Repositories Found")
table.add_column("Repository", style="cyan", no_wrap=True)
table.add_column("Branch", style="magenta")
table.add_column("Changes", justify="right")
table.add_column("Unpushed", justify="right")
table.add_column("Stashes", justify="right")
# Use ThreadPoolExecutor to run Git commands in parallel (much faster for 100+ repos)
with Progress() as progress:
task = progress.add_task("[green]Checking status...", total=len(repo_paths))
with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
future_to_repo = {executor.submit(get_single_repo_data, path): path for path in repo_paths}
for future in concurrent.futures.as_completed(future_to_repo):
data = future.result()
# Format the row
change_str = f"[bold red]Yes ({data['changes']})" if data['changes'] > 0 else "[green]Clean"
push_str = f"[bold yellow]{data['unpushed']}" if data['unpushed'] > 0 else "[dim]0"
stash_str = f"{data['stashes']}" if data['stashes'] > 0 else "[dim]0"
table.add_row(data['name'], data['branch'], change_str, push_str, stash_str)
progress.update(task, advance=1)
console.print(table)
if __name__ == "__main__":
main()