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]


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_git_version():
    """⚡ Bolt: Get 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 get_default_branch():
    """⚡ Bolt: Detect default branch name (main or master)."""
    for branch in ["main", "master"]:
        res = run_git_command(["rev-parse", "--verify", branch], capture_output=True)
        if res and res.returncode == 0:
            return branch
    return "main"


def get_branch_info():
    """⚡ Bolt: Optimized branch info retrieval using a single bulk Git metadata call when supported."""
    print("=" * 80)
    print("BRANCH REVIEW")
    print("=" * 80)
    print()

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

    local_branches = []
    remote_branches = []
    merged_branches = []
    unmerged_branches = []
    current_branch = ""

    # Track current branch from 'branch' output for UI consistency
    branch_res = run_git_command(["branch", "--show-current"])
    if branch_res and branch_res.returncode == 0:
        current_branch = branch_res.stdout.strip()

    if use_ahead_behind:
        # ⚡ Bolt: Use git for-each-ref with %(ahead-behind) atom (Git 2.41+) to fetch
        # all branch information (local, remote, merged status) in a single subprocess call.
        # This reduces O(N) subprocess overhead to O(1).
        fmt = "%(refname:short)|%(ahead-behind:" + default_branch + ")"
        result = run_git_command(["for-each-ref", f"--format={fmt}", "refs/heads", "refs/remotes/origin"])

        if result and result.returncode == 0:
            for line in result.stdout.strip().split("\n"):
                if not line: continue
                parts = line.split("|")
                if len(parts) < 2: continue

                name = parts[0]
                if "HEAD" in name or name == "origin": continue

                counts = parts[1].split()
                is_merged = (len(counts) == 2 and int(counts[0]) == 0)

                if name.startswith("origin/"):
                    if name == f"origin/{default_branch}": continue
                    remote_branches.append(name)
                    if is_merged: merged_branches.append(name)
                    else: unmerged_branches.append(name)
                else:
                    local_branches.append(name)
        else:
            use_ahead_behind = False # Fallback if for-each-ref failed

    if not use_ahead_behind:
        # Standard O(N) fallback for older Git or failures
        res = run_git_command(["branch"])
        if res and res.returncode == 0:
            local_branches = [b.strip().replace("*", "").strip() for b in res.stdout.strip().split("\n") if b.strip()]
        
        res = run_git_command(["branch", "-r"])
        if res and res.returncode == 0:
            remote_branches = [b.strip() for b in res.stdout.strip().split("\n") if b.strip() and "HEAD" not in b and b != f"origin/{default_branch}"]
        
        res = run_git_command(["branch", "-r", "--merged", default_branch])
        if res and res.returncode == 0:
            merged_branches = [b.strip() for b in res.stdout.strip().split("\n") if b.strip() and "origin/" in b and b != f"origin/{default_branch}"]

        unmerged_branches = [b for b in remote_branches if b not in merged_branches]

    # Local branches
    print(f"📌 Local Branches: {len(local_branches)}")
    for branch in local_branches:
        marker = "*" if branch == current_branch else " "
        print(f"  {marker} {branch}")
    print()

    # 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 {default_branch}: {len(merged_branches)} branches")
        print("   (These can potentially be deleted)")
    print()
    
    # Unmerged branches
    if unmerged_branches:
        print(f"⚠️  Not merged into {default_branch}: {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():
    """Get current working tree status."""
    print("=" * 80)
    print("WORKING TREE STATUS")
    print("=" * 80)
    print()
    
    result = run_git_command(["status", "--short"])
    if result and result.returncode == 0:
        if result.stdout.strip():
            print("⚠️  Uncommitted changes:")
            print(result.stdout)
        else:
            print("✅ Working tree is clean")
    print()
    
    # Check if ahead/behind
    result = run_git_command(["status", "-sb"])
    if result and result.returncode == 0:
        status_lines = result.stdout.strip().split("\n")
        for line in status_lines:
            if "ahead" in line or "behind" in line:
                print(f"📊 {line}")
    print()


def get_unpushed_commits():
    """Get commits that haven't been pushed."""
    print("=" * 80)
    print("UNPUSHED COMMITS")
    print("=" * 80)
    print()
    
    result = run_git_command(["log", "--oneline", "origin/main..HEAD"])
    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")
    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."""
    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()