feature: pack folder
This commit is contained in:
@@ -139,6 +139,7 @@ class Packer:
|
||||
) # type: typing.DefaultDict[pathlib.Path, AssetAction]
|
||||
self.missing_files = set() # type: typing.Set[pathlib.Path]
|
||||
self._new_location_paths = set() # type: typing.Set[pathlib.Path]
|
||||
self._top_level_paths = set() # type: typing.Set[pathlib.Path]
|
||||
self._output_path = None # type: typing.Optional[pathlib.PurePath]
|
||||
self._output_paths = [] # type: typing.List[pathlib.PurePath]
|
||||
|
||||
@@ -254,6 +255,7 @@ class Packer:
|
||||
|
||||
self._progress_cb.pack_start()
|
||||
self._new_location_paths = set()
|
||||
self._top_level_paths = set()
|
||||
self._output_paths = []
|
||||
|
||||
for bf in self.blendfiles:
|
||||
@@ -272,8 +274,14 @@ class Packer:
|
||||
self._output_paths.append(bfile_pp)
|
||||
|
||||
act = self._actions[bfile_path]
|
||||
act.path_action = PathAction.KEEP_PATH
|
||||
# This file may already have been seen as a dependency of another
|
||||
# blend file we're packing. In that case it needs to keep its
|
||||
# FIND_NEW_LOCATION action, so that the file referring to it is
|
||||
# rewritten; only its location in the pack is decided here.
|
||||
if act.path_action != PathAction.FIND_NEW_LOCATION:
|
||||
act.path_action = PathAction.KEEP_PATH
|
||||
act.new_path = bfile_pp
|
||||
self._top_level_paths.add(bfile_path)
|
||||
|
||||
self._check_aborted()
|
||||
for usage in trace.deps(bf, self._progress_cb):
|
||||
@@ -371,6 +379,12 @@ class Packer:
|
||||
act = self._actions[path]
|
||||
assert isinstance(act, AssetAction)
|
||||
|
||||
if path in self._top_level_paths:
|
||||
# This is one of the blend files we're packing, so it already
|
||||
# has its place in the pack. Files referring to it are still
|
||||
# rewritten, they'll just point at that place.
|
||||
continue
|
||||
|
||||
relpath = bpathlib.strip_root(path)
|
||||
if self.keep_hierarchy:
|
||||
act.new_path = pathlib.Path(self._target_path, relpath)
|
||||
|
||||
@@ -0,0 +1,227 @@
|
||||
# ***** BEGIN GPL LICENSE BLOCK *****
|
||||
#
|
||||
# This program is free software; you can redistribute it and/or
|
||||
# modify it under the terms of the GNU General Public License
|
||||
# as published by the Free Software Foundation; either version 2
|
||||
# of the License, or (at your option) any later version.
|
||||
#
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with this program; if not, write to the Free Software Foundation,
|
||||
# Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
|
||||
#
|
||||
# ***** END GPL LICENCE BLOCK *****
|
||||
"""Scan a directory tree for blend files, to feed a folder-wide BAT pack.
|
||||
|
||||
This module deliberately has no Blender dependencies, so that it can be used
|
||||
both from the commandline (``bat pack-folder``) and from the add-on.
|
||||
"""
|
||||
import fnmatch
|
||||
import logging
|
||||
import os
|
||||
import pathlib
|
||||
import re
|
||||
import typing
|
||||
|
||||
from blender_asset_tracer.blendfile import magic_compression
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
BLENDFILE_SUFFIX = ".blend"
|
||||
|
||||
# Matches filenames like scene_v02.blend (single integer version only).
|
||||
# Does NOT match: _v02.1.blend, _v2-1.blend, or files without _vNN suffix.
|
||||
VERSION_RE = re.compile(r"_v(\d+)\.blend$", re.IGNORECASE)
|
||||
|
||||
# Splits a versioned filename into the parts that identify its version series,
|
||||
# so that scene_v02.blend, scene_v02_light.blend and scene_v02_work.blend are
|
||||
# each versioned independently. The greedy prefix makes the *last* '_vNNN' of
|
||||
# the name the version.
|
||||
VERSION_SERIES_RE = re.compile(
|
||||
r"^(?P<prefix>.*)_v(?P<version>\d+)(?P<suffix>.*)\.blend$", re.IGNORECASE
|
||||
)
|
||||
|
||||
# Directory name globs that are skipped by default while scanning.
|
||||
DEFAULT_IGNORE_DIRS = (".*", "__pycache__")
|
||||
|
||||
|
||||
class ScanResult(typing.NamedTuple):
|
||||
"""The outcome of scanning a directory tree for blend files."""
|
||||
|
||||
blendfiles: typing.List[pathlib.Path]
|
||||
"""Blend files to pack, in deterministic (sorted) order."""
|
||||
|
||||
skipped: typing.List[typing.Tuple[pathlib.Path, str]]
|
||||
"""Files that were found but not packed, as (path, reason) tuples."""
|
||||
|
||||
unreadable_dirs: typing.List[typing.Tuple[pathlib.Path, str]]
|
||||
"""Directories that could not be scanned, as (path, error) tuples."""
|
||||
|
||||
|
||||
def find_blend_files(
|
||||
folder: pathlib.Path,
|
||||
*,
|
||||
follow_symlinks: bool = False,
|
||||
ignore_dirs: typing.Sequence[str] = DEFAULT_IGNORE_DIRS,
|
||||
latest_only: bool = False,
|
||||
validate: bool = True,
|
||||
) -> ScanResult:
|
||||
"""Recursively find the blend files in ``folder``.
|
||||
|
||||
Backup files (``.blend1``, ``.blend2``, …) are never returned, as only the
|
||||
``.blend`` suffix is matched.
|
||||
|
||||
:param folder: Directory to scan.
|
||||
:param follow_symlinks: Whether to descend into symlinked directories.
|
||||
Already-visited directories are skipped, so symlink loops are safe.
|
||||
:param ignore_dirs: Globs of directory names to skip entirely.
|
||||
:param latest_only: Only keep the highest ``_vNNN`` file of each series.
|
||||
:param validate: Check that each file actually starts with a blend file
|
||||
magic (optionally compressed). Files that don't are reported as
|
||||
skipped, instead of aborting the entire pack later on.
|
||||
"""
|
||||
blendfiles = [] # type: typing.List[pathlib.Path]
|
||||
skipped = [] # type: typing.List[typing.Tuple[pathlib.Path, str]]
|
||||
unreadable_dirs = [] # type: typing.List[typing.Tuple[pathlib.Path, str]]
|
||||
|
||||
def on_error(ex: OSError) -> None:
|
||||
path = pathlib.Path(ex.filename) if ex.filename else folder
|
||||
reason = ex.strerror or str(ex)
|
||||
log.warning("Unable to scan %s: %s", path, reason)
|
||||
unreadable_dirs.append((path, reason))
|
||||
|
||||
visited_dirs = set() # type: typing.Set[typing.Tuple[int, int]]
|
||||
if follow_symlinks:
|
||||
_first_visit(folder, visited_dirs)
|
||||
|
||||
for dirpath_str, dirnames, filenames in os.walk(
|
||||
str(folder), onerror=on_error, followlinks=follow_symlinks
|
||||
):
|
||||
dirpath = pathlib.Path(dirpath_str)
|
||||
|
||||
# Prune ignored directories, and keep the walk order deterministic.
|
||||
subdirs = sorted(
|
||||
dirname for dirname in dirnames if not _matches_any(dirname, ignore_dirs)
|
||||
)
|
||||
if follow_symlinks:
|
||||
subdirs = [
|
||||
dirname
|
||||
for dirname in subdirs
|
||||
if _first_visit(dirpath / dirname, visited_dirs)
|
||||
]
|
||||
dirnames[:] = subdirs
|
||||
|
||||
for filename in sorted(filenames):
|
||||
path = dirpath / filename
|
||||
if path.suffix.lower() != BLENDFILE_SUFFIX:
|
||||
continue
|
||||
|
||||
if validate:
|
||||
reason = why_not_a_blendfile(path)
|
||||
if reason is not None:
|
||||
log.warning("Skipping %s: %s", path, reason)
|
||||
skipped.append((path, reason))
|
||||
continue
|
||||
|
||||
blendfiles.append(path)
|
||||
|
||||
if latest_only:
|
||||
blendfiles, superseded = keep_latest_versions(blendfiles)
|
||||
for path in superseded:
|
||||
log.debug("Skipping %s: superseded by a newer version", path)
|
||||
skipped.extend((path, "superseded by a newer version") for path in superseded)
|
||||
|
||||
return ScanResult(
|
||||
blendfiles=blendfiles, skipped=skipped, unreadable_dirs=unreadable_dirs
|
||||
)
|
||||
|
||||
|
||||
def why_not_a_blendfile(path: pathlib.Path) -> typing.Optional[str]:
|
||||
"""Return None when the file looks like a blend file, else why it doesn't."""
|
||||
|
||||
try:
|
||||
with path.open("rb") as fileobj:
|
||||
compression = magic_compression.find_compression_type(fileobj)
|
||||
except OSError as ex:
|
||||
return "unable to read: %s" % (ex.strerror or ex)
|
||||
|
||||
if compression == magic_compression.Compression.UNRECOGNISED:
|
||||
return "not a blend file"
|
||||
return None
|
||||
|
||||
|
||||
def keep_latest_versions(
|
||||
blendfiles: typing.Iterable[pathlib.Path],
|
||||
) -> typing.Tuple[typing.List[pathlib.Path], typing.List[pathlib.Path]]:
|
||||
"""Reduce each ``name_vNNN.blend`` series to its highest version.
|
||||
|
||||
Files whose name contains no ``_vNNN`` are always kept. Versions are
|
||||
compared per directory and per name, so identically-named files in
|
||||
different directories don't shadow each other, and variants such as
|
||||
``shot_v003_light.blend`` form their own series.
|
||||
|
||||
:return: (files to keep, superseded files), both sorted.
|
||||
"""
|
||||
# Maps (directory, name without version) to the best (version, path) so far.
|
||||
best = (
|
||||
{}
|
||||
) # type: typing.Dict[typing.Tuple[pathlib.Path, str, str], typing.Tuple[int, pathlib.Path]]
|
||||
kept = [] # type: typing.List[pathlib.Path]
|
||||
superseded = [] # type: typing.List[pathlib.Path]
|
||||
|
||||
for path in blendfiles:
|
||||
match = VERSION_SERIES_RE.match(path.name)
|
||||
if match is None:
|
||||
kept.append(path)
|
||||
continue
|
||||
|
||||
key = (
|
||||
path.parent,
|
||||
match.group("prefix").lower(),
|
||||
match.group("suffix").lower(),
|
||||
)
|
||||
version = int(match.group("version"))
|
||||
previous = best.get(key)
|
||||
|
||||
if previous is None:
|
||||
best[key] = (version, path)
|
||||
elif version > previous[0]:
|
||||
best[key] = (version, path)
|
||||
superseded.append(previous[1])
|
||||
else:
|
||||
superseded.append(path)
|
||||
|
||||
kept.extend(path for _, path in best.values())
|
||||
kept.sort()
|
||||
superseded.sort()
|
||||
return kept, superseded
|
||||
|
||||
|
||||
def _matches_any(name: str, patterns: typing.Sequence[str]) -> bool:
|
||||
return any(fnmatch.fnmatch(name, pattern) for pattern in patterns)
|
||||
|
||||
|
||||
def _first_visit(
|
||||
dirpath: pathlib.Path, visited: typing.Set[typing.Tuple[int, int]]
|
||||
) -> bool:
|
||||
"""Return True when this directory wasn't walked before.
|
||||
|
||||
Used to keep ``follow_symlinks`` from looping forever.
|
||||
"""
|
||||
try:
|
||||
stat = dirpath.stat()
|
||||
except OSError as ex:
|
||||
log.warning("Unable to inspect %s: %s", dirpath, ex.strerror or ex)
|
||||
return False
|
||||
|
||||
key = (stat.st_dev, stat.st_ino)
|
||||
if key in visited:
|
||||
log.warning("Not descending into %s again (symlink loop?)", dirpath)
|
||||
return False
|
||||
|
||||
visited.add(key)
|
||||
return True
|
||||
Reference in New Issue
Block a user