"""File storage and Manager. directory structure ------------------- :: STORAGE_ROOT (set by -C) |_ storage | |_ .export.cache.txt | |_ DATE | | |_ RECORDING_FILES | |_ external | | |_ LINK | |_ .trash | |_ DATE | |_ RECORDING_FILES |_ export (EXPORT_ROOT, set by -E) | |_ DATE | |_ EXPORT_FILES |_ user_repository (set by -U) |_ USER |_ .repository |_ SETUP_FILES Command line ------------ Usage :: env PYTHONPATH=RaspBerryPi3/python python -m biopro.file.manager -C .BPS/ ... debugging User Account System :: ... --debug-account-system user ... debugging verbose output :: ... -v ... Class ----- """ from datetime import date from pprint import pprint from shutil import rmtree from biopro.util.console import CYAN from biopro.util.lock import Synchronized, synchronized from biopro.util.logger import LoggerFlag, logging_verbose from .api import * from .export import * from .user.manager import UserSettingManager from .util import * from json import loads as json_parse, dumps as _json_stringify import json def json_stringify(o) -> str: return _json_stringify(o, separators=(',', ':')) class FileManager(FileAPI, LoggerFlag, Synchronized): """file manager.""" DEFAULT_EXPORT_CACHE_FILE = '.export.cache.txt' def __init__(self, options: FileOptions): """ :param options: file manager options """ LoggerFlag.__init__(self, 'FileManager', CYAN) Synchronized.__init__(self) if options.storage_root is None: raise RuntimeError('storage root not set') s = Path(options.storage_root).absolute() if options.export_root is None: e = s / 'export' else: e = Path(options.export_root).absolute() if options.user_repository is None: options.user_repository = s / 'user_repository' self._storage_root = s / 'storage' self._export_root = e self._external_root = (s / EXTERNAL_STORAGE_DIRECTORY) self._user_setting = UserSettingManager(options) self._current_storage_root = None self._current_used_files = {} self._export_manager = FileExporter() self._export_storage_file = None self._export_storage_save_request = False self._mount_external = [] @property def storage_root(self) -> Path: return self._storage_root @property def export_root(self) -> Path: return self._export_root @property def external_root(self) -> Path: return self._external_root @synchronized @logging_verbose def setup_storage(self): """setup storage directory. :raises NotADirectoryError: if storage root or export root not a directory. """ storage_root = self._storage_root export_root = self._export_root # storage root if not storage_root.exists(): if storage_root.is_symlink(): storage_root.unlink() self.log_info('create directory', storage_root) storage_root.mkdir(parents=True) elif not storage_root.is_dir(): raise NotADirectoryError(str(storage_root)) # export root if not export_root.exists(): if export_root.is_symlink(): export_root.unlink() self.log_info('create directory', export_root) export_root.mkdir(parents=True) elif not export_root.is_dir(): raise NotADirectoryError(str(export_root)) # export cache if self._export_storage_file is None: self._export_storage_file = self._storage_root / self.DEFAULT_EXPORT_CACHE_FILE if self._export_storage_file.exists(): self._load_export_cache() # logging self.log_info('storage root', storage_root) self.log_info('export root', export_root) self.log_info('user repository', self._user_setting.user_repository) @synchronized @logging_verbose def update_storage(self): """update default storage path :return: """ storage_root = self._storage_root t = datetime.now() d = '%d-%02d-%02d' % (t.year, t.month, t.day) self._current_storage_root = current_root = storage_root / d self.log_verbose('set save directory', current_root) # storage root if not current_root.exists(): self.log_info('create directory', current_root) current_root.mkdir(parents=True) elif not current_root.is_dir(): raise NotADirectoryError(str(storage_root)) @synchronized @logging_verbose def mount_external_auto(self): if not self._external_root.exists(): self._external_root.mkdir(parents=True) for p in Partition.list_block_device(sync=True): if p.is_external_storage: try: q = p.mount_with_label(self._external_root) except BaseException as e: self.log_warn('cannot mount ' + p.fs_path + ' : ' + str(e)) else: self.log_info('mount', str(q)) self._mount_external.append(PartitionInfo(p)) @synchronized def mount_external(self, name: str, path: Path): if not path.exists(): self.log_warn('mount path not found', str(path)) return if name == '': name = path.name self.log_verbose('mount_external', name, '->', str(path)) if name == '*': if path.is_dir(): for p in path.iterdir(): self.mount_external('', p) else: self.log_warn('mount path not directory', str(path)) else: if not self._external_root.exists(): self._external_root.mkdir(parents=True) f = self._external_root / name if f.exists(): if not f.is_symlink(): raise ValueError('file existed') f.unlink() f.symlink_to(path.absolute(), target_is_directory=True) @synchronized def umount_external(self): self.log_verbose('umount_external') if not self._external_root.exists(): return for p in self._mount_external: m = Path(p.partition.mount_path) self.log_verbose('umount', self.shadow_path(m)) p.partition.umount() if m.exists(): try: m.rmdir() except OSError: pass self._mount_external.clear() for p in list(self._external_root.iterdir()): if p.is_symlink(): p.unlink() @synchronized @logging_verbose def shutdown(self, umount_external=False): self._export_manager.shutdown() self._user_setting.shutdown() if umount_external: self.umount_external() if self._export_storage_save_request: self._save_export_cache() def _load_export_cache(self): cache = self._export_storage_file if cache is None: return if not cache.exists(): return self.log_verbose('load_export_cache', cache) tmp = {} with cache.open('r') as f: for line in f: line = line.strip() if len(line) > 0: part = line.split('\t', 3) try: file_path = part[0] file_info = tmp.get(file_path, None) export_type = part[1] export_path = Path(part[2]) except IndexError: pass else: if file_info is None: try: info_path = self.validate_path_root(file_path) if info_path.is_file(): tmp[file_path] = file_info = FileInfo.get_info(self.storage_root, info_path) else: continue except FileNotFoundError: continue try: export_path = (self.export_root / export_path).resolve() except FileNotFoundError: pass else: if export_path.exists() and path_start_with(export_path, self.export_root): file_info.add_export_path(export_type, export_path.relative_to(self.export_root)) def _save_export_cache(self): cache = self._export_storage_file if cache is None: return self.log_verbose('save_export_cache', cache) with cache.open('w') as f: for file_info in FileInfo.CACHE.values(): if file_info.is_meta_file: for export_type, export_path in file_info.export.items(): if export_path is not None: file_path = file_info.file_path if len(file_path) > 0: try: print(file_path, export_type, export_path, sep='\t', file=f) except ValueError: pass self._export_storage_save_request = False def _validate_directory_not_used(self, path: Path): """check none of files under directory **path** is opened :param path: directory path :raise RuntimeError: file in directory is opened """ for p in self._current_used_files.values(): if path_start_with(p, path): raise RuntimeError('directory in use : ' + self.shadow_path(path)) def _validate_file_not_used(self, path: Path): """check file **path** is not opened :param path: file path :raise RuntimeError: file is opened """ for p in self._current_used_files.values(): if p == path: raise RuntimeError('file in use : ' + self.shadow_path(path)) def _validate_path_not_external(self, path: Path): if path_in_external(path): raise RuntimeError('file in external storage : ' + self.shadow_path(path)) @property def user_setting(self) -> UserSettingManager: return self._user_setting @synchronized @logging_verbose def index(self) -> RootInfo: return RootInfo(self.storage_root) @synchronized @logging_verbose @validate_path_root def info(self, path: str) -> FileInfo: path = resolve_path_if_trashed(path) result = FileInfo.get_info(self.storage_root, path) return result @synchronized @logging_verbose @validate_path_root def meta(self, path: str) -> RecordingMetaFile: path = resolve_path_if_trashed(path) info = FileInfo.get_info(self.storage_root, path) meta = info.meta_file if meta is None: raise RuntimeError('not a meta file', self.shadow_path(path)) return meta @synchronized @logging_verbose @validate_path_root def export_info(self, path: Union[str, Path]) -> Dict[str, Optional[Path]]: info = FileInfo.get_info(self.storage_root, path) if info.meta_file is None: raise RuntimeError('not a meta file', self.shadow_path(path)) return info.export @synchronized @logging_verbose @validate_path_root @validate_path_is_directory(resolve_in_trash=True) def list(self, path: Union[str, str]) -> DirectoryInfo: ### file list return DirectoryInfo(self.storage_root, path) @synchronized @logging_verbose @validate_path_root @validate_path_is_directory(resolve_in_trash=True) def list_range(self, path: Union[str, str], start, end) -> DirectoryInfo: ### file list return DirectoryInfo(self.storage_root, path) @synchronized @logging_verbose @validate_path_root def mkdir(self, path: Union[str, Path]) -> DirectoryInfo: if path.exists(): if path.is_file(): raise FileExistsError(self.shadow_path(path)) else: path.mkdir(parents=True) return DirectoryInfo(self.storage_root, path) def use(self, device: Union[int, Device]) -> FileInfo: if isinstance(device, Device): device = device.device_id self.log_verbose('use', 'device', device) path = self._current_used_files.get(device, None) def _remove_used(): try: del self._current_used_files[device] except KeyError: pass if path is not None: try: info = FileInfo.get_info(self.storage_root, path) except FileNotFoundError: _remove_used() else: if not info.is_meta_file or not info.file_path_origin.exists(): _remove_used() else: return info return FileInfo.get_info(self.storage_root, self._current_storage_root) def unset(self, device: Union[int, Device]): if isinstance(device, Device): device = device.device_id self.log_verbose('unset', 'device', device) try: del self._current_used_files[device] except KeyError: pass @synchronized def save(self, device: CompletedDevice, filepath: Union[str, Path, FileInfo], overwrite=False) -> FileInfo: if isinstance(filepath, FileInfo): filepath = filepath.file_path overwrite = True elif isinstance(filepath, Path): filepath = str(filepath) self.log_verbose('save', filepath, '(overwrite)' if overwrite else '') # create meta file if "/" in filepath: filepath = self.validate_path_root(filepath) meta = RecordingMetaFile(filepath) else: # use default directory meta = RecordingMetaFile(self._current_storage_root / filepath) # check file path if meta.filepath.exists() and not overwrite: raise FileExistsError(self.shadow_path(meta.filepath)) p = meta.filepath.parent if not p.exists(): p.mkdir(parents=True) # set meta and save meta.device = json_stringify(device.as_json()) meta.configuration = device.configuration meta.write() # file info load info = FileInfo.get_info(self.storage_root, meta.filepath) info.meta_file = meta # set used self._current_used_files[device.device_id] = meta.filepath return info @synchronized @logging_verbose @validate_path_root def rename(self, path: Union[str, Path], name: str) -> FileInfo: source_path = path target_path = source_path.with_name(name) if source_path.is_dir(): self._validate_directory_not_used(source_path) self._validate_directory_not_used(target_path) # XXX overwrite checking? source_path.rename(target_path) FileInfo.invalid_cache(source_path) else: self._validate_file_not_used(source_path) self._validate_file_not_used(target_path) info = FileInfo.get_info(self.storage_root, source_path) if info.is_meta_file: # XXX overwrite checking? target_path = RecordingMetaFile.load(source_path).change_filepath(target_path) FileInfo.invalid_cache(source_path) else: raise RuntimeError('unsupported rename for file type : ' + info.file_type) return FileInfo.get_info(self.storage_root, target_path) @logging_verbose @validate_path_root def delete(self, path: Union[str, Path]) -> bool: self._validate_path_not_external(path) path = resolve_path_if_trashed(path) return self._delete(path) @synchronized def _delete(self, path: Path) -> bool: if path_start_with(path, self._external_root): return False if not path.exists(): return False if RecordingFile.is_recording_file(path): return False self.log_verbose('deleted', self.shadow_path(path)) if path.is_dir() and not path.is_symlink(): rmtree(str(path), ignore_errors=True) elif RecordingMetaFile.is_recording_meta_file(path): RecordingMetaFile.remove_meta_file(path) else: path.unlink() # invalid the file under this directory in cache FileInfo.invalid_cache(path) return True @logging_verbose @validate_path_root def trash(self, path: Union[str, Path]) -> bool: self._validate_path_not_external(path) if not path.exists(): return False return self._trash(path) @synchronized def _trash(self, path: Path) -> bool: if path_in_trash(path): return False if path_start_with(path, self._external_root): return False if RecordingFile.is_recording_file(path): return False trash_root = path.parent / TRASH_DIRECTORY if not trash_root.exists(): trash_root.mkdir() target_path = trash_root / path.name self.log_verbose(self.shadow_path(path), '->', self.shadow_path(target_path)) if RecordingMetaFile.is_recording_meta_file(path): RecordingMetaFile.load(path).change_filepath(target_path) else: path.rename(target_path) # invalid the file under this directory in cache FileInfo.invalid_cache(path) if target_path.is_dir(): self._untrash(target_path) return True def _untrash(self, path: Path) -> bool: trash_dir = path / TRASH_DIRECTORY if trash_dir.exists(): for p in list(trash_dir.iterdir()): if RecordingFile.is_recording_file(p): continue t = path / p.name if t.exists(): i = 0 while t.exists(): t = t.with_name(p.name + '_%d' % i) i += 1 self.log_verbose(self.shadow_path(p), '->', self.shadow_path(t)) if RecordingMetaFile.is_recording_meta_file(p): RecordingMetaFile.load(p).change_filepath(t) else: p.rename(t) rmtree(str(trash_dir), ignore_errors=True) for p in path.iterdir(): if p.is_dir() and not p.is_symlink(): self._untrash(p) return True @logging_verbose @validate_path_root def restore(self, path: Union[str, Path]) -> bool: self.log_verbose('restore', str(path)) path = resolve_path_if_trashed(path) if not path_in_trash(path): return False else: return self._restore(path) @synchronized def _restore(self, path: Path) -> bool: if RecordingFile.is_recording_file(path): return False target_path = resolve_path_un_trashed(path) # trashed_root = get_trashed_root(path) if target_path.exists(): self.log_verbose('file existed', self.shadow_path(target_path)) return False target_directory = target_path.parent if not target_directory.exists(): target_directory.mkdir(parents=True) self.log_verbose(self.shadow_path(path), '->', self.shadow_path(target_path)) if RecordingMetaFile.is_recording_meta_file(path): RecordingMetaFile.load(path).change_filepath(target_path) else: path.rename(target_path) # invalid the file under this directory in cache FileInfo.invalid_cache(target_path.parent) return True @logging_verbose def perform_trashing(self, expire: int): self._perform_trashing(self.storage_root, time() - 86400 * expire) def _perform_trashing(self, path: Path, expire_time: float): next_dir = [] for p in list(path.iterdir()): if not p.exists(): # been trash, not existed. continue if p.name == TRASH_DIRECTORY: continue if path_start_with(p, self._external_root): # do not do this in external root continue mt = p.stat().st_mtime if mt < expire_time: self._trash(p) elif p.is_dir() and not p.is_symlink(): next_dir.append(p) for p in next_dir: self._perform_trashing(p, expire_time) @synchronized @logging_verbose def clean_trash(self, expire: Optional[int] = None): if expire is None: expire_time = None else: expire_time = time() - 86400 * expire self._trash_clean(self.storage_root, expire_time, in_trash=False) def _trash_clean(self, path: Path, expire_time: Optional[int], in_trash: bool): if path_start_with(path, self._external_root): return if in_trash: for p in list(path.iterdir()): if p.exists(): mt = p.stat().st_mtime if mt < expire_time: self._delete(p) else: for p in path.iterdir(): if p.is_dir(): if p.name == TRASH_DIRECTORY: self._trash_clean(p, expire_time, in_trash=True) else: self._trash_clean(p, expire_time, in_trash=False) @logging_verbose def perform_archive(self, expire: Optional[int] = None): # XXX when or what period should we use? if expire is None: expire_time = None else: expire_time = time() - 86400 * expire self._perform_archive(self.storage_root, expire_time) def _perform_archive(self, path: Path, expire_time: float): next_dir = [] for p in list(path.iterdir()): if p.name == TRASH_DIRECTORY: # do not perform archive in trashing continue if p.is_dir() and not p.is_symlink(): next_dir.append(p) elif RecordingMetaFile.is_recording_meta_file(p): mt = p.stat().st_mtime if mt < expire_time: self._compress_archive(p) for p in next_dir: self._perform_archive(p, expire_time) # def _compress_archive(self, path: Path): # from biopro.recording.tar import compress_meta_file # compress_meta_file(RecordingMetaFile.load(path), delete_source=True) # FileInfo.invalid_cache(path) # def _extra_archive(self, path: Path): # from biopro.recording.tar import extra_meta_file # # XXX how do we perform file action on a archive? # extra_meta_file(path, overwrite=True, delete_source=True) # FileInfo.invalid_cache(path) @synchronized @logging_verbose def clean_storage(self) -> bool: self.umount_external() rmtree(str(self.storage_root), ignore_errors=True) rmtree(str(self.export_root), ignore_errors=True) FileInfo.invalid_cache_all() self._export_storage_save_request = False self._storage_root.mkdir() self._export_root.mkdir() self._external_root.mkdir() return True def get_export_functions(self) -> Tuple[ExternalExportFunction, ...]: return self._export_manager.get_export_functions() def get_export_parameters(self, export_type: str) -> Tuple[ExternalParameter, ...]: return self._export_manager.get_export_parameters(export_type) @validate_path_root(path='source_path') def export_file_path(self, export_type: str, source_path: Union[str, Path]) -> Path: return self._export_file_path(export_type, source_path) def _export_file_path(self, export_type: str, source_path: Path) -> Path: export_func = self._export_manager.get_export_func(export_type) if not source_path.is_file(): raise FileNotFoundError(self.shadow_path(source_path)) info = FileInfo.get_info(self.storage_root, source_path) if not info.is_meta_file: raise RuntimeError('not a meta file : ' + self.shadow_path(source_path)) source_directory = source_path.parent.relative_to(self._storage_root) export_directory = self._export_root / source_directory return (export_directory / source_path.name).with_suffix(export_func.ext) @validate_path_root(path='source_path') def export_file(self, source_path: Union[str, Path], export_path: Union[None, str, Path], export_type: str, async_call=False, overwrite=False, **options) -> Path: self.log_verbose('export_file', export_type, source_path) source_path = source_path.with_suffix(RecordingMetaFile.FILE_EXT) _export_path = self._export_file_path(export_type, source_path) # create export target directory if export_path is None: export_path = _export_path else: if isinstance(export_path, str): export_path = Path(export_path) export_path = export_path.absolute() if export_path.exists() and not overwrite: return export_path.relative_to(self.export_root) export_directory = export_path.parent if not export_directory.exists(): export_directory.mkdir(parents=True) # set cache process info = FileInfo.get_info(self.storage_root, source_path) info.add_export_path(export_type, None) self._export_storage_save_request = True try: ret = self._export_manager.export_file(source_path, export_path, export_type, async_call=async_call, overwrite=overwrite, **options) except BaseException: info.del_export_path(export_type) raise else: ret = ret.relative_to(self.export_root) info.add_export_path(export_type, ret) return ret @validate_path_is_file(path='import_path') def import_file(self, import_path: Union[str, Path], target_path: Union[None, str, Path], import_type: str, async_call=False, overwrite=False, **options) -> Path: self._export_manager.get_import_func(import_type) # create import result directory if target_path is None: import_root = self._storage_root / 'import' import_directory = import_root / str(date.today()) if not import_directory.exists(): import_directory.mkdir(parents=True) target_path = (import_directory / import_path.name) else: target_path = self.validate_path_root(target_path) target_path = target_path.with_suffix(RecordingMetaFile.FILE_EXT) if target_path.exists() and not overwrite: return target_path.relative_to(self.storage_root) ret = self._export_manager.import_file(import_path, target_path, import_type, async_call=async_call, overwrite=overwrite, **options) return ret.relative_to(self.storage_root) @synchronized #@validate_export_root(path='export_path') @validate_path_root(path='export_path') def segment_info(self, export_path: Union[str, Path], create=False) -> Optional[FileSegment]: print('*****3', export_path) return FileSegment.get(export_path, create=create) # noinspection PyUnusedLocal class Main(CliMain, FileOptions, metaclass=CliCommandDelegateMacro(FileAPI, 'file_manager', '_result')): """Controller Server File Manager. """ def __init__(self): super().__init__() FileOptions.__init__(self) self.storage_root = '.' self.output_format = 'raw' self.file_action = None self.file_path = None self.file_argv = [] self._file_manager = None @cli_flags('-h', '--help', force_return=True) def _help(self, opt: str): """print help document""" self.print_help() @cli_flags('-v', '--verbose') def _verbose(self, opt: str): """verbose mode, enable logger""" LoggerFlag.set_formatter(LoggerFlag.OPTIONS.STYLE_DEFAULT) @cli_options('-f', '--format', value='FORMAT') def _output_format(self, opt: str, value: str): """output format. could be: raw : raw data format detail : detail raw data format json : json text """ if value not in ('raw', 'detail', 'json'): raise RuntimeError('illegal output format : ' + value) self.output_format = value @property def file_manager(self) -> FileManager: return self._file_manager def __enter__(self): self._file_manager = FileManager(self) self._file_manager.setup_storage() return self def __exit__(self, exc_type, exc_val, exc_tb): if self._file_manager is not None: self._file_manager.shutdown() self._file_manager = None def _result(self, command: str, result: Any): if isinstance(result, list): for o in result: self._result(command, o) print() elif self.output_format == 'raw': if isinstance(result, BaseInfo): result.print() else: print(result) elif self.output_format == 'detail': if isinstance(result, BaseInfo): result.print(detail_mode=True) else: print(result) elif self.output_format == 'json': if isinstance(result, JsonSerialize): result = result.as_json() pprint(result) @cli_command('help') def _help_command(self, command: str, argv: List[str]): """print help document""" self.print_help() @cli_command('user') def _user_setting(self, command: str, argv: List[str]): """user setting""" from .user.main import Main with self: Main(self.file_manager.user_setting).main(argv) @cli_command("export", value='...') def _export_file(self, command: str, argv: List[str]): """export file through the file manager""" return FileExportMain(command, self) @cli_command("import", value='...') def _import_file(self, command: str, argv: List[str]): """import file through the manager""" return FileImportMain(command, self) def run(self): self.print_help() if __name__ == '__main__': LoggerFlag.set_formatter(LoggerFlag.OPTIONS.STYLE_QUIET) Main().main()