# -*- coding: utf-8 -*- """Bootstrap OpenPype repositories.""" from __future__ import annotations import logging as log import os import re import shutil import sys import tempfile from pathlib import Path from typing import Union, Callable, List, Tuple import hashlib import platform from zipfile import ZipFile, BadZipFile from appdirs import user_data_dir from speedcopy import copyfile import semver from .user_settings import ( OpenPypeSecureRegistry, OpenPypeSettingsRegistry ) from .tools import ( get_openpype_global_settings, get_openpype_path_from_settings, get_expected_studio_version_str, get_local_openpype_path_from_settings ) LOG_INFO = 0 LOG_WARNING = 1 LOG_ERROR = 3 def sanitize_long_path(path): """Sanitize long paths (260 characters) when on Windows. Long paths are not capatible with ZipFile or reading a file, so we can shorten the path to use. Args: path (str): path to either directory or file. Returns: str: sanitized path """ if platform.system().lower() != "windows": return path path = os.path.abspath(path) if path.startswith("\\\\"): path = "\\\\?\\UNC\\" + path[2:] else: path = "\\\\?\\" + path return path def sha256sum(filename): """Calculate sha256 for content of the file. Args: filename (str): Path to file. Returns: str: hex encoded sha256 """ h = hashlib.sha256() b = bytearray(128 * 1024) mv = memoryview(b) with open(filename, 'rb', buffering=0) as f: for n in iter(lambda: f.readinto(mv), 0): h.update(mv[:n]) return h.hexdigest() class ZipFileLongPaths(ZipFile): def _extract_member(self, member, targetpath, pwd): return ZipFile._extract_member( self, member, sanitize_long_path(targetpath), pwd ) class OpenPypeVersion(semver.VersionInfo): """Class for storing information about OpenPype version. Attributes: path (str): path to OpenPype """ path = None _local_openpype_path = None # this should match any string complying with https://semver.org/ _VERSION_REGEX = re.compile(r"(?P0|[1-9]\d*)\.(?P0|[1-9]\d*)\.(?P0|[1-9]\d*)(?:-(?P[a-zA-Z\d\-.]*))?(?:\+(?P[a-zA-Z\d\-.]*))?") # noqa: E501 _installed_version = None def __init__(self, *args, **kwargs): """Create OpenPype version. .. deprecated:: 3.0.0-rc.2 `client` and `variant` are removed. Args: major (int): version when you make incompatible API changes. minor (int): version when you add functionality in a backwards-compatible manner. patch (int): version when you make backwards-compatible bug fixes. prerelease (str): an optional prerelease string build (str): an optional build string version (str): if set, it will be parsed and will override parameters like `major`, `minor` and so on. path (Path): path to version location. """ self.path = None if "version" in kwargs.keys(): if not kwargs.get("version"): raise ValueError("Invalid version specified") v = OpenPypeVersion.parse(kwargs.get("version")) kwargs["major"] = v.major kwargs["minor"] = v.minor kwargs["patch"] = v.patch kwargs["prerelease"] = v.prerelease kwargs["build"] = v.build kwargs.pop("version") if kwargs.get("path"): if isinstance(kwargs.get("path"), str): self.path = Path(kwargs.get("path")) elif isinstance(kwargs.get("path"), Path): self.path = kwargs.get("path") else: raise TypeError("Path must be str or Path") kwargs.pop("path") if "path" in kwargs.keys(): kwargs.pop("path") super().__init__(*args, **kwargs) def __repr__(self): return f"<{self.__class__.__name__}: {str(self)} - path={self.path}>" def __lt__(self, other: OpenPypeVersion): result = super().__lt__(other) # prefer path over no path if self == other and not self.path and other.path: return True if self == other and self.path and other.path and \ other.path.is_dir() and self.path.is_file(): return True if self.finalize_version() == other.finalize_version() and \ self.prerelease == other.prerelease: return True return result def get_main_version(self) -> str: """Return main version component. This returns x.x.x part of version from possibly more complex one like x.x.x-foo-bar. .. deprecated:: 3.0.0-rc.2 use `finalize_version()` instead. Returns: str: main version component """ return str(self.finalize_version()) @staticmethod def version_in_str(string: str) -> Union[None, OpenPypeVersion]: """Find OpenPype version in given string. Args: string (str): string to search. Returns: OpenPypeVersion: of detected or None. """ # strip .zip ext if present string = re.sub(r"\.zip$", "", string, flags=re.IGNORECASE) m = re.search(OpenPypeVersion._VERSION_REGEX, string) if not m: return None version = OpenPypeVersion.parse(string[m.start():m.end()]) return version def __hash__(self): return hash(self.path) if self.path else hash(str(self)) @staticmethod def is_version_in_dir( dir_item: Path, version: OpenPypeVersion) -> Tuple[bool, str]: """Test if path item is OpenPype version matching detected version. If item is directory that might (based on it's name) contain OpenPype version, check if it really does contain OpenPype and that their versions matches. Args: dir_item (Path): Directory to test. version (OpenPypeVersion): OpenPype version detected from name. Returns: Tuple: State and reason, True if it is valid OpenPype version, False otherwise. """ try: # add one 'openpype' level as inside dir there should # be many other repositories. version_str = OpenPypeVersion.get_version_string_from_directory( dir_item) # noqa: E501 version_check = OpenPypeVersion(version=version_str) except ValueError: return False, f"cannot determine version from {dir_item}" version_main = version_check.get_main_version() detected_main = version.get_main_version() if version_main != detected_main: return False, (f"dir version ({version}) and " f"its content version ({version_check}) " "doesn't match. Skipping.") return True, "Versions match" @staticmethod def is_version_in_zip( zip_item: Path, version: OpenPypeVersion) -> Tuple[bool, str]: """Test if zip path is OpenPype version matching detected version. Open zip file, look inside and parse version from OpenPype inside it. If there is none, or it is different from version specified in file name, skip it. Args: zip_item (Path): Zip file to test. version (OpenPypeVersion): Pype version detected from name. Returns: Tuple: State and reason, True if it is valid OpenPype version, False otherwise. """ # skip non-zip files if zip_item.suffix.lower() != ".zip": return False, "Not a zip" try: with ZipFile(zip_item, "r") as zip_file: with zip_file.open( "openpype/version.py") as version_file: zip_version = {} exec(version_file.read(), zip_version) try: version_check = OpenPypeVersion( version=zip_version["__version__"]) except ValueError as e: return False, str(e) version_main = version_check.get_main_version() # # noqa: E501 detected_main = version.get_main_version() # noqa: E501 if version_main != detected_main: return False, (f"zip version ({version}) " f"and its content version " f"({version_check}) " "doesn't match. Skipping.") except BadZipFile: return False, f"{zip_item} is not a zip file" except KeyError: return False, "Zip does not contain OpenPype" return True, "Versions match" @staticmethod def get_version_string_from_directory(repo_dir: Path) -> Union[str, None]: """Get version of OpenPype in given directory. Note: in frozen OpenPype installed in user data dir, this must point one level deeper as it is: `openpype-version-v3.0.0/openpype/version.py` Args: repo_dir (Path): Path to OpenPype repo. Returns: str: version string. None: if OpenPype is not found. """ # try to find version version_file = Path(repo_dir) / "openpype" / "version.py" if not version_file.exists(): return None version = {} with version_file.open("r") as fp: exec(fp.read(), version) return version['__version__'] @classmethod def get_openpype_path(cls): """Path to openpype zip directory. Path can be set through environment variable 'OPENPYPE_PATH' which is set during start of OpenPype if is not available. """ return os.getenv("OPENPYPE_PATH") @classmethod def get_local_openpype_path(cls): """Path to unzipped versions. By default it should be user appdata, but could be overridden by settings. """ if cls._local_openpype_path: return cls._local_openpype_path settings = get_openpype_global_settings(os.environ["OPENPYPE_MONGO"]) data_dir = get_local_openpype_path_from_settings(settings) if not data_dir: data_dir = Path(user_data_dir("openpype", "pypeclub")) cls._local_openpype_path = data_dir return data_dir @classmethod def openpype_path_is_set(cls): """Path to OpenPype zip directory is set.""" if cls.get_openpype_path(): return True return False @classmethod def openpype_path_is_accessible(cls): """Path to OpenPype zip directory is accessible. Exists for this machine. """ # First check if is set if not cls.openpype_path_is_set(): return False # Validate existence if Path(cls.get_openpype_path()).exists(): return True return False @classmethod def get_local_versions(cls) -> List: """Get all versions available on this machine. Returns: list: of compatible versions available on the machine. """ dir_to_search = cls.get_local_openpype_path() versions = cls.get_versions_from_directory(dir_to_search) return list(sorted(set(versions))) @classmethod def get_remote_versions(cls) -> List: """Get all versions available in OpenPype Path. Returns: list of OpenPypeVersions: Versions found in OpenPype path. """ # Return all local versions if arguments are set to None dir_to_search = None if cls.openpype_path_is_accessible(): dir_to_search = Path(cls.get_openpype_path()) else: registry = OpenPypeSettingsRegistry() try: registry_dir = Path(str(registry.get_item("openPypePath"))) if registry_dir.exists(): dir_to_search = registry_dir except ValueError: # nothing found in registry, we'll use data dir pass if not dir_to_search: return [] versions = cls.get_versions_from_directory(dir_to_search) return list(sorted(set(versions))) @staticmethod def get_versions_from_directory( openpype_dir: Path) -> List: """Get all detected OpenPype versions in directory. Args: openpype_dir (Path): Directory to scan. Returns: list of OpenPypeVersion Throws: ValueError: if invalid path is specified. """ openpype_versions = [] if not openpype_dir.exists() and not openpype_dir.is_dir(): return openpype_versions # iterate over directory in first level and find all that might # contain OpenPype. for item in openpype_dir.iterdir(): # if the item is directory with major.minor version, dive deeper if item.is_dir() and re.match(r"^\d+\.\d+$", item.name): _versions = OpenPypeVersion.get_versions_from_directory( item) if _versions: openpype_versions += _versions # if file exists, strip extension, in case of dir don't. name = item.name if item.is_dir() else item.stem result = OpenPypeVersion.version_in_str(name) if result: detected_version: OpenPypeVersion detected_version = result if item.is_dir() and not OpenPypeVersion.is_version_in_dir( item, detected_version )[0]: continue if item.is_file() and not OpenPypeVersion.is_version_in_zip( item, detected_version )[0]: continue detected_version.path = item openpype_versions.append(detected_version) return sorted(openpype_versions) @staticmethod def get_installed_version_str() -> str: """Get version of local OpenPype.""" version = {} path = Path(os.environ["OPENPYPE_ROOT"]) / "openpype" / "version.py" with open(path, "r") as fp: exec(fp.read(), version) return version["__version__"] @classmethod def get_installed_version(cls): """Get version of OpenPype inside build.""" if cls._installed_version is None: installed_version_str = cls.get_installed_version_str() if installed_version_str: cls._installed_version = OpenPypeVersion( version=installed_version_str, path=Path(os.environ["OPENPYPE_ROOT"]) ) return cls._installed_version @staticmethod def get_latest_version( local: bool = None, remote: bool = None ) -> Union[OpenPypeVersion, None]: """Get the latest available version. The version does not contain information about path and source. This is utility version to get the latest version from all found. Arguments 'local' and 'remote' define if local and remote repository versions are used. All versions are used if both are not set (or set to 'None'). If only one of them is set to 'True' the other is disabled. It is possible to set both to 'True' (same as both set to None) and to 'False' in that case only build version can be used. Args: local (bool, optional): List local versions if True. remote (bool, optional): List remote versions if True. Returns: Latest OpenPypeVersion or None """ if local is None and remote is None: local = True remote = True elif local is None and not remote: local = True elif remote is None and not local: remote = True installed_version = OpenPypeVersion.get_installed_version() local_versions = OpenPypeVersion.get_local_versions() if local else [] remote_versions = OpenPypeVersion.get_remote_versions() if remote else [] # noqa: E501 all_versions = local_versions + remote_versions + [installed_version] all_versions.sort() return all_versions[-1] @classmethod def get_expected_studio_version(cls, staging=False, global_settings=None): """Expected OpenPype version that should be used at the moment. If version is not defined in settings the latest found version is used. Using precached global settings is needed for usage inside OpenPype. Args: staging (bool): Staging version or production version. global_settings (dict): Optional precached global settings. Returns: OpenPypeVersion: Version that should be used. """ result = get_expected_studio_version_str(staging, global_settings) if not result: return None return OpenPypeVersion(version=result) def is_compatible(self, version: OpenPypeVersion): """Test build compatibility. This will simply compare major and minor versions (ignoring patch and the rest). Args: version (OpenPypeVersion): Version to check compatibility with. Returns: bool: if the version is compatible """ return self.major == version.major and self.minor == version.minor class BootstrapRepos: """Class for bootstrapping local OpenPype installation. Attributes: data_dir (Path): local OpenPype installation directory. registry (OpenPypeSettingsRegistry): OpenPype registry object. zip_filter (list): List of files to exclude from zip openpype_filter (list): list of top level directories to include in zip in OpenPype repository. """ def __init__(self, progress_callback: Callable = None, message=None): """Constructor. Args: progress_callback (callable): Optional callback method to report progress. message (QtCore.Signal, optional): Signal to report messages back. """ # vendor and app used to construct user data dir self._message = message self._log = log.getLogger(str(__class__)) self.set_data_dir(None) self.secure_registry = OpenPypeSecureRegistry("mongodb") self.registry = OpenPypeSettingsRegistry() self.zip_filter = [".pyc", "__pycache__"] self.openpype_filter = [ "openpype", "LICENSE" ] # dummy progress reporter def empty_progress(x: int): """Progress callback dummy.""" return x if not progress_callback: progress_callback = empty_progress self._progress_callback = progress_callback def set_data_dir(self, data_dir): if not data_dir: self.data_dir = Path(user_data_dir("openpype", "pypeclub")) else: self._print(f"overriding local folder: {data_dir}") self.data_dir = data_dir @staticmethod def get_version_path_from_list( version: str, version_list: list) -> Union[Path, None]: """Get path for specific version in list of OpenPype versions. Args: version (str): Version string to look for (1.2.4-nightly.1+test) version_list (list of OpenPypeVersion): list of version to search. Returns: Path: Path to given version. """ for v in version_list: if str(v) == version: return v.path return None @staticmethod def get_version(repo_dir: Path) -> Union[str, None]: """Get version of OpenPype in given directory. Note: in frozen OpenPype installed in user data dir, this must point one level deeper as it is: `openpype-version-v3.0.0/openpype/version.py` Args: repo_dir (Path): Path to OpenPype repo. Returns: str: version string. None: if OpenPype is not found. """ # try to find version version_file = Path(repo_dir) / "openpype" / "version.py" if not version_file.exists(): return None version = {} with version_file.open("r") as fp: exec(fp.read(), version) return version['__version__'] def create_version_from_live_code( self, repo_dir: Path = None) -> Union[OpenPypeVersion, None]: """Copy zip created from OpenPype repositories to user data dir. This detects OpenPype version either in local "live" OpenPype repository or in user provided path. Then it will zip it in temporary directory, and finally it will move it to destination which is user data directory. Existing files will be replaced. Args: repo_dir (Path, optional): Path to OpenPype repository. Returns: Path: path of installed repository file. """ # if repo dir is not set, we detect local "live" OpenPype repository # version and use it as a source. Otherwise, repo_dir is user # entered location. if repo_dir: version = self.get_version(repo_dir) else: installed_version = OpenPypeVersion.get_installed_version() version = str(installed_version) repo_dir = installed_version.path if not version: self._print("OpenPype not found.", LOG_ERROR) return # create destination directory destination = self.data_dir / f"{installed_version.major}.{installed_version.minor}" # noqa if not destination.exists(): destination.mkdir(parents=True) # create zip inside temporary directory. with tempfile.TemporaryDirectory() as temp_dir: temp_zip = \ Path(temp_dir) / f"openpype-v{version}.zip" self._print(f"creating zip: {temp_zip}") self._create_openpype_zip(temp_zip, repo_dir) if not os.path.exists(temp_zip): self._print("make archive failed.", LOG_ERROR) return None destination = self._move_zip_to_data_dir(temp_zip) return OpenPypeVersion(version=version, path=Path(destination)) def _move_zip_to_data_dir(self, zip_file) -> Union[None, Path]: """Move zip with OpenPype version to user data directory. Args: zip_file (Path): Path to zip file. Returns: None if move fails. Path to moved zip on success. """ version = OpenPypeVersion.version_in_str(zip_file.name) destination_dir = self.data_dir / f"{version.major}.{version.minor}" if not destination_dir.exists(): destination_dir.mkdir(parents=True) destination = destination_dir / zip_file.name if destination.exists(): self._print( f"Destination file {destination} exists, removing.", LOG_WARNING) try: destination.unlink() except Exception as e: self._print(str(e), LOG_ERROR, exc_info=True) return None if not destination_dir.exists(): destination_dir.mkdir(parents=True) elif not destination_dir.is_dir(): self._print( "Destination exists but is not directory.", LOG_ERROR) return None try: shutil.move(zip_file.as_posix(), destination_dir.as_posix()) except shutil.Error as e: self._print(str(e), LOG_ERROR, exc_info=True) return None return destination def _filter_dir(self, path: Path, path_filter: List) -> List[Path]: """Recursively crawl over path and filter.""" result = [] for item in path.iterdir(): if item.name in path_filter: continue if item.name.startswith('.'): continue if item.is_dir(): result.extend(self._filter_dir(item, path_filter)) else: result.append(item) return result def create_version_from_frozen_code(self) -> Union[None, OpenPypeVersion]: """Create OpenPype version from *frozen* code distributed by installer. This should be real edge case for those wanting to try out OpenPype without setting up whole infrastructure but is strongly discouraged in studio setup as this use local version independent of others that can be out of date. Returns: :class:`OpenPypeVersion` zip file to be installed. """ frozen_root = Path(sys.executable).parent openpype_list = [] for f in self.openpype_filter: if (frozen_root / f).is_dir(): openpype_list += self._filter_dir( frozen_root / f, self.zip_filter) else: openpype_list.append(frozen_root / f) version = self.get_version(frozen_root) # create zip inside temporary directory. with tempfile.TemporaryDirectory() as temp_dir: temp_zip = \ Path(temp_dir) / f"openpype-v{version}.zip" self._print(f"creating zip: {temp_zip}") with ZipFile(temp_zip, "w") as zip_file: progress = 0 openpype_inc = 98.0 / float(len(openpype_list)) file: Path for file in openpype_list: progress += openpype_inc self._progress_callback(int(progress)) arc_name = file.relative_to(frozen_root.parent) # we need to replace first part of path which starts with # something like `exe.win/linux....` with `openpype` as # this is expected by OpenPype in zip archive. arc_name = Path().joinpath(*arc_name.parts[1:]) zip_file.write(file, arc_name) destination = self._move_zip_to_data_dir(temp_zip) return OpenPypeVersion(version=version, path=destination) def _create_openpype_zip(self, zip_path: Path, openpype_path: Path) -> None: """Pack repositories and OpenPype into zip. We are using :mod:`ZipFile` instead :meth:`shutil.make_archive` because we need to decide what file and directories to include in zip and what not. They are determined by :attr:`zip_filter` on file level and :attr:`openpype_filter` on top level directory in OpenPype repository. Args: zip_path (Path): Path to zip file. openpype_path (Path): Path to OpenPype sources. """ # get filtered list of file in Pype repository # openpype_list = self._filter_dir(openpype_path, self.zip_filter) openpype_list = [] for f in self.openpype_filter: if (openpype_path / f).is_dir(): openpype_list += self._filter_dir( openpype_path / f, self.zip_filter) else: openpype_list.append(openpype_path / f) openpype_files = len(openpype_list) openpype_inc = 98.0 / float(openpype_files) with ZipFile(zip_path, "w") as zip_file: progress = 0 openpype_root = openpype_path.resolve() # generate list of filtered paths dir_filter = [openpype_root / f for f in self.openpype_filter] checksums = [] file: Path for file in openpype_list: progress += openpype_inc self._progress_callback(int(progress)) # if file resides in filtered path, skip it is_inside = None df: Path for df in dir_filter: try: is_inside = file.resolve().relative_to(df) except ValueError: pass if not is_inside: continue processed_path = file self._print(f"- processing {processed_path}") checksums.append( ( sha256sum(sanitize_long_path(file.as_posix())), file.resolve().relative_to(openpype_root) ) ) zip_file.write( file, file.resolve().relative_to(openpype_root)) checksums_str = "" for c in checksums: file_str = c[1] if platform.system().lower() == "windows": file_str = c[1].as_posix().replace("\\", "/") checksums_str += "{}:{}\n".format(c[0], file_str) zip_file.writestr("checksums", checksums_str) # test if zip is ok zip_file.testzip() self._progress_callback(100) def validate_openpype_version(self, path: Path) -> tuple: """Validate version directory or zip file. This will load `checksums` file if present, calculate checksums of existing files in given path and compare. It will also compare lists of files together for missing files. Args: path (Path): Path to OpenPype version to validate. Returns: tuple(bool, str): with version validity as first item and string with reason as second. """ if os.getenv("OPENPYPE_DONT_VALIDATE_VERSION"): return True, "Disabled validation" if not path.exists(): return False, "Path doesn't exist" if path.is_file(): return self._validate_zip(path) return self._validate_dir(path) @staticmethod def _validate_zip(path: Path) -> tuple: """Validate content of zip file.""" with ZipFile(path, "r") as zip_file: # read checksums try: checksums_data = str(zip_file.read("checksums")) except IOError: # FIXME: This should be set to False sometimes in the future return True, "Cannot read checksums for archive." # split it to the list of tuples checksums = [ tuple(line.split(":")) for line in checksums_data.split("\n") if line ] # get list of files in zip minus `checksums` file itself # and turn in to set to compare against list of files # from checksum file. If difference exists, something is # wrong files_in_zip = set(zip_file.namelist()) files_in_zip.remove("checksums") files_in_checksum = {file[1] for file in checksums} diff = files_in_zip.difference(files_in_checksum) if diff: return False, f"Missing files {diff}" # calculate and compare checksums in the zip file for file_checksum, file_name in checksums: if platform.system().lower() == "windows": file_name = file_name.replace("/", "\\") h = hashlib.sha256() try: h.update(zip_file.read(file_name)) except FileNotFoundError: return False, f"Missing file [ {file_name} ]" if h.hexdigest() != file_checksum: return False, f"Invalid checksum on {file_name}" return True, "All ok" @staticmethod def _validate_dir(path: Path) -> tuple: """Validate checksums in a given path. Args: path (Path): path to folder to validate. Returns: tuple(bool, str): returns status and reason as a bool and str in a tuple. """ checksums_file = Path(path / "checksums") if not checksums_file.exists(): # FIXME: This should be set to False sometimes in the future return True, "Cannot read checksums for archive." checksums_data = checksums_file.read_text() checksums = [ tuple(line.split(":")) for line in checksums_data.split("\n") if line ] # compare file list against list of files from checksum file. # If difference exists, something is wrong and we invalidate directly files_in_dir = set( file.relative_to(path).as_posix() for file in path.iterdir() if file.is_file() ) files_in_dir.remove("checksums") files_in_checksum = {file[1] for file in checksums} diff = files_in_dir.difference(files_in_checksum) if diff: return False, f"Missing files {diff}" # calculate and compare checksums for file_checksum, file_name in checksums: if platform.system().lower() == "windows": file_name = file_name.replace("/", "\\") try: current = sha256sum( sanitize_long_path((path / file_name).as_posix()) ) except FileNotFoundError: return False, f"Missing file [ {file_name} ]" if file_checksum != current: return False, f"Invalid checksum on {file_name}" return True, "All ok" @staticmethod def add_paths_from_archive(archive: Path) -> None: """Add first-level directory and 'repos' as paths to :mod:`sys.path`. This will enable Python to import OpenPype and modules in `repos` submodule directory in zip file. Adding to both `sys.path` and `PYTHONPATH`, skipping duplicates. Args: archive (Path): path to archive. .. deprecated:: 3.0 we don't use zip archives directly """ if not archive.is_file() and not archive.exists(): raise ValueError("Archive is not file.") archive_path = str(archive) sys.path.insert(0, archive_path) pythonpath = os.getenv("PYTHONPATH", "") python_paths = pythonpath.split(os.pathsep) python_paths.insert(0, archive_path) os.environ["PYTHONPATH"] = os.pathsep.join(python_paths) @staticmethod def add_paths_from_directory(directory: Path) -> None: """Add repos first level directories as paths to :mod:`sys.path`. This works the same as :meth:`add_paths_from_archive` but in specified directory. Adding to both `sys.path` and `PYTHONPATH`, skipping duplicates. Args: directory (Path): path to directory. """ sys.path.insert(0, directory.as_posix()) @staticmethod def find_openpype_version( version: Union[str, OpenPypeVersion] ) -> Union[OpenPypeVersion, None]: """Find location of specified OpenPype version. Args: version (Union[str, OpenPypeVersion): Version to find. Returns: requested OpenPypeVersion. """ installed_version = OpenPypeVersion.get_installed_version() if isinstance(version, str): version = OpenPypeVersion(version=version) if installed_version == version: return installed_version local_versions = OpenPypeVersion.get_local_versions() zip_version = None for local_version in local_versions: if local_version == version: if local_version.path.suffix.lower() == ".zip": zip_version = local_version else: return local_version if zip_version is not None: return zip_version remote_versions = OpenPypeVersion.get_remote_versions() return next( ( remote_version for remote_version in remote_versions if remote_version == version ), None) @staticmethod def find_latest_openpype_version() -> Union[OpenPypeVersion, None]: """Find the latest available OpenPype version in all location. Returns: Latest OpenPype version on None if nothing was found. """ installed_version = OpenPypeVersion.get_installed_version() local_versions = OpenPypeVersion.get_local_versions() remote_versions = OpenPypeVersion.get_remote_versions() all_versions = local_versions + remote_versions + [installed_version] if not all_versions: return None all_versions.sort() latest_version = all_versions[-1] if latest_version == installed_version: return latest_version if not latest_version.path.is_dir(): for version in local_versions: if version == latest_version and version.path.is_dir(): latest_version = version break return latest_version def find_openpype( self, openpype_path: Union[Path, str] = None, include_zips: bool = False ) -> Union[List[OpenPypeVersion], None]: """Get ordered dict of detected OpenPype version. Resolution order for OpenPype is following: 1) First we test for ``OPENPYPE_PATH`` environment variable 2) We try to find ``openPypePath`` in registry setting 3) We use user data directory Args: openpype_path (Path or str, optional): Try to find OpenPype on the given path or url. include_zips (bool, optional): If set True it will try to find OpenPype in zip files in given directory. Returns: dict of Path: Dictionary of detected OpenPype version. Key is version, value is path to zip file. None: if OpenPype is not found. Todo: implement git/url support as OpenPype location, so it would be possible to enter git url, OpenPype would check it out and if it is ok install it as normal version. """ if openpype_path and not isinstance(openpype_path, Path): raise NotImplementedError( ("Finding OpenPype in non-filesystem locations is" " not implemented yet.")) # if checks bellow for OPENPYPE_PATH and registry fails, use data_dir # DEPRECATED: lookup in root of this folder is deprecated in favour # of major.minor sub-folders. dirs_to_search = [self.data_dir] if openpype_path: dirs_to_search = [openpype_path] elif os.getenv("OPENPYPE_PATH") \ and Path(os.getenv("OPENPYPE_PATH")).exists(): # first try OPENPYPE_PATH and if that is not available, # try registry. dirs_to_search = [Path(os.getenv("OPENPYPE_PATH"))] else: try: registry_dir = Path( str(self.registry.get_item("openPypePath"))) if registry_dir.exists(): dirs_to_search = [registry_dir] except ValueError: # nothing found in registry, we'll use data dir pass openpype_versions = [] for dir_to_search in dirs_to_search: try: openpype_versions += self.get_openpype_versions( dir_to_search) except ValueError: # location is invalid, skip it pass if not include_zips: openpype_versions = [ v for v in openpype_versions if v.path.suffix != ".zip" ] # remove duplicates openpype_versions = sorted(list(set(openpype_versions))) return openpype_versions def process_entered_location(self, location: str) -> Union[Path, None]: """Process user entered location string. It decides if location string is mongodb url or path. If it is mongodb url, it will connect and load ``OPENPYPE_PATH`` from there and use it as path to OpenPype. In it is _not_ mongodb url, it is assumed we have a path, this is tested and zip file is produced and installed using :meth:`create_version_from_live_code`. Args: location (str): User entered location. Returns: Path: to OpenPype zip produced from this location. None: Zipping failed. """ openpype_path = None # try to get OpenPype path from mongo. if location.startswith("mongodb"): global_settings = get_openpype_global_settings(location) openpype_path = get_openpype_path_from_settings(global_settings) if not openpype_path: self._print("cannot find OPENPYPE_PATH in settings.") return None # if not successful, consider location to be fs path. if not openpype_path: openpype_path = Path(location) # test if this path does exist. if not openpype_path.exists(): self._print(f"{openpype_path} doesn't exists.") return None # test if entered path isn't user data dir if self.data_dir == openpype_path: self._print("cannot point to user data dir", LOG_ERROR) return None # find openpype zip files in location. There can be # either "live" OpenPype repository, or multiple zip files or even # multiple OpenPype version directories. This process looks into zip # files and directories and tries to parse `version.py` file. versions = self.find_openpype(openpype_path, include_zips=True) if versions: self._print(f"found OpenPype in [ {openpype_path} ]") self._print(f"latest version found is [ {versions[-1]} ]") return self.install_version(versions[-1]) # if we got here, it means that location is "live" # OpenPype repository. We'll create zip from it and move it to user # data dir. live_openpype = self.create_version_from_live_code(openpype_path) if not live_openpype.path.exists(): self._print(f"installing zip {live_openpype} failed.", LOG_ERROR) return None # install it return self.install_version(live_openpype) def _print(self, message: str, level: int = LOG_INFO, exc_info: bool = False): """Helper function passing logs to UI and to logger. Supporting 3 levels of logs defined with `LOG_INFO`, `LOG_WARNING` and `LOG_ERROR` constants. Args: message (str): Message to log. level (int, optional): Log level to use. exc_info (bool, optional): Exception info object to pass to logger. """ if self._message: self._message.emit(message, level == LOG_ERROR) if level == LOG_WARNING: self._log.warning(message, exc_info=exc_info) return if level == LOG_ERROR: self._log.error(message, exc_info=exc_info) return self._log.info(message, exc_info=exc_info) def extract_openpype(self, version: OpenPypeVersion) -> Union[Path, None]: """Extract zipped OpenPype version to user data directory. Args: version (OpenPypeVersion): Version of OpenPype. Returns: Path: path to extracted version. None: if something failed. """ if not version.path: raise ValueError( f"version {version} is not associated with any file") destination = self.data_dir / f"{version.major}.{version.minor}" / version.path.stem # noqa if destination.exists() and destination.is_dir(): try: shutil.rmtree(destination) except OSError as e: msg = f"!!! Cannot remove already existing {destination}" self._print(msg, LOG_ERROR, exc_info=True) raise e destination.mkdir(parents=True) # extract zip there self._print("Extracting zip to destination ...") with ZipFileLongPaths(version.path, "r") as zip_ref: zip_ref.extractall(destination) self._print(f"Installed as {version.path.stem}") return destination def is_inside_user_data(self, path: Path) -> bool: """Test if version is located in user data dir. Args: path (Path) Path to test. Returns: True if path is inside user data dir. """ is_inside = False try: is_inside = path.resolve().relative_to( self.data_dir) except ValueError: # if relative path cannot be calculated, OpenPype version is not # inside user data dir pass return is_inside def install_version(self, openpype_version: OpenPypeVersion, force: bool = False) -> Path: """Install OpenPype version to user data directory. Args: openpype_version (OpenPypeVersion): OpenPype version to install. force (bool, optional): Force overwrite existing version. Returns: Path: Path to installed OpenPype. Raises: OpenPypeVersionExists: If not forced and this version already exist in user data directory. OpenPypeVersionInvalid: If version to install is invalid. OpenPypeVersionIOError: If copying or zipping fail. """ if self.is_inside_user_data(openpype_version.path) and not openpype_version.path.is_file(): # noqa raise OpenPypeVersionExists( "OpenPype already inside user data dir") # determine destination directory name # for zip file strip suffix, in case of dir use whole dir name if openpype_version.path.is_dir(): dir_name = openpype_version.path.name else: dir_name = openpype_version.path.stem destination = self.data_dir / f"{openpype_version.major}.{openpype_version.minor}" / dir_name # noqa # test if destination directory already exist, if so lets delete it. if destination.exists() and force: self._print("removing existing directory") try: shutil.rmtree(destination) except OSError as e: self._print( f"cannot remove already existing {destination}", LOG_ERROR, exc_info=True) raise OpenPypeVersionIOError( f"cannot remove existing {destination}") from e elif destination.exists() and not force: self._print("destination directory already exists") raise OpenPypeVersionExists(f"{destination} already exist.") else: # create destination parent directories even if they don't exist. destination.mkdir(parents=True) remove_source_file = False # version is directory if openpype_version.path.is_dir(): # create zip inside temporary directory. self._print("Creating zip from directory ...") self._progress_callback(0) with tempfile.TemporaryDirectory() as temp_dir: temp_zip = \ Path(temp_dir) / f"openpype-v{openpype_version}.zip" self._print(f"creating zip: {temp_zip}") self._create_openpype_zip(temp_zip, openpype_version.path) if not os.path.exists(temp_zip): self._print("make archive failed.", LOG_ERROR) raise OpenPypeVersionIOError("Zip creation failed.") # set zip as version source openpype_version.path = temp_zip if self.is_inside_user_data(openpype_version.path): raise OpenPypeVersionInvalid( "Version is in user data dir.") openpype_version.path = self._copy_zip( openpype_version.path, destination) elif openpype_version.path.is_file(): # check if file is zip (by extension) if openpype_version.path.suffix.lower() != ".zip": raise OpenPypeVersionInvalid("Invalid file format") if not self.is_inside_user_data(openpype_version.path): self._progress_callback(35) openpype_version.path = self._copy_zip( openpype_version.path, destination) # Mark zip to be deleted when done remove_source_file = True # extract zip there self._print("extracting zip to destination ...") with ZipFileLongPaths(openpype_version.path, "r") as zip_ref: self._progress_callback(75) zip_ref.extractall(destination) self._progress_callback(100) # Remove zip file copied to local app data if remove_source_file: os.remove(openpype_version.path) return destination def _copy_zip(self, source: Path, destination: Path) -> Path: try: # copy file to destination self._print("Copying zip to destination ...") _destination_zip = destination.parent / source.name # noqa: E501 copyfile( source.as_posix(), _destination_zip.as_posix()) except OSError as e: self._print( "cannot copy version to user data directory", LOG_ERROR, exc_info=True) raise OpenPypeVersionIOError(( f"can't copy version {source.as_posix()} " f"to destination {destination.parent.as_posix()}")) from e return _destination_zip def _is_openpype_in_dir(self, dir_item: Path, detected_version: OpenPypeVersion) -> bool: """Test if path item is OpenPype version matching detected version. If item is directory that might (based on it's name) contain OpenPype version, check if it really does contain OpenPype and that their versions matches. Args: dir_item (Path): Directory to test. detected_version (OpenPypeVersion): OpenPype version detected from name. Returns: True if it is valid OpenPype version, False otherwise. """ try: # add one 'openpype' level as inside dir there should # be many other repositories. version_str = BootstrapRepos.get_version(dir_item) version_check = OpenPypeVersion(version=version_str) except ValueError: self._print( f"cannot determine version from {dir_item}", True) return False version_main = version_check.get_main_version() detected_main = detected_version.get_main_version() if version_main != detected_main: self._print( (f"dir version ({detected_version}) and " f"its content version ({version_check}) " "doesn't match. Skipping.")) return False return True def _is_openpype_in_zip(self, zip_item: Path, detected_version: OpenPypeVersion) -> bool: """Test if zip path is OpenPype version matching detected version. Open zip file, look inside and parse version from OpenPype inside it. If there is none, or it is different from version specified in file name, skip it. Args: zip_item (Path): Zip file to test. detected_version (OpenPypeVersion): Pype version detected from name. Returns: True if it is valid OpenPype version, False otherwise. """ # skip non-zip files if zip_item.suffix.lower() != ".zip": return False try: with ZipFile(zip_item, "r") as zip_file: with zip_file.open( "openpype/version.py") as version_file: zip_version = {} exec(version_file.read(), zip_version) try: version_check = OpenPypeVersion( version=zip_version["__version__"]) except ValueError as e: self._print(str(e), True) return False version_main = version_check.get_main_version() # noqa: E501 detected_main = detected_version.get_main_version() # noqa: E501 if version_main != detected_main: self._print( (f"zip version ({detected_version}) " f"and its content version " f"({version_check}) " "doesn't match. Skipping."), True) return False except BadZipFile: self._print(f"{zip_item} is not a zip file", True) return False except KeyError: self._print("Zip does not contain OpenPype", True) return False return True def get_openpype_versions(self, openpype_dir: Path) -> list: """Get all detected OpenPype versions in directory. Args: openpype_dir (Path): Directory to scan. Returns: list of OpenPypeVersion Throws: ValueError: if invalid path is specified. """ if not openpype_dir.exists() and not openpype_dir.is_dir(): raise ValueError(f"specified directory {openpype_dir} is invalid") openpype_versions = [] # iterate over directory in first level and find all that might # contain OpenPype. for item in openpype_dir.iterdir(): # if the item is directory with major.minor version, dive deeper if item.is_dir() and re.match(r"^\d+\.\d+$", item.name): _versions = self.get_openpype_versions(item) if _versions: openpype_versions += _versions # if it is file, strip extension, in case of dir don't. name = item.name if item.is_dir() else item.stem result = OpenPypeVersion.version_in_str(name) if result: detected_version: OpenPypeVersion detected_version = result if item.is_dir() and not self._is_openpype_in_dir( item, detected_version ): continue if item.is_file() and not self._is_openpype_in_zip( item, detected_version ): continue detected_version.path = item openpype_versions.append(detected_version) return sorted(openpype_versions) class OpenPypeVersionExists(Exception): """Exception for handling existing OpenPype version.""" pass class OpenPypeVersionInvalid(Exception): """Exception for handling invalid OpenPype version.""" pass class OpenPypeVersionIOError(Exception): """Exception for handling IO errors in OpenPype version.""" pass