from pathlib import Path import os from modelspace.Essentials import SetsDict from modelspace.ModelPackage import ModelPackage from modelspace.ModelPackageSelector import ModelPackageSelector from modelspace.ModelSpaceDatabase import ModelSpaceDatabase from modelspace.Repository import Repository from modelspace.PathMap import map class ModelSpace: def __init__(self, repo: Repository, path, layout): self.layout = map.get(layout, None) if not self.layout: raise RuntimeError(f"No such layout: {layout}") self.repo = repo self.path = path self.config_dir = Path(self.path) / '.vaiola' os.makedirs(self.config_dir, exist_ok=True) self._copier = self._hard_link_copier self.db = ModelSpaceDatabase(str(self.config_dir)) self.selector = ModelPackageSelector(self.repo.model_sub_repo, set(self.installed_packages)) self.temp_installed_resources = self.installed_resources self.temp_files_pending = SetsDict() self.temp_file_conflicts = SetsDict() def _reset_selector(self): self.selector = ModelPackageSelector(self.repo.model_sub_repo, set(self.installed_packages)) self.temp_installed_resources = self.installed_resources self.temp_files_pending = SetsDict() self.temp_file_conflicts = SetsDict() def check_for_file_conflicts(self, pkg_id, path: str): path = Path(path) if path.exists() or path in self.temp_files_pending.index: self.temp_file_conflicts.add(pkg_id, path) self.temp_files_pending.add(pkg_id, path) def get_dest_dir(self, package: ModelPackage, no_lineage = False): type_subdir = self.layout.get(package.package_type, None) if not type_subdir: raise RuntimeError(f"This type ({package.package_type}) of packages not supported by current modelspace") lineage_subdir = package.lineage if no_lineage: return str(type_subdir) return str(Path(type_subdir) / lineage_subdir) def _register_package_in_temp(self, package: ModelPackage): for resource in package.provides: self.temp_installed_resources.add(resource, package.uuid) def _deptree(self, resource: str, answer = None, depth = 0) -> set[ModelPackage]: selected_packages = set() packages = self.selector.run(resource, answer) deps = self.repo.model_sub_repo.deps_from_pkg_list(packages) deps = deps - set(self.temp_installed_resources.keys) for package in packages: self._register_package_in_temp(package) for dep in deps: selected_packages = selected_packages | self._deptree(dep, answer, depth + 1) return selected_packages | packages def install(self, resources: str | list[str], answer = None, depth = 0) -> None: if depth == 0: self._reset_selector() if isinstance(resources, str): resources = [resources] selected_packages = set() for resource in resources: selected_packages = selected_packages | self._deptree(resource, answer) for package in selected_packages: for file in package.files: self.check_for_file_conflicts(package.uuid, Path(self.path) / self.get_dest_dir(package, self.layout.no_lineage) / file) if len(self.temp_file_conflicts.index) != 0: raise RuntimeError("File conflicts detected") for package in selected_packages: self.db.create_manual(package.uuid) # TODO add deps separation from manual packages for file in package.files: self.db.create_file(package.uuid, str(Path(self.get_dest_dir(package, self.layout.no_lineage)) / file)) self._copier(package.path / "files", str(Path(self.path) / self.get_dest_dir(package, self.layout.no_lineage)),[file]) @staticmethod def _hard_link_copier(source_dir, dest_dir, file_paths): """Создает жесткие ссылки для файлов из источника в директорию назначения Args: source_dir (str): Путь к директории-источнику dest_dir (str): Путь к директории-назначению file_paths (list): Список относительных путей файлов в директории источника """ for file_path in file_paths: # Формируем полный путь к исходному файлу source_file = os.path.join(source_dir, file_path) # Формируем полный путь к целевому файлу dest_file = os.path.join(dest_dir, file_path) # Создаем необходимые поддиректории в директории назначения dest_dir_name = os.path.dirname(dest_file) if dest_dir_name and not os.path.exists(dest_dir_name): os.makedirs(dest_dir_name) # Создаем жесткую ссылку os.link(source_file, dest_file) @property def available_packages(self): return self.repo.model_sub_repo.packages @property def available_resources(self): return self.repo.model_sub_repo.resources @property def installed_manuals(self): return [self.repo.model_sub_repo.package_by_id(m[0]) if m and len(m) > 0 else None for m in self.db.get_all_manuals()] if self.db.get_all_manuals() else list() @property def installed_deps(self): return [self.repo.model_sub_repo.package_by_id(m[0]) if m and len(m) > 0 else None for m in self.db.get_all_deps()] if self.db.get_all_deps() else list() @property def installed_packages(self): return self.installed_manuals + self.installed_deps @property def installed_resources(self): return self.repo.model_sub_repo.resources_from_pkg_list(p.uuid for p in self.installed_packages)