bin/gh-interdiff

#!/usr/bin/env python3
'''show the diff between two pushes of a PR'''

import argparse
import datetime
import json
import shlex
import subprocess
import sys
import termios
import tty
import typing

QUERY = '''
query($owner: String!, $name: String!, $number: Int!, $cursor: String) {
	repository(owner: $owner, name: $name) {
		pullRequest(number: $number) {
			createdAt
			baseRefOid
			timelineItems(first: 100, after: $cursor, itemTypes: [HEAD_REF_FORCE_PUSHED_EVENT]) {
				nodes {
					... on HeadRefForcePushedEvent {
						afterCommit { oid }
						beforeCommit { oid }
						createdAt
					}
				}
				pageInfo {
					endCursor
					hasNextPage
				}
			}
		}
	}
}
'''

def main() -> None:
	parser = argparse.ArgumentParser(description=__doc__)
	parser.add_argument('pr', type=int, help='pull request number')
	args = parser.parse_args()

	name_with_owner = run(['gh', 'repo', 'view', '--json', 'nameWithOwner', '--jq', '.nameWithOwner'],
			capture_output=True, text=True).stdout.strip()
	owner, name = name_with_owner.split('/', maxsplit=1)

	events = []
	cursor = None
	while True:
		command = ['gh', 'api', 'graphql',
			 '-f', f'query={QUERY}', '-f', f'owner={owner}', '-f', f'name={name}', '-F', f'number={args.pr}']
		if cursor:
			command.extend(['-f', f'cursor={cursor}'])

		response = json.loads(run(command, capture_output=True, text=True).stdout)
		pull_request = response['data']['repository']['pullRequest']
		if pull_request is None:
			parser.error(f'pull request #{args.pr} was not found in {name_with_owner}')

		timeline = pull_request['timelineItems']
		events.extend(timeline['nodes'])
		if not timeline['pageInfo']['hasNextPage']:
			break
		cursor = timeline['pageInfo']['endCursor']

	if len(events) == 0:
		print(f'no pushes for {name_with_owner}#{args.pr}')
		return
	# add a fake event for base branch → first push
	events.insert(0, {
		'createdAt': pull_request['createdAt'],
		'beforeCommit': {'oid': pull_request['baseRefOid']},
		'afterCommit': events[0]['beforeCommit'],
	})

	print(f'push events for {name_with_owner}#{args.pr}:')
	options = []
	for index, event in enumerate(events):
		before = event['beforeCommit']
		after = event['afterCommit']
		before_oid = before['oid'][:12]
		after_oid = after['oid'][:12]
		options.append(f'{index}. {local_time(event['createdAt'])}: {before_oid} → {after_oid}')

	while True:
		index = menu(options)
		if index is None:
			break
		event = events[index]
		before = event['beforeCommit']['oid']
		after = event['afterCommit']['oid']
		run(['git', 'fetch', 'origin', before, after])
		run(['git', 'update-ref', 'refs/heads/gh-interdiff-' + before, before])
		run(['git', 'update-ref', 'refs/heads/gh-interdiff-' + after, after])
		try:
			run(['jj', 'interdiff', '-f', before, '-t', after])
		finally:
			run(['git', 'update-ref', '-d', 'refs/heads/gh-interdiff-' + before])
			run(['git', 'update-ref', '-d', 'refs/heads/gh-interdiff-' + after])
	run(['jj', 'git', 'import', '--ignore-working-copy']) # abandon unreachable commits

def run(command: typing.Sequence[str], **kwargs) -> str:
	print('\033[90m$ ' + shlex.join(command) + '\033[0m')
	return subprocess.run(command, check=True, **kwargs)

def local_time(dt_str: str) -> str:
	dt = datetime.datetime.fromisoformat(dt_str.replace('Z', '+00:00')).astimezone()
	return dt.strftime('%Y-%m-%d %H:%M:%S %Z')

def menu(options: typing.Sequence[str]) -> int | None:
	selected = 0
	menu_height = len(options)

	original_settings = termios.tcgetattr(sys.stdin)
	try:
		tty.setraw(sys.stdin.fileno())

		print('\n' * menu_height, end='', flush=True)

		while True:
			# move the cursor up to the beginning of the existing menu
			sys.stdout.write(f'\x1b[{menu_height}A')
			for index, option in enumerate(options):
				if index == selected:
					option = f'\x1b[7m{option}\x1b[0m' # invert colors
				sys.stdout.write(f'\r\x1b[2K{option}\n')
			sys.stdout.write('\r')
			sys.stdout.flush()

			key = read_key()
			if key in ('\x1b[A', 'k'):
				if selected > 0:
					selected -= 1
			elif key in ('\x1b[B', 'j'):
				if selected < len(options) - 1:
					selected += 1
			elif key in ('\r', '\n'):
				return selected
			elif key == 'q':
				return None
	finally:
		termios.tcsetattr(sys.stdin, termios.TCSADRAIN, original_settings)

def read_key() -> str:
	key = sys.stdin.read(1)
	if key == '\x1b':
		key += sys.stdin.read(2) # arrow keys are 3 bytes: ESC, [, A/B
	return key

if __name__ == '__main__':
	main()