Repository: OCR-D/core (python)

Issue:
allow filtering by `mets:div/@TYPE`
In case the METS already contains a **logical structMap** (or we run a processor like `ocrd-anybaseocr-layout-analysis` as first step), there will be information on the type of every page (content vs title vs empty vs cover etc). This in turn might be very useful to allow filtering the page range when processing – complementary to the `--page-id` selection mechanism we already have (i.e. list or regex of physical page `@ID`, `@ORDER`, `@ORDERLABEL`, `@LABEL`, or `@CONTENTIDS`).

We might tie the list of types to consider to [DFG Strukturdatenset](https://dfg-viewer.de/strukturdatenset) – but in many cases, libraries extend that anyway (or use different spellings, like titlecase).

For example, a user may want to rule out processing `cover_front,cover_back,binding` pages. (Not only to avoid wasting resources, but because some processors may not handle these well, or even crash.)

Our current syntax does not allow for negative selection, so we'd need some new operator. I suggest `~`, since `!` might be interpreted by the shell via history expansion (even when quoted to prevent pathname expansion).

For `..` range expressions, negation only makes sense when applied to the entire expression. For comma-separated lists, every atom could be negated individually. And for regexes, the entire expression.

So how about `--page-id ~PHYS_0001..PHYS_0004`, `--page-id ~cover_front,~cover_back,~binding`, `--page-id "~//(cover_(front|back)|binding)"`?

(We might even want to support selecting logical div `@ID` in addition to `@TYPE`, e.g. `--page-id ~LOG_0001..LOG_0005` ...)

Pull request:
--page-id negation operator and logical structmap attribute matching
fixes #1328, depends on https://github.com/OCR-D/assets/pull/104

Patch:
diff --git a/repo/spec b/repo/spec
index de0e4269..506b3393 160000
--- a/repo/spec
+++ b/repo/spec
@@ -1,1 +1,1 @@
-Subproject commit de0e426987605c2dc278d422a0176c6069819230
+Subproject commit 506b33936d89080a683fa8a26837f2a23b23e5e2
diff --git a/requirements.txt b/requirements.txt
index 3e2dc689..5b9ef927 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -8,7 +8,7 @@ elementpath
 fastapi>=0.78.0
 filetype
 Flask
-frozendict>=2.3.4
+frozendict>=2.4.0
 gdown
 httpx>=0.22.0
 importlib_metadata ; python_version < '3.8'
diff --git a/src/ocrd_models/constants.py b/src/ocrd_models/constants.py
index 08bf10d9..f7f71f8f 100644
--- a/src/ocrd_models/constants.py
+++ b/src/ocrd_models/constants.py
@@ -1,7 +1,11 @@
 """
 Constants for ocrd_models.
 """
+from re import Pattern
 from enum import Enum, auto
+from dataclasses import dataclass, field
+from abc import ABC, abstractmethod
+from typing import Any, List, Optional, Union
 from ocrd_utils import resource_string
 
 __all__ = [
@@ -28,6 +32,10 @@ __all__ = [
     'TAG_PAGE_TEXTEQUIV',
     'TAG_PAGE_TEXTREGION',
     'METS_PAGE_DIV_ATTRIBUTE',
+    'METS_STRUCT_DIV_ATTRIBUTE',
+    'METS_DIV_ATTRIBUTE_ATOM_PATTERN',
+    'METS_DIV_ATTRIBUTE_RANGE_PATTERN',
+    'METS_DIV_ATTRIBUTE_REGEX_PATTERN',
     'PAGE_REGION_TYPES',
     'PAGE_ALTIMG_FEATURES',
 ]
@@ -89,6 +97,7 @@ PAGE_ALTIMG_FEATURES = [
 
 
 class METS_PAGE_DIV_ATTRIBUTE(Enum):
+    """page selection attributes of PHYSICAL mets:structMap//mets:div"""
     ID = auto()
     ORDER = auto()
     ORDERLABEL = auto()
@@ -98,3 +107,99 @@ class METS_PAGE_DIV_ATTRIBUTE(Enum):
     @classmethod
     def names(cls):
         return [x.name for x in cls]
+    @classmethod
+    def type_prefix(cls):
+        """disambiguation prefix to use for all subtypes"""
+        return "physical:"
+    def prefix(self):
+        """disambiguation prefix to use for this attribute type"""
+        return self.type_prefix() + self.name.lower() + ":"
+
+class METS_STRUCT_DIV_ATTRIBUTE(Enum):
+    """page selection attributes of LOGICAL mets:structMap//mets:div"""
+    ID = auto()
+    DMDID = auto()
+    TYPE = auto()
+    LABEL = auto()
+
+    @classmethod
+    def names(cls):
+        return [x.name for x in cls]
+    @classmethod
+    def type_prefix(cls):
+        """disambiguation prefix to use for all subtypes"""
+        return "logical:"
+    def prefix(self):
+        """disambiguation prefix to use for this attribute type"""
+        return self.type_prefix() + self.name.lower() + ":"
+
+@dataclass
+class METS_DIV_ATTRIBUTE_PATTERN(ABC):
+    """page selection pattern (abstract supertype)"""
+
+    expr: Any
+    """pattern value to match a mets:div against"""
+    attr: List[Union[METS_PAGE_DIV_ATTRIBUTE, METS_STRUCT_DIV_ATTRIBUTE]] = field(
+        default_factory=lambda: list(METS_PAGE_DIV_ATTRIBUTE) + list(METS_STRUCT_DIV_ATTRIBUTE))
+    """attribute type(s) to match a mets:div for
+    (pre-disambiguated with prefix syntax, or filled upon first match)
+    """
+    has_matched: bool = field(init=False, default=False)
+    """whether this pattern has already been matched"""
+
+    def attr_prefix(self):
+        """attribute type disambiguation prefix corresponding to the current state of disambiguation"""
+        if self.attr == list(METS_PAGE_DIV_ATTRIBUTE) + list(METS_STRUCT_DIV_ATTRIBUTE):
+            return ""
+        if self.attr == list(METS_PAGE_DIV_ATTRIBUTE):
+            return METS_PAGE_DIV_ATTRIBUTE.type_prefix()
+        if self.attr == list(METS_STRUCT_DIV_ATTRIBUTE):
+            return METS_STRUCT_DIV_ATTRIBUTE.type_prefix()
+        assert len(self.attr) == 1, "unexpected type ambiguity: %s" % repr(self.attr)
+        return self.attr[0].prefix()
+
+    @abstractmethod
+    def _matches(self, input) -> bool:
+        return
+    def matches(self, input) -> bool:
+        """does the selection pattern match on the given attribute value?"""
+        if (matched := self._matches(input)):
+            self.has_matched = True
+        return matched
+
+@dataclass
+class METS_DIV_ATTRIBUTE_ATOM_PATTERN(METS_DIV_ATTRIBUTE_PATTERN):
+    """page selection pattern for literal (single value) matching"""
+
+    expr: str
+    def __repr__(self):
+        return "%s%s" % (self.attr_prefix(), self.expr)
+    def _matches(self, input):
+        return input == self.expr
+
+@dataclass
+class METS_DIV_ATTRIBUTE_RANGE_PATTERN(METS_DIV_ATTRIBUTE_PATTERN):
+    """page selection pattern for interval (list expansion) matching"""
+
+    expr: List[str]
+    start: str = field(init=False)
+    """first value of the range after expansion, before matching-exhausting"""
+    stop: str = field(init=False)
+    """last value of the range after expansion, before matching-exhausting"""
+    def __post_init__(self):
+        self.start = self.expr[0]
+        self.stop = self.expr[-1]
+    def __repr__(self):
+        return "%s%s..%s" % (self.attr_prefix(), self.start, self.stop)
+    def _matches(self, input):
+        return input in self.expr
+
+@dataclass
+class METS_DIV_ATTRIBUTE_REGEX_PATTERN(METS_DIV_ATTRIBUTE_PATTERN):
+    """page selection pattern for regular expression matching"""
+
+    expr: Pattern
+    def __repr__(self):
+        return "%s//%s" % (self.attr_prefix(), self.expr.pattern)
+    def _matches(self, input):
+        return bool(self.expr.fullmatch(input))
diff --git a/src/ocrd_models/ocrd_mets.py b/src/ocrd_models/ocrd_mets.py
index de068567..152ab49d 100644
--- a/src/ocrd_models/ocrd_mets.py
+++ b/src/ocrd_models/ocrd_mets.py
@@ -29,7 +29,12 @@ from .constants import (
     IDENTIFIER_PRIORITY,
     TAG_MODS_IDENTIFIER,
     METS_XML_EMPTY,
-    METS_PAGE_DIV_ATTRIBUTE
+    METS_PAGE_DIV_ATTRIBUTE,
+    METS_STRUCT_DIV_ATTRIBUTE,
+    METS_DIV_ATTRIBUTE_PATTERN,
+    METS_DIV_ATTRIBUTE_ATOM_PATTERN,
+    METS_DIV_ATTRIBUTE_RANGE_PATTERN,
+    METS_DIV_ATTRIBUTE_REGEX_PATTERN,
 )
 
 from .ocrd_xml_base import OcrdXmlDocument, ET      # type: ignore
@@ -43,9 +48,11 @@ class OcrdMets(OcrdXmlDocument):
     API to a single METS file
     """
     _cache_flag : bool
-    # Cache for the pages (mets:div)
-    # The dictionary's Key: 'div.ID'
-    # The dictionary's Value: a 'div' object at some memory location
+    # Cache for the physical pages (mets:div) - two nested dictionaries
+    # The outer dictionary's key: attribute type
+    # The outer dictionary's value: inner dictionary
+    # The inner dictionary's key: attribute value (str)
+    # The inner dictionary's value: a 'div' object at some memory location
     _page_cache : Dict[METS_PAGE_DIV_ATTRIBUTE, Dict[str, ET._Element]]
     # Cache for the files (mets:file) - two nested dictionaries
     # The outer dictionary's Key: 'fileGrp.USE'
@@ -59,6 +66,12 @@ class OcrdMets(OcrdXmlDocument):
     # The inner dictionary's Key: 'fptr.FILEID'
     # The inner dictionary's Value: a 'fptr' object at some memory location
     _fptr_cache : Dict[str, Dict[str, ET._Element]]
+    # Cache for the logical structural divs (mets:div) - two nested dictionaries
+    # The outer dictionary's key: attribute type
+    # The outer dictionary's value: inner dictionary
+    # The inner dictionary's key: attribute value (str)
+    # The inner dictionary's value: a list of corresponding physical div.ID
+    _struct_cache : Dict[METS_STRUCT_DIV_ATTRIBUTE, Dict[str, List[str]]]
 
     @staticmethod
     def empty_mets(now : Optional[str] = None, cache_flag : bool = False):
@@ -111,7 +124,6 @@ class OcrdMets(OcrdXmlDocument):
             return
 
         log = getLogger('ocrd.models.ocrd_mets._fill_caches-files')
-
         for el_fileGrp in el_fileSec.findall('mets:fileGrp', NS):
             fileGrp_use = el_fileGrp.get('USE')
 
@@ -124,10 +136,10 @@ class OcrdMets(OcrdXmlDocument):
                 # log.info("File added to the cache: %s" % file_id)
 
         # Fill with pages
+        log = getLogger('ocrd.models.ocrd_mets._fill_caches-pages')
         el_div_list = tree_root.findall(".//mets:div[@TYPE='page']", NS)
         if len(el_div_list) == 0:
             return
-        log = getLogger('ocrd.models.ocrd_mets._fill_caches-pages')
 
         for el_div in el_div_list:
             div_id = el_div.get('ID')
@@ -148,11 +160,30 @@ class OcrdMets(OcrdXmlDocument):
         # log.info("Len of page_cache: %s" % len(self._page_cache[METS_PAGE_DIV_ATTRIBUTE.ID]))
         # log.info("Len of fptr_cache: %s" % len(self._fptr_cache))
 
+        # Fill with logical divs
+        log = getLogger('ocrd.models.ocrd_mets._fill_caches-structs')
+        el_struct_list = tree_root.findall("mets:structMap[@TYPE='LOGICAL']//mets:div", NS)
+        el_smlink_list = tree_root.findall("mets:structLink/mets:smLink", NS)
+        if len(el_struct_list) == 0 or len(el_smlink_list) == 0:
+            return
+        smlink_map = {}
+        for link in el_smlink_list:
+            link_log = link.get('{%s}from' % NS['xlink'])
+            link_phy = link.get('{%s}to' % NS['xlink'])
+            smlink_map.setdefault(link_log, list()).append(link_phy)
+        for el_div in el_struct_list:
+            for attr in METS_STRUCT_DIV_ATTRIBUTE:
+                val = self._struct_cache[attr].setdefault(str(el_div.get(attr.name)), list())
+                val.extend(smlink_map.get(el_div.get('ID'), []))
+
+        # log.info("Len of struct_cache: %s" % len(self._struct_cache[METS_STRUCT_DIV_ATTRIBUTE.ID]))
+
     def _initialize_caches(self) -> None:
         self._file_cache = {}
         # NOTE we can only guarantee uniqueness for @ID and @ORDER
         self._page_cache = {k : {} for k in METS_PAGE_DIV_ATTRIBUTE}
         self._fptr_cache = {}
+        self._struct_cache = {k : {} for k in METS_STRUCT_DIV_ATTRIBUTE}
 
     def _refresh_caches(self) -> None:
         if self._cache_flag:
@@ -253,12 +284,20 @@ class OcrdMets(OcrdXmlDocument):
         :py:attr:`url` and :py:attr:`mimetype` parameters can each be either a
         literal string, or a regular expression if the string starts with
         ``//`` (double slash).
+
         If it is a regex, the leading ``//`` is removed and candidates are matched
         against the regex with `re.fullmatch`. If it is a literal string, comparison
         is done with string equality.
-        The :py:attr:`pageId` parameter supports the numeric range operator ``..``. For
-        example, to find all files in pages ``PHYS_0001`` to ``PHYS_0003``,
-        ``PHYS_0001..PHYS_0003`` will be expanded to ``PHYS_0001,PHYS_0002,PHYS_0003``.
+
+        The :py:attr:`pageId` parameter also supports comma-separated lists, as well
+        as the numeric range operator ``..`` and the negation operator ``~``.
+
+        For example, to find all files in pages ``PHYS_0001`` to ``PHYS_0003``, the
+        both expressions ``PHYS_0001..PHYS_0003`` and ``PHYS_0001,PHYS_0002,PHYS_0003``
+        will be expanded to the same 3 pages. To find all files above that subrange,
+        both expressions ``~PHYS_0001..PHYS_0003`` and ``~PHYS_0001,~PHYS_0002,~PHYS_0003``
+        will be expanded to ``PHYS_0004`` and upwards.
+
         Keyword Args:
             ID (string) : ``@ID`` of the ``mets:file``
             fileGrp (string) : ``@USE`` of the ``mets:fileGrp`` to list files of
@@ -609,101 +648,73 @@ class OcrdMets(OcrdXmlDocument):
 
             return self.physical_pages
 
-        # log = getLogger('ocrd.models.ocrd_mets.get_physical_pages')
+        log = getLogger('ocrd.models.ocrd_mets.get_physical_pages')
         if for_pageIds is not None:
-            ret = []
             page_attr_patterns = []
-            page_attr_patterns_raw = re.split(r',', for_pageIds)
-            for pageId_token in page_attr_patterns_raw:
+            page_attr_antipatterns = []
+            for pageId_token in re.split(r',', for_pageIds):
+                pageId_token_raw = pageId_token
+                # prefix for disambiguation of attribute?
+                attr = list(METS_PAGE_DIV_ATTRIBUTE) + list(METS_STRUCT_DIV_ATTRIBUTE)
+                for attr_type in [METS_STRUCT_DIV_ATTRIBUTE, METS_PAGE_DIV_ATTRIBUTE]:
+                    if pageId_token.startswith(attr_type.type_prefix()):
+                        for attr_val in list(attr_type):
+                            if pageId_token.startswith(attr_val.prefix()):
+                                # disambiguated to e.g. "logical:label:"
+                                attr = [attr_val]
+                                pageId_token = pageId_token[len(attr_val.prefix()):]
+                                break
+                        if len(attr) > 1:
+                            # just "logical:" or "physical:"
+                            attr = list(attr_type)
+                            pageId_token = pageId_token[len(attr_type.type_prefix()):]
+                        break
+                if not pageId_token:
+                    raise ValueError("invalid pageId syntax '%s': empty after type prefix" % pageId_token_raw)
+                # negation prefix
+                if pageId_token.startswith('~'):
+                    page_attr_xpatterns = page_attr_antipatterns
+                    pageId_token = pageId_token[1:]
+                else:
+                    page_attr_xpatterns = page_attr_patterns
+                if not pageId_token:
+                    raise ValueError("invalid pageId syntax '%s': empty after negator prefix" % pageId_token_raw)
+                # operator prefix
                 if pageId_token.startswith(REGEX_PREFIX):
-                    page_attr_patterns.append((None, re.compile(pageId_token[REGEX_PREFIX_LEN:])))
+                    pageId_token = pageId_token[REGEX_PREFIX_LEN:]
+                    if not pageId_token:
+                        raise ValueError("invalid pageId syntax '%s': empty after regex prefix" % pageId_token_raw)
+                    val_expr = re.compile(pageId_token)
+                    page_attr_xpatterns.append(
+                        METS_DIV_ATTRIBUTE_REGEX_PATTERN(val_expr, attr))
                 elif '..' in pageId_token:
-                    val_range = generate_range(*pageId_token.split('..', 1))
-                    page_attr_patterns.append(val_range)
+                    try:
+                        val_range = generate_range(*pageId_token.split('..', 1))
+                    except ValueError as e:
+                        raise ValueError("invalid pageId syntax '%s': %s" % (pageId_token_raw, str(e))) from None
+                    page_attr_xpatterns.append(
+                        METS_DIV_ATTRIBUTE_RANGE_PATTERN(val_range, attr))
                 else:
-                    page_attr_patterns.append(pageId_token)
-            if not page_attr_patterns:
+                    if not pageId_token:
+                        raise ValueError("invalid pageId syntax '%s': empty" % pageId_token_raw)
+                    page_attr_xpatterns.append(
+                        METS_DIV_ATTRIBUTE_ATOM_PATTERN(pageId_token, attr))
+                log.debug("parsed pattern '%s' to %s", pageId_token_raw, page_attr_xpatterns[-1])
+            if not page_attr_patterns and not page_attr_antipatterns:
                 return []
-            range_patterns_first_last = [(x[0], x[-1]) if isinstance(x, list) else None for x in page_attr_patterns]
-            page_attr_patterns_copy = list(page_attr_patterns)
-            if self._cache_flag:
-                for pat in page_attr_patterns:
-                    try:
-                        attr : METS_PAGE_DIV_ATTRIBUTE
-                        if isinstance(pat, str):
-                            attr = next(a for a in list(METS_PAGE_DIV_ATTRIBUTE) if pat in self._page_cache[a])
-                            cache_keys = [pat]
-                        elif isinstance(pat, list):
-                            attr = next(a for a in list(METS_PAGE_DIV_ATTRIBUTE) if any(x in self._page_cache[a] for x in pat))
-                            cache_keys = [v for v in pat if v in self._page_cache[attr]]
-                            for k in cache_keys:
-                                pat.remove(k)
-                        elif isinstance(pat, tuple):
-                            _, re_pat = pat
-                            attr = next(a for a in list(METS_PAGE_DIV_ATTRIBUTE) for v in self._page_cache[a] if re_pat.fullmatch(v))
-                            cache_keys = [v for v in self._page_cache[attr] if re_pat.fullmatch(v)]
-                        else:
-                            raise ValueError
-                        if return_divs:
-                            ret += [self._page_cache[attr][v] for v in cache_keys]
-                        else:
-                            ret += [self._page_cache[attr][v].get('ID') for v in cache_keys]
-                    except StopIteration:
-                        raise ValueError(f"{pat} matches none of the keys of any of the _page_caches.")
+            if page_attr_patterns:
+                divs = self.get_physical_page_patterns(page_attr_patterns)
             else:
-                page_attr_patterns_matched = []
-                for page in self._tree.getroot().xpath(
-                        'mets:structMap[@TYPE="PHYSICAL"]/mets:div[@TYPE="physSequence"]/mets:div[@TYPE="page"]',
-                        namespaces=NS):
-                    patterns_exhausted = []
-                    for pat_idx, pat in enumerate(page_attr_patterns):
-                        try:
-                            if isinstance(pat, str):
-                                attr = next(a for a in list(METS_PAGE_DIV_ATTRIBUTE) if pat == page.get(a.name))
-                                ret.append(page if return_divs else page.get('ID'))
-                                patterns_exhausted.append(pat)
-                            elif isinstance(pat, list):
-                                if not isinstance(pat[0], METS_PAGE_DIV_ATTRIBUTE):
-                                    pat.insert(0, next(a for a in list(METS_PAGE_DIV_ATTRIBUTE) if any(x == page.get(a.name) for x in pat)))
-                                attr_val = page.get(pat[0].name)
-                                if attr_val in pat:
-                                    pat.remove(attr_val)
-                                    ret.append(page if return_divs else page.get('ID'))
-                                if len(pat) == 1:
-                                    patterns_exhausted.append(pat)
-                            elif isinstance(pat, tuple):
-                                attr, re_pat = pat
-                                if not attr:
-                                    attr = next(a for a in list(METS_PAGE_DIV_ATTRIBUTE) if re_pat.fullmatch(page.get(a.name) or ''))
-                                    page_attr_patterns[pat_idx] = (attr, re_pat)
-                                if re_pat.fullmatch(page.get(attr.name) or ''):
-                                    ret.append(page if return_divs else page.get('ID'))
-                            else:
-                                raise ValueError
-                            page_attr_patterns_matched.append(pat)
-                        except StopIteration:
-                            continue
-                    for p in patterns_exhausted:
-                        page_attr_patterns.remove(p)
-                unmatched = [x for x in page_attr_patterns_copy if x not in page_attr_patterns_matched]
-                if unmatched:
-                    raise ValueError(f"Patterns {unmatched} match none of the pages")
-
-            ranges_without_start_match = []
-            ranges_without_last_match = []
-            for idx, pat in enumerate(page_attr_patterns_copy):
-                if isinstance(pat, list):
-                    start, last = range_patterns_first_last[idx]
-                    if start in pat:
-                        print(pat, start, last)
-                        ranges_without_start_match.append(page_attr_patterns_raw[idx])
-                    # if last in pat:
-                    #     ranges_without_last_match.append(page_attr_patterns_raw[idx])
-            if ranges_without_start_match:
-                raise ValueError(f"Start of range patterns {ranges_without_start_match} not matched - invalid range")
-            # if ranges_without_last_match:
-            #     raise ValueError(f"End of range patterns {ranges_without_last_match} not matched - invalid range")
-            return ret
+                all_pages = [METS_DIV_ATTRIBUTE_REGEX_PATTERN(
+                    re.compile(".*"), [METS_PAGE_DIV_ATTRIBUTE.ID])]
+                divs = self.get_physical_page_patterns(all_pages)
+            if page_attr_antipatterns:
+                antidivs = self.get_physical_page_patterns(page_attr_antipatterns)
+                divs = [div for div in divs if div not in antidivs]
+            if return_divs:
+                return divs
+            else:
+                return [div.get('ID') for div in divs]
 
         if for_fileIds == []:
             return []
@@ -731,6 +742,129 @@ class OcrdMets(OcrdXmlDocument):
                             ret[index] = page.get('ID')
         return ret
 
+    def get_physical_page_patterns(self, page_attr_patterns: List[METS_DIV_ATTRIBUTE_PATTERN]) -> List[ET._Element]:
+        log = getLogger('ocrd.models.ocrd_mets.get_physical_pages')
+        ret = []
+        page_attr_patterns_copy = list(page_attr_patterns)
+        if self._cache_flag:
+            for pat in page_attr_patterns:
+                for attr in pat.attr:
+                    if isinstance(attr, METS_PAGE_DIV_ATTRIBUTE):
+                        cache = self._page_cache[attr]
+                    else:
+                        cache = self._struct_cache[attr]
+                    if (isinstance(pat, METS_DIV_ATTRIBUTE_RANGE_PATTERN) and
+                        # @TYPE makes no sense in range expressions
+                        # @LABEL makes no sense in range expressions
+                        attr in [METS_STRUCT_DIV_ATTRIBUTE.TYPE,
+                                 METS_STRUCT_DIV_ATTRIBUTE.LABEL]):
+                        continue
+                    if cache_keys := [v for v in cache if pat.matches(v)]:
+                        if isinstance(attr, METS_PAGE_DIV_ATTRIBUTE):
+                            ret += [cache[v] for v in cache_keys]
+                            log.debug('physical matches for %s: %s', pat, str(cache_keys))
+                        else:
+                            for v in cache_keys:
+                                ret += [self._page_cache[METS_PAGE_DIV_ATTRIBUTE.ID][p]
+                                        for p in cache[v]]
+                            log.debug('logical matches for %s: %s', pat, str(cache_keys))
+                        if isinstance(pat, METS_DIV_ATTRIBUTE_RANGE_PATTERN):
+                            # remove matches for final range check
+                            for v in cache_keys:
+                                pat.expr.remove(v)
+                        break
+                if not cache_keys:
+                    raise ValueError(f"{pat} matches none of the keys of any of the _page_caches and _struct_caches.")
+        else:
+            # cache logical structmap:
+            el_struct_list = self._tree.getroot().findall("mets:structMap[@TYPE='LOGICAL']//mets:div", NS)
+            el_smlink_list = self._tree.getroot().findall("mets:structLink/mets:smLink", NS)
+            smlink_map = {}
+            for link in el_smlink_list:
+                link_log = link.get('{%s}from' % NS['xlink'])
+                link_phy = link.get('{%s}to' % NS['xlink'])
+                smlink_map.setdefault(link_log, list()).append(link_phy)
+            struct_cache = {k: {} for k in METS_STRUCT_DIV_ATTRIBUTE}
+            for el_div in el_struct_list:
+                for attr in METS_STRUCT_DIV_ATTRIBUTE:
+                    if not el_div.get(attr.name):
+                        # avoid mapping None indiscriminately
+                        continue
+                    val = struct_cache[attr].setdefault(str(el_div.get(attr.name)), list())
+                    val.extend(smlink_map.get(el_div.get('ID'), []))
+            log.debug("found %d smLink entries for %d logical divs", len(el_smlink_list), len(el_struct_list))
+            page_attr_patterns_matched = []
+            for page in self._tree.getroot().xpath(
+                    'mets:structMap[@TYPE="PHYSICAL"]/mets:div[@TYPE="physSequence"]/mets:div[@TYPE="page"]',
+                    namespaces=NS):
+                patterns_exhausted = []
+                for pat in page_attr_patterns:
+                    for attr in pat.attr:
+                        if isinstance(attr, METS_PAGE_DIV_ATTRIBUTE):
+                            cache = [page.get(attr.name) or '']
+                        else:
+                            cache = struct_cache[attr]
+                        if (isinstance(pat, METS_DIV_ATTRIBUTE_RANGE_PATTERN) and
+                            # @TYPE makes no sense in range expressions
+                            # @LABEL makes no sense in range expressions
+                            attr in [METS_STRUCT_DIV_ATTRIBUTE.TYPE,
+                                     METS_STRUCT_DIV_ATTRIBUTE.LABEL]):
+                            continue
+                        if cache_keys := [v for v in cache if pat.matches(v)]:
+                            pat.attr = [attr] # disambiguate next
+                            if isinstance(attr, METS_PAGE_DIV_ATTRIBUTE):
+                                ret.append(page)
+                                log.debug('physical match for %s on page %s', pat, page.get('ID'))
+                                if isinstance(pat, METS_DIV_ATTRIBUTE_ATOM_PATTERN):
+                                    patterns_exhausted.append(pat)
+                                elif isinstance(pat, METS_DIV_ATTRIBUTE_RANGE_PATTERN):
+                                    # remove for efficiency and final range check
+                                    pat.expr.remove(cache_keys[0])
+                                    if not pat.expr:
+                                        patterns_exhausted.append(pat)
+                            elif cache_key := next((v for v in cache_keys
+                                                    if page.get('ID') in cache[v]), None):
+                                ret.append(page)
+                                log.debug('logical match for %s on page %s', pat, page.get('ID'))
+                                cache[cache_key].remove(page.get('ID'))
+                                # remove for efficiency and final range check
+                                if not cache[cache_key]:
+                                    if isinstance(pat, METS_DIV_ATTRIBUTE_ATOM_PATTERN):
+                                        patterns_exhausted.append(pat)
+                                    elif isinstance(pat, METS_DIV_ATTRIBUTE_RANGE_PATTERN):
+                                        pat.expr.remove(cache_key)
+                                        if not pat.expr:
+                                            patterns_exhausted.append(pat)
+                            break # no more attributes for this pattern
+                    # keep matching in order to exhaust and consume pattern list
+                    #if page in ret:
+                    #    break # no more patterns for this page
+                for p in patterns_exhausted:
+                    page_attr_patterns.remove(p)
+            unmatched = [pat for pat in page_attr_patterns_copy
+                         if not pat.has_matched]
+            if unmatched:
+                raise ValueError(f"Patterns {unmatched} match none of the pages")
+
+        ranges_without_start_match = []
+        ranges_without_stop_match = []
+        for pat in page_attr_patterns_copy:
+            if isinstance(pat, METS_DIV_ATTRIBUTE_RANGE_PATTERN):
+                # range expression, expanded to pattern list
+                # list items get consumed (pat.expr.remove) when matched,
+                # exhausted patterns also get consumed (page_attr_patterns.remove)
+                # (but top-level list copy references the same list objects)
+                if pat.start in pat.expr:
+                    log.debug((pat, pat.expr))
+                    ranges_without_start_match.append(pat)
+                # if pat.stop in pat.expr:
+                #     ranges_without_stop_match.append(pat)
+        if ranges_without_start_match:
+            raise ValueError(f"Start of range patterns {ranges_without_start_match} not matched - invalid range")
+        # if ranges_without_stop_match:
+        #     raise ValueError(f"End of range patterns {ranges_without_stop_match} not matched - invalid range")
+        return ret
+
     def set_physical_page_for_file(self, pageId : str, ocrd_file : OcrdFile, 
                                    order : Optional[str] = None, orderlabel : Optional[str] = None) -> None:
         """


Test patch:
diff --git a/tests/cli/test_workspace.py b/tests/cli/test_workspace.py
index a214cb9f..f0b34e98 100644
--- a/tests/cli/test_workspace.py
+++ b/tests/cli/test_workspace.py
@@ -411,6 +411,10 @@ class TestCli(TestCase):
             self.assertEqual(result.exit_code, 0)
             result = self.runner.invoke(workspace_cli, ['-d', tempdir, 'find', '--page-id', 'PHYS_0005,PHYS_0001', '-k', 'local_filename'])
             self.assertEqual(len(result.stdout.split('\n')), 19)
+            self.assertEqual(result.exit_code, 0)
+            result = self.runner.invoke(workspace_cli, ['-d', tempdir, 'find', '--page-id', '~PHYS_0002', '-k', 'local_filename'])
+            self.assertEqual(len(result.stdout.split('\n')), 19)
+            self.assertEqual(result.exit_code, 0)
 
     def test_mets_basename(self):
         with TemporaryDirectory() as tempdir:
diff --git a/tests/model/test_ocrd_mets.py b/tests/model/test_ocrd_mets.py
index 79ecee19..15c4665a 100644
--- a/tests/model/test_ocrd_mets.py
+++ b/tests/model/test_ocrd_mets.py
@@ -16,7 +16,8 @@ from tests.base import (
 
 from ocrd_utils import (
     VERSION,
-    MIMETYPE_PAGE
+    MIMETYPE_PAGE,
+    setOverrideLogLevel
 )
 from ocrd_models import (
     OcrdMets
@@ -28,14 +29,18 @@ CACHING_ENABLED = [False, True]
 
 
 @pytest.fixture(name='sbb_sample_01', params=CACHING_ENABLED)
-def _fixture(request):
+def _fixture(request, pytestconfig):
+    if pytestconfig.getoption('verbose') > 0:
+            setOverrideLogLevel('DEBUG')
     mets = OcrdMets(filename=assets.url_of(
         'SBB0000F29300010000/data/mets.xml'), cache_flag=request.param)
     yield mets
 
 
 @pytest.fixture(name='sbb_directory_ocrd_mets', params=CACHING_ENABLED)
-def _fixture_sbb(tmp_path, request):
+def _fixture_sbb(tmp_path, request, pytestconfig):
+    if pytestconfig.getoption('verbose') > 0:
+            setOverrideLogLevel('DEBUG')
     src_path = assets.path_to('SBB0000F29300010000/data')
     dst_path = tmp_path / 'SBB_directory'
     shutil.copytree(src_path, dst_path)
@@ -43,6 +48,15 @@ def _fixture_sbb(tmp_path, request):
     yield OcrdMets(filename=mets_path, cache_flag=request.param)
 
 
+@pytest.fixture(name='pembroke_werke', params=CACHING_ENABLED)
+def _fixture_pembroke(request, pytestconfig):
+    if pytestconfig.getoption('verbose') > 0:
+            setOverrideLogLevel('DEBUG')
+    mets = OcrdMets(filename=assets.url_of(
+        'pembroke_werke_1766/data/mets.xml'), cache_flag=request.param)
+    yield mets
+
+
 def test_unique_identifier():
     mets = OcrdMets(filename=assets.url_of('SBB0000F29300010000/data/mets.xml'))
     assert mets.unique_identifier == 'http://resolver.staatsbibliothek-berlin.de/SBB0000F29300010000', 'Right identifier'
@@ -88,7 +102,12 @@ def test_find_all_files(sbb_sample_01):
     assert len(mets.find_all_files(local_filename='OCR-D-IMG/FILE_0005_IMAGE.tif')) == 1, '1 FILE xlink:href="OCR-D-IMG/FILE_0005_IMAGE.tif"'
     assert len(mets.find_all_files(url='https://github.com/OCR-D/assets/raw/master/data/SBB0000F29300010000/data/00000001_DESKEW.tif')) == 1, '1 URL xlink:href="https://github.com/OCR-D/assets/raw/master/data/SBB0000F29300010000/00000001_DESKEW.tif"'
     assert len(mets.find_all_files(pageId='PHYS_0001..PHYS_0005')) == 35, '35 files for page "PHYS_0001..PHYS_0005"'
-    assert len(mets.find_all_files(pageId='//PHYS_000(1|2)')) == 34, '34 files in PHYS_001 and PHYS_0002'
+    assert len(mets.find_all_files(pageId='~PHYS_0001..PHYS_0002')) == 1, '1 file for page "~PHYS_0001..PHYS_0002"'
+    assert len(mets.find_all_files(pageId='PHYS_0001,~PHYS_0002')) == 17, '17 files for page "PHYS_0001,~PHYS_0002"'
+    assert len(mets.find_all_files(pageId='PHYS_0001..PHYS_0005,~PHYS_0002')) == 18, '18 files for page "PHYS_0001..PHYS_0005,~PHYS_0002"'
+    assert len(mets.find_all_files(pageId='~PHYS_0002')) == 18, '18 files for page "~PHYS_0002"'
+    assert len(mets.find_all_files(pageId='//PHYS_000(1|2)')) == 34, '34 files in PHYS_0001 and PHYS_0002'
+    assert len(mets.find_all_files(pageId='~//PHYS_000(1|2)')) == 1, '1 file in PHYS_0005'
     assert len(mets.find_all_files(pageId='//PHYS_0001,//PHYS_0005')) == 18, '18 files in PHYS_001 and PHYS_0005 (two regexes)'
     assert len(mets.find_all_files(pageId='//PHYS_0005,PHYS_0001..PHYS_0002')) == 35, '35 files in //PHYS_0005,PHYS_0001..PHYS_0002'
     assert len(mets.find_all_files(pageId='//PHYS_0005,PHYS_0001..PHYS_0002')) == 35, '35 files in //PHYS_0005,PHYS_0001..PHYS_0002'
@@ -104,6 +123,8 @@ def test_find_all_files(sbb_sample_01):
         mets.find_all_files(pageId='PHYS_0006..PHYS_0029')
     with pytest.raises(ValueError, match=re.compile(f'match(es)? none')):
         mets.find_all_files(pageId='PHYS_0001-NOTEXIST')
+    with pytest.raises(ValueError, match=re.compile(f'match(es)? none')):
+        mets.find_all_files(pageId='~PHYS_0001-NOTEXIST')
     with pytest.raises(ValueError, match=re.compile(f'match(es)? none')):
         mets.find_all_files(pageId='1..5,PHYS_0006..PHYS_0029')
     with pytest.raises(ValueError, match=re.compile(f'match(es)? none')):
@@ -111,10 +132,73 @@ def test_find_all_files(sbb_sample_01):
     with pytest.raises(ValueError, match=re.compile(f'Start of range pattern')):
         mets.find_all_files(pageId='PHYS_0000..PHYS_0004')
 
+def test_find_all_files_with_prefix(sbb_sample_01):
+    mets = sbb_sample_01
+    assert len(mets.find_all_files(pageId='physical:PHYS_0001')) == 17, '17 files for page "PHYS_0001"'
+    assert len(mets.find_all_files(pageId='physical:id:PHYS_0001')) == 17, '17 files for page "PHYS_0001"'
+    assert len(mets.find_all_files(pageId='physical:PHYS_0001..PHYS_0005')) == 35, '35 files for page "PHYS_0001..PHYS_0005"'
+    assert len(mets.find_all_files(pageId='physical:id:PHYS_0001..PHYS_0005')) == 35, '35 files for page "PHYS_0001..PHYS_0005"'
+    assert len(mets.find_all_files(pageId='physical:~PHYS_0001..PHYS_0002')) == 1, '1 file for page "~PHYS_0001..PHYS_0002"'
+    assert len(mets.find_all_files(pageId='physical:id:~PHYS_0001..PHYS_0002')) == 1, '1 file for page "~PHYS_0001..PHYS_0002"'
+    assert len(mets.find_all_files(pageId='physical:PHYS_0001,physical:~PHYS_0002')) == 17, '17 files for page "PHYS_0001,~PHYS_0002"'
+    assert len(mets.find_all_files(pageId='physical:id:PHYS_0001,physical:id:~PHYS_0002')) == 17, '17 files for page "PHYS_0001,~PHYS_0002"'
+    assert len(mets.find_all_files(pageId='physical:PHYS_0001..PHYS_0005,physical:~PHYS_0002')) == 18, '18 files for page "PHYS_0001..PHYS_0005,~PHYS_0002"'
+    assert len(mets.find_all_files(pageId='physical:id:PHYS_0001..PHYS_0005,physical:id:~PHYS_0002')) == 18, '18 files for page "PHYS_0001..PHYS_0005,~PHYS_0002"'
+    assert len(mets.find_all_files(pageId='physical:~PHYS_0002')) == 18, '18 files for page "~PHYS_0002"'
+    assert len(mets.find_all_files(pageId='physical:id:~PHYS_0002')) == 18, '18 files for page "~PHYS_0002"'
+    assert len(mets.find_all_files(pageId='physical://PHYS_000(1|2)')) == 34, '34 files in PHYS_0001 and PHYS_0002'
+    assert len(mets.find_all_files(pageId='physical:id://PHYS_000(1|2)')) == 34, '34 files in PHYS_0001 and PHYS_0002'
+    assert len(mets.find_all_files(pageId='physical:~//PHYS_000(1|2)')) == 1, '1 file in PHYS_0005'
+    assert len(mets.find_all_files(pageId='physical:id:~//PHYS_000(1|2)')) == 1, '1 file in PHYS_0005'
+    assert len(mets.find_all_files(pageId='physical://PHYS_0001,physical://PHYS_0005')) == 18, '18 files in PHYS_001 and PHYS_0005 (two regexes)'
+    assert len(mets.find_all_files(pageId='physical:id://PHYS_0001,physical:id://PHYS_0005')) == 18, '18 files in PHYS_001 and PHYS_0005 (two regexes)'
+    assert len(mets.find_all_files(pageId='physical://PHYS_0005,physical:PHYS_0001..PHYS_0002')) == 35, '35 files in //PHYS_0005,PHYS_0001..PHYS_0002'
+    assert len(mets.find_all_files(pageId='physical:id://PHYS_0005,physical:id:PHYS_0001..PHYS_0002')) == 35, '35 files in //PHYS_0005,PHYS_0001..PHYS_0002'
+    assert len(mets.find_all_files(pageId='physical://PHYS_0005,physical:PHYS_0001..PHYS_0002')) == 35, '35 files in //PHYS_0005,PHYS_0001..PHYS_0002'
+    assert len(mets.find_all_files(pageId='physical:id://PHYS_0005,physical:id:PHYS_0001..PHYS_0002')) == 35, '35 files in //PHYS_0005,PHYS_0001..PHYS_0002'
+    assert len(mets.find_all_files(pageId='physical:1..10')) == 35, '35 files in @ORDER range 1..10'
+    assert len(mets.find_all_files(pageId='physical:order:1..10')) == 35, '35 files in @ORDER range 1..10'
+    assert len(mets.find_all_files(pageId='physical:1..5')) == 35, '35 files in @ORDER range 1..10'
+    assert len(mets.find_all_files(pageId='physical:order:1..5')) == 35, '35 files in @ORDER range 1..10'
+    assert len(mets.find_all_files(pageId='physical:page 1..page 2,physical:5')) == 35, '35 in PHYS_0001,PHYS_0002,PHYS_0005'
+    assert len(mets.find_all_files(pageId='physical:orderlabel:page 1..page 2,physical:order:5')) == 35, '35 in PHYS_0001,PHYS_0002,PHYS_0005'
+
 def test_find_all_files_local_only(sbb_sample_01):
     assert len(sbb_sample_01.find_all_files(pageId='PHYS_0001',
                local_only=True)) == 14, '14 local files for page "PHYS_0001"'
 
+def test_find_all_files_logical_pages(pembroke_werke):
+    mets = pembroke_werke
+    assert len(mets.find_all_files(fileGrp='DEFAULT')) == 195, '195 files total'
+    assert len(mets.find_all_files(fileGrp='DEFAULT',pageId='binding,title_page')) == 10, '10 pages of type binding and title_page (PHYS_0001..4,PHYS_0007..8,PHYS_0191..4)'
+    assert len(mets.find_all_files(fileGrp='DEFAULT',pageId='LOG_0010..LOG_0014')) == 85, '85 pages between LOG_0010 and LOG_0014'
+    assert len(mets.find_all_files(fileGrp='DEFAULT',pageId='//(section|chapter)')) == 179, '179 pages for //(section|chapter)'
+    assert len(mets.find_all_files(fileGrp='DEFAULT',pageId='~binding,~title_page,~colour_checker,~illustration,~table')) == 179, '179 pages for ~binding,~title_page,~colour_checker,~illustration,~table'
+    assert len(mets.find_all_files(fileGrp='DEFAULT',pageId='~LOG_0010..LOG_0014')) == 110, '110 pages not between LOG_0010 and LOG_0014'
+    assert len(mets.find_all_files(fileGrp='DEFAULT',pageId='~//(binding|title_page|colour_checker|illustration|table)')) == 179, '179 pages for ~//(binding|title_page|colour_checker|illustration|table)'
+    with pytest.raises(ValueError, match=re.compile(f'match(es)? none')):
+        mets.find_all_files(pageId='LOG_0001-NOTEXIST')
+    assert len(mets.find_all_files(fileGrp='DEFAULT',pageId='//^Einleitung.*')) == 1, '1 pages for //^Einleitung.*'
+
+def test_find_all_files_logical_pages_with_prefix(pembroke_werke):
+    mets = pembroke_werke
+    assert len(mets.find_all_files(fileGrp='DEFAULT',pageId='logical:binding,logical:title_page')) == 10, '10 pages of type binding and title_page (PHYS_0001..4,PHYS_0007..8,PHYS_0191..4)'
+    assert len(mets.find_all_files(fileGrp='DEFAULT',pageId='logical:type:binding,logical:type:title_page')) == 10, '10 pages of type binding and title_page (PHYS_0001..4,PHYS_0007..8,PHYS_0191..4)'
+    assert len(mets.find_all_files(fileGrp='DEFAULT',pageId='logical:LOG_0010..LOG_0014')) == 85, '85 pages between LOG_0010 and LOG_0014'
+    assert len(mets.find_all_files(fileGrp='DEFAULT',pageId='logical:id:LOG_0010..LOG_0014')) == 85, '85 pages between LOG_0010 and LOG_0014'
+    assert len(mets.find_all_files(fileGrp='DEFAULT',pageId='//(section|chapter)')) == 179, '179 pages for //(section|chapter)'
+    assert len(mets.find_all_files(fileGrp='DEFAULT',pageId='logical://(section|chapter)')) == 179, '179 pages for //(section|chapter)'
+    assert len(mets.find_all_files(fileGrp='DEFAULT',pageId='logical:type://(section|chapter)')) == 179, '179 pages for //(section|chapter)'
+    assert len(mets.find_all_files(fileGrp='DEFAULT',pageId='logical:~binding,logical:~title_page,logical:~colour_checker,logical:~illustration,logical:~table')) == 179, '179 pages for ~binding,~title_page,~colour_checker,~illustration,~table'
+    assert len(mets.find_all_files(fileGrp='DEFAULT',pageId='logical:type:~binding,logical:type:~title_page,logical:type:~colour_checker,logical:type:~illustration,logical:type:~table')) == 179, '179 pages for ~binding,~title_page,~colour_checker,~illustration,~table'
+    assert len(mets.find_all_files(fileGrp='DEFAULT',pageId='logical:~LOG_0010..LOG_0014')) == 110, '110 pages not between LOG_0010 and LOG_0014'
+    assert len(mets.find_all_files(fileGrp='DEFAULT',pageId='logical:id:~LOG_0010..LOG_0014')) == 110, '110 pages not between LOG_0010 and LOG_0014'
+    assert len(mets.find_all_files(fileGrp='DEFAULT',pageId='logical:~//(binding|title_page|colour_checker|illustration|table)')) == 179, '179 pages for ~//(binding|title_page|colour_checker|illustration|table)'
+    assert len(mets.find_all_files(fileGrp='DEFAULT',pageId='logical:type:~//(binding|title_page|colour_checker|illustration|table)')) == 179, '179 pages for ~//(binding|title_page|colour_checker|illustration|table)'
+    with pytest.raises(ValueError, match=re.compile(f'match(es)? none')):
+        mets.find_all_files(pageId='logical:LOG_0001-NOTEXIST')
+    assert len(mets.find_all_files(fileGrp='DEFAULT',pageId='logical://^Einleitung.*')) == 1, '1 pages for //^Einleitung.*'
+    assert len(mets.find_all_files(fileGrp='DEFAULT',pageId='logical:label://^Einleitung.*')) == 1, '1 pages for //^Einleitung.*'
 
 def test_physical_pages(sbb_sample_01):
     assert len(sbb_sample_01.physical_pages) == 3, '3 physical pages'
diff --git a/tests/processor/test_processor.py b/tests/processor/test_processor.py
index 8d496623..cd61996e 100644
--- a/tests/processor/test_processor.py
+++ b/tests/processor/test_processor.py
@@ -375,7 +375,8 @@ class TestProcessor(TestCase):
             ws.add_file('GRP2', mimetype='application/alto+xml', file_id='foobar2', page_id='phys_0001')
             ws.add_file('GRP1', mimetype=MIMETYPE_PAGE, file_id='foobar3', page_id='phys_0002')
             ws.add_file('GRP2', mimetype=MIMETYPE_PAGE, file_id='foobar4', page_id='phys_0002')
-            for page_id in [None, 'phys_0001,phys_0002']:
+            ws.add_file('GRP3', mimetype=MIMETYPE_PAGE, file_id='foobar5', page_id='phys_0003')
+            for page_id in [None, 'phys_0001,phys_0002', '~phys_0003']:
                 with self.subTest(page_id=page_id):
                     proc = ZipTestProcessor(None)
                     proc.workspace = ws
@@ -453,7 +454,8 @@ class TestProcessor(TestCase):
             ws = self.resolver.workspace_from_nothing(directory=tempdir)
             ws.add_file('GRP1', mimetype=MIMETYPE_PAGE, file_id='foobar1', page_id=None)
             ws.add_file('GRP2', mimetype=MIMETYPE_PAGE, file_id='foobar2', page_id='phys_0001')
-            for page_id in [None, 'phys_0001']:
+            ws.add_file('GRP3', mimetype=MIMETYPE_PAGE, file_id='foobar3', page_id='phys_0002')
+            for page_id in [None, 'phys_0001', '~phys_0002']:
                 with self.subTest(page_id=page_id):
                     proc = ZipTestProcessor(None)
                     proc.workspace = ws