check_for_updated_refs.py (1483B)
1 import json 2 import os 3 import re 4 import sys 5 from typing import IO, Container, Dict, Iterable, Optional 6 7 8 GIT_PUSH = re.compile( 9 r"^(?P<flag>.)\t(?P<from_ref>[^\t:]*):(?P<to_ref>[^\t:]*)\t(?P<summary>.*?)(?: \((?P<reason>.*)\))?\n$" 10 ) 11 12 13 def parse_push(fd: IO[str]) -> Iterable[Dict[str, Optional[str]]]: 14 for line in fd: 15 m = GIT_PUSH.match(line) 16 if m is not None: 17 yield m.groupdict() 18 19 20 def process_push(fd: IO[str], refs: Container[str]) -> Dict[str, Optional[str]]: 21 updated_refs = {} 22 23 for ref_status in parse_push(fd): 24 flag = ref_status["flag"] 25 if flag not in (" ", "+", "-", "*"): 26 continue 27 28 to_ref = ref_status["to_ref"] 29 summary = ref_status["summary"] 30 assert to_ref is not None 31 assert summary is not None 32 33 if to_ref in refs: 34 sha = None 35 36 if flag in (" ", "+"): 37 if "..." in summary: 38 _, sha = summary.split("...", maxsplit=1) 39 elif ".." in summary: 40 _, sha = summary.split("..", maxsplit=1) 41 42 updated_refs[to_ref] = sha 43 44 return updated_refs 45 46 47 def main() -> None: 48 git_push_output = os.environ["GIT_PUSH_OUTPUT"] 49 refs = json.loads(os.environ["REFS"]) 50 51 with open(git_push_output, "r") as fd: 52 updated_refs = process_push(fd, refs) 53 54 json.dump(updated_refs, sys.stdout, indent=2) 55 sys.stdout.write("\n") 56 57 58 if __name__ == "__main__": 59 main()