tracker.py

import os
import sys
import json
import csv
import argparse
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

# Folder to search (default is where the script is)
DEFAULT_ROOT_DIR = os.path.dirname(os.path.abspath(__file__))
DEFAULT_MAX_DEPTH = 3

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, root_dir):
    """Gathers data for one specific repository."""
    name = os.path.basename(path)
    # Check if we are in a subfolder, show relative path if so
    try:
        rel_path = os.path.relpath(path, root_dir)
    except ValueError:
        rel_path = path
    
    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"])
    
    changes_count = len(status.splitlines()) if status else 0
    unpushed_count = len(unpushed.splitlines()) if unpushed else 0
    stashes_count = len(stashes.splitlines()) if stashes else 0

    return {
        "name": rel_path if rel_path != "." else name,
        "path": path,
        "branch": branch or "N/A",
        "changes": changes_count,
        "unpushed": unpushed_count,
        "stashes": stashes_count,
        "is_clean": changes_count == 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 export_data(data, filepath):
    """Exports repository data to JSON, CSV, MD, or TXT format."""
    ext = os.path.splitext(filepath)[1].lower()
    
    # Ensure export directory exists if path specified
    dirname = os.path.dirname(os.path.abspath(filepath))
    if dirname and not os.path.exists(dirname):
        os.makedirs(dirname, exist_ok=True)

    if ext == ".json":
        clean_data = [
            {
                "name": item["name"],
                "branch": item["branch"],
                "changes": item["changes"],
                "unpushed": item["unpushed"],
                "stashes": item["stashes"],
                "status": "Clean" if item["is_clean"] else "Dirty"
            }
            for item in data
        ]
        with open(filepath, "w", encoding="utf-8") as f:
            json.dump(clean_data, f, indent=2)
            
    elif ext == ".csv":
        with open(filepath, "w", newline="", encoding="utf-8") as f:
            writer = csv.writer(f)
            writer.writerow(["Repository", "Branch", "Status", "Changes", "Unpushed", "Stashes"])
            for item in data:
                status = "Clean" if item["is_clean"] else "Dirty"
                writer.writerow([item["name"], item["branch"], status, item["changes"], item["unpushed"], item["stashes"]])
                
    elif ext in [".md", ".txt"]:
        with open(filepath, "w", encoding="utf-8") as f:
            f.write("| Repository | Branch | Status | Changes | Unpushed | Stashes |\n")
            f.write("|---|---|---|---|---|---|\n")
            for item in data:
                status = "Clean" if item["is_clean"] else "Dirty"
                f.write(f"| {item['name']} | {item['branch']} | {status} | {item['changes']} | {item['unpushed']} | {item['stashes']} |\n")
    else:
        console.print(f"[bold red]Unsupported export format: {ext}. Supported: .json, .csv, .md, .txt[/bold red]")
        return False

    console.print(f"[bold green]Successfully exported {len(data)} repo records to:[/bold green] {filepath}")
    return True

def main():
    parser = argparse.ArgumentParser(description="Multi-Repo Command Center Tracker")
    parser.add_argument(
        "-f", "--filter",
        choices=["all", "clean", "red", "dirty", "unpushed"],
        default="all",
        help="Filter repositories by status: 'clean' (green/no changes), 'red'/'dirty' (has uncommitted changes), 'unpushed', or 'all' (default: all)"
    )
    parser.add_argument(
        "-e", "--export",
        metavar="FILEPATH",
        help="Export summary to file (.csv, .json, .md, .txt)"
    )
    parser.add_argument(
        "-p", "--path",
        default=DEFAULT_ROOT_DIR,
        help=f"Root directory to search (default: {DEFAULT_ROOT_DIR})"
    )
    parser.add_argument(
        "-d", "--depth",
        type=int,
        default=DEFAULT_MAX_DEPTH,
        help=f"Max directory search depth (default: {DEFAULT_MAX_DEPTH})"
    )

    args = parser.parse_args()

    search_path = os.path.abspath(args.path)
    console.print(f"[bold blue]Scanning for Repositories in:[/bold blue] {search_path} (Filter: [cyan]{args.filter}[/cyan])\n")
    
    repo_paths = find_repos(search_path, args.depth)
    
    if not repo_paths:
        console.print("[bold red]No repositories found![/bold red]")
        return

    all_repo_data = []

    # Use ThreadPoolExecutor to run Git commands in parallel
    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, search_path): path for path in repo_paths}
            
            for future in concurrent.futures.as_completed(future_to_repo):
                data = future.result()
                all_repo_data.append(data)
                progress.update(task, advance=1)

    # Sort repo data by name
    all_repo_data.sort(key=lambda x: x["name"])

    # Apply Filter
    filter_mode = args.filter.lower()
    if filter_mode == "clean":
        filtered_data = [r for r in all_repo_data if r["is_clean"]]
    elif filter_mode in ["red", "dirty"]:
        filtered_data = [r for r in all_repo_data if not r["is_clean"]]
    elif filter_mode == "unpushed":
        filtered_data = [r for r in all_repo_data if r["unpushed"] > 0]
    else:
        filtered_data = all_repo_data

    table = Table(title=f"Tracker: {len(filtered_data)} of {len(all_repo_data)} Repositories Shown (Filter: {args.filter})")
    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")

    for data in filtered_data:
        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)

    console.print(table)

    if args.export:
        export_data(filtered_data, args.export)

if __name__ == "__main__":
    main()