228 lines
8.0 KiB
Python
228 lines
8.0 KiB
Python
# ***** 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
|