141 lines
5.7 KiB
Python
Executable File
141 lines
5.7 KiB
Python
Executable File
from __future__ import annotations
|
|
from typing import Iterable, List, Tuple, Union
|
|
import logging
|
|
import os
|
|
import stat
|
|
|
|
from ..SAOLogging import perror
|
|
|
|
class FileSystem():
|
|
def __init__(self, adb_arguments: List[str]) -> None:
|
|
self.adb_arguments = adb_arguments
|
|
|
|
def _get_files_tree(self, tree_path: str, tree_path_stat: os.stat_result, follow_links: bool = False):
|
|
# the reason to have two functions instead of one purely recursive one is to use self.lstat_in_dir ie ls
|
|
# which is much faster than individually stat-ing each file. Hence we have get_files_tree's special first lstat
|
|
if stat.S_ISLNK(tree_path_stat.st_mode):
|
|
if not follow_links:
|
|
logging.warning(f"Ignoring symlink {tree_path}")
|
|
return None
|
|
logging.debug(f"Following symlink {tree_path}")
|
|
try:
|
|
tree_path_realpath = self.realpath(tree_path)
|
|
tree_path_stat_realpath = self.lstat(tree_path_realpath)
|
|
except (FileNotFoundError, NotADirectoryError, PermissionError) as e:
|
|
perror(f"Skipping symlink {tree_path}", e)
|
|
return None
|
|
return self._get_files_tree(tree_path_realpath, tree_path_stat_realpath, follow_links = follow_links)
|
|
elif stat.S_ISDIR(tree_path_stat.st_mode):
|
|
tree = {".": (60 * (int(tree_path_stat.st_atime) // 60), 60 * (int(tree_path_stat.st_mtime) // 60))}
|
|
for filename, stat_object_child, in self.lstat_in_dir(tree_path):
|
|
if filename in [".", ".."]:
|
|
continue
|
|
tree[filename] = self._get_files_tree(
|
|
self.join(tree_path, filename),
|
|
stat_object_child,
|
|
follow_links = follow_links)
|
|
return tree
|
|
elif stat.S_ISREG(tree_path_stat.st_mode):
|
|
return (60 * (int(tree_path_stat.st_atime) // 60), 60 * (int(tree_path_stat.st_mtime) // 60)) # minute resolution
|
|
else:
|
|
raise NotImplementedError
|
|
|
|
def get_files_tree(self, tree_path: str, follow_links: bool = False):
|
|
statObject = self.lstat(tree_path)
|
|
return self._get_files_tree(tree_path, statObject, follow_links = follow_links)
|
|
|
|
def remove_tree(self, tree_path: str, tree: Union[Tuple[int, int], dict], dry_run: bool = True) -> None:
|
|
if isinstance(tree, tuple):
|
|
logging.info(f"Removing {tree_path}")
|
|
if not dry_run:
|
|
self.unlink(tree_path)
|
|
elif isinstance(tree, dict):
|
|
remove_folder = tree.pop(".", False)
|
|
for key, value in tree.items():
|
|
self.remove_tree(self.normpath(self.join(tree_path, key)), value, dry_run = dry_run)
|
|
if remove_folder:
|
|
logging.info(f"Removing folder {tree_path}")
|
|
if not dry_run:
|
|
self.rmdir(tree_path)
|
|
else:
|
|
raise NotImplementedError
|
|
|
|
def push_tree_here(self,
|
|
tree_path: str,
|
|
relative_tree_path: str, # for logging paths of files / folders copied relative to the source root / destination root
|
|
# nicely instead of repeating the root every time; rsync does this nice logging
|
|
tree: Union[Tuple[int, int], dict],
|
|
destination_root: str,
|
|
fs_source: FileSystem,
|
|
dry_run: bool = True,
|
|
show_progress: bool = False
|
|
) -> None:
|
|
if isinstance(tree, tuple):
|
|
if dry_run:
|
|
logging.info(f"{relative_tree_path}")
|
|
else:
|
|
if not show_progress:
|
|
# log this instead of letting adb display output
|
|
logging.info(f"{relative_tree_path}")
|
|
self.push_file_here(tree_path, destination_root, show_progress = show_progress)
|
|
self.utime(destination_root, tree)
|
|
elif isinstance(tree, dict):
|
|
try:
|
|
tree.pop(".") # directory needs making
|
|
logging.info(f"{relative_tree_path}{self.sep}")
|
|
if not dry_run:
|
|
self.makedirs(destination_root)
|
|
except KeyError:
|
|
pass
|
|
for key, value in tree.items():
|
|
self.push_tree_here(
|
|
fs_source.normpath(fs_source.join(tree_path, key)),
|
|
fs_source.join(relative_tree_path, key),
|
|
value,
|
|
self.normpath(self.join(destination_root, key)),
|
|
fs_source,
|
|
dry_run = dry_run,
|
|
show_progress = show_progress
|
|
)
|
|
else:
|
|
raise NotImplementedError
|
|
|
|
# Abstract methods below implemented in Local.py and Android.py
|
|
|
|
@property
|
|
def sep(self) -> str:
|
|
raise NotImplementedError
|
|
|
|
def unlink(self, path: str) -> None:
|
|
raise NotImplementedError
|
|
|
|
def rmdir(self, path: str) -> None:
|
|
raise NotImplementedError
|
|
|
|
def makedirs(self, path: str) -> None:
|
|
raise NotImplementedError
|
|
|
|
def realpath(self, path: str) -> str:
|
|
raise NotImplementedError
|
|
|
|
def lstat(self, path: str) -> os.stat_result:
|
|
raise NotImplementedError
|
|
|
|
def lstat_in_dir(self, path: str) -> Iterable[Tuple[str, os.stat_result]]:
|
|
raise NotImplementedError
|
|
|
|
def utime(self, path: str, times: Tuple[int, int]) -> None:
|
|
raise NotImplementedError
|
|
|
|
def join(self, base: str, leaf: str) -> str:
|
|
raise NotImplementedError
|
|
|
|
def split(self, path: str) -> Tuple[str, str]:
|
|
raise NotImplementedError
|
|
|
|
def normpath(self, path: str) -> str:
|
|
raise NotImplementedError
|
|
|
|
def push_file_here(self, source: str, destination: str, show_progress: bool = False) -> None:
|
|
raise NotImplementedError
|