tegwick
8bcb3414d7
All checks were successful
synthetic / suite (push) Successful in 1m2s
Assistant: grok Assistant-Session: 01a1531b-6698-7732-8587-f28ea604bcdf
228 lines
9.6 KiB
Python
228 lines
9.6 KiB
Python
"""Quality counts for a published snapshot and an ingest report.
|
|
|
|
The counts come from the documents passed in. A failure class is present even
|
|
when its count is zero, so a missing license, a dangling reference, or a
|
|
duplicate identity cannot disappear by omission.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import date
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from content_explorer.ingest import ALLOWED_LICENSES
|
|
|
|
FAILURE_CLASSES = (
|
|
"rights",
|
|
"missing-license",
|
|
"dangling-reference",
|
|
"duplicate-identity",
|
|
"unresolved-reference",
|
|
"missing-relation-label",
|
|
"unknown-kind",
|
|
"stale",
|
|
)
|
|
|
|
|
|
def _failure(kind: str, item_id: str, detail: str) -> dict[str, str]:
|
|
return {"class": kind, "id": item_id, "detail": detail}
|
|
|
|
|
|
def _days_after(earlier: str, later: str) -> int | None:
|
|
try:
|
|
start = date.fromisoformat(earlier)
|
|
end = date.fromisoformat(later)
|
|
except ValueError:
|
|
return None
|
|
return (end - start).days
|
|
|
|
|
|
def _licensed_source(source: dict) -> bool:
|
|
license_name = str(source.get("license") or "")
|
|
return bool(source.get("id") and source.get("uri") and source.get("title") and license_name in ALLOWED_LICENSES)
|
|
|
|
|
|
def _from_ingest(report: dict) -> list[dict[str, str]]:
|
|
failures: list[dict[str, str]] = []
|
|
for item in report.get("quarantined") or []:
|
|
reason = item.get("reason")
|
|
if reason == "rights":
|
|
license_name = str(item.get("license") or "")
|
|
if license_name in ("", "(missing)"):
|
|
failures.append(_failure("missing-license", str(item.get("id") or ""), license_name or "(missing)"))
|
|
else:
|
|
failures.append(_failure("rights", str(item.get("id") or ""), item.get("detail") or license_name))
|
|
elif reason == "missing-relation-label":
|
|
failures.append(_failure("missing-relation-label", "", ",".join(item.get("predicates") or [])))
|
|
elif reason == "unknown-kind":
|
|
failures.append(_failure("unknown-kind", str(item.get("key") or ""), str(item.get("kind") or "")))
|
|
elif reason:
|
|
failures.append(_failure(str(reason), str(item.get("key") or item.get("id") or ""), str(item.get("detail") or "")))
|
|
for item in report.get("dangling") or []:
|
|
failures.append(_failure(
|
|
"dangling-reference",
|
|
str(item.get("from") or ""),
|
|
str(item.get("to") or ""),
|
|
))
|
|
for item in report.get("conflicts") or []:
|
|
reason = item.get("reason") or "conflict"
|
|
kind = "duplicate-identity" if reason == "duplicate-identity" else str(reason)
|
|
failures.append(_failure(kind, str(item.get("key") or ""), str(item.get("count") or reason)))
|
|
return failures
|
|
|
|
|
|
def quality_report(
|
|
snapshot: dict | None = None,
|
|
manifest: dict | None = None,
|
|
ingest_report: dict | None = None,
|
|
*,
|
|
today: str | None = None,
|
|
stale_days: int = 365,
|
|
) -> dict[str, Any]:
|
|
"""Summarise provenance, staleness, references, licenses, and failure classes."""
|
|
failures = _from_ingest(ingest_report) if ingest_report else []
|
|
unresolved: list[dict[str, str]] = []
|
|
stale_ids: list[str] = []
|
|
incomplete: list[str] = []
|
|
missing_license: list[str] = []
|
|
licenses: list[str] = []
|
|
records = 0
|
|
complete = 0
|
|
source_count = 0
|
|
sources_complete = 0
|
|
dataspace_id = ""
|
|
|
|
if snapshot:
|
|
dataspace_id = (snapshot.get("dataspace") or {}).get("id") or ""
|
|
entities = list(snapshot.get("entities") or [])
|
|
relations = list(snapshot.get("relations") or [])
|
|
claims = list(snapshot.get("claims") or [])
|
|
sources = list(snapshot.get("sources") or [])
|
|
source_count = len(sources)
|
|
by_source = {}
|
|
for source in sources:
|
|
source_id = str(source.get("id") or "")
|
|
license_name = str(source.get("license") or "")
|
|
if license_name and license_name not in licenses:
|
|
licenses.append(license_name)
|
|
if _licensed_source(source):
|
|
sources_complete += 1
|
|
by_source[source_id] = source
|
|
else:
|
|
missing_license.append(source_id or "(missing)")
|
|
if not license_name:
|
|
failures.append(_failure("missing-license", source_id, "(missing)"))
|
|
else:
|
|
failures.append(_failure("rights", source_id, license_name))
|
|
|
|
for label, items in (("entity", entities), ("claim", claims), ("relation", relations)):
|
|
seen_here: dict[str, str] = {}
|
|
for item in items:
|
|
item_id = str(item.get("id") or "")
|
|
if not item_id:
|
|
continue
|
|
if item_id in seen_here:
|
|
failures.append(_failure("duplicate-identity", item_id, label))
|
|
else:
|
|
seen_here[item_id] = label
|
|
|
|
entity_ids = {str(entity.get("id")) for entity in entities if entity.get("id")}
|
|
claim_ids = {str(claim.get("id")) for claim in claims if claim.get("id")}
|
|
known_source_ids = {str(source.get("id")) for source in sources if source.get("id")}
|
|
source_ids = set(by_source)
|
|
|
|
def provenance_ok(refs: list) -> bool:
|
|
return any(ref in source_ids for ref in refs)
|
|
|
|
for entity in entities:
|
|
entity_id = str(entity.get("id") or "")
|
|
records += 1
|
|
refs = list(entity.get("sourceRefs") or [])
|
|
missing_refs = [ref for ref in refs if ref not in known_source_ids]
|
|
for ref in missing_refs:
|
|
unresolved.append({"id": entity_id, "field": "sourceRefs", "target": str(ref)})
|
|
failures.append(_failure("unresolved-reference", entity_id, str(ref)))
|
|
if provenance_ok(refs) and not missing_refs:
|
|
complete += 1
|
|
else:
|
|
incomplete.append(entity_id)
|
|
if entity.get("state") == "stale":
|
|
stale_ids.append(entity_id)
|
|
failures.append(_failure("stale", entity_id, "state"))
|
|
elif today and entity.get("asOf"):
|
|
age = _days_after(str(entity["asOf"]), today)
|
|
if age is not None and age > stale_days:
|
|
stale_ids.append(entity_id)
|
|
failures.append(_failure("stale", entity_id, str(entity["asOf"])))
|
|
for claim_id in entity.get("claimRefs") or []:
|
|
if claim_id not in claim_ids:
|
|
unresolved.append({"id": entity_id, "field": "claimRefs", "target": str(claim_id)})
|
|
failures.append(_failure("unresolved-reference", entity_id, str(claim_id)))
|
|
|
|
for relation in relations:
|
|
relation_id = str(relation.get("id") or "")
|
|
records += 1
|
|
refs = list(relation.get("sourceRefs") or [])
|
|
if provenance_ok(refs):
|
|
complete += 1
|
|
else:
|
|
incomplete.append(relation_id)
|
|
for field in ("from", "to"):
|
|
target = relation.get(field)
|
|
if target not in entity_ids:
|
|
unresolved.append({"id": relation_id, "field": field, "target": str(target)})
|
|
failures.append(_failure("dangling-reference", relation_id, f"{field}:{target}"))
|
|
for ref in refs:
|
|
if ref not in known_source_ids:
|
|
unresolved.append({"id": relation_id, "field": "sourceRefs", "target": str(ref)})
|
|
failures.append(_failure("unresolved-reference", relation_id, str(ref)))
|
|
|
|
for claim in claims:
|
|
claim_id = str(claim.get("id") or "")
|
|
records += 1
|
|
refs = list(claim.get("evidenceRefs") or [])
|
|
if provenance_ok(refs):
|
|
complete += 1
|
|
else:
|
|
incomplete.append(claim_id)
|
|
subject = claim.get("subjectId")
|
|
if subject not in entity_ids:
|
|
unresolved.append({"id": claim_id, "field": "subjectId", "target": str(subject)})
|
|
failures.append(_failure("dangling-reference", claim_id, f"subject:{subject}"))
|
|
for ref in refs:
|
|
if ref not in known_source_ids:
|
|
unresolved.append({"id": claim_id, "field": "evidenceRefs", "target": str(ref)})
|
|
failures.append(_failure("unresolved-reference", claim_id, str(ref)))
|
|
|
|
counts = {name: 0 for name in FAILURE_CLASSES}
|
|
for failure in failures:
|
|
counts[failure["class"]] = counts.get(failure["class"], 0) + 1
|
|
|
|
return {
|
|
"dataspace": dataspace_id or (manifest or {}).get("id") or "",
|
|
"provenance": {"records": records, "complete": complete, "incomplete": incomplete},
|
|
"stale": {"count": len(stale_ids), "ids": stale_ids},
|
|
"unresolvedReferences": unresolved,
|
|
"failureClasses": counts,
|
|
"failures": failures,
|
|
"sources": {
|
|
"count": source_count,
|
|
"complete": sources_complete,
|
|
"missingLicense": missing_license,
|
|
"licenses": licenses,
|
|
},
|
|
}
|
|
|
|
|
|
def report_dataspaces(root: Path, *, today: str | None = None, stale_days: int = 365) -> list[dict[str, Any]]:
|
|
"""One quality report for each example dataspace directory."""
|
|
reports = []
|
|
for snapshot_path in sorted(root.glob("*/snapshot.json")):
|
|
directory = snapshot_path.parent
|
|
snapshot = json.loads(snapshot_path.read_text(encoding="utf-8"))
|
|
manifest_path = directory / "manifest.json"
|
|
manifest = json.loads(manifest_path.read_text(encoding="utf-8")) if manifest_path.is_file() else None
|
|
reports.append(quality_report(snapshot, manifest, today=today, stale_days=stale_days))
|
|
return reports
|