moved 'get_last_workfile_with_version' and 'get_last_workfile' to 'openpype.pipeline.workfile'

This commit is contained in:
Jakub Trllo 2022-08-09 12:56:50 +02:00
parent b89e99e890
commit 02007784fa
6 changed files with 153 additions and 90 deletions

View file

@ -1,9 +1,6 @@
import os
from openpype.lib import (
StringTemplate,
get_last_workfile_with_version,
)
from openpype.lib import StringTemplate
from openpype.pipeline import (
registered_host,
legacy_io,
@ -11,6 +8,7 @@ from openpype.pipeline import (
)
from openpype.pipeline.workfile import (
get_workfile_template_key_from_context,
get_last_workfile_with_version,
)
from openpype.pipeline.template_data import get_template_data_with_names
from openpype.hosts.tvpaint.api import lib, pipeline, plugin

View file

@ -27,7 +27,6 @@ from openpype.settings.constants import (
from . import PypeLogger
from .profiles_filtering import filter_profiles
from .local_settings import get_openpype_username
from .avalon_context import get_last_workfile
from .python_module_tools import (
modules_from_path,
@ -1728,7 +1727,10 @@ def _prepare_last_workfile(data, workdir):
if not last_workfile_path:
extensions = HOST_WORKFILE_EXTENSIONS.get(app.host_name)
if extensions:
from openpype.pipeline import get_workfile_template_key
from openpype.pipeline.workfile import (
get_workfile_template_key,
get_last_workfile
)
anatomy = data["anatomy"]
project_settings = data["project_settings"]

View file

@ -1696,6 +1696,7 @@ def get_custom_workfile_template(template_profiles):
)
@deprecated("openpype.pipeline.workfile.get_last_workfile_with_version")
def get_last_workfile_with_version(
workdir, file_template, fill_data, extensions
):
@ -1711,78 +1712,15 @@ def get_last_workfile_with_version(
tuple: Last workfile<str> with version<int> if there is any otherwise
returns (None, None).
"""
if not os.path.exists(workdir):
return None, None
# Fast match on extension
filenames = [
filename
for filename in os.listdir(workdir)
if os.path.splitext(filename)[1] in extensions
]
from openpype.pipeline.workfile import get_last_workfile_with_version
# Build template without optionals, version to digits only regex
# and comment to any definable value.
_ext = []
for ext in extensions:
if not ext.startswith("."):
ext = "." + ext
# Escape dot for regex
ext = "\\" + ext
_ext.append(ext)
ext_expression = "(?:" + "|".join(_ext) + ")"
# Replace `.{ext}` with `{ext}` so we are sure there is not dot at the end
file_template = re.sub(r"\.?{ext}", ext_expression, file_template)
# Replace optional keys with optional content regex
file_template = re.sub(r"<.*?>", r".*?", file_template)
# Replace `{version}` with group regex
file_template = re.sub(r"{version.*?}", r"([0-9]+)", file_template)
file_template = re.sub(r"{comment.*?}", r".+?", file_template)
file_template = StringTemplate.format_strict_template(
file_template, fill_data
return get_last_workfile_with_version(
workdir, file_template, fill_data, extensions
)
# Match with ignore case on Windows due to the Windows
# OS not being case-sensitive. This avoids later running
# into the error that the file did exist if it existed
# with a different upper/lower-case.
kwargs = {}
if platform.system().lower() == "windows":
kwargs["flags"] = re.IGNORECASE
# Get highest version among existing matching files
version = None
output_filenames = []
for filename in sorted(filenames):
match = re.match(file_template, filename, **kwargs)
if not match:
continue
file_version = int(match.group(1))
if version is None or file_version > version:
output_filenames[:] = []
version = file_version
if file_version == version:
output_filenames.append(filename)
output_filename = None
if output_filenames:
if len(output_filenames) == 1:
output_filename = output_filenames[0]
else:
last_time = None
for _output_filename in output_filenames:
full_path = os.path.join(workdir, _output_filename)
mod_time = os.path.getmtime(full_path)
if last_time is None or last_time < mod_time:
output_filename = _output_filename
last_time = mod_time
return output_filename, version
@deprecated("openpype.pipeline.workfile.get_last_workfile")
def get_last_workfile(
workdir, file_template, fill_data, extensions, full_path=False
):
@ -1800,22 +1738,12 @@ def get_last_workfile(
Returns:
str: Last or first workfile as filename of full path to filename.
"""
filename, version = get_last_workfile_with_version(
workdir, file_template, fill_data, extensions
from openpype.pipeline.workfile import get_last_workfile
return get_last_workfile(
workdir, file_template, fill_data, extensions, full_path
)
if filename is None:
data = copy.deepcopy(fill_data)
data["version"] = 1
data.pop("comment", None)
if not data.get("ext"):
data["ext"] = extensions[0]
data["ext"] = data["ext"].replace('.', '')
filename = StringTemplate.format_strict_template(file_template, data)
if full_path:
return os.path.normpath(os.path.join(workdir, filename))
return filename
@with_pipeline_io

View file

@ -3,6 +3,9 @@ from .path_resolving import (
get_workfile_template_key,
get_workdir_with_workdir_data,
get_workdir,
get_last_workfile_with_version,
get_last_workfile,
)
@ -11,4 +14,7 @@ __all__ = (
"get_workfile_template_key",
"get_workdir_with_workdir_data",
"get_workdir",
"get_last_workfile_with_version",
"get_last_workfile",
)

View file

@ -1,6 +1,11 @@
import os
import re
import copy
import platform
from openpype.client import get_asset_by_name
from openpype.settings import get_project_settings
from openpype.lib import filter_profiles
from openpype.lib import filter_profiles, StringTemplate
from openpype.pipeline import Anatomy
from openpype.pipeline.template_data import get_template_data
@ -177,3 +182,127 @@ def get_workdir(
template_key,
project_settings
)
def get_last_workfile_with_version(
workdir, file_template, fill_data, extensions
):
"""Return last workfile version.
Args:
workdir(str): Path to dir where workfiles are stored.
file_template(str): Template of file name.
fill_data(Dict[str, Any]): Data for filling template.
extensions(Iterable[str]): All allowed file extensions of workfile.
Returns:
Tuple[Union[str, None], Union[int, None]]: Last workfile with version
if there is any workfile otherwise None for both.
"""
if not os.path.exists(workdir):
return None, None
# Fast match on extension
filenames = [
filename
for filename in os.listdir(workdir)
if os.path.splitext(filename)[1] in extensions
]
# Build template without optionals, version to digits only regex
# and comment to any definable value.
_ext = []
for ext in extensions:
if not ext.startswith("."):
ext = "." + ext
# Escape dot for regex
ext = "\\" + ext
_ext.append(ext)
ext_expression = "(?:" + "|".join(_ext) + ")"
# Replace `.{ext}` with `{ext}` so we are sure there is not dot at the end
file_template = re.sub(r"\.?{ext}", ext_expression, file_template)
# Replace optional keys with optional content regex
file_template = re.sub(r"<.*?>", r".*?", file_template)
# Replace `{version}` with group regex
file_template = re.sub(r"{version.*?}", r"([0-9]+)", file_template)
file_template = re.sub(r"{comment.*?}", r".+?", file_template)
file_template = StringTemplate.format_strict_template(
file_template, fill_data
)
# Match with ignore case on Windows due to the Windows
# OS not being case-sensitive. This avoids later running
# into the error that the file did exist if it existed
# with a different upper/lower-case.
kwargs = {}
if platform.system().lower() == "windows":
kwargs["flags"] = re.IGNORECASE
# Get highest version among existing matching files
version = None
output_filenames = []
for filename in sorted(filenames):
match = re.match(file_template, filename, **kwargs)
if not match:
continue
file_version = int(match.group(1))
if version is None or file_version > version:
output_filenames[:] = []
version = file_version
if file_version == version:
output_filenames.append(filename)
output_filename = None
if output_filenames:
if len(output_filenames) == 1:
output_filename = output_filenames[0]
else:
last_time = None
for _output_filename in output_filenames:
full_path = os.path.join(workdir, _output_filename)
mod_time = os.path.getmtime(full_path)
if last_time is None or last_time < mod_time:
output_filename = _output_filename
last_time = mod_time
return output_filename, version
def get_last_workfile(
workdir, file_template, fill_data, extensions, full_path=False
):
"""Return last workfile filename.
Returns file with version 1 if there is not workfile yet.
Args:
workdir(str): Path to dir where workfiles are stored.
file_template(str): Template of file name.
fill_data(Dict[str, Any]): Data for filling template.
extensions(Iterable[str]): All allowed file extensions of workfile.
full_path(bool): Full path to file is returned if set to True.
Returns:
str: Last or first workfile as filename of full path to filename.
"""
filename, version = get_last_workfile_with_version(
workdir, file_template, fill_data, extensions
)
if filename is None:
data = copy.deepcopy(fill_data)
data["version"] = 1
data.pop("comment", None)
if not data.get("ext"):
data["ext"] = extensions[0]
data["ext"] = data["ext"].replace('.', '')
filename = StringTemplate.format_strict_template(file_template, data)
if full_path:
return os.path.normpath(os.path.join(workdir, filename))
return filename

View file

@ -5,11 +5,11 @@ import logging
from Qt import QtWidgets, QtCore
from openpype.lib import get_last_workfile_with_version
from openpype.pipeline import (
registered_host,
legacy_io,
)
from openpype.pipeline.workfile import get_last_workfile_with_version
from openpype.pipeline.template_data import get_template_data_with_names
from openpype.tools.utils import PlaceholderLineEdit