diff --git a/client/ayon_core/pipeline/workfile/utils.py b/client/ayon_core/pipeline/workfile/utils.py index 0c3a50446d..29d636fd7d 100644 --- a/client/ayon_core/pipeline/workfile/utils.py +++ b/client/ayon_core/pipeline/workfile/utils.py @@ -13,6 +13,7 @@ from ayon_core.settings import get_project_settings from ayon_core.host.interfaces import ( SaveWorkfileOptionalData, ListWorkfilesOptionalData, + CopyWorkfileOptionalData, ) from ayon_core.pipeline.version_start import get_versioning_start from ayon_core.pipeline.template_data import get_template_data @@ -516,6 +517,124 @@ def save_next_version( ) +def copy_workfile_to_context( + src_workfile_path: str, + folder_entity: dict[str, Any], + task_entity: dict[str, Any], + *, + version: Optional[int] = None, + comment: Optional[str] = None, + description: Optional[str] = None, + open_workfile: bool = True, + prepared_data: Optional[CopyWorkfileOptionalData] = None, +) -> None: + """Copy workfile to a context. + + Copy workfile to a specified folder and task. Destination path is + calculated based on passed information. + + Args: + src_workfile_path (str): Source workfile path. + folder_entity (dict[str, Any]): Target folder entity. + task_entity (dict[str, Any]): Target task entity. + version (Optional[int]): Workfile version. Use next version if not + passed. + comment (optional[str]): Workfile comment. + description (Optional[str]): Workfile description. + prepared_data (Optional[CopyWorkfileOptionalData]): Prepared data + for speed enhancements. Rootless path is calculated in this + function. + + """ + from ayon_core.pipeline import Anatomy + from ayon_core.pipeline.context_tools import registered_host + + host = registered_host() + project_name = host.get_current_project_name() + + anatomy = prepared_data.anatomy + if anatomy is None: + if prepared_data.project_entity is None: + prepared_data.project_entity = ayon_api.get_project( + project_name + ) + anatomy = Anatomy( + project_name, project_entity=prepared_data.project_entity + ) + prepared_data.anatomy = anatomy + + project_settings = prepared_data.project_settings + if project_settings is None: + project_settings = get_project_settings(project_name) + prepared_data.project_settings = project_settings + + if version is None: + list_prepared_data = None + if prepared_data is not None: + list_prepared_data = ListWorkfilesOptionalData( + project_entity=prepared_data.project_entity, + anatomy=prepared_data.anatomy, + project_settings=prepared_data.project_settings, + workfile_entities=prepared_data.workfile_entities, + ) + + workfiles = host.list_workfiles( + project_name, + folder_entity, + task_entity, + prepared_data=list_prepared_data + ) + if workfiles: + version = max( + workfile.version + for workfile in workfiles + ) + 1 + else: + version = get_versioning_start( + project_name, + host.name, + task_name=task_entity["name"], + task_type=task_entity["taskType"], + product_type="workfile" + ) + + task_type = task_entity["taskType"] + template_key = get_workfile_template_key( + project_name, + task_type, + host.name, + project_settings=prepared_data.project_settings + ) + + template_data = get_template_data( + prepared_data.project_entity, + folder_entity, + task_entity, + host.name, + prepared_data.project_settings, + ) + template_data["version"] = version + if comment: + template_data["comment"] = comment + + workfile_template = anatomy.get_template_item( + "work", template_key, "path" + ) + workfile_path = workfile_template.format_strict(template_data) + prepared_data.rootless_path = workfile_path.rootless + host.copy_workfile( + src_workfile_path, + workfile_path, + folder_entity, + task_entity, + version=version, + comment=comment, + description=description, + open_workfile=open_workfile, + prepared_data=prepared_data, + ) + + def find_workfile_rootless_path( workfile_path: str, project_name: str,