From c97743e65f25e7f3b3a956ebeb28449444098687 Mon Sep 17 00:00:00 2001 From: Nicholas FitzRoy-Dale Date: Sat, 22 Jun 2024 16:36:36 +1000 Subject: [PATCH] (Start to) refactor zipped filesystems --- rime/filesystem/android.py | 18 +++--------- rime/filesystem/ios.py | 55 +++++++++++------------------------ rime/filesystem/zipsupport.py | 21 +++++++++++++ 3 files changed, 42 insertions(+), 52 deletions(-) create mode 100644 rime/filesystem/zipsupport.py diff --git a/rime/filesystem/android.py b/rime/filesystem/android.py index 3d02a17..e88ee7b 100644 --- a/rime/filesystem/android.py +++ b/rime/filesystem/android.py @@ -11,6 +11,7 @@ from .base import DeviceFilesystem from .devicesettings import DeviceSettings from .fslibfilesystem import FSLibFilesystem +from . import zipsupport class AndroidDeviceFilesystem(DeviceFilesystem): @@ -88,17 +89,6 @@ class AndroidZippedDeviceFilesystem(DeviceFilesystem): Zipped filesystem of an Android device. Currently supports only read mode for the data. - The class assumes that there is one directory in the .zip file - and all the other files and directories are located withn that directory. - - file.zip - |- main_dir - |- _rime_settings.db - |- sdcard - |- ... - |- data - |- ... - The contents of the .zip file are extracted in a temporary directory and then the (only) directory from within the temporary directory (the `main_dir`) is used to instantiate a filesystem. All queries @@ -112,8 +102,8 @@ def __init__(self, id_: str, root: str, metadata_db_path: str): self.temp_root = tempfile.TemporaryDirectory() with zipfile.ZipFile(root) as zp: + main_dir = zipsupport.get_zipfile_main_dir(zp) zp.extractall(path=self.temp_root.name) - main_dir, = zipfile.Path(zp).iterdir() # instantiate a filesystem from the temporary directory self._fs = fs.osfs.OSFS(os.path.join(self.temp_root.name, main_dir.name)) @@ -127,8 +117,8 @@ def is_device_filesystem(cls, path): with zipfile.ZipFile(path) as zp: # get the main directory contained in the .zip container file - main_dir, = zipfile.Path(zp).iterdir() - return zipfile.Path(zp, os.path.join(main_dir.name, 'data', 'data', 'android/')).exists() + main_dir = zipsupport.get_zipfile_main_dir(zp) + return (main_dir / 'data' / 'data' / 'android').exists() @classmethod def create(cls, id_: str, root: str, metadata_db_path: str, template: Optional['DeviceFilesystem'] = None)\ diff --git a/rime/filesystem/ios.py b/rime/filesystem/ios.py index 70b5b11..c52e374 100644 --- a/rime/filesystem/ios.py +++ b/rime/filesystem/ios.py @@ -19,6 +19,7 @@ from .exceptions import NoPassphraseError, NotDecryptedError, WrongPassphraseError from .ensuredir import ensuredir from . import metadata +from . import zipsupport from ..sql import Table, Query, get_field_indices, sqlite3_connect_filename as sqlite3_connect_with_regex_support log = getLogger(__name__) @@ -300,29 +301,7 @@ class IosZippedDeviceFilesystem(DeviceFilesystem, IosDeviceFilesystemBase): """ Zipped filesystem of an iOS device. Currently supports only read mode for the data. - - The class assumes there is only one directory in the .zip file - and all the other files and directories are located withn that directory. - - file.zip - |- main_dir - |- Manifest.db - |- Info.plist - |- 7c - |- 7c7fba66680ef796b916b067077cc246adacf01d """ - - @staticmethod - def get_main_dir(zp: zipfile.ZipFile) -> zipfile.Path: - """ - Get the main directory from within the zip file. The zip file - should contain one directory and all other files should be in - that directory. - """ - root = zipfile.Path(zp) - main_dir, = root.iterdir() - return main_dir - def __init__(self, id_: str, root: str, metadata_db_path: str): self.id_ = id_ @@ -336,12 +315,12 @@ def __init__(self, id_: str, root: str, metadata_db_path: str): with zipfile.ZipFile(self.root) as zp: # get the main directory contained in the .zip container file - main_dir = self.get_main_dir(zp) + main_dir = zipsupport.get_zipfile_main_dir(zp) - with zp.open(os.path.join(main_dir.name, 'Manifest.db')) as zf: + with zp.open(main_dir / 'Manifest.db') as zf: self.temp_manifest.write(zf.read()) - with zp.open(os.path.join(main_dir.name, '_rime_settings.db')) as zf: + with zp.open(main_dir / '_rime_settings.db') as zf: self.temp_settings.write(zf.read()) self.manifest = sqlite3_connect_with_regex_support(self.temp_manifest.name, read_only=True) @@ -360,10 +339,10 @@ def is_device_filesystem(cls, path) -> bool: with zipfile.ZipFile(path) as zp: # get the main directory contained in the .zip container file - main_dir = cls.get_main_dir(zp) + main_dir = zipsupport.get_zipfile_main_dir(zp) return ( - zipfile.Path(zp, os.path.join(main_dir.name, 'Manifest.db')).exists() - and zipfile.Path(zp, os.path.join(main_dir.name, 'Info.plist')).exists() + zipfile.Path(zp, main_dir / 'Manifest.db').exists() + and zipfile.Path(zp, main_dir / 'Info.plist').exists() ) @classmethod @@ -388,22 +367,22 @@ def exists(self, path) -> bool: # contains the `real_path with zipfile.ZipFile(self.root) as zp: # get the main directory contained in the .zip container file - main_dir = self.get_main_dir(zp) - return zipfile.Path(zp, os.path.join(main_dir.name, real_path)).exists() + main_dir = zipsupport.get_zipfile_main_dir(zp) + return zipfile.Path(zp, main_dir.name / real_path).exists() def getsize(self, path) -> int: with zipfile.ZipFile(self.root) as zp: # get the main directory contained in the .zip container file - main_dir = self.get_main_dir(zp) - return zp.getinfo(os.path.join(str(main_dir), self._converter.get_hashed_pathname(path))).file_size + main_dir = zipsupport.get_zipfile_main_dir(zp) + return zp.getinfo(main_dir / self._converter.get_hashed_pathname(path)).file_size def ios_open_raw(self, path, mode): # TODO: mode tmp_copy = tempfile.NamedTemporaryFile(mode='w+b') with zipfile.ZipFile(self.root) as zp: # get the main directory contained in the .zip container file - main_dir = self.get_main_dir(zp) - with zp.open(os.path.join(main_dir.name, path)) as zf: + main_dir = zipsupport.get_zipfile_main_dir(zp) + with zp.open(main_dir / path) as zf: tmp_copy.write(zf.read()) return tmp_copy @@ -418,8 +397,8 @@ def sqlite3_connect(self, path, read_only=True): with zipfile.ZipFile(self.root) as zp: # get the main directory contained in the .zip container file - main_dir = self.get_main_dir(zp) - with zp.open(os.path.join(main_dir.name, self._converter.get_hashed_pathname(path))) as zf: + main_dir = zipsupport.get_zipfile_main_dir(zp) + with zp.open(main_dir / self._converter.get_hashed_pathname(path)) as zf: tmp_copy.write(zf.read()) log.debug(f"iOS connecting to {tmp_copy.name}") @@ -435,8 +414,8 @@ def lock(self, locked: bool): # for persistent settings preferenses with zipfile.ZipFile(self.root, 'w') as zp: # get the main directory contained in the .zip container file - main_dir = self.get_main_dir(zp) - with zp.open(os.path.join(str(main_dir), '_rime_settings.db'), 'w') as zf: + main_dir = zipsupport.get_zipfile_main_dir(zp) + with zp.open(main_dir / '_rime_settings.db', 'w') as zf: zf.write(self.temp_settings.read()) def is_locked(self) -> bool: diff --git a/rime/filesystem/zipsupport.py b/rime/filesystem/zipsupport.py new file mode 100644 index 0000000..2cc515b --- /dev/null +++ b/rime/filesystem/zipsupport.py @@ -0,0 +1,21 @@ +from zipfile import ZipFile, Path + +def get_zipfile_main_dir(zf: ZipFile) -> Path: + """ + Zipped filesystem support assumes that there is one directory in the .zip file + and all the other files and directories are located withn that directory. + + file.zip + |- main_dir + |- _rime_settings.db + |- sdcard + |- ... + |- data + |- ... + """ + path = Path(zf) + for element in path.iterdir(): + if element.is_dir(): + return element + + raise ValueError("The zipfile does not contain a single directory.")