Tweak ADS API, implement new-style metadata preference

diff --git a/fs.py b/fs.py
index 58ea4de20134d1677f93a4e6c0b95cc7a7dcad21..f2357f047c50f959da678be4324d3d122b35cca7 100644 (file)
--- a/fs.py
+++ b/fs.py
@@ -14,12 +14,13 @@
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
 from collections import namedtuple
+from enum import Enum
 import json
 import pathlib
 import re
 
 
-class WinPath(pathlib.WindowsPath):
+def get_ads(path: pathlib.WindowsPath, name):
        r"""
        This constructs a path for an NTFS alternate data stream on Windows.
        In most cases, for directories, a trailing slash is permitted and optional:
@@ -36,13 +37,17 @@ class WinPath(pathlib.WindowsPath):
        we need to manually add a separator to ensure the resulting path actually
        points to an ADS. This makes the path absolute, but that's okay because there
        doesn't seem to be any way to get an ADS on a relative path with a drive.
+
+       Incidentally, this is essentially cross-platform, since there are Linux NTFS
+       implementations with ADS support using the same path syntax. But if the
+       application were targeting other platforms there would probably be platform-
+       specific implementations of this function.
        """
-       def get_ads(self, name):
-               full_name = self.name + ':' + name
-               if self.name:
-                       return self.with_name(full_name)
-               else:
-                       return self.joinpath('/', full_name)
+       full_name = path.name + ':' + name
+       if path.name:
+               return path.with_name(full_name)
+       else:
+               return path.joinpath('/', full_name)
 
 
 def clean_win_path(seg):
@@ -73,8 +78,16 @@ def content_path(course_path, segments):
 BB_META_STREAM_NAME = '8f3b98ea-e227-478f-bb58-5c31db476409'
 
 
-VersionInfo = namedtuple('ParseResult', ['bb_id', 'version'])
-VersionInfo.next = lambda self: VersionInfo(self.bb_id, self.version + 1)
+VersionInfo = namedtuple('ParseResult', ['meta_type', 'bb_id', 'version'])
+VersionInfo.MetaType = Enum('MetaType', ['OLD', 'NEW'])
+
+def _vinfo_next(self):
+       return VersionInfo(
+               self.MetaType.NEW,
+               self.bb_id,
+               self.version + 1,
+       )
+VersionInfo.next = _vinfo_next
 
 
 def _extract_version(path):
@@ -82,7 +95,7 @@ def _extract_version(path):
                return None
 
        info = None
-       stream_path = WinPath(path).get_ads(BB_META_STREAM_NAME)
+       stream_path = fs.get_ads(path, BB_META_STREAM_NAME)
        if stream_path.exists():
                # NTFS ADS metadata
                with stream_path.open() as f:
@@ -95,7 +108,7 @@ def _extract_version(path):
                                version = metadata.get('version')
                                version_typecheck = lambda v: isinstance(v, int) if path.is_file() else v is None
                                if isinstance(bb_id, str) and version_typecheck(version):
-                                       info = VersionInfo(bb_id, version)
+                                       info = VersionInfo(VersionInfo.MetaType.NEW, bb_id, version)
        else:
                # old in-filename metadata
                (stem, _) = _split_name(path.name)
@@ -103,7 +116,7 @@ def _extract_version(path):
                if match:
                        version = int(match.group('version')) if match.group('version') else None
                        if (version is None) == path.is_dir():
-                               info = VersionInfo(match.group('id'), version)
+                               info = VersionInfo(VersionInfo.MetaType.OLD, match.group('id'), version)
 
        return info
 
@@ -129,6 +142,15 @@ def join_content_path(path, content_doc):
        ver_map = get_child_versions(path)
        versions = [v for v in ver_map if v.bb_id == content_doc['id']]
        if versions:
+               if len(versions) > 1:
+                       # attempt to disambiguate by preferring directories with new-style metadata
+                       versions = [
+                               info for info in versions
+                               if info.meta_type is VersionInfo.MetaType.NEW
+                       ]
+
+               # Either destructure could fail based on filesystem contents,
+               # but neither failure is likely to happen by accident.
                [info] = versions
                [child_path] = ver_map[info]
                return child_path
@@ -137,7 +159,11 @@ def join_content_path(path, content_doc):
                new_path.mkdir(exist_ok = True)
                write_metadata(
                        new_path,
-                       VersionInfo(bb_id = content_doc['id'], version = None),
+                       VersionInfo(
+                               VersionInfo.MetaType.NEW,
+                               bb_id = content_doc['id'],
+                               version = None
+                       ),
                )
                return new_path
 
@@ -163,5 +189,5 @@ def get_new_path(parent, base_name):
 
 
 def write_metadata(path, version_info):
-       with WinPath(path).get_ads(BB_META_STREAM_NAME).open('x') as f:
+       with fs.get_ads(path, BB_META_STREAM_NAME).open('x') as f:
                json.dump({'contentId': version_info.bb_id, 'version': version_info.version}, f)

diff --git a/main.py b/main.py
index e2d78db773824ea2165a8f86f6f47f12f97c11f1..eafa2b8444f2750066cb7a15020bc2f2aeeba5f7 100644 (file)
--- a/main.py
+++ b/main.py
@@ -84,7 +84,7 @@ with StorageManager(Path('auth_cache')) as storage_mgr:
        [course_id] = course_ids
        [content_id] = content_ids
 
-       local_course_root = fs.WinPath(cfg_section['base_path'])
+       local_course_root = Path(cfg_section['base_path'])
        local_course_root.mkdir(parents = True, exist_ok = True)
        meta_path = local_course_root.get_ads(fs.BB_META_STREAM_NAME)
        if meta_path.is_file():
@@ -142,7 +142,8 @@ with StorageManager(Path('auth_cache')) as storage_mgr:
 
                                my_versions = [info for info in versions.keys() if info.bb_id == att_id]
                                if my_versions:
-                                       latest = max(my_versions, key = attrgetter('version'))
+                                       sort_key = lambda v: (v.version, v.meta_type is v.MetaType.NEW)
+                                       latest = max(my_versions, key = sort_key)
                                        latest_paths = versions[latest]
                                        if len(latest_paths) == 1:
                                                [latest_path] = latest_paths