Spaces:
Sleeping
Sleeping
File size: 4,908 Bytes
b92b744 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 | """Deduplicate PMID/NCT links while preferring registry primary-outcome numbers."""
from collections import defaultdict
NUMERIC_FIELDS = {
"n", "estimate", "ci_low", "ci_high", "p_value", "effect_type", "ci_level", "ci_sides",
"p_value_operator", "p_value_raw", "effect_type_raw", "statistical_method", "numeric_source",
"outcome", "outcome_unit", "intervention", "comparator", "analysis_group_ids", "evidence_tier",
}
UNION_FIELDS = {"pmids", "nct_ids", "result_pmids", "referenced_works"}
def link_studies(studies: list[dict]) -> list[dict]:
"""Merge papers into a canonical trial and remove duplicate source records.
A paper can mention several trials. Such a paper is attached to each trial,
without collapsing distinct NCT IDs into one trial; the publication row is
removed. Reviews are never proof that an individual trial reported results.
"""
unique = {study["id"]: dict(study) for study in studies}
trials = {identifier: row for identifier, row in unique.items()
if row.get("source") in {"ctgov", "merged"}}
papers = {identifier: row for identifier, row in unique.items()
if row.get("source") == "openalex"}
by_pmid: dict[str, set[str]] = defaultdict(set)
by_nct: dict[str, set[str]] = defaultdict(set)
for identifier, trial in trials.items():
for value in trial.get("result_pmids", []):
by_pmid[str(value)].add(identifier)
for value in trial.get("nct_ids", []):
by_nct[value].add(identifier)
consumed: set[str] = set()
for paper_id, paper in papers.items():
if paper.get("is_review"):
continue
targets: set[str] = set()
for value in paper.get("pmids", []):
targets.update(by_pmid.get(str(value), set()))
reports_outcome = paper.get("result_label") in {"positive", "null", "mixed"} or any(
paper.get(key) is not None for key in ("estimate", "ci_low", "p_value")
)
# NCT mentions in protocols/background are not proof that results were published.
if reports_outcome:
for value in paper.get("nct_ids", []):
targets.update(by_nct.get(value, set()))
for target in sorted(targets):
trial = trials[target]
registry_numbers = {key: trial[key] for key in NUMERIC_FIELDS if trial.get(key) is not None}
registry_has_effect = any(trial.get(key) is not None for key in ("estimate", "ci_low", "p_value"))
if registry_has_effect:
registry_numbers["evidence_span"] = trial.get("evidence_span", "")
trial.setdefault("linked_papers", []).append({
"id": paper_id, "title": paper.get("title"), "url": paper.get("url"),
"result_label": paper.get("result_label"), "evidence_span": paper.get("evidence_span"),
})
# The first linked paper supplies searchable literature text, never registry effects.
if trial.get("source") != "merged":
trial["registry_title"] = trial.get("title", "")
trial["registry_abstract"] = trial.get("abstract", "")
trial["registry_url"] = trial.get("url", "")
for key in ("title", "abstract", "authors", "venue", "embedding", "embedding_model",
"abstract_available", "work_type", "snapshot_provenance",
"result_label", "null_score", "evidence_span", "classification_method"):
if key in paper:
trial[key] = paper[key]
if not registry_has_effect:
for key in NUMERIC_FIELDS:
if paper.get(key) is not None:
trial[key] = paper[key]
trial.update(registry_numbers)
for key in UNION_FIELDS:
# Only the trial's own NCT ID belongs to this canonical row.
if key != "nct_ids":
trial[key] = sorted(set(trial.get(key, [])) | set(paper.get(key, [])))
trial["is_retracted"] = bool(trial.get("is_retracted") or paper.get("is_retracted"))
trial["source"] = "merged"
trial["has_linked_publication"] = True
trial["paper_result_label"] = paper.get("result_label", "no_result_stated")
# Disagreement is descriptive; statistical equivalence is determined later.
p = trial.get("p_value")
null_registry = (p is not None and p >= 0.05 and
trial.get("p_value_operator", "=") in {"=", ">", ">="})
trial["possible_abstract_spin"] = bool(null_registry and paper.get("result_label") == "positive")
if targets:
consumed.add(paper_id)
return [trials.get(identifier, row) for identifier, row in unique.items() if identifier not in consumed]
|