From bd4381fe9b22ede0b70bbb4b0343c9e19caa3d73 Mon Sep 17 00:00:00 2001 From: Petr Kalis Date: Mon, 13 Oct 2025 13:45:24 +0200 Subject: [PATCH] Moved fill_sequence_gaps_with_previous --- client/ayon_core/lib/plugin_tools.py | 96 ------------------------ client/ayon_core/pipeline/publish/lib.py | 95 ++++++++++++++++++++++- 2 files changed, 94 insertions(+), 97 deletions(-) diff --git a/client/ayon_core/lib/plugin_tools.py b/client/ayon_core/lib/plugin_tools.py index 676d3e43b5..b19fe1e200 100644 --- a/client/ayon_core/lib/plugin_tools.py +++ b/client/ayon_core/lib/plugin_tools.py @@ -1,17 +1,9 @@ # -*- coding: utf-8 -*- """AYON plugin tools.""" import os -import logging import re import collections -from typing import Optional, Any -import clique -import speedcopy -from ayon_api import get_last_version_by_product_name, get_representations - - -log = logging.getLogger(__name__) CAPITALIZE_REGEX = re.compile(r"[a-zA-Z0-9]") @@ -157,91 +149,3 @@ def source_hash(filepath, *args): time = str(os.path.getmtime(filepath)) size = str(os.path.getsize(filepath)) return "|".join([file_name, time, size] + list(args)).replace(".", ",") - - -def fill_sequence_gaps_with_previous( - collection: str, - staging_dir: str, - instance: pyblish.plugin.Instance, - current_repre_name: str, - start_frame: int, - end_frame: int -) -> tuple[Optional[dict[str, Any]], Optional[dict[int, str]]]: - """Tries to replace missing frames from ones from last version""" - used_version_entity, repre_file_paths = _get_last_version_files( - instance, current_repre_name - ) - if repre_file_paths is None: - # issues in getting last version files - return (None, None) - - prev_collection = clique.assemble( - repre_file_paths, - patterns=[clique.PATTERNS["frames"]], - minimum_items=1 - )[0][0] - prev_col_format = prev_collection.format("{head}{padding}{tail}") - - added_files = {} - anatomy = instance.context.data["anatomy"] - col_format = collection.format("{head}{padding}{tail}") - for frame in range(start_frame, end_frame + 1): - if frame in collection.indexes: - continue - hole_fpath = os.path.join(staging_dir, col_format % frame) - - previous_version_path = prev_col_format % frame - previous_version_path = anatomy.fill_root(previous_version_path) - if not os.path.exists(previous_version_path): - log.warning( - "Missing frame should be replaced from " - f"'{previous_version_path}' but that doesn't exist. " - ) - return (None, None) - - log.warning( - f"Replacing missing '{hole_fpath}' with " - f"'{previous_version_path}'" - ) - speedcopy.copyfile(previous_version_path, hole_fpath) - added_files[frame] = hole_fpath - - return (used_version_entity, added_files) - - -def _get_last_version_files( - instance: pyblish.plugin.Instance, - current_repre_name: str, -) -> tuple[Optional[dict[str, Any], Optional[list[str]]]: - product_name = instance.data["productName"] - project_name = instance.data["projectEntity"]["name"] - folder_entity = instance.data["folderEntity"] - - version_entity = get_last_version_by_product_name( - project_name, - product_name, - folder_entity["id"], - fields={"id", "attrib"} - ) - - if not version_entity: - return None, None - - matching_repres = get_representations( - project_name, - version_ids=[version_entity["id"]], - representation_names=[current_repre_name], - fields={"files"} - ) - - matching_repre = next(matching_repres, None) - if not matching_repre: - return None, None - - - repre_file_paths = [ - file_info["path"] - for file_info in matching_repre["files"] - ] - - return (version_entity, repre_file_paths) diff --git a/client/ayon_core/pipeline/publish/lib.py b/client/ayon_core/pipeline/publish/lib.py index 3b82d961f8..4d555ae48b 100644 --- a/client/ayon_core/pipeline/publish/lib.py +++ b/client/ayon_core/pipeline/publish/lib.py @@ -7,7 +7,10 @@ import copy import warnings import hashlib import xml.etree.ElementTree -from typing import TYPE_CHECKING, Optional, Union, List +from typing import TYPE_CHECKING, Optional, Union, List, Any +import clique +import speedcopy +import logging import ayon_api import pyblish.util @@ -27,6 +30,7 @@ from .constants import ( DEFAULT_PUBLISH_TEMPLATE, DEFAULT_HERO_PUBLISH_TEMPLATE, ) +from ayon_api import get_last_version_by_product_name, get_representations if TYPE_CHECKING: from ayon_core.pipeline.traits import Representation @@ -34,6 +38,8 @@ if TYPE_CHECKING: TRAIT_INSTANCE_KEY: str = "representations_with_traits" +log = logging.getLogger(__name__) + def get_template_name_profiles( project_name, project_settings=None, logger=None @@ -1143,3 +1149,90 @@ def get_trait_representations( """ return instance.data.get(TRAIT_INSTANCE_KEY, []) + + +def fill_sequence_gaps_with_previous( + collection: str, + staging_dir: str, + instance: pyblish.plugin.Instance, + current_repre_name: str, + start_frame: int, + end_frame: int +) -> tuple[Optional[dict[str, Any]], Optional[dict[int, str]]]: + """Tries to replace missing frames from ones from last version""" + used_version_entity, repre_file_paths = _get_last_version_files( + instance, current_repre_name + ) + if repre_file_paths is None: + # issues in getting last version files + return (None, None) + + prev_collection = clique.assemble( + repre_file_paths, + patterns=[clique.PATTERNS["frames"]], + minimum_items=1 + )[0][0] + prev_col_format = prev_collection.format("{head}{padding}{tail}") + + added_files = {} + anatomy = instance.context.data["anatomy"] + col_format = collection.format("{head}{padding}{tail}") + for frame in range(start_frame, end_frame + 1): + if frame in collection.indexes: + continue + hole_fpath = os.path.join(staging_dir, col_format % frame) + + previous_version_path = prev_col_format % frame + previous_version_path = anatomy.fill_root(previous_version_path) + if not os.path.exists(previous_version_path): + log.warning( + "Missing frame should be replaced from " + f"'{previous_version_path}' but that doesn't exist. " + ) + return (None, None) + + log.warning( + f"Replacing missing '{hole_fpath}' with " + f"'{previous_version_path}'" + ) + speedcopy.copyfile(previous_version_path, hole_fpath) + added_files[frame] = hole_fpath + + return (used_version_entity, added_files) + + +def _get_last_version_files( + instance: pyblish.plugin.Instance, + current_repre_name: str, +) -> tuple[Optional[dict[str, Any]], Optional[list[str]]]: + product_name = instance.data["productName"] + project_name = instance.data["projectEntity"]["name"] + folder_entity = instance.data["folderEntity"] + + version_entity = get_last_version_by_product_name( + project_name, + product_name, + folder_entity["id"], + fields={"id", "attrib"} + ) + + if not version_entity: + return None, None + + matching_repres = get_representations( + project_name, + version_ids=[version_entity["id"]], + representation_names=[current_repre_name], + fields={"files"} + ) + + matching_repre = next(matching_repres, None) + if not matching_repre: + return None, None + + repre_file_paths = [ + file_info["path"] + for file_info in matching_repre["files"] + ] + + return (version_entity, repre_file_paths)