Flexible manifest input

compass/pipeline/collection/persistence.py

@@ -1,8 +1,11 @@
 """Persistence for collected documents"""
 
+import os
 import json
 import asyncio
 from pathlib import Path
+from glob import glob
+from itertools import chain
 from statistics import median
 from collections import Counter
 from warnings import warn
@@ -123,25 +126,61 @@ async def write_collection_manifest_shard(shard_dir, collection_info):
     )
 
 
-async def load_collection_manifest(manifest_fp, expected_tech):
-    """Load a collection manifest from disk
+async def load_collection_manifest_jurisdictions(manifest_fp, expected_tech):
+    """Load jurisdictions from one or more collection manifest(s)
 
     Parameters
     ----------
-    manifest_fp : path-like
-        Path to the collection manifest file to be loaded.
+    manifest_fp : path-like or list of path-like
+        Path to the collection manifest file to be loaded. Can be a
+        single path or a list of paths, any of which may include glob
+        patterns.
     expected_tech : str
         Technology specified in the pipeline request, used to validate
         compatibility with the manifest.
 
     Returns
     -------
     dict
-        Loaded collection manifest as a dictionary.
+        Mapping of FIPS codes to jurisdiction infos from the collection
+        manifest(s).
+
+    Raises
+    ------
+    COMPASSValueError
+        If a duplicate jurisdiction is found in the manifest(s).
     """
-    return await GenericFuncRunner.call(
-        _load_collection_manifest, manifest_fp, expected_tech
-    )
+    if isinstance(manifest_fp, (str, os.PathLike)):
+        manifest_fp = [str(manifest_fp)]
+
+    task_fps = []
+    for maybe_glob in manifest_fp:
+        # ruff: ignore[glob]
+        new_fps = [
+            Path(match) for match in glob(str(maybe_glob), recursive=True)
+        ]
+        task_fps.extend(new_fps or [maybe_glob])
+
+    tasks = [
+        GenericFuncRunner.call(_load_collection_manifest, fp, expected_tech)
+        for fp in task_fps
+    ]
+    manifests = await asyncio.gather(*tasks)
+
+    jurisdictions_by_fips = {}
+    for jurisdiction in chain.from_iterable(
+        manifest.get("jurisdictions", []) for manifest in manifests
+    ):
+        if jurisdiction is None:
+            continue
+
+        fips = jurisdiction.get("FIPS")
+        if fips in jurisdictions_by_fips:
+            msg = f"Duplicate collection manifest entry for FIPS '{fips}'"
+            raise COMPASSValueError(msg)
+        jurisdictions_by_fips[fips] = jurisdiction
+
+    return jurisdictions_by_fips
 
 
 async def load_specific_collection_manifest_shard(shard_dir, jurisdiction):
@@ -191,7 +230,9 @@ def _write_collection_manifest_shard(shard_dir, collection_info):
 def _load_collection_manifest(manifest_fp, expected_tech):
     """Load a collection manifest from disk"""
     try:
-        manifest = load_config(manifest_fp, file_name="Collection manifest")
+        manifest = load_config(
+            manifest_fp, resolve_paths=True, file_name="Collection manifest"
+        )
     except COMPASSFileNotFoundError:
         manifest = _load_collection_manifest_from_shards(
             manifest_fp, expected_tech
@@ -223,6 +264,8 @@ def _load_specific_collection_manifest_shard(shard_dir, jurisdiction):
 
     return load_config(
         shard_fp,
+        # paths are NOT relative to the shard directory, so should not
+        # be resolved here
         resolve_paths=False,
         file_name="Collection manifest shard",
     )
@@ -421,6 +464,9 @@ def _load_collection_manifest_from_shards(manifest_fp, expected_tech):
     for shard_fp in shard_fps:
         collection_info = load_config(
             shard_fp,
+            # paths are NOT relative to the shard directory, so should
+            # not be resolved here; they are resolved using the
+            # `resolve_all_paths` function call below
             resolve_paths=False,
             file_name="Collection manifest shard",
         )

compass/pipeline/coordinator.py

@@ -25,7 +25,7 @@
 from compass.pipeline.collection.persistence import (
     build_collection_manifest,
     write_collection_manifest,
-    load_collection_manifest,
+    load_collection_manifest_jurisdictions,
 )
 from compass.pipeline import BaseRequest
 from compass.pipeline.runtime import PipelineRuntime
@@ -300,10 +300,12 @@ async def run(self, jurisdictions_df):
         logger.debug(
             "Manifest path(s): %s", self.runtime.request.collection_manifest_fp
         )
-        manifest = await load_collection_manifest(
-            self.runtime.request.collection_manifest_fp, self.runtime.tech
+        collection_infos_by_fips = (
+            await load_collection_manifest_jurisdictions(
+                self.runtime.request.collection_manifest_fp, self.runtime.tech
+            )
         )
-        jurisdictions = manifest.get("jurisdictions", [])
+
         logger.info(
             "Extracting structured data for %d jurisdiction(s)",
             len(jurisdictions_df),
@@ -312,12 +314,8 @@ async def run(self, jurisdictions_df):
         tasks = []
         start_date = datetime.now(UTC)
         for jurisdiction in jurisdictions_from_df(jurisdictions_df):
-            collection_info = [
-                info
-                for info in jurisdictions
-                if info is not None and info.get("FIPS") == jurisdiction.code
-            ]
-            if not collection_info:
+            collection_info = collection_infos_by_fips.get(jurisdiction.code)
+            if collection_info is None:
                 logger.warning(
                     "No collection info found for %s; skipping extraction",
                     jurisdiction.full_name,
@@ -330,7 +328,7 @@ async def run(self, jurisdictions_df):
             workflow = self._create(jurisdiction, usage_tracker=usage_tracker)
             tasks.append(
                 asyncio.create_task(
-                    workflow.run_extraction_with_logging(collection_info[0]),
+                    workflow.run_extraction_with_logging(collection_info),
                     name=jurisdiction.full_name,
                 )
             )

compass/pipeline/data_classes.py

@@ -624,14 +624,16 @@ def __init__(  # ruff:ignore[too-many-arguments]
             terminal. If ``True``, all of the unordered records are
             written to a "all.log" file in the `log_dir` directory.
             By default, ``False``.
-        collection_manifest_fp : path-like, optional
+        collection_manifest_fp : path-like or list of path-like, optional
             Path to the JSON collection manifest created by the document
-            collection step. The manifest must contain the persisted
-            document information needed to reload each collected
-            document for extraction. Only needed if running in
+            collection step. This can be a single path or a list of
+            paths for multiple collection manifests, any of which may
+            include glob patterns. Each collection manifest must contain
+            the persisted document information needed to reload each
+            collected document for extraction. Only needed if running in
             extraction mode with a separate collection step.
             By default, ``None``.
-        """
+        """  # ruff:ignore[doc-line-too-long]
         self.tech = tech
         self.jurisdiction_fp = jurisdiction_fp
         self.perform_se_search = perform_se_search
@@ -1092,11 +1094,13 @@ def __init__(  # ruff:ignore[too-many-arguments]
             name of the subdivision, and the "Jurisdiction Type" should
             be a string identifying the type of subdivision (e.g.,
             "City", "Township", etc.)
-        collection_manifest_fp : path-like
+        collection_manifest_fp : path-like or list of path-like, optional
             Path to the JSON collection manifest created by the document
-            collection step. The manifest must contain the persisted
-            document information needed to reload each collected
-            document for extraction.
+            collection step. This can be a single path or a list of
+            paths for multiple collection manifests, any of which may
+            include glob patterns. Each collection manifest must contain
+            the persisted document information needed to reload each
+            collected document for extraction. By default, ``None``.
         model : str or list of dict, default="gpt-4o-mini"
             LLM model(s) to use for scraping and parsing ordinance
             documents. If a string is provided, it is assumed to be the
@@ -1228,7 +1232,7 @@ def __init__(  # ruff:ignore[too-many-arguments]
             terminal. If ``True``, all of the unordered records are
             written to a "all.log" file in the `log_dir` directory.
             By default, ``False``.
-        """
+        """  # ruff:ignore[doc-line-too-long]
 
         super().__init__(
             out_dir=out_dir,