xBar gerrit review monitor

002-gerrit_incoming_crs.2m.py

#!/usr/bin/env python3
##!/usr/bin/python3
# -*- coding: utf-8 -*-

import os
import sys
import json
import subprocess
import logging
import glob
import re


gerrit_user = 'drewgulino'
gerrit_host = 'gerrit.relaydev.sh'
gerrit_port = '29418'

def find_ssh_auth_sock():
    """Dynamically locates the active macOS ssh-agent socket."""
    
    # 0. Check if SSH_AUTH_SOCK is already in the environment and exists
    env_sock = os.environ.get('SSH_AUTH_SOCK')
    if env_sock and os.path.exists(env_sock):
        return env_sock

    # 1. Search common patterns for the active agent socket
    # - ~/.keychain/*.s (keychain ssh-agent sockets)
    # - /var/folders/.../T/ssh-XXXXXX/agent.XXXX (macOS default launchd sockets)
    # - /tmp/ssh-XXXXXX/agent.XXXX (standard ssh-agent sockets)
    patterns = [
        os.path.expanduser('~/.keychain/*.s'),
        '/var/folders/*/*/*/ssh-*/agent.*',
        '/tmp/ssh-*/agent.*'
    ]
    
    for pattern in patterns:
        sockets = glob.glob(pattern)
        if sockets:
            # Sort by modification time to get the most recently created socket
            sockets.sort(key=os.path.getmtime, reverse=True)
            return sockets[0]
            
    # 2. Check keychain shell configuration files as fallback
    keychain_sh_pattern = os.path.expanduser('~/.keychain/*-sh')
    keychain_sh_files = glob.glob(keychain_sh_pattern)
    if keychain_sh_files:
        keychain_sh_files.sort(key=os.path.getmtime, reverse=True)
        try:
            with open(keychain_sh_files[0], 'r') as f:
                content = f.read()
                # Parse "SSH_AUTH_SOCK=...;"
                match = re.search(r'SSH_AUTH_SOCK=([^;]+);', content)
                if match:
                    sock_path = match.group(1).strip()
                    if os.path.exists(sock_path):
                        return sock_path
        except Exception:
            pass

    return None

# --- Apply the socket to the environment ---
sock_path = find_ssh_auth_sock()
if sock_path:
    os.environ['SSH_AUTH_SOCK'] = sock_path
    #print(f"DEBUG: Found socket at {sock_path}")  # Viewable in xbar error logs
else:
    print("DEBUG: No SSH socket found.", file=sys.stderr)


def query_gerrit():
    gerrit_query = ['ssh', '%s@%s' % (gerrit_user, gerrit_host), '-p', gerrit_port,
                    'gerrit', 'query', '--format=JSON', '--all-approvals',
                    'status:open', 'reviewer:"%s"' % gerrit_user]
    try:
        results = subprocess.check_output(
            gerrit_query, stderr=subprocess.STDOUT, universal_newlines=True)
        return strip_unused_results(results)
    except Exception:
        logging.error('Unable to query gerrit')
        logging.error('---')
        logging.error("Ensure you've uploaded your SSH key to gerrit")
        sys.exit(1)
def strip_unused_results(results):
    # last line is blank
    # second to last line is an aggregate
    results = results.split('\n')
    results = results[:-2] if len(results) > 2 else []
    return [json.loads(x) for x in results]


def is_self(result):
    return result.get('owner', {}).get('username', '') == gerrit_user

def filter_self(results):
    return [x for x in results if not is_self(x)]

def is_approved(result):
    approval_total = 0
    patch_sets = result.get("patchSets", {})
    for patch in patch_sets:
        approvals = patch.get("approvals",[])
        for approval in approvals:
            value = int(approval.get("value",0))
            approval_total += value
    if approval_total >= 2:
        return True
    else:
        return False

def filter_approved(results):
    return [x for x in results if not is_approved(x)]

def print_results(results):
    if len(results) > 0:
        print('⚠️ %s CRs' % len(results))
        print('---')
        for r in results:
            subj = r.get('subject')
            url = r.get('url')
            print('%s | href=%s' % (subj, url))
    else:
        print('x')
        print('---')

def main():
    all_unapproved = []
    all_incoming = filter_self(query_gerrit())
    all_unapproved = filter_approved(all_incoming)
    print_results(all_unapproved)

if __name__ == "__main__":
    main()