1import os 

2import time 

3import tempfile 

4import io 

5from logging import getLogger 

6from typing import Generator, Tuple 

7from .iface import Input 

8import git 

9 

10 

11log = getLogger(__name__) 

12 

13 

14class GitInput(Input): 

15 def __init__(self, ifn: str, name: str, branch: str = "master", tmpdir: str = None): 

16 log.debug("ifn=%s, name=%s, branch=%s, tmpdir=%s", 

17 ifn, name, branch, tmpdir) 

18 super().__init__(ifn) 

19 if tmpdir is None: 

20 self.tmpd = tempfile.TemporaryDirectory() 

21 self.tmpdir = self.tmpd.name 

22 else: 

23 self.tmpdir = tmpdir 

24 dirn = os.path.join(self.tmpdir, name) 

25 if os.path.isdir(dirn): 

26 self.repo = git.repo.Repo(dirn) 

27 self.repo.remote().fetch(refspec=branch, depth=1) 

28 else: 

29 self.repo = git.repo.Repo.clone_from(self.ifn, dirn, bare=True, 

30 branch=branch, depth=1) 

31 self.ts = time.time() 

32 self.files = {} 

33 

34 def walk(self) -> Generator[Tuple[str, int, float], None, None]: 

35 for x in self.repo.tree().traverse(): 

36 if x.type != "blob": 

37 continue 

38 self.files[x.path] = x 

39 yield x.path, x.mode, self.ts 

40 

41 def readfile(self, fn: str) -> str: 

42 try: 

43 return io.TextIOWrapper(io.BytesIO(self.files.get(fn).data_stream.read())).read() 

44 except UnicodeDecodeError as e: 

45 log.warning("cannot decode", e) 

46 return ""