tracker_cached.py

import os
import sys
import json
import argparse
import datetime
import concurrent.futures
from rich.console import Console
from rich.progress import Progress

from tracker import (
    DEFAULT_ROOT_DIR,
    DEFAULT_MAX_DEPTH,
    console,
    find_repos,
    parse_exclude_patterns,
    get_single_repo_data,
    export_data,
    filter_repos,
    sort_repos,
    render_table
)

DEFAULT_CACHE_FILE = os.path.join(DEFAULT_ROOT_DIR, ".tracker_cache.json")

def load_cache(cache_file):
    """Loads cached repository data if it exists and is valid."""
    if not os.path.exists(cache_file):
        return None
    try:
        with open(cache_file, "r", encoding="utf-8") as f:
            return json.load(f)
    except Exception as e:
        console.print(f"[dim yellow]Warning: Failed to read cache ({e}). A fresh scan will run.[/dim yellow]")
        return None

def save_cache(cache_file, repos, search_path, depth, exclude=None):
    """Saves repository data and scan metadata to cache file."""
    cache_payload = {
        "timestamp": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        "search_path": os.path.abspath(search_path),
        "depth": depth,
        "exclude": list(sorted(exclude)) if exclude else [],
        "total_repos": len(repos),
        "repos": repos
    }
    try:
        dirname = os.path.dirname(os.path.abspath(cache_file))
        if dirname and not os.path.exists(dirname):
            os.makedirs(dirname, exist_ok=True)
        with open(cache_file, "w", encoding="utf-8") as f:
            json.dump(cache_payload, f, indent=2)
    except Exception as e:
        console.print(f"[dim yellow]Warning: Could not save cache to {cache_file} ({e})[/dim yellow]")

def scan_repositories(search_path, depth, exclude=None):
    """Scans the disk for git repositories and gathers git statuses in parallel, skipping excluded folders."""
    console.print(f"[bold blue]Scanning for Repositories in:[/bold blue] {search_path}")
    if exclude:
        console.print(f"[dim]Excluded folder patterns: {', '.join(sorted(exclude))}[/dim]")
    console.print()

    repo_paths = find_repos(search_path, depth, exclude=exclude)

    if not repo_paths:
        console.print("[bold red]No repositories found![/bold red]")
        return []

    all_repo_data = []
    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)

    all_repo_data.sort(key=lambda x: x["name"])
    return all_repo_data

def main():
    parser = argparse.ArgumentParser(
        description="Multi-Repo Command Center Tracker with Instant Cache Filtering"
    )
    parser.add_argument(
        "-f", "--filter",
        choices=["all", "clean", "red", "dirty", "unpushed", "behind"],
        default="all",
        help="Filter repositories: '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(
        "-r", "--refresh",
        action="store_true",
        help="Force a fresh scan of repositories and update cache"
    )
    parser.add_argument(
        "-c", "--cached",
        action="store_true",
        help="Strictly read from cache (exits with error if no cache exists)"
    )
    parser.add_argument(
        "--no-cache",
        action="store_true",
        help="Run live scan without reading or writing cache"
    )
    parser.add_argument(
        "--clear-cache",
        action="store_true",
        help="Delete existing cache file and exit"
    )
    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)"
    )
    parser.add_argument(
        "--cache-file",
        default=DEFAULT_CACHE_FILE,
        help=f"Custom cache file path (default: {DEFAULT_CACHE_FILE})"
    )

    args = parser.parse_args()

    # Handle clearing cache
    if args.clear_cache:
        if os.path.exists(args.cache_file):
            try:
                os.remove(args.cache_file)
                console.print(f"[bold green]Successfully deleted cache file:[/bold green] {args.cache_file}")
            except Exception as e:
                console.print(f"[bold red]Failed to delete cache file:[/bold red] {e}")
        else:
            console.print(f"[dim yellow]No cache file found at:[/dim yellow] {args.cache_file}")
        return

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

    # Check cache if not explicitly refreshing or disabling cache
    if not args.refresh and not args.no_cache:
        cache_data = load_cache(args.cache_file)
        if cache_data:
            # If a custom path was provided and doesn't match cache, warn unless strictly --cached
            cached_path = cache_data.get("search_path")
            cached_exclude = set(cache_data.get("exclude", []))

            if cached_path and cached_path.lower() != search_path.lower() and not args.cached:
                console.print(
                    f"[dim yellow]Cached path ({cached_path}) does not match requested path ({search_path}). "
                    f"Running fresh scan...[/dim yellow]\n"
                )
            elif (args.exclude or exclude_patterns) and cached_exclude != exclude_patterns and not args.cached:
                console.print(
                    f"[dim yellow]Exclude settings changed since last cached scan. "
                    f"Running fresh scan...[/dim yellow]\n"
                )
            else:
                all_repo_data = cache_data.get("repos", [])
                timestamp = cache_data.get("timestamp", "Unknown time")
                console.print(
                    f"[bold green][CACHE] Loaded {len(all_repo_data)} repo(s) from cache[/bold green] "
                    f"[dim](scanned: {timestamp})[/dim]\n"
                    f"[dim italic]Tip: Run with --refresh (-r) whenever you want a fresh scan.[/dim italic]\n"
                )
        elif args.cached:
            console.print(f"[bold red]Error: No cache file found at {args.cache_file}.[/bold red] Run without --cached first.")
            return

    # If no cached data available or refresh requested, run scan
    if all_repo_data is None:
        all_repo_data = scan_repositories(search_path, args.depth, exclude=exclude_patterns)
        if not all_repo_data:
            return
        if not args.no_cache:
            save_cache(args.cache_file, all_repo_data, search_path, args.depth, exclude=exclude_patterns)
            console.print(f"[dim green]Saved scan results to cache: {args.cache_file}[/dim green]\n")

    # 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)

    # Export if requested
    if args.export:
        export_data(sorted_data, args.export)

if __name__ == "__main__":
    main()