1# copy directory 

2import os 

3import time 

4from typing import Generator, Tuple 

5from .iface import Input, Output 

6from logging import getLogger 

7 

8log = getLogger(__name__) 

9 

10 

11class DirInput(Input): 

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

13 for root, dirs, files in os.walk(self.ifn): 

14 relpath = os.path.relpath(root, self.ifn) 

15 if relpath == ".": 

16 relpath = "" 

17 for fn in sorted(files): 

18 fromfn = os.path.join(root, fn) 

19 st = os.stat(fromfn) 

20 mode = st.st_mode 

21 yield os.path.join(relpath, fn), mode, st.st_mtime 

22 

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

24 with open(os.path.join(self.ifn, fn)) as ifp: 

25 return ifp.read() 

26 

27 

28class SingleInput(Input): 

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

30 st = os.stat(self.ifn) 

31 yield os.path.basename(self.ifn), st.st_mode, st.st_mtime 

32 

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

34 with open(self.ifn) as ifp: 

35 return ifp.read() 

36 

37 

38class DirOutput(Output): 

39 def __init__(self, ofn: str): 

40 super().__init__(ofn) 

41 self.ofn = os.path.abspath(self.ofn) 

42 log.debug("ofn is %s -> %s", ofn, self.ofn) 

43 

44 def writefile(self, fn: str, content: str, mode: int, ts: float = None): 

45 fname = os.path.join(self.ofn, fn) 

46 if not os.path.abspath(fname).startswith(self.ofn): 

47 log.warning("%s is not in %s", fname, self.ofn) 

48 return 

49 dirname = os.path.dirname(fname) 

50 os.makedirs(dirname, exist_ok=True) 

51 with open(fname, "w") as ofp: 

52 ofp.write(content) 

53 os.chmod(fname, mode) 

54 if ts is not None: 

55 os.utime(fname, (ts, ts)) 

56 

57 

58class ListOutput(Output): 

59 def writefile(self, fn: str, content: str, mode: int, ts: float = None): 

60 tsstr = "YYYY-mm-dd HH:MM" 

61 if ts is not None: 

62 tsstr = time.strftime("%Y-%m-%d %H:%M", time.localtime(ts)) 

63 print("%o %d %s %s" % (mode, len(content), tsstr, fn))