implement replace_root

conductor/lib/file_utils.py

@@ -3,6 +3,7 @@
 import re
 import sys
 import glob
+from itertools import izip_longest
 
 from conductor.lib import exceptions
 
@@ -40,6 +41,13 @@
     RX_ASTERISK,
 )
 
+# https://regex101.com/r/GTumsD/1/
+# Detects path components - e.g.:
+# //192.168.0.1/renders/light/render_0001_large.exr
+# becomes:
+# Result: ['//192.168.0.1', '/renders', '/light', '/render_0001_large.exr']
+PATH_SPLIT_REGEX = re.compile(r"((?:\/\/|[a-zA-Z]:\/|\/)[^\/]+)")
+
 logger = logging.getLogger(__name__)
 
 
@@ -487,3 +495,57 @@ def strip_drive_letter(filepath):
     '''
     rx_drive = r'^[a-z]:'
     return re.sub(rx_drive, "", filepath, flags=re.I)
+
+
+def _common_tail_parts(*parts):
+    """
+    find consecutive common items in iterables, working from the back. 
+
+    [1,7,4,9,3]
+    [1,8,4,9,3]
+    yields:
+    [4,9,3]
+    """
+    tail_parts = []
+    parts = [part[::-1] for part in parts]
+    for level_tuple in izip_longest(*parts):
+        if all(item == level_tuple[0] for item in level_tuple):
+            tail_parts.insert(0,level_tuple[0])
+        else:
+            return tail_parts
+
+def replace_root(path1, path2):
+    """
+    Replace the root of the first path with the root of the second.
+
+    Useful when some plugin is used to dynamically replace drives. Paths should
+    be absolute (on one or other platform) and use only forward slashes.
+
+    See tests for examples. conductor/native/tests/test_file_utils.py
+    """ 
+
+    all_parts1 = PATH_SPLIT_REGEX.findall(path1)
+    all_parts2 = PATH_SPLIT_REGEX.findall(path2)
+    if not (len(all_parts1)>1 and len(all_parts2)>1):
+        return path1
+
+    # Assume that if the first part of each path is the same, then we don't
+    # want to do any replacement, even if there are differences after. WHY?
+    # Because a difference in some middle part could be because of a frame
+    # expression that we don't want to replace. However, it IS likely that if
+    # the first parts are different, then we DO want to replace all the not-same
+    # root parts with those from path 2, because it could very well be a multi
+    # part replacement, e.g. C:/projects/foo/bar -> /Volumes/my/share/c/foo/bar
+    if all_parts1[0] == all_parts2[0]:
+        return path1
+
+    # dont consider the filename tail itself. We are interested in directories.
+    # Just reattach from path1 afterwards.
+    root_parts1 = all_parts1[:-1]
+    root_parts2 = all_parts2[:-1]
+
+    tail_parts = _common_tail_parts(root_parts1,root_parts2)
+    if len(tail_parts):
+        root_parts2 = root_parts2[:-len(tail_parts)]
+
+    return "".join(root_parts2 + tail_parts + [all_parts1[-1]])

conductor/lib/nuke_utils.py

@@ -4,6 +4,7 @@
 
 import nuke
 from conductor.lib import package_utils
+from conductor.lib import file_utils
 
 logger = logging.getLogger(__name__)
 
@@ -174,6 +175,19 @@ def resolve_knob_path(knob):
     # linux?)
     path = os.path.abspath(path).replace('\\', "/")
 
+    # Some customers use a dynamic path replacement tool to change drives for
+    # portability. Our code needs to make the same switch in order to find
+    # assets. However, getValue() only gets the raw value - before replacement.
+    # The path replacer value is only visible when you evaluate the knob. The
+    # reason we don't do that (I guess) is so we don't inadvertently replace
+    # frame expressions with a single current frame.
+    try:
+        path = file_utils.replace_root(path, knob.evaluate().replace('\\', "/"))
+    except BaseException:
+        # If anything goes wrong, we'll just use the orig path value
+        pass
+
+
     logger.debug("Resolved to: %s", path)
     return path
 

conductor/native/tests/test_file_utils.py

@@ -0,0 +1,95 @@
+""" test file_utils
+   isort:skip_file
+"""
+from conductor.lib import file_utils as fu
+import unittest
+
+
+class ReplaceRootTest(unittest.TestCase):
+    def test_replace_simple_same_level_root(self):
+        p1 = '/a/renders/render_0001_large.exr'
+        p2 = '/b/renders/render_0001_large.exr'
+        expected = '/b/renders/render_0001_large.exr'
+        result = fu.replace_root(p1, p2)
+        self.assertEqual(result, expected)
+
+    def test_replace_simple_differing_level_root(self):
+        p1 = '/a/renders/render_0001_large.exr'
+        p2 = '/b/renders/c/d/render_0001_large.exr'
+        expected = '/b/renders/c/d/render_0001_large.exr'
+        result = fu.replace_root(p1, p2)
+        self.assertEqual(result, expected)
+
+    def test_replace_root_retains_original_filename(self):
+        p1 = '/a/renders/render_0001_large.0%4d.exr'
+        p2 = '/b/renders/c/d/render_0001_large.0786.exr'
+        expected = '/b/renders/c/d/render_0001_large.0%4d.exr'
+        result = fu.replace_root(p1, p2)
+        self.assertEqual(result, expected)
+
+    def test_replace_root_when_orig_contain_drive_letter(self):
+        p1 = 'Z:/a/renders/render_0001_large.exr'
+        p2 = '/b/renders/c/d/render_0001_large.exr'
+        expected = '/b/renders/c/d/render_0001_large.exr'
+        result = fu.replace_root(p1, p2)
+        self.assertEqual(result, expected)
+
+    def test_replace_root_when_replacement_contains_drive_letter(self):
+        p1 = '/a/renders/render_0001_large.exr'
+        p2 = 'C:/b/renders/c/d/render_0001_large.exr'
+        expected = 'C:/b/renders/c/d/render_0001_large.exr'
+        result = fu.replace_root(p1, p2)
+        self.assertEqual(result, expected)
+
+    def test_replace_root_when_both_contain_drive_letter(self):
+        p1 = 'Z:/a/renders/render_0001_large.exr'
+        p2 = 'C:/b/renders/c/d/render_0001_large.exr'
+        expected = 'C:/b/renders/c/d/render_0001_large.exr'
+        result = fu.replace_root(p1, p2)
+        self.assertEqual(result, expected)
+
+    def test_replace_root_when_orig_contain_unc_ip(self):
+        p1 = '//192.168.0.1/renders/render_0001_large.exr'
+        p2 = '/Volumes/renders/c/d/render_0001_large.exr'
+        expected = '/Volumes/renders/c/d/render_0001_large.exr'
+        result = fu.replace_root(p1, p2)
+        self.assertEqual(result, expected)
+
+    def test_replace_root_when_replacement_contain_unc_ip(self):
+        p1 = '/Volumes/renders/render_0001_large.exr'
+        p2 = '//192.168.0.1/renders/c/d/render_0001_large.exr'
+        expected = '//192.168.0.1/renders/c/d/render_0001_large.exr'
+        result = fu.replace_root(p1, p2)
+        self.assertEqual(result, expected)
+
+    def test_replace_root_when_both_contain_unc_ip(self):
+        p1 = '//192.168.0.1/renders/render_0001_large.exr'
+        p2 = '//192.168.0.2/renders/c/d/render_0001_large.exr'
+        expected = '//192.168.0.2/renders/c/d/render_0001_large.exr'
+        result = fu.replace_root(p1, p2)
+        self.assertEqual(result, expected)
+
+    def test_dont_replace_when_different_roots_but_first_part_the_same(self):
+        p1 = '//192.168.0.1/renders/0%4d/render_0001_large.exr'
+        p2 = '//192.168.0.1/renders/0023/render_0001_large.exr'
+        expected = '//192.168.0.1/renders/0%4d/render_0001_large.exr'
+        result = fu.replace_root(p1, p2)
+        self.assertEqual(result, expected)
+
+    def test_replace_root_when_everything_different(self):
+        p1 = '/Volumes/e/f/render_0001_large.0%4d.exr'
+        p2 = '//192.168.0.1/renders/c/d/render_0001_large.0001.exr'
+        expected = '//192.168.0.1/renders/c/d/render_0001_large.0%4d.exr'
+        result = fu.replace_root(p1, p2)
+        self.assertEqual(result, expected)
+
+    def test_replace_root_when_single_root_part_and_different_filename(self):
+        p1 = '/Volumes/render_0001_large.0%4d.exr'
+        p2 = '//192.168.0.1/render_0001_large.0001.exr'
+        expected = '//192.168.0.1/render_0001_large.0%4d.exr'
+        result = fu.replace_root(p1, p2)
+        self.assertEqual(result, expected)
+
+
+if __name__ == "__main__":
+    unittest.main()