Verifying an archive

⟨check the document versions of a document⟩


⟨check an archive and list its problems⟩


⟨the verify command⟩

def verify(args):
    errors = verify_errors()
    if getattr(args, "remote", False):
        from . import collections
        for obj_id, d, inv in objects():
            if read_json(d, inv, OBJ)["kind"] == "collection" and not mirror_of(d, inv):
                errors += [f"{obj_id}: {e}" for e in
                           collections.check_remote(read_json(d, inv, "members.json")["members"])]
        errors += collections.check_accepted_copies()
    for e in errors:
        print(e)
    print("OK" if not errors else f"{len(errors)} problem(s)")
    sys.exit(1 if errors else 0)

def verify_errors():
    errors = []
    ⟨check against the JSON Schemas of the specification, when jsonschema is installed⟩

    ⟨check the discovery document of an archive against its storage root⟩

    for obj_id, (d, inv) in found.items():
        ⟨check the inventory and the stored files of an object⟩
        ⟨check every version of an object⟩
        errors += document_version_errors(obj_id, d, inv)

    ⟨check the storage root with an independent OCFL validator⟩

    ⟨check that the index shows the pins of the main collection⟩
    return errors

try:
    import jsonschema
except ImportError:
    jsonschema = None

def check_schema(where, data, name):
    if jsonschema is None:
        return
    schema = json.loads((SCHEMAS / name).read_text())
    v = jsonschema.Draft202012Validator(schema, format_checker=jsonschema.FormatChecker())
    for e in v.iter_errors(data):
        errors.append(f"{where}: {e.message}")

disc_path = ROOT / "opensheaf.json"
if not disc_path.exists():
    return ["opensheaf.json is missing"]
disc = json.loads(disc_path.read_text())
check_schema("opensheaf.json", disc, "opensheaf-archive-0.1.schema.json")
if disc.get("storageLayout") != LAYOUT or not (OCFL / "0=ocfl_1.1").exists():
    errors.append("storage root missing or with another layout")
listed = {o["id"]: o for o in disc.get("objects", [])}
found = {obj_id: (d, inv) for obj_id, d, inv in objects()}
for obj_id in set(listed) ^ set(found):
    errors.append(f"opensheaf.json and the storage root disagree about {obj_id}")
for obj_id, (d, inv) in found.items():
    if obj_id in listed and listed[obj_id].get("mirrorOf") != mirror_of(d, inv):
        errors.append(f"opensheaf.json: {obj_id} " + (f"is a mirror of {mirror_of(d, inv)}, without mirrorOf"
                      if mirror_of(d, inv) else "is listed as a mirror, and is not one"))

where = f"ocfl/{obj_id}"
if inv.get("id") != obj_id or not ID_RE.match(obj_id) or len(obj_id) > 64:
    errors.append(f"{where}: bad identifier")
if inv.get("digestAlgorithm") != "sha512":
    errors.append(f"{where}: digest algorithm is not sha512")
side = (d / "inventory.json.sha512").read_text().split()[0]
if side != digest(d / "inventory.json"):
    errors.append(f"{where}: inventory checksum file does not match")
lst = listed.get(obj_id)
if lst and (lst["head"] != inv["head"] or lst["inventorySha512"] != side):
    errors.append(f"opensheaf.json: stale entry for {obj_id}")
for dig, paths in inv["manifest"].items():
    for cp in paths:
        if not (d / cp).exists():
            errors.append(f"{where}: missing {cp}")
        elif digest(d / cp) != dig:
            errors.append(f"{where}: digest mismatch for {cp}")

for ver, vinfo in inv["versions"].items():
    ⟨check the inventory copy, the user and the logical paths of an object version⟩
    ⟨check the two descriptions of an object version⟩
    ⟨check the files of a document version or the members of a collection version⟩

if not (d / ver / "inventory.json").exists() or not (d / ver / "inventory.json.sha512").exists():
    errors.append(f"{where}: {ver} has no inventory copy")
if not vinfo.get("user", {}).get("name") or not vinfo.get("message"):
    errors.append(f"{where}: {ver} lacks a user name or a message")
logical = {p for ps in vinfo["state"].values() for p in ps}
for p in logical:
    parts = p.split("/")
    if p.startswith("/") or ".." in parts or "." in parts:
        errors.append(f"{where}: bad logical path {p}")
missing = [n for n in ("datacite.json", OBJ) if n not in logical]
for n in missing:
    errors.append(f"{where}: {ver} has no {n}")
if missing:
    continue

obj = read_json(d, inv, OBJ, ver)
dc = read_json(d, inv, "datacite.json", ver)
check_schema(f"{where}/{ver}/{OBJ}", obj, "opensheaf-object-0.1.schema.json")
if obj.get("objectId") != obj_id:
    errors.append(f"{where}: {ver} object.json names another object")
elif obj.get("archiveId") != (mirror_of(d, inv) or disc.get("archiveId")):
    errors.append(f"{where}: {ver} object.json names another archive")
if str(obj.get("documentVersion")) != dc.get("version"):
    errors.append(f"{where}: {ver} datacite.json and object.json disagree on the version")
if obj.get("status") == "withdrawn" and date_of(dc, "Withdrawn") != obj.get("withdrawal", {}).get("date"):
    errors.append(f"{where}: {ver} withdrawal dates disagree")
for k in ("titles", "creators", "publicationYear", "types", "version", "publisher"):
    if k not in dc:
        errors.append(f"{where}: {ver} datacite.json lacks {k}")
for r in dc.get("rightsList", []):
    if r.get("rightsIdentifier") is None and r.get("rights", "").lower().startswith(("creative", "arxiv")) \
            and not r.get("rightsUri"):
        errors.append(f"{where}: {ver} licence without SPDX identifier and without rightsUri")
for note in [obj.get("changes"), obj.get("sourcesMissing"), obj.get("licence")] + \
        obj.get("notices", []) + [obj.get("withdrawal", {}).get("reason")]:
    if note and note.get("path") and note["path"] not in logical:
        errors.append(f"{where}: {ver} note file {note['path']} not in the version")

if obj.get("kind") == "document":
    if obj.get("mainSource") and obj["mainSource"] not in logical:
        errors.append(f"{where}: {ver} main source not in the version")
    if obj.get("sourcesMissing") and obj.get("mainSource"):
        errors.append(f"{where}: {ver} sourcesMissing together with a main source")
    for r in obj.get("renditions", []):
        if r["path"] not in logical:
            errors.append(f"{where}: {ver} rendition {r['path']} not in the version")
elif "members.json" not in logical:
    errors.append(f"{where}: {ver} collection without members.json")
else:
    mem = read_json(d, inv, "members.json", ver)
    check_schema(f"{where}/{ver}/members.json", mem, "opensheaf-collection-0.1.schema.json")
    ⟨check the type of a member list and the keys of its members⟩
    ⟨check the pins of the members of this archive⟩

kind = mem.get("type")
if kind not in ("archive", "selection"):
    errors.append(f"{where}: {ver} members.json has no valid type (archive or selection)")
for m in mem.get("members", []):
    who = f"{where}: {ver} member {m.get('id')}"
    if kind == "archive" and (not m.get("version") or not m.get("inventorySha512") or "revisions" in m):
        errors.append(f"{who}: a member of an archive collection needs version and inventorySha512, "
                      "and no revisions")
    if kind == "selection":
        revs = m.get("revisions")
        if any(k in m for k in ("version", "inventorySha512", "follow")):
            errors.append(f"{who}: a member of a selection has no version, inventorySha512 or follow")
        if not isinstance(revs, list) or not revs or any(
                not all(r.get(k) for k in ("documentVersion", "version", "inventorySha512")) for r in revs):
            errors.append(f"{who}: a member of a selection needs revisions, each with documentVersion, "
                          "version and inventorySha512")
        elif len({r["documentVersion"] for r in revs}) != len(revs):
            errors.append(f"{who}: a document version is vetted twice")

for m in mem.get("members", []):
    if m.get("archive") != disc.get("archiveId") or mirror_of(d, inv):
        continue  # members of other archives: resolved as in section 6.4
    for pv, pin, n in member_pins(m):
        md = OCFL / str(m.get("id")) / str(pv) / "inventory.json"
        if not md.exists():
            errors.append(f"{where}: {ver} member {m['id']} {pv}: unavailable")
        elif digest(md) != pin:
            errors.append(f"{where}: {ver} member {m['id']} {pv}: mismatch")
        elif n is not None and doc_version_of(OCFL / m["id"], load_inventory(OCFL / m["id"]), pv) != n:
            errors.append(f"{where}: {ver} member {m['id']} {pv} does not hold document version {n}")
    if kind == "selection" and (OCFL / str(m.get("id"))).exists():
        md = OCFL / m["id"]
        order = [doc_version_of(md, load_inventory(md), h) for h in history_of(md, load_inventory(md))]
        vetted = [r.get("documentVersion") for r in m.get("revisions") or []]
        pos = [order.index(x) for x in vetted if x in order]
        if pos != sorted(pos):
            errors.append(f"{where}: {ver} member {m['id']}: revisions not in the order of the history")

def document_version_errors(obj_id, d, inv):
    """Section 7.3: numbering without gaps, the same files for the same
    document version (a version with sourcesMissing may be completed by adding
    files), the history, and notices kept unchanged."""
    errors, where = [], f"ocfl/{obj_id}"
    seen, files_of, missing_of, highest = {}, {}, {}, 0
    had_history, notices = False, []
    for ver, vinfo in inv["versions"].items():
        ⟨the files of an object version that belong to its document version⟩
        ⟨check the number and the files of a document version⟩
        ⟨check the history recorded in an object version⟩
        ⟨check that an object version keeps the earlier notices⟩
    return errors

obj = read_json(d, inv, OBJ, ver)
if obj is None or obj.get("kind") != "document":
    return errors
n = obj["documentVersion"]
notes = {x["path"] for x in obj.get("notices", []) if x.get("path")}
if obj.get("withdrawal", {}).get("reason", {}).get("path"):
    notes.add(obj["withdrawal"]["reason"]["path"])
own = {p: dig for dig, ps in vinfo["state"].items() for p in ps
       if p not in ("datacite.json", OBJ) and p not in notes}
change = (obj.get("changes") or {}).get("text")

if n in seen:
    before, mis = files_of[n], missing_of[n]
    if mis:
        if any(before.get(p) not in (None, dig) for p, dig in own.items()) or not set(before) <= set(own):
            errors.append(f"{where}: {ver} changes files of document version {n} (only additions allowed)")
    elif own != before or change != seen[n]:
        errors.append(f"{where}: {ver} has other files than document version {n}")
    missing_of[n] = bool(obj.get("sourcesMissing"))
    files_of[n] = own
else:
    if n != highest + 1:
        errors.append(f"{where}: {ver} document version {n} breaks the numbering (expected {highest + 1})")
    highest = max(highest, n)
    seen[n], files_of[n], missing_of[n] = change, own, bool(obj.get("sourcesMissing"))

hist = obj.get("history")
if had_history and hist is None:
    errors.append(f"{where}: {ver} lacks the history of earlier versions")
if hist is not None:
    had_history = True
    known = [v for v in inv["versions"] if int(v[1:]) <= int(ver[1:])]
    if any(h not in known for h in hist):
        errors.append(f"{where}: {ver} history names unknown object versions")
    else:
        docs = [read_json(d, inv, OBJ, h)["documentVersion"] for h in hist]
        if len(set(docs)) != len(docs) or set(docs) != set(range(1, highest + 1)):
            errors.append(f"{where}: {ver} history does not list each document version once")
        for h, k in zip(hist, docs):
            later = [v for v in known if int(v[1:]) > int(h[1:])
                     and read_json(d, inv, OBJ, v)["documentVersion"] == k]
            if later:
                errors.append(f"{where}: {ver} history entry {h} is not the last object version of {k}")

cur = obj.get("notices", [])
if any(x not in cur for x in notices):
    errors.append(f"{where}: {ver} drops or changes an earlier notice")
notices = cur

# an independent OCFL validator, when installed
try:
    import ocfl
    from ocfl.validator import Validator
except ImportError:
    ocfl = None
if ocfl is not None:
    for obj_id, (d, inv) in found.items():
        v = Validator(log_warnings=False, log_errors=True)
        if not v.validate_object(str(d)):
            errors.append(f"ocfl/{obj_id}: OCFL validator: {v.status_str()}")
    if not ocfl.StorageRoot(root=str(OCFL)).validate(log_warnings=False):
        errors.append("ocfl: the OCFL validator rejects the storage root")

# the presentation shows the current pins
coll = found.get(disc.get("mainCollection"))
if coll and (ROOT / "index.html").exists():
    html = (ROOT / "index.html").read_text(encoding="utf-8", errors="replace")
    mem = json.loads(read_logical(coll[0], coll[1], "members.json").read_text())
    for m in mem["members"]:
        if (member_pins(m)[-1][1] or "")[:32] not in html:
            errors.append(f"index.html does not show the pin of {m['id']}")