503 lines
18 KiB
Python
503 lines
18 KiB
Python
|
"""Allowlist-only OnCo reference adapter.
|
||
|
|
|
||
|
|
Reads enumerated records from a local upstream checkout and writes a snapshot
|
||
|
|
under the gitignored reference directory. It does not publish into
|
||
|
|
``examples/`` and it does not import UI code. Edges that leave the allowlist
|
||
|
|
are reported and are not stored as relations.
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import json
|
||
|
|
from pathlib import Path
|
||
|
|
from typing import Any
|
||
|
|
|
||
|
|
from content_explorer.attribution import attribution_gaps
|
||
|
|
from content_explorer.contract import validate_documents
|
||
|
|
from content_explorer.ingest import build_manifest
|
||
|
|
|
||
|
|
PIN = "b594d7433bf4e408b44215e25e5304696d6f840e"
|
||
|
|
LICENSE = "CC-BY-NC-4.0"
|
||
|
|
LICENSE_URL = "https://creativecommons.org/licenses/by-nc/4.0/"
|
||
|
|
ATTRIBUTION = "Data from OnCo (onco.cc)"
|
||
|
|
MODIFICATION = (
|
||
|
|
"Projected into a content-explorer snapshot. Titles, summaries, and "
|
||
|
|
"in-slice relations keep the pinned OnCo wording. Edges outside the "
|
||
|
|
"allowlist are listed in the import report and are not stored."
|
||
|
|
)
|
||
|
|
RELATION_FIELDS = {
|
||
|
|
"targets": "targets",
|
||
|
|
"drugs": "drugs",
|
||
|
|
"trials": "trials",
|
||
|
|
"terms": "terms",
|
||
|
|
"technologies": "technologies",
|
||
|
|
"pathways": "pathways",
|
||
|
|
"companies": "companies",
|
||
|
|
"institutions": "institutions",
|
||
|
|
"cancers": "cancers",
|
||
|
|
"related": "related",
|
||
|
|
"people": "people",
|
||
|
|
"papers": "papers",
|
||
|
|
"keyPapers": "key-papers",
|
||
|
|
}
|
||
|
|
LOCAL_ROOT_NAME = Path("tests") / "onco-reference" / "local"
|
||
|
|
|
||
|
|
|
||
|
|
def load_authorization(path: Path) -> dict[str, Any]:
|
||
|
|
document = json.loads(path.read_text(encoding="utf-8"))
|
||
|
|
if not isinstance(document, dict) or document.get("granted") is not True:
|
||
|
|
raise RuntimeError("OnCo reference use is not authorized")
|
||
|
|
if document.get("distribution_allowed") is not False or document.get("ci_allowed") is not False:
|
||
|
|
raise RuntimeError("this grant does not allow redistribution or public CI")
|
||
|
|
return document
|
||
|
|
|
||
|
|
|
||
|
|
def _skip_string(text: str, index: int) -> int:
|
||
|
|
quote = text[index]
|
||
|
|
index += 1
|
||
|
|
length = len(text)
|
||
|
|
while index < length:
|
||
|
|
char = text[index]
|
||
|
|
if char == "\\":
|
||
|
|
index += 2
|
||
|
|
continue
|
||
|
|
if quote == "`" and char == "$" and index + 1 < length and text[index + 1] == "{":
|
||
|
|
depth = 1
|
||
|
|
index += 2
|
||
|
|
while index < length and depth:
|
||
|
|
if text[index] in ('"', "'", "`"):
|
||
|
|
index = _skip_string(text, index)
|
||
|
|
continue
|
||
|
|
if text[index] == "{":
|
||
|
|
depth += 1
|
||
|
|
elif text[index] == "}":
|
||
|
|
depth -= 1
|
||
|
|
index += 1
|
||
|
|
continue
|
||
|
|
if char == quote:
|
||
|
|
return index + 1
|
||
|
|
index += 1
|
||
|
|
return index
|
||
|
|
|
||
|
|
|
||
|
|
def _skip_comment(text: str, index: int) -> int:
|
||
|
|
if text.startswith("//", index):
|
||
|
|
newline = text.find("\n", index)
|
||
|
|
return length if (length := len(text)) and newline < 0 else newline + 1
|
||
|
|
if text.startswith("/*", index):
|
||
|
|
end = text.find("*/", index + 2)
|
||
|
|
return len(text) if end < 0 else end + 2
|
||
|
|
return index
|
||
|
|
|
||
|
|
|
||
|
|
def find_object(text: str, entity_id: str) -> str:
|
||
|
|
"""Return the object literal whose own ``id`` is ``entity_id``."""
|
||
|
|
length = len(text)
|
||
|
|
index = 0
|
||
|
|
depth = 0
|
||
|
|
starts: list[int] = []
|
||
|
|
pending: tuple[int, int] | None = None
|
||
|
|
while index < length:
|
||
|
|
char = text[index]
|
||
|
|
if char in ('"', "'", "`"):
|
||
|
|
index = _skip_string(text, index)
|
||
|
|
continue
|
||
|
|
if char == "/" and index + 1 < length and text[index + 1] in "/*":
|
||
|
|
index = _skip_comment(text, index)
|
||
|
|
continue
|
||
|
|
if char == "{":
|
||
|
|
starts.append(index)
|
||
|
|
depth += 1
|
||
|
|
index += 1
|
||
|
|
continue
|
||
|
|
if char == "}":
|
||
|
|
start = starts.pop() if starts else 0
|
||
|
|
if pending and pending[0] == depth:
|
||
|
|
return text[start:index + 1]
|
||
|
|
depth -= 1
|
||
|
|
index += 1
|
||
|
|
continue
|
||
|
|
if text.startswith("id", index) and (index == 0 or not (text[index - 1].isalnum() or text[index - 1] == "_")):
|
||
|
|
cursor = index + 2
|
||
|
|
while cursor < length and text[cursor].isspace():
|
||
|
|
cursor += 1
|
||
|
|
if cursor < length and text[cursor] == ":":
|
||
|
|
cursor += 1
|
||
|
|
while cursor < length and text[cursor].isspace():
|
||
|
|
cursor += 1
|
||
|
|
if cursor < length and text[cursor] in ('"', "'"):
|
||
|
|
end = _skip_string(text, cursor)
|
||
|
|
literal = text[cursor + 1:end - 1]
|
||
|
|
if literal == entity_id and starts:
|
||
|
|
pending = (depth, starts[-1])
|
||
|
|
index += 1
|
||
|
|
raise ValueError(f"{entity_id} was not found")
|
||
|
|
|
||
|
|
|
||
|
|
def _string_literal(text: str, index: int) -> tuple[str, int]:
|
||
|
|
end = _skip_string(text, index)
|
||
|
|
return text[index + 1:end - 1], end
|
||
|
|
|
||
|
|
|
||
|
|
def object_fields(text: str, constants: dict[str, str]) -> dict[str, Any]:
|
||
|
|
"""Read depth-1 keys. Nested objects are skipped. String arrays are kept."""
|
||
|
|
fields: dict[str, Any] = {}
|
||
|
|
length = len(text)
|
||
|
|
index = 0
|
||
|
|
depth = 0
|
||
|
|
while index < length:
|
||
|
|
char = text[index]
|
||
|
|
if char in ('"', "'", "`"):
|
||
|
|
index = _skip_string(text, index)
|
||
|
|
continue
|
||
|
|
if char == "/" and index + 1 < length and text[index + 1] in "/*":
|
||
|
|
index = _skip_comment(text, index)
|
||
|
|
continue
|
||
|
|
if char == "{":
|
||
|
|
depth += 1
|
||
|
|
index += 1
|
||
|
|
continue
|
||
|
|
if char == "}":
|
||
|
|
depth -= 1
|
||
|
|
index += 1
|
||
|
|
continue
|
||
|
|
if depth != 1 or not (char.isalpha() or char == "_"):
|
||
|
|
index += 1
|
||
|
|
continue
|
||
|
|
start = index
|
||
|
|
while index < length and (text[index].isalnum() or text[index] == "_"):
|
||
|
|
index += 1
|
||
|
|
key = text[start:index]
|
||
|
|
while index < length and text[index].isspace():
|
||
|
|
index += 1
|
||
|
|
if index < length and text[index] in ",}":
|
||
|
|
if key in constants:
|
||
|
|
fields[key] = constants[key]
|
||
|
|
continue
|
||
|
|
if index >= length or text[index] != ":":
|
||
|
|
continue
|
||
|
|
index += 1
|
||
|
|
while index < length and text[index].isspace():
|
||
|
|
index += 1
|
||
|
|
if index >= length:
|
||
|
|
break
|
||
|
|
if text[index] in ('"', "'", "`"):
|
||
|
|
value, index = _string_literal(text, index)
|
||
|
|
fields[key] = value
|
||
|
|
continue
|
||
|
|
if text[index] == "[":
|
||
|
|
index += 1
|
||
|
|
values: list[str] = []
|
||
|
|
while index < length and text[index] != "]":
|
||
|
|
if text[index] in ('"', "'", "`"):
|
||
|
|
value, index = _string_literal(text, index)
|
||
|
|
values.append(value)
|
||
|
|
continue
|
||
|
|
if text[index] in ('"', "'", "`"):
|
||
|
|
index = _skip_string(text, index)
|
||
|
|
continue
|
||
|
|
if text[index] == "{":
|
||
|
|
nested = 1
|
||
|
|
index += 1
|
||
|
|
while index < length and nested:
|
||
|
|
if text[index] in ('"', "'", "`"):
|
||
|
|
index = _skip_string(text, index)
|
||
|
|
continue
|
||
|
|
if text[index] == "{":
|
||
|
|
nested += 1
|
||
|
|
elif text[index] == "}":
|
||
|
|
nested -= 1
|
||
|
|
index += 1
|
||
|
|
continue
|
||
|
|
index += 1
|
||
|
|
fields[key] = values
|
||
|
|
index += 1
|
||
|
|
continue
|
||
|
|
ident_start = index
|
||
|
|
while index < length and (text[index].isalnum() or text[index] == "_"):
|
||
|
|
index += 1
|
||
|
|
ident = text[ident_start:index]
|
||
|
|
if ident in constants:
|
||
|
|
fields[key] = constants[ident]
|
||
|
|
continue
|
||
|
|
if index == ident_start:
|
||
|
|
index += 1
|
||
|
|
continue
|
||
|
|
return fields
|
||
|
|
|
||
|
|
|
||
|
|
def file_constants(text: str) -> dict[str, str]:
|
||
|
|
constants: dict[str, str] = {}
|
||
|
|
marker = "const asOf = "
|
||
|
|
at = text.find(marker)
|
||
|
|
if at >= 0:
|
||
|
|
cursor = at + len(marker)
|
||
|
|
if cursor < len(text) and text[cursor] in ('"', "'"):
|
||
|
|
value, _end = _string_literal(text, cursor)
|
||
|
|
constants["asOf"] = value
|
||
|
|
return constants
|
||
|
|
|
||
|
|
|
||
|
|
OPTIONAL_FIELDS = ("aka", "status")
|
||
|
|
|
||
|
|
|
||
|
|
def read_record(upstream: Path, entry: dict[str, str]) -> dict[str, Any]:
|
||
|
|
path = upstream / entry["file"]
|
||
|
|
text = path.read_text(encoding="utf-8")
|
||
|
|
obj = find_object(text, entry["id"])
|
||
|
|
fields = object_fields(obj, file_constants(text))
|
||
|
|
if fields.get("id") != entry["id"]:
|
||
|
|
raise ValueError(f"{entry['id']} object did not keep its id")
|
||
|
|
title = str(fields.get("name") or "").strip()
|
||
|
|
tldr = str(fields.get("tldr") or "").strip()
|
||
|
|
if not title or not tldr:
|
||
|
|
raise ValueError(f"{entry['id']} is missing a name or tldr")
|
||
|
|
relations: list[tuple[str, str]] = []
|
||
|
|
for field, predicate in RELATION_FIELDS.items():
|
||
|
|
for target in fields.get(field) or []:
|
||
|
|
if isinstance(target, str) and target:
|
||
|
|
relations.append((predicate, target))
|
||
|
|
return {
|
||
|
|
"id": entry["id"],
|
||
|
|
"kind": entry["kind"],
|
||
|
|
"title": title,
|
||
|
|
"tldr": tldr,
|
||
|
|
"summary": str(fields.get("summary") or tldr).strip(),
|
||
|
|
"asOf": str(fields.get("asOf") or "2026-09-04"),
|
||
|
|
"relations": relations,
|
||
|
|
"file": entry["file"],
|
||
|
|
"missingFields": [
|
||
|
|
name for name in OPTIONAL_FIELDS
|
||
|
|
if name not in fields or fields.get(name) in ("", None, [])
|
||
|
|
],
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def project_slice(
|
||
|
|
records: list[dict[str, Any]],
|
||
|
|
*,
|
||
|
|
source: dict[str, Any],
|
||
|
|
kinds: list[dict[str, str]],
|
||
|
|
relation_labels: dict[str, dict[str, str]],
|
||
|
|
require_size: bool = True,
|
||
|
|
) -> dict[str, Any]:
|
||
|
|
"""Build a snapshot. Outside-allowlist edges are reported, not stored."""
|
||
|
|
gaps = attribution_gaps(source)
|
||
|
|
if gaps:
|
||
|
|
raise RuntimeError("refusing to project a source with attribution gaps: " + ", ".join(gaps))
|
||
|
|
by_upstream = {record["id"]: record for record in records}
|
||
|
|
if len(by_upstream) != len(records):
|
||
|
|
raise RuntimeError("duplicate upstream id")
|
||
|
|
outside: list[dict[str, str]] = []
|
||
|
|
relations: list[dict[str, Any]] = []
|
||
|
|
entities: list[dict[str, Any]] = []
|
||
|
|
claims: list[dict[str, Any]] = []
|
||
|
|
predicates: list[str] = []
|
||
|
|
record_sources: list[dict[str, Any]] = []
|
||
|
|
dataspace = "onco-reference"
|
||
|
|
for record in records:
|
||
|
|
kind = record["kind"]
|
||
|
|
key = record["id"]
|
||
|
|
entity_id = f"urn:ce:{dataspace}:{kind}:{key}"
|
||
|
|
claim_id = f"claim:{kind}:{key}:tldr"
|
||
|
|
record_source = dict(source)
|
||
|
|
record_source["id"] = f"src:onco:{kind}:{key}"
|
||
|
|
record_source["uri"] = f"https://onco.cc/{kind}s/{key}"
|
||
|
|
record_source["title"] = record["title"]
|
||
|
|
record_gaps = attribution_gaps(record_source)
|
||
|
|
if record_gaps:
|
||
|
|
raise RuntimeError("record source is missing attribution: " + ", ".join(record_gaps))
|
||
|
|
record_sources.append(record_source)
|
||
|
|
entities.append({
|
||
|
|
"id": entity_id,
|
||
|
|
"kindId": kind,
|
||
|
|
"title": record["title"],
|
||
|
|
"tldr": record["tldr"],
|
||
|
|
"summary": record["summary"],
|
||
|
|
"aliases": [key],
|
||
|
|
"tags": [kind],
|
||
|
|
"sourceRefs": [source["id"], record_source["id"]],
|
||
|
|
"claimRefs": [claim_id],
|
||
|
|
"asOf": record["asOf"],
|
||
|
|
})
|
||
|
|
claims.append({
|
||
|
|
"id": claim_id,
|
||
|
|
"subjectId": entity_id,
|
||
|
|
"key": "tldr",
|
||
|
|
"value": record["tldr"],
|
||
|
|
"status": "asserted",
|
||
|
|
"evidenceRefs": [source["id"]],
|
||
|
|
"observedAt": record["asOf"],
|
||
|
|
})
|
||
|
|
seen: set[tuple[str, str]] = set()
|
||
|
|
index = 0
|
||
|
|
for predicate, target in record["relations"]:
|
||
|
|
if (predicate, target) in seen:
|
||
|
|
continue
|
||
|
|
seen.add((predicate, target))
|
||
|
|
if target not in by_upstream:
|
||
|
|
outside.append({"from": key, "predicate": predicate, "to": target})
|
||
|
|
continue
|
||
|
|
index += 1
|
||
|
|
target_kind = by_upstream[target]["kind"]
|
||
|
|
predicates.append(predicate)
|
||
|
|
relations.append({
|
||
|
|
"id": f"{kind}-{key}-{index}",
|
||
|
|
"predicate": predicate,
|
||
|
|
"from": entity_id,
|
||
|
|
"to": f"urn:ce:{dataspace}:{target_kind}:{target}",
|
||
|
|
"sourceRefs": [source["id"], record_source["id"]],
|
||
|
|
"asOf": record["asOf"],
|
||
|
|
})
|
||
|
|
entities.sort(key=lambda item: item["id"])
|
||
|
|
relations.sort(key=lambda item: item["id"])
|
||
|
|
claims.sort(key=lambda item: item["id"])
|
||
|
|
manifest, missing = build_manifest(
|
||
|
|
dataspace_id=dataspace,
|
||
|
|
heading="OnCo reference slice",
|
||
|
|
kinds=kinds,
|
||
|
|
relation_labels=relation_labels,
|
||
|
|
predicates=predicates,
|
||
|
|
fact_policy="show-all",
|
||
|
|
)
|
||
|
|
if missing or manifest is None:
|
||
|
|
raise RuntimeError("missing relation labels: " + ", ".join(missing))
|
||
|
|
for view in manifest["kindViews"].values():
|
||
|
|
view["columns"] = ["name", "tldr", "tags"]
|
||
|
|
view["facets"] = ["tags", "kind"]
|
||
|
|
record_sources.sort(key=lambda item: item["id"])
|
||
|
|
snapshot = {
|
||
|
|
"schemaVersion": "0.1",
|
||
|
|
"dataspace": {
|
||
|
|
"id": dataspace,
|
||
|
|
"title": "OnCo reference slice",
|
||
|
|
"description": "A local allowlisted OnCo conformance slice. Not medical advice. Not for redistribution.",
|
||
|
|
},
|
||
|
|
"kinds": kinds,
|
||
|
|
"entities": entities,
|
||
|
|
"relations": relations,
|
||
|
|
"sources": [source, *record_sources],
|
||
|
|
"claims": claims,
|
||
|
|
}
|
||
|
|
errors = validate_documents(snapshot, manifest, require_size=require_size)
|
||
|
|
if errors:
|
||
|
|
raise RuntimeError("projection failed validation: " + "; ".join(errors))
|
||
|
|
source_edges = sum(len({(predicate, target) for predicate, target in record["relations"]}) for record in records)
|
||
|
|
if len(relations) + len(outside) != source_edges:
|
||
|
|
raise RuntimeError("an edge was lost between the extract and the report")
|
||
|
|
report = {
|
||
|
|
"connectorVersion": "onco-reference/0.1.0",
|
||
|
|
"published": True,
|
||
|
|
"candidates": len(records),
|
||
|
|
"accepted": len(entities),
|
||
|
|
"outsideAllowlist": outside,
|
||
|
|
"storedRelations": len(relations),
|
||
|
|
"sourceEdges": source_edges,
|
||
|
|
"missingOptional": [
|
||
|
|
{"id": record["id"], "fields": list(record.get("missingFields") or [])}
|
||
|
|
for record in records
|
||
|
|
if record.get("missingFields")
|
||
|
|
],
|
||
|
|
"source": {"id": source["id"], "license": source["license"], "originSha": source.get("originSha")},
|
||
|
|
}
|
||
|
|
return {"snapshot": snapshot, "manifest": manifest, "report": report}
|
||
|
|
|
||
|
|
|
||
|
|
def assert_local_destination(destination: Path, repo: Path) -> None:
|
||
|
|
local = (repo / LOCAL_ROOT_NAME).resolve()
|
||
|
|
target = destination.resolve()
|
||
|
|
if local not in target.parents and target != local:
|
||
|
|
raise RuntimeError(f"refusing to write the OnCo slice outside {local}")
|
||
|
|
examples = (repo / "examples").resolve()
|
||
|
|
if examples in target.parents or target == examples:
|
||
|
|
raise RuntimeError("refusing to publish the OnCo slice into examples")
|
||
|
|
|
||
|
|
|
||
|
|
def write_projection(destination: Path, result: dict[str, Any], repo: Path) -> None:
|
||
|
|
assert_local_destination(destination, repo)
|
||
|
|
destination.mkdir(parents=True, exist_ok=True)
|
||
|
|
for name, document in (
|
||
|
|
("snapshot.json", result["snapshot"]),
|
||
|
|
("manifest.json", result["manifest"]),
|
||
|
|
("import-report.json", result["report"]),
|
||
|
|
):
|
||
|
|
(destination / name).write_text(json.dumps(document, indent=2) + "\n", encoding="utf-8")
|
||
|
|
|
||
|
|
|
||
|
|
def upstream_sha(upstream: Path) -> str:
|
||
|
|
import subprocess
|
||
|
|
completed = subprocess.run(
|
||
|
|
["git", "-C", str(upstream), "rev-parse", "HEAD"],
|
||
|
|
check=True,
|
||
|
|
capture_output=True,
|
||
|
|
text=True,
|
||
|
|
)
|
||
|
|
return completed.stdout.strip()
|
||
|
|
|
||
|
|
|
||
|
|
def extract_allowlist(upstream: Path, allowlist: list[dict[str, str]], authorization: Path) -> list[dict[str, Any]]:
|
||
|
|
"""Read the allowlist only after a local grant that refuses redistribution."""
|
||
|
|
load_authorization(authorization)
|
||
|
|
sha = upstream_sha(upstream)
|
||
|
|
if sha != PIN:
|
||
|
|
raise RuntimeError(f"upstream pin is {sha}, expected {PIN}")
|
||
|
|
return [read_record(upstream, entry) for entry in allowlist]
|
||
|
|
|
||
|
|
|
||
|
|
def relation_labels() -> dict[str, dict[str, str]]:
|
||
|
|
wording = {
|
||
|
|
"targets": ("Targets", "Targeted by"),
|
||
|
|
"drugs": ("Drugs", "Drug for"),
|
||
|
|
"trials": ("Trials", "Trial of"),
|
||
|
|
"terms": ("Terms", "Term for"),
|
||
|
|
"technologies": ("Technologies", "Technology for"),
|
||
|
|
"pathways": ("Pathways", "Pathway for"),
|
||
|
|
"companies": ("Companies", "Company for"),
|
||
|
|
"institutions": ("Institutions", "Institution for"),
|
||
|
|
"cancers": ("Cancers", "Cancer for"),
|
||
|
|
"related": ("Related", "Related to"),
|
||
|
|
"people": ("People", "Person for"),
|
||
|
|
"papers": ("Papers", "Paper for"),
|
||
|
|
"key-papers": ("Key papers", "Key paper for"),
|
||
|
|
}
|
||
|
|
return {
|
||
|
|
predicate: {"forward": forward, "inverse": inverse}
|
||
|
|
for predicate, (forward, inverse) in wording.items()
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def kinds_for(entries: list[dict[str, str]]) -> list[dict[str, str]]:
|
||
|
|
labels = {
|
||
|
|
"cancer": ("Cancer", "Cancers"),
|
||
|
|
"drug": ("Drug", "Drugs"),
|
||
|
|
"target": ("Target", "Targets"),
|
||
|
|
"technology": ("Technology", "Technologies"),
|
||
|
|
"term": ("Term", "Terms"),
|
||
|
|
"trial": ("Trial", "Trials"),
|
||
|
|
"company": ("Company", "Companies"),
|
||
|
|
"institution": ("Institution", "Institutions"),
|
||
|
|
"paper": ("Paper", "Papers"),
|
||
|
|
}
|
||
|
|
ordered: list[str] = []
|
||
|
|
for entry in entries:
|
||
|
|
if entry["kind"] not in ordered:
|
||
|
|
ordered.append(entry["kind"])
|
||
|
|
return [
|
||
|
|
{"id": kind, "label": labels[kind][0], "plural": labels[kind][1]}
|
||
|
|
for kind in ordered
|
||
|
|
]
|
||
|
|
|
||
|
|
|
||
|
|
def onco_source(retrieved_at: str) -> dict[str, str]:
|
||
|
|
return {
|
||
|
|
"id": "src:onco:reference-slice",
|
||
|
|
"uri": "https://onco.cc",
|
||
|
|
"title": "OnCo",
|
||
|
|
"license": LICENSE,
|
||
|
|
"licenseUrl": LICENSE_URL,
|
||
|
|
"attribution": ATTRIBUTION,
|
||
|
|
"originSha": PIN,
|
||
|
|
"modificationNotice": MODIFICATION,
|
||
|
|
"retrievedAt": retrieved_at,
|
||
|
|
}
|