tracker.py

import os
import sys
import re
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
from rich.panel import Panel

# 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"])
    behind = run_git_cmd(path, ["log", "HEAD..@{u}", "--oneline"])
    stashes = run_git_cmd(path, ["stash", "list"])
    commits_raw = run_git_cmd(path, ["rev-list", "--count", "HEAD"])
    loc_raw = run_git_cmd(path, ["diff", "--shortstat", "4b825dc642cb6eb9a060e54bf8d69288fbee4904", "HEAD"])
    last_active = run_git_cmd(path, ["log", "-1", "--format=%cr"])
    
    changes_count = len(status.splitlines()) if status else 0
    unpushed_count = len(unpushed.splitlines()) if unpushed else 0
    behind_count = len(behind.splitlines()) if behind else 0
    stashes_count = len(stashes.splitlines()) if stashes else 0
    commits_count = int(commits_raw) if (commits_raw and commits_raw.isdigit()) else 0

    loc_count = 0
    if loc_raw:
        match = re.search(r"(\d+)\s+insertion", loc_raw)
        if match:
            loc_count = int(match.group(1))

    return {
        "name": rel_path if rel_path != "." else name,
        "path": path,
        "branch": branch or "N/A",
        "changes": changes_count,
        "commits": commits_count,
        "loc": loc_count,
        "unpushed": unpushed_count,
        "behind": behind_count,
        "stashes": stashes_count,
        "last_active": last_active or "No commits",
        "is_clean": changes_count == 0
    }

def parse_exclude_patterns(exclude_list=None, root_dir=None):
    """Parses folder exclusion names/patterns from CLI arguments and optional .trackerignore file."""
    patterns = set()
    
    # Check for .trackerignore file in target root or working directory
    candidate_files = []
    if root_dir:
        candidate_files.append(os.path.join(root_dir, ".trackerignore"))
    cwd_file = os.path.join(os.getcwd(), ".trackerignore")
    if cwd_file not in candidate_files:
        candidate_files.append(cwd_file)

    for ifile in candidate_files:
        if os.path.exists(ifile):
            try:
                with open(ifile, "r", encoding="utf-8") as f:
                    for line in f:
                        line = line.strip()
                        if line and not line.startswith("#"):
                            patterns.add(line.replace("\\", "/").strip("/").lower())
            except Exception:
                pass

    if exclude_list:
        for item in exclude_list:
            for part in item.split(","):
                clean = part.strip().replace("\\", "/").strip("/")
                if clean:
                    patterns.add(clean.lower())

    return patterns

def find_repos(root, max_depth, exclude=None):
    """Finds all git repositories up to a certain depth, skipping excluded folders."""
    repos = []
    root = os.path.abspath(root)
    base_depth = root.count(os.sep)
    exclude_patterns = set(p.lower().replace("\\", "/").strip("/") for p in (exclude or []))
    
    for dirpath, dirnames, filenames in os.walk(root):
        rel_dir = os.path.relpath(dirpath, root).replace("\\", "/").lower()
        dir_name = os.path.basename(dirpath).lower()

        # If current directory itself is excluded, skip it and all its children
        if exclude_patterns:
            if dir_name in exclude_patterns or (rel_dir != "." and (rel_dir in exclude_patterns or any(rel_dir.startswith(pat + "/") for pat in exclude_patterns))):
                del dirnames[:]
                continue

        # Prune excluded subdirectories so os.walk does not enter them
        if exclude_patterns:
            filtered = []
            for d in dirnames:
                full_child = os.path.join(dirpath, d)
                rel_child = os.path.relpath(full_child, root).replace("\\", "/").lower()
                d_lower = d.lower()
                
                is_excluded = False
                for pat in exclude_patterns:
                    if d_lower == pat or rel_child == pat or rel_child.startswith(pat + "/"):
                        is_excluded = True
                        break
                
                if not is_excluded:
                    filtered.append(d)
            dirnames[:] = filtered

        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 filter_repos(repo_list, filter_mode):
    """Filters repository list by status."""
    mode = filter_mode.lower()
    if mode == "clean":
        return [r for r in repo_list if r.get("is_clean", False)]
    elif mode in ["red", "dirty"]:
        return [r for r in repo_list if not r.get("is_clean", True)]
    elif mode == "unpushed":
        return [r for r in repo_list if r.get("unpushed", 0) > 0]
    elif mode == "behind":
        return [r for r in repo_list if r.get("behind", 0) > 0]
    return repo_list

def sort_repos(repo_list, sort_by):
    """Sorts repository list by specified column."""
    if not sort_by:
        return sorted(repo_list, key=lambda x: x["name"].lower())
    
    sort_key = sort_by.lower()
    if sort_key in ["commits", "loc", "changes", "unpushed", "behind", "stashes"]:
        return sorted(repo_list, key=lambda x: x.get(sort_key, 0), reverse=True)
    elif sort_key == "name":
        return sorted(repo_list, key=lambda x: x["name"].lower())
    elif sort_key == "branch":
        return sorted(repo_list, key=lambda x: x.get("branch", "").lower())
    return repo_list

def render_table(filtered_data, total_count, filter_mode):
    """Renders the formatted rich summary table and KPI summary panel."""
    table = Table(title=f"Tracker: {len(filtered_data)} of {total_count} Repositories Shown (Filter: {filter_mode})")
    table.add_column("Repository", style="cyan", no_wrap=True)
    table.add_column("Branch", style="magenta")
    table.add_column("Changes", justify="right")
    table.add_column("Commits", justify="right", style="green")
    table.add_column("LOC", justify="right", style="blue")
    table.add_column("Unpushed", justify="right")
    table.add_column("Behind", justify="right")
    table.add_column("Stashes", justify="right")
    table.add_column("Last Active", style="dim")

    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"
        behind_str = f"[bold cyan]{data.get('behind', 0)}" if data.get('behind', 0) > 0 else "[dim]0"
        stash_str = f"{data['stashes']}" if data['stashes'] > 0 else "[dim]0"
        commits_str = f"{data.get('commits', 0):,}"
        loc_str = f"{data.get('loc', 0):,}"
        last_active_str = data.get("last_active", "N/A")

        table.add_row(
            data['name'],
            data['branch'],
            change_str,
            commits_str,
            loc_str,
            push_str,
            behind_str,
            stash_str,
            last_active_str
        )

    console.print(table)

    # KPI Summary Panel
    total_commits = sum(r.get("commits", 0) for r in filtered_data)
    total_loc = sum(r.get("loc", 0) for r in filtered_data)
    dirty_count = sum(1 for r in filtered_data if not r.get("is_clean", True))
    unpushed_count = sum(1 for r in filtered_data if r.get("unpushed", 0) > 0)
    behind_count = sum(1 for r in filtered_data if r.get("behind", 0) > 0)

    summary_text = (
        f"[bold]Summary ({len(filtered_data)} Repos):[/bold] "
        f"[green]{total_commits:,}[/green] Commits  |  "
        f"[blue]{total_loc:,}[/blue] Lines of Code  |  "
        f"[red]{dirty_count}[/red] Dirty  |  "
        f"[yellow]{unpushed_count}[/yellow] Unpushed  |  "
        f"[cyan]{behind_count}[/cyan] Behind"
    )
    console.print(Panel(summary_text, expand=False, border_style="dim blue"))

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"],
                "status": "Clean" if item["is_clean"] else "Dirty",
                "changes": item["changes"],
                "commits": item.get("commits", 0),
                "loc": item.get("loc", 0),
                "unpushed": item["unpushed"],
                "behind": item.get("behind", 0),
                "stashes": item["stashes"],
                "last_active": item.get("last_active", "N/A")
            }
            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", "Commits", "LOC", "Unpushed", "Behind", "Stashes", "Last Active"])
            for item in data:
                status = "Clean" if item["is_clean"] else "Dirty"
                writer.writerow([
                    item["name"],
                    item["branch"],
                    status,
                    item["changes"],
                    item.get("commits", 0),
                    item.get("loc", 0),
                    item["unpushed"],
                    item.get("behind", 0),
                    item["stashes"],
                    item.get("last_active", "N/A")
                ])
                
    elif ext in [".md", ".txt"]:
        with open(filepath, "w", encoding="utf-8") as f:
            f.write("| Repository | Branch | Status | Changes | Commits | LOC | Unpushed | Behind | Stashes | Last Active |\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']} | "
                    f"{item.get('commits', 0):,} | {item.get('loc', 0):,} | "
                    f"{item['unpushed']} | {item.get('behind', 0)} | {item['stashes']} | {item.get('last_active', 'N/A')} |\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", "behind"],
        default="all",
        help="Filter repositories by status: 'clean', 'red'/'dirty', 'unpushed', 'behind', or 'all' (default: all)"
    )
    parser.add_argument(
        "-s", "--sort",
        choices=["name", "commits", "loc", "changes", "unpushed", "behind", "stashes"],
        default="name",
        help="Sort repositories by field (default: name). Numeric fields sort highest-first."
    )
    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})"
    )
    parser.add_argument(
        "-x", "--exclude",
        nargs="*",
        default=[],
        help="Folder names or paths to exclude from scanning (e.g. -x venv node_modules temp)"
    )

    args = parser.parse_args()

    search_path = os.path.abspath(args.path)
    exclude_patterns = parse_exclude_patterns(args.exclude, search_path)

    console.print(f"[bold blue]Scanning for Repositories in:[/bold blue] {search_path} (Filter: [cyan]{args.filter}[/cyan], Sort: [cyan]{args.sort}[/cyan])")
    if exclude_patterns:
        console.print(f"[dim]Excluded folder patterns: {', '.join(sorted(exclude_patterns))}[/dim]")
    console.print()
    
    repo_paths = find_repos(search_path, args.depth, exclude=exclude_patterns)
    
    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)

    # Apply Filter
    filtered_data = filter_repos(all_repo_data, args.filter)

    # Apply Sort
    sorted_data = sort_repos(filtered_data, args.sort)

    # Render Table & KPI Summary Panel
    render_table(sorted_data, len(all_repo_data), args.filter)

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

if __name__ == "__main__":
    main()