gh-interdiff

gh-interdiff

@@ -0,0 +1,151 @@
+#!/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()