MQL5-Google-Onedrive/scripts/review_working_trees.py

#!/usr/bin/env python3
"""
Working Tree Review Script
Reviews all git branches, stashes, and working trees for the repository
"""

import subprocess
import sys
from pathlib import Path
from datetime import datetime
from collections import defaultdict

REPO_ROOT = Path(__file__).resolve().parents[1]

# ⚡ Bolt: Global cache for branch metadata to avoid redundant git subprocess calls.
BRANCH_METADATA_CACHE = {}
CURRENT_BRANCH = "HEAD"


def get_git_version():
    """Get the git version as a tuple of integers."""
    res = run_git_command(["version"])
    if not res or res.returncode != 0:
        return (0, 0, 0)
    try:
        ver_str = res.stdout.strip().split()[-1]
        return tuple(int(x) for x in ver_str.split('.')[:3] if x.isdigit())
    except (IndexError, ValueError):
        return (0, 0, 0)


def populate_metadata_cache():
    """⚡ Bolt: Fetch metadata for all local and remote branches in a single bulk call."""
    global BRANCH_METADATA_CACHE, CURRENT_BRANCH

    git_ver = get_git_version()
    use_ahead_behind = git_ver >= (2, 41, 0)

    delim = "|||"
    # Use %(HEAD) to detect current branch and %(ahead-behind:origin/main) for unpushed changes detection.
    fmt = f"%(refname){delim}%(HEAD){delim}%(ahead-behind:origin/main)" if use_ahead_behind else f"%(refname){delim}%(HEAD)"

    # Fetch both local branches and remote branches
    result = run_git_command(["for-each-ref", f"--format={fmt}", "refs/heads", "refs/remotes"])
    if not result or result.returncode != 0:
        return

    for line in result.stdout.strip().split('\n'):
        if not line: continue

        parts = line.split(delim)
        ref_info = parts[0]
        is_head = (parts[1] == "*")

        # Distinguish between local and remote
        is_remote = ref_info.startswith("refs/remotes/")
        is_local = ref_info.startswith("refs/heads/")

        if is_remote:
            short_name = ref_info[len("refs/remotes/"):]
        elif is_local:
            short_name = ref_info[len("refs/heads/"):]
        else:
            continue

        if is_head:
            CURRENT_BRANCH = short_name

        metadata = {
            "short_name": short_name,
            "is_remote": is_remote,
            "is_local": is_local,
            "is_merged": False,
            "ahead": None, # Use None to indicate unknown for older Git
            "behind": None
        }

        if use_ahead_behind and len(parts) >= 3:
            counts = parts[2].split()
            if len(counts) == 2:
                ahead = int(counts[0])
                metadata["ahead"] = ahead
                metadata["behind"] = int(counts[1])
                metadata["is_merged"] = (ahead == 0)

        BRANCH_METADATA_CACHE[short_name] = metadata

    # Fallback for older Git versions
    if not use_ahead_behind:
        merged_res = run_git_command(["branch", "-a", "--merged", "main"])
        if merged_res and merged_res.returncode == 0:
            for b in merged_res.stdout.strip().split("\n"):
                b = b.strip().replace("*", "").strip()
                if b.startswith("remotes/"):
                    b = b[len("remotes/"):]
                if b in BRANCH_METADATA_CACHE:
                    BRANCH_METADATA_CACHE[b]["is_merged"] = True


def run_git_command(cmd, capture_output=True):
    """Run a git command and return the result."""
    try:
        result = subprocess.run(
            ["git"] + cmd,
            cwd=REPO_ROOT,
            capture_output=capture_output,
            text=True,
            timeout=30,
            encoding='utf-8',
            errors='replace'
        )
        return result
    except subprocess.TimeoutExpired:
        return None
    except Exception as e:
        print(f"Error running git command: {e}", file=sys.stderr)
        return None


def get_branch_info():
    """⚡ Bolt: Get information about all branches using cached metadata."""
    print("=" * 80)
    print("BRANCH REVIEW")
    print("=" * 80)
    print()
    
    # ⚡ Bolt: Use cached metadata to avoid redundant git subprocess calls.
    local_branches = sorted([m["short_name"] for m in BRANCH_METADATA_CACHE.values() if m["is_local"]])
    remote_branches = sorted([m["short_name"] for m in BRANCH_METADATA_CACHE.values() if m["is_remote"] and "HEAD" not in m["short_name"]])
    merged_branches = sorted([m["short_name"] for m in BRANCH_METADATA_CACHE.values() if m["is_remote"] and m["is_merged"] and "origin/main" not in m["short_name"] and "HEAD" not in m["short_name"]])
    unmerged_branches = sorted([m["short_name"] for m in BRANCH_METADATA_CACHE.values() if m["is_remote"] and not m["is_merged"] and "origin/main" not in m["short_name"] and "HEAD" not in m["short_name"]])

    # Local branches
    if local_branches:
        print(f"📌 Local Branches: {len(local_branches)}")
        for branch in local_branches:
            current = "*" if branch == CURRENT_BRANCH else " "
            print(f"  {current} {branch}")
    print()
    
    # Remote branches
    if remote_branches:
        print(f"🌐 Remote Branches: {len(remote_branches)}")
        
        # Group by prefix
        branch_groups = defaultdict(list)
        for branch in remote_branches:
            parts = branch.replace("origin/", "").split("/")
            prefix = parts[0] if len(parts) > 1 else "other"
            branch_groups[prefix].append(branch)
        
        for prefix, branches in sorted(branch_groups.items()):
            print(f"\n  {prefix.upper()}: {len(branches)} branches")
            for branch in branches[:5]:  # Show first 5
                print(f"    - {branch}")
            if len(branches) > 5:
                print(f"    ... and {len(branches) - 5} more")
    print()
    
    # Merged branches
    if merged_branches:
        print(f"✅ Merged into main: {len(merged_branches)} branches")
        print("   (These can potentially be deleted)")
    print()
    
    # Unmerged branches
    if unmerged_branches:
        print(f"⚠️  Not merged into main: {len(unmerged_branches)} branches")
        print("   (These may contain unmerged changes)")
    print()


def get_stash_info():
    """Get information about stashes."""
    print("=" * 80)
    print("STASH REVIEW")
    print("=" * 80)
    print()
    
    result = run_git_command(["stash", "list"])
    if result and result.returncode == 0 and result.stdout.strip():
        stashes = result.stdout.strip().split("\n")
        print(f"📦 Stashes: {len(stashes)}")
        for stash in stashes:
            print(f"  - {stash}")
    else:
        print("📦 No stashes found")
    print()


def get_worktree_info():
    """Get information about git worktrees."""
    print("=" * 80)
    print("WORKTREE REVIEW")
    print("=" * 80)
    print()
    
    result = run_git_command(["worktree", "list"])
    if result and result.returncode == 0:
        worktrees = [w.strip() for w in result.stdout.strip().split("\n") if w.strip()]
        print(f"🌳 Worktrees: {len(worktrees)}")
        for worktree in worktrees:
            print(f"  - {worktree}")
    else:
        print("🌳 No additional worktrees found")
    print()


def get_status_info():
    """⚡ Bolt: Get current working tree status in a single consolidated call."""
    print("=" * 80)
    print("WORKING TREE STATUS")
    print("=" * 80)
    print()
    
    # ⚡ Bolt: Use 'git status -sb' to get both branch status and modified files in one call.
    result = run_git_command(["status", "-sb"])
    if result and result.returncode == 0:
        lines = result.stdout.strip().split('\n')
        if not lines: return

        # First line is always branch info like '## main...origin/main [ahead 1]'
        branch_line = lines[0]
        if "ahead" in branch_line or "behind" in branch_line:
            print(f"📊 {branch_line}")

        # Remaining lines are modified files (equivalent to --short)
        modified_files = lines[1:]
        if modified_files:
            print("⚠️  Uncommitted changes:")
            for f in modified_files:
                print(f)
        else:
            print("✅ Working tree is clean")
    print()


def get_unpushed_commits():
    """⚡ Bolt: Get commits that haven't been pushed using cached metadata."""
    print("=" * 80)
    print("UNPUSHED COMMITS")
    print("=" * 80)
    print()
    
    # ⚡ Bolt: Check cached ahead-behind status for current branch (HEAD) before running git log.
    meta = BRANCH_METADATA_CACHE.get(CURRENT_BRANCH, {})
    ahead = meta.get("ahead")

    # If ahead is None (older Git) or > 0, we run git log to be sure.
    if ahead is None or ahead > 0:
        result = run_git_command(["log", "--oneline", f"origin/main..{CURRENT_BRANCH}"])
        if result and result.returncode == 0 and result.stdout.strip():
            commits = result.stdout.strip().split("\n")
            print(f"📤 Unpushed commits: {len(commits)}")
            for commit in commits:
                print(f"  - {commit}")
        else:
            print("✅ All commits are pushed")
    else:
        print("✅ All commits are pushed")
    print()


def get_recent_activity():
    """Get recent commit activity."""
    print("=" * 80)
    print("RECENT ACTIVITY")
    print("=" * 80)
    print()
    
    result = run_git_command(["log", "--all", "--oneline", "--graph", "--decorate", "-15"])
    if result and result.returncode == 0:
        print(result.stdout)
    print()


def main():
    """Main review function."""
    populate_metadata_cache()
    print("\n" + "=" * 80)
    print(f"WORKING TREE REVIEW REPORT")
    print(f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print("=" * 80)
    print()
    
    get_status_info()
    get_unpushed_commits()
    get_branch_info()
    get_stash_info()
    get_worktree_info()
    get_recent_activity()
    
    print("=" * 80)
    print("REVIEW COMPLETE")
    print("=" * 80)
    print()
    print("Recommendations:")
    print("1. Push any unpushed commits")
    print("2. Review and potentially delete merged branches")
    print("3. Consider merging or closing unmerged branches")
    print("4. Clean up old stashes if no longer needed")
    print()


if __name__ == "__main__":
    main()