#!/usr/bin/env python3 """A11 restricted, offline metadata audit and artificial-data design tools. No participant-data endpoint, remote writes, ORCID authentication, DOI registration, or resource-license certification is implemented here. Audit reads the frozen, privacy-minimized DataCite metadata projection supplied with this article. """ import argparse import collections import hashlib import json import math from pathlib import Path import re import sys sys.dont_write_bytecode = True VERSION_RELATIONS = {"HasVersion", "IsVersionOf", "IsNewVersionOf", "IsPreviousVersionOf"} def canonical_doi(value): value = str(value or "").strip().lower() value = re.sub(r"^https?://(?:dx\.)?doi\.org/", "", value) return value def valid_orcid(value): value = re.sub(r"^https?://orcid\.org/", "", str(value or "").strip(), flags=re.I) if not re.fullmatch(r"\d{4}-\d{4}-\d{4}-\d{3}[\dX]", value): return False digits = value.replace("-", "") total = 0 for c in digits[:-1]: total = (total + int(c)) * 2 result = (12 - total % 11) % 11 return digits[-1] == ("X" if result == 10 else str(result)) def project(responses): """Preserve audit-bearing metadata, omit names, abstracts and ORCID values.""" rows = [] for response in responses: for item in response["data"]: a = item["attributes"] ids = [n for c in a.get("creators", []) for n in c.get("nameIdentifiers", []) if str(n.get("nameIdentifierScheme", "")).lower() == "orcid"] rows.append({ "doi": canonical_doi(a["doi"]), "resourceTypeGeneral": a.get("types", {}).get("resourceTypeGeneral"), "publicationYear": a.get("publicationYear"), "titlePresent": any(bool(str(t.get("title", "")).strip()) for t in a.get("titles", [])), "creatorCount": len(a.get("creators", [])), "creatorOrcidIdentifierCount": len(ids), "creatorOrcidChecksumValidCount": sum(valid_orcid(n.get("nameIdentifier")) for n in ids), "publisher": a.get("publisher"), "version": a.get("version"), "metadataVersion": a.get("metadataVersion"), "schemaVersion": a.get("schemaVersion"), "rightsList": a.get("rightsList", []), "relatedIdentifiers": a.get("relatedIdentifiers", []), "dates": a.get("dates", []), "created": a.get("created"), "registered": a.get("registered"), "published": a.get("published"), "updated": a.get("updated"), "language": a.get("language"), "formats": a.get("formats", []), "urlPresent": bool(a.get("url")), "contentUrlPresent": bool(a.get("contentUrl")), "state": a.get("state"), "repositoryId": item.get("relationships", {}).get("client", {}).get("data", {}).get("id") }) return {"projectionVersion": "1.0", "source": "DataCite REST API", "rows": rows} def audit(projection): rows = projection["rows"] dois = [r["doi"] for r in rows] if len(set(dois)) != len(dois): raise ValueError("Duplicate DOI in bounded sample") if any(r["publicationYear"] not in (2024, 2025) for r in rows): raise ValueError("Publication year outside frozen query") if any(r["creatorOrcidChecksumValidCount"] > r["creatorOrcidIdentifierCount"] for r in rows): raise ValueError("ORCID counts inconsistent") summaries = {} for group in ["ALL"] + sorted({r["resourceTypeGeneral"] for r in rows}): rr = rows if group == "ALL" else [r for r in rows if r["resourceTypeGeneral"] == group] c = collections.Counter() reltypes = collections.Counter() publishers = collections.Counter() for r in rr: c["records"] += 1 c["creatorEntries"] += r["creatorCount"] c["recordsWithCreatorOrcid"] += r["creatorOrcidIdentifierCount"] > 0 c["recordsWithChecksumValidCreatorOrcid"] += r["creatorOrcidChecksumValidCount"] > 0 c["creatorOrcidIdentifierEntries"] += r["creatorOrcidIdentifierCount"] c["checksumValidCreatorOrcidEntries"] += r["creatorOrcidChecksumValidCount"] c["versionPresent"] += bool(str(r.get("version") or "").strip()) c["rightsListPresent"] += bool(r["rightsList"]) c["rightsUriPresent"] += any(bool(x.get("rightsUri")) for x in r["rightsList"]) c["languagePresent"] += bool(r["language"]) c["formatsPresent"] += bool(r["formats"]) c["landingUrlPresent"] += r["urlPresent"] c["contentUrlPresent"] += r["contentUrlPresent"] c["resourcePublishedTimestampPresent"] += bool(r["published"]) c["registeredTimestampPresent"] += bool(r["registered"]) c["createdTimestampPresent"] += bool(r["created"]) c["datedIssuedPresent"] += any(x.get("dateType") == "Issued" for x in r["dates"]) c["relatedIdentifierPresent"] += bool(r["relatedIdentifiers"]) c["versionRelationPresent"] += any(x.get("relationType") in VERSION_RELATIONS for x in r["relatedIdentifiers"]) c["sourceDerivationRelationPresent"] += any(x.get("relationType") in {"IsDerivedFrom", "IsSourceOf"} for x in r["relatedIdentifiers"]) c["selfRelationPresent"] += any(canonical_doi(x.get("relatedIdentifier")) == r["doi"] for x in r["relatedIdentifiers"] if x.get("relatedIdentifierType") == "DOI") c["inSampleDoiRelationPresent"] += any(canonical_doi(x.get("relatedIdentifier")) in set(dois) for x in r["relatedIdentifiers"] if x.get("relatedIdentifierType") == "DOI") for x in r["relatedIdentifiers"]: reltypes[x.get("relationType", "MISSING")] += 1 pub = r["publisher"] publishers[pub.get("name", "") if isinstance(pub, dict) else str(pub or "")] += 1 summaries[group] = {"counts": dict(sorted(c.items())), "relationTypeEntries": dict(sorted(reltypes.items())), "publisherCounts": dict(sorted(publishers.items()))} return {"analysis": "Completed descriptive metadata feasibility audit", "sample": "Bounded first 100 DOI names per type; not random or exhaustive", "recordUnit": "DOI record, not distinct person, research project, paper or learner", "summaries": summaries, "identityAuthenticationAssessed": False, "linkedFullTextRightsVerified": False, "learningOutcomesAssessed": False, "inferentialPopulationEstimates": False} def paired_summary(differences, permutations=9999, seed=20261016): import numpy as np from scipy.stats import t d = np.asarray(differences, dtype=float) if d.ndim != 1 or len(d) < 3 or not np.isfinite(d).all() or isinstance(permutations, bool) or not isinstance(permutations, int) or not 1 <= permutations <= 1000000: raise ValueError("At least three finite pair differences and positive permutation count required") estimate = float(d.mean()) se = float(d.std(ddof=1) / math.sqrt(len(d))) if se <= 0: raise ValueError("Positive standard error required") critical = float(t.ppf(.975, len(d)-1)) signs = np.random.default_rng(seed).choice([-1, 1], size=(permutations, len(d))) perm = np.abs((signs*d).mean(axis=1)) return {"pairs": len(d), "estimate": estimate, "standardError": se, "confidenceInterval95": [estimate-critical*se, estimate+critical*se], "tPValue": float(2*t.sf(abs(estimate/se), len(d)-1)), "fisherSharpNullMonteCarloP": float((1+(perm >= abs(estimate)-1e-12).sum())/(permutations+1)), "note": "Complete-pair diagnostic only. Does not implement missing-outcome primary model."} def simulate(replicates=20000, seed=20261016): import numpy as np import scipy from scipy.stats import t if replicates < 100 or replicates > 1000000: raise ValueError("Replicates must be between 100 and 1000000") rng = np.random.default_rng(seed) out = [] # Scale is individual within-pair change-score SD; shared pair effect cancels. for rho in [.05, .12, .20]: for retention in [1., .85, .70]: pairs = 48 n = rng.binomial(12, retention, size=(replicates, pairs, 2)) n = np.maximum(n, 1) cluster_error = rng.normal(size=(replicates, pairs, 2))*math.sqrt(rho) individual_mean_error = rng.normal(size=n.shape)*np.sqrt((1-rho)/n) d0 = (cluster_error+individual_mean_error)[:, :, 1]-(cluster_error+individual_mean_error)[:, :, 0] se = d0.std(axis=1, ddof=1)/math.sqrt(pairs) critical = t.ppf(.975, pairs-1) null = float(np.mean(np.abs(d0.mean(axis=1)/se) > critical)) power = float(np.mean(np.abs((d0.mean(axis=1)+.30)/se) > critical)) coverage = float(np.mean(np.abs(d0.mean(axis=1)) <= critical*se)) out.append({"pairs":pairs,"clusters":96,"baselineStudentsPerCluster":12, "changeScoreIcc":rho,"independentStudentRetention":retention, "standardizedEffect":.30,"power":power,"powerMonteCarloSE":math.sqrt(power*(1-power)/replicates), "nullRejection":null,"nullMonteCarloSE":math.sqrt(null*(1-null)/replicates), "coverage95":coverage,"meanRetainedPerCluster":float(n.mean())}) return {"analysis":"Executed artificial-data sensitivity study; not participant findings", "seed":seed,"replicatesPerScenario":replicates, "software":{"python":sys.version.split()[0],"numpy":np.__version__,"scipy":scipy.__version__}, "assumptions":["Gaussian cluster and student mean errors", "Independent retention unrelated to outcomes", "At least one learner retained per cluster by construction", "No cluster loss or interference", "Equal pair weights", "Fixed constant effect 0.30 within-pair change-score SD", "Shared matched-pair effects cancel"], "limitations":["Not a model of informative missingness", "No validated rubric SD or empirical ICC exists for this new instrument", "Not a recruitment guarantee"], "scenarios":out} def main(): parser = argparse.ArgumentParser() parser.add_argument("operation", choices=["audit", "simulate", "pair-diagnostic"]) parser.add_argument("--input", type=Path) parser.add_argument("--output", type=Path, required=True) args = parser.parse_args() if args.operation == "audit": result = audit(json.loads(args.input.read_text())) elif args.operation == "simulate": result = simulate() else: result = paired_summary(json.loads(args.input.read_text())["pairDifferences"]) args.output.write_text(json.dumps(result, indent=2)+"\n") if __name__ == "__main__": main()