diff --git a/client/ayon_core/host/interfaces/workfiles.py b/client/ayon_core/host/interfaces/workfiles.py index 97985b754a..456ba0b9d4 100644 --- a/client/ayon_core/host/interfaces/workfiles.py +++ b/client/ayon_core/host/interfaces/workfiles.py @@ -7,6 +7,7 @@ import typing from typing import Optional, Any import ayon_api +import arrow if typing.TYPE_CHECKING: from ayon_core.pipeline import Anatomy @@ -77,6 +78,70 @@ class WorkfileInfo: return WorkfileInfo(**data) +@dataclass +class PublishedWorkfileInfo: + folder_id: str + task_id: Optional[str] + representation_id: str + filepath: str + created_at: float + author: str + available: bool + file_size: Optional[float] + file_created: Optional[float] + file_modified: Optional[float] + + @classmethod + def new( + cls, + folder_id: str, + task_id: Optional[str], + repre_entity: dict[str, Any], + filepath: str, + author: str, + available: bool, + file_size: Optional[float], + file_modified: Optional[float], + file_created: Optional[float], + ): + created_at = arrow.get(repre_entity["createdAt"]).to("local") + + return cls( + folder_id=folder_id, + task_id=task_id, + representation_id=repre_entity["id"], + filepath=filepath, + created_at=created_at.float_timestamp, + author=author, + available=available, + file_size=file_size, + file_created=file_created, + file_modified=file_modified, + ) + + def to_data(self): + """Converts file item to data. + + Returns: + dict[str, Any]: Workfile item data. + + """ + return asdict(self) + + @classmethod + def from_data(self, data): + """Converts data to workfile item. + + Args: + data (dict[str, Any]): Workfile item data. + + Returns: + WorkfileInfo: File item. + + """ + return WorkfileInfo(**data) + + class IWorkfileHost: """Implementation requirements to be able use workfile utils and tool.""" @@ -264,6 +329,110 @@ class IWorkfileHost: return items + def list_published_workfiles( + self, + project_name: str, + folder_id: str, + anatomy: Optional["Anatomy"] = None, + version_entities: Optional[list[dict[str, Any]]] = None, + repre_entities: Optional[list[dict[str, Any]]] = None, + ) -> list[PublishedWorkfileInfo]: + """List published workfiles for given folder. + + Default implementation looks for products with 'workfile' + product type. + + Pre-fetched entities have mandatory fields to be fetched. + - Version: 'id', 'author', 'taskId' + - Representation: 'id', 'versionId', 'files' + + Args: + project_name (str): Project name. + folder_id (str): Folder id. + anatomy (Anatomy): Project anatomy. + version_entities (Optional[list[dict[str, Any]]]): Pre-fetched + version entities. + repre_entities (Optional[list[dict[str, Any]]]): Pre-fetched + representation entities. + + Returns: + list[PublishedWorkfileInfo]: Published workfile information for + given context. + + """ + from ayon_core.pipeline import Anatomy + + # Get all representations of the folder + ( + version_entities, + repre_entities + ) = self._fetch_workfile_entities( + project_name, + folder_id, + version_entities, + repre_entities, + ) + if not repre_entities: + return [] + + if anatomy is None: + anatomy = Anatomy(project_name) + + versions_by_id = { + version_entity["id"]: version_entity + for version_entity in version_entities + } + extensions = self.get_workfile_extensions() + items = [] + for repre_entity in repre_entities: + version_id = repre_entity["versionId"] + version_entity = versions_by_id[version_id] + task_id = version_entity["taskId"] + + # Filter by extension + workfile_path = None + for repre_file in repre_entity["files"]: + ext = ( + os.path.splitext(repre_file["name"])[1] + .lower() + .lstrip(".") + ) + if ext in extensions: + workfile_path = repre_file["path"] + break + + if not workfile_path: + continue + + try: + workfile_path = workfile_path.format(root=anatomy.roots) + except Exception as exc: + print(f"Failed to format workfile path: {exc}") + + is_available = False + file_size = file_modified = file_created = None + if workfile_path and os.path.exists(workfile_path): + filestat = os.stat(workfile_path) + is_available = True + file_size = filestat.st_size + file_created = filestat.st_ctime + file_modified = filestat.st_mtime + + workfile_item = PublishedWorkfileInfo.new( + folder_id, + task_id, + repre_entity, + workfile_path, + version_entity["author"], + is_available, + file_size, + file_created, + file_modified, + ) + items.append(workfile_item) + + return items + # --- Deprecated method names --- def file_extensions(self): """Deprecated variant of 'get_workfile_extensions'. @@ -308,3 +477,53 @@ class IWorkfileHost: """ return self.workfile_has_unsaved_changes() + + def _fetch_workfile_entities( + self, + project_name: str, + folder_id: str, + version_entities: Optional[list[dict[str, Any]]], + repre_entities: Optional[list[dict[str, Any]]], + ) -> tuple[ + list[dict[str, Any]], + list[dict[str, Any]] + ]: + if repre_entities is not None and version_entities is None: + # Get versions of representations + version_ids = {r["versionId"] for r in repre_entities} + version_entities = list(ayon_api.get_versions( + project_name, + version_ids=version_ids, + fields={"id", "author", "taskId"}, + )) + + if version_entities is None: + # Get product entities of folder + product_entities = ayon_api.get_products( + project_name, + folder_ids={folder_id}, + product_types={"workfile"}, + fields={"id", "name"} + ) + + version_entities = [] + product_ids = {product["id"] for product in product_entities} + if product_ids: + # Get version docs of products with their families + version_entities = list(ayon_api.get_versions( + project_name, + product_ids=product_ids, + fields={"id", "author", "taskId"}, + )) + + # Fetch representations of filtered versions and add filter for + # extension + if repre_entities is None: + repre_entities = [] + if version_entities: + repre_entities = list(ayon_api.get_representations( + project_name, + version_ids={v["id"] for v in version_entities} + )) + + return version_entities, repre_entities