feat: add circle archiving in the BagIt (RFC 8493) format

Author: paulmueller

CHANGELOG

@@ -1,4 +1,5 @@
-1.0.1
+1.1.0
+ - feat: add circle archiving in the BagIt (RFC 8493) format
  - fix: handle resource downloads whose target directory got deleted
  - fix: check for existence of persistent job list in upload
 1.0.0

dcoraid/bagit/__init__.py

@@ -0,0 +1 @@
+from .archive import bag_circle  # noqa: F401

dcoraid/bagit/archive.py

@@ -0,0 +1,198 @@
+import hashlib
+import json
+import pathlib
+import shutil
+import threading
+from typing import Callable
+
+from ..api import CKANAPI
+from ..dbmodel import APIInterrogator
+from ..download import DownloadJob
+from . import info, manifest
+
+
+def bag_circle(api: CKANAPI,
+               circle_name: str,
+               target_path: pathlib.Path,
+               abort_event: threading.Event = None,
+               callback: Callable = None):
+    """Download an entire circle to a target directory in BagIt format
+
+    The format follows RFC 8493 "The BagIt File Packaging Format (V1.0)".
+
+    To validate the BagIt bags:
+
+        pip install bagit
+        bagit.py --validate --quiet target_path/*
+
+    Parameters
+    ----------
+    api
+        CKANAPI for connecting to the DCOR instance
+    circle_name
+        Name of the circle to archive
+    target_path
+        Download location
+    abort_event
+        Specify a `threading.Event` to be able to abort archiving;
+        when you wish to abort `.set()` the event.
+    callback
+        Method for progress tracking (returns a float between 0 and 1)
+    """
+    # fetch total list of active datasets
+    ai = APIInterrogator(api)
+    dataset_dicts = ai.search_dataset_via_api(circles=[circle_name],
+                                              limit=0,
+                                              ret_db_extract=False)
+
+    num_datasets = len(dataset_dicts)
+
+    # sort datasets according to creation date
+    dataset_dicts = sorted(dataset_dicts, key=lambda x: x["metadata_created"])
+
+    # compute sha256 hash of all dataset IDs
+    hasher = hashlib.sha256()
+    for ds_dict in dataset_dicts:
+        hasher.update(ds_dict["id"].encode(encoding="utf-8"))
+    sha256_hash = hasher.hexdigest()
+
+    # Check whether there is already a list of dataset IDs in the directory,
+    # and if yes, compute the MD5 hash and compare it. If the comparison
+    # fails, then the user has to choose a different `target_path`, because
+    # we cannot guarantee data integrity.
+    circle_jsonlines_path = target_path / "circle.jsonlines"
+    if circle_jsonlines_path.exists():
+        lines = circle_jsonlines_path.read_text().split("\n")
+        hasher2 = hashlib.sha256()
+        for line in lines:
+            hasher2.update(json.loads(line)["id"].encode(encoding="utf-8"))
+        sha256_hash2 = hasher2.hexdigest()
+        if sha256_hash != sha256_hash2:
+            raise ValueError(
+                f"A previous attempt to archive circle {circle_name} was made "
+                f"in directory {target_path}. However, the number of datasets "
+                f"changed since then. Therefore, it is not possible to "
+                f"archive this circle to that directory.")
+    else:
+        # save list of datasets as jsonlines
+        with circle_jsonlines_path.open("w", encoding="utf-8") as f:
+            for ds_dict in dataset_dicts:
+                f.write(json.dumps(ds_dict) + "\n")
+
+    # number of digits for enumeration
+    max_digits = len(str(num_datasets))
+
+    # bag all dataset
+    for ii, ds_dict in enumerate(dataset_dicts):
+        if callback:
+            callback(ii / num_datasets)
+
+        dataset_index = ii+1
+        prefix = str(dataset_index).zfill(max_digits)
+        bag_path = target_path / f"{prefix}_{ds_dict['name']}"
+
+        if not manifest.is_bagged(bag_path):
+            bag_dataset(api=api,
+                        ds_dict=ds_dict,
+                        dataset_index=dataset_index,
+                        num_datasets=num_datasets,
+                        bag_path=bag_path,
+                        abort_event=abort_event)
+        if abort_event is not None and abort_event.is_set():
+            return
+    if callback:
+        callback(1)
+
+
+def bag_dataset(api: CKANAPI,
+                ds_dict: dict,
+                bag_path: pathlib.Path,
+                abort_event: threading.Event = None,
+                dataset_index: int = 1,
+                num_datasets: int = 1,
+                ):
+    """Download a dataset to a target directory in BagIt format
+
+    Parameters
+    ----------
+    api
+        CKANAPI for connecting to the DCOR instance
+    ds_dict
+        CKAN dataset dictionary
+    bag_path
+        Path of the bag
+    abort_event
+        Event for aborting (see :func:`bag_circle`)
+    dataset_index
+        Index of this dataset in the circle
+    num_datasets
+        Total number of datasets in the circle
+    """
+    # clear/create download directory
+    if bag_path.exists():
+        shutil.rmtree(bag_path)
+    bag_path.mkdir(parents=True, exist_ok=True)
+    data_path = bag_path / "data"
+    data_path.mkdir(parents=True, exist_ok=True)
+
+    # dataset dictionary
+    meta = json.dumps(ds_dict, indent=2, sort_keys=True)
+    (data_path / "dataset.json").write_text(meta)
+
+    # download all resources
+    for res_dict in ds_dict["resources"]:
+        if abort_event is not None and abort_event.is_set():
+            return
+
+        # resource
+        download_resource(api=api,
+                          bag_path=bag_path,
+                          res_dict=res_dict,
+                          condensed=False)
+
+        # condensed resource
+        if res_dict["name"].endswith(".rtdc"):
+            download_resource(api=api,
+                              bag_path=bag_path,
+                              res_dict=res_dict,
+                              condensed=True)
+
+    # create BagIt files
+    info.write_bag_info(bag_path=bag_path,
+                        bag_index=dataset_index,
+                        num_bags=num_datasets,
+                        ds_dict=ds_dict)
+
+    # create BagIt manifest files
+    manifest.write_manifest(bag_path=bag_path,
+                            ds_dict=ds_dict)
+
+
+def download_resource(api: CKANAPI,
+                      bag_path: pathlib.Path,
+                      res_dict: dict,
+                      condensed: bool):
+    """Download and verify a resource from DCOR
+
+    Parameters
+    api
+        CKANAPI for connecting to the DCOR instance
+    bag_path
+        Path of the bag
+    res_dict
+        CKAN resource dictionary
+    condensed
+        Whether to download the condensed resource (or the original resource)
+    """
+    data_path = bag_path / "data"
+    data_path.mkdir(parents=True, exist_ok=True)
+    dl_path = data_path / res_dict["name"]
+    if condensed:
+        dl_path = dl_path.with_name(dl_path.stem + "_condensed.rtdc")
+    dj = DownloadJob(api=api,
+                     resource_id=res_dict["id"],
+                     download_path=dl_path,
+                     condensed=condensed
+                     )
+    dj.task_download_resource()
+    dj.task_verify_resource()

dcoraid/bagit/info.py

@@ -0,0 +1,60 @@
+import pathlib
+
+import time
+
+
+from .._version import version
+
+
+def get_payload_oxum(bag_path: pathlib.Path) -> str:
+    """Return the Payload-Oxum
+
+    The "octetstream sum" of the payload, which is
+    intended for the purpose of quickly detecting incomplete bags
+    before performing checksum validation.  This is strictly an
+    optimization, and implementations MUST perform the standard
+    checksum validation process before proclaiming a bag to be valid.
+    This element MUST NOT be present more than once and, if present,
+    MUST be in the form "_OctetCount_._StreamCount_", where
+    _OctetCount_ is the total number of octets (8-bit bytes) across
+    all payload file content and _StreamCount_ is the total number of
+    payload files.  This metadata element MUST NOT be repeated.
+    """
+    num_files = 0
+    size_files = 0
+    for pp in (bag_path / "data").rglob("*"):
+        if pp.is_file():
+            num_files += 1
+            size_files += pp.stat().st_size
+    return f"{size_files}.{num_files}"
+
+
+def write_bag_info(bag_path: pathlib.Path,
+                   ds_dict: dict,
+                   bag_index: int = 1,
+                   num_bags: int = 1,
+                   ):
+    """Write the bag-info.txt file for a bag
+
+    The content of the info file looks like this:
+
+        Bag-Software-Agent: dcoraid 1.1 <https://github.com/DCOR-dev/DCOR-Aid>
+        Bagging-Date: 2026-02-12
+        Payload-Oxum: 2307.10
+        Bag-Group-Identifier: Name of the Circle
+        Bag-Count: 23 of 8472
+        External-Identifier: DATASET IDENTIFIER
+        Internal-Sender-Identifier: DATASET TITLE
+    """
+    webloc = "https://github.com/DCOR-dev/DCOR-Aid"
+    lines = [
+        f"Bag-Software-Agent: dcoraid {version} <{webloc}>",
+        time.strftime("Bagging-Date: %Y-%m-%d"),
+        f"Payload-Oxum: {get_payload_oxum(bag_path)}",
+        f"Bag-Group-Identifier: {ds_dict['organization']['name']}",
+        f"Bag-Count: {bag_index} of {num_bags}",
+        f"External-Identifier: {ds_dict["name"]}",
+        f"Internal-Sender-Identifier: {ds_dict["title"]}",
+    ]
+    (bag_path / "bag-info.txt").write_text("\n".join(lines),
+                                           encoding="utf-8")

dcoraid/bagit/manifest.py

@@ -0,0 +1,109 @@
+import hashlib
+import warnings
+
+
+class ManifestValidationError(BaseException):
+    pass
+
+
+def is_bagged(bag_path):
+    """Check whether a dataset has been bagged
+
+    This checks for the existence of the tagmanifest file and
+    validates it.
+    """
+    for name in ["bagit.txt", "bag-info.txt", "manifest-sha256.txt"]:
+        if not (bag_path / name).exists():
+            return False
+
+    tagmanifest_path = bag_path / "tagmanifest-sha256.txt"
+    if not tagmanifest_path.exists():
+        return False
+    else:
+        try:
+            validate_tag_manifest(bag_path)
+        except ManifestValidationError:
+            return False
+        else:
+            return True
+
+
+def hash_file(path):
+    hasher = hashlib.sha256()
+    with path.open("rb") as fd:
+        while chunk := fd.read(1024 ** 2):
+            hasher.update(chunk)
+    return hasher.hexdigest()
+
+
+def write_manifest(bag_path, ds_dict):
+    """Write a manifest-sha256.txt to the bag directory
+
+    Manifest files look like this:
+
+        e91f941be5973ff71f1dccb743da38b1689  data/file_1.rtdc
+        ee792190d28d8a7abdb0ec805a2618e4573  data/file_2.txt
+    """
+    manifest = []
+    for res_dict in ds_dict["resources"]:
+        manifest.append(f"{res_dict['sha256']}  data/{res_dict['name']}")
+        if res_dict.get("mimetype") == "RT-DC":
+            pp_res = bag_path / "data" / res_dict["name"]
+            pp_cond = pp_res.with_name(f"{pp_res.stem}_condensed.rtdc")
+            # compute that hash for the condensed file
+            if pp_cond.exists():
+                manifest.append(f"{hash_file(pp_cond)}  data/{pp_cond.name}")
+            else:
+                warnings.warn(f"No condensed file found for {pp_res}")
+
+    # add dataset.json
+    json_hash = hash_file(bag_path / "data" / "dataset.json")
+    manifest.append(f"{json_hash}  data/dataset.json")
+
+    (bag_path / "manifest-sha256.txt").write_text("\n".join(manifest))
+
+    # write bag declaration
+    (bag_path / "bagit.txt").write_text(
+        "\n".join(["BagIt-Version: 0.97",
+                   "Tag-File-Character-Encoding: UTF-8"]),
+        encoding="utf-8")
+
+    write_tag_manifest(bag_path)
+
+
+def write_tag_manifest(bag_path):
+    """Write a tagmanifest-sha256.txt file
+
+    This is just a manifest file for the other txt metadata files
+    """
+    tag_manifest = []
+    for pp in bag_path.glob("*.txt"):
+        pp_hash = hashlib.sha256(pp.read_bytes()).hexdigest()
+        tag_manifest.append(f"{pp_hash}  {pp.name}")
+    (bag_path / "tagmanifest-sha256.txt").write_text("\n".join(tag_manifest),
+                                                     encoding="utf-8")
+
+
+def validate_tag_manifest(bag_path):
+    """Validate the tagmanifest-sha256.txt file
+
+    Raises
+    ------
+    ManifestValidationError
+        When a hash is missing or a hash does not match
+    """
+    # read hashes
+    hash_dict = {}
+    lines = (bag_path / "tagmanifest-sha256.txt").read_text().split("\n")
+    for line in lines:
+        pp_hash, pp_name = line.split("  ", 1)
+        hash_dict[pp_name] = pp_hash
+
+    for pp in bag_path.glob("*.txt"):
+        if pp.name == "tagmanifest-sha256.txt":
+            # tagmanifest does not have a hash
+            continue
+        if pp.name not in hash_dict:
+            raise ManifestValidationError(f"No hash associated with {pp}")
+        if hash_dict[pp.name] != hash_file(pp):
+            raise ManifestValidationError(f"Hash mismatch for {pp}")