Derive Clinical Trials source links from each trial's own registration id

src/igh_data_transform/transformations/clinical_trials.py

@@ -1,5 +1,8 @@
 """Clinical trials table transformation (vin_clinicaltrials)."""
 
+import re
+from urllib.parse import quote
+
 import pandas as pd
 
 from igh_data_transform.transformations.cleanup import (
@@ -95,6 +98,124 @@
 }
 
 
+# =========================================================
+# Clinical trial source-link construction
+# =========================================================
+#
+# Each trial's authoritative registration id lives in ``vin_name`` (in
+# registry-native format). The raw ``vin_source`` column is unreliable —
+# bulk-import batches share a single constant URL, and there are blanks and
+# literal "CT.gov"/"N/A" values — so we rebuild the source link from
+# ``vin_name`` whenever we recognize the registry, and only fall back to
+# ``vin_source`` when the registration id is unrecognized.
+
+
+# Registries whose public trial page embeds the registration id verbatim (or
+# via a simple slug). Each builder receives the trimmed id.
+def _nct(id_):
+    return f"https://clinicaltrials.gov/study/{id_}"
+
+
+def _isrctn(id_):
+    return f"https://www.isrctn.com/{id_}"
+
+
+def _actrn(id_):
+    return f"https://anzctr.org.au/{id_}.aspx"
+
+
+def _tctr(id_):
+    return f"https://www.thaiclinicaltrials.org/show/{id_}"
+
+
+def _drks(id_):
+    return f"https://drks.de/search/en/trial/{id_}"
+
+
+def _jrct(id_):
+    return f"https://jrct.niph.go.jp/en-latest-detail/{id_}"
+
+
+def _slctr(id_):
+    # SLCTR/2016/015 -> slctr-2016-015
+    slug = id_.lower().replace("/", "-")
+    return f"https://slctr.lk/trials/{slug}"
+
+
+def _eudract_classic(id_):
+    return (
+        "https://www.clinicaltrialsregister.eu/ctr-search/search"
+        f"?query=eudract_number:{id_}"
+    )
+
+
+def _euct(id_):
+    return f"https://euclinicaltrials.eu/search-for-clinical-trials/?lang=en&EUCT={id_}"
+
+
+# Ordered (pattern, builder) pairs. Patterns are mutually exclusive by prefix,
+# so order is not significant, but EU CT (4 groups) must be distinguished from
+# classic EudraCT (3 groups) by the trailing "-NN".
+_NATIVE_REGISTRIES = [
+    (re.compile(r"^NCT\d+$", re.I), _nct),
+    (re.compile(r"^ISRCTN\d+$", re.I), _isrctn),
+    (re.compile(r"^ACTRN\d+$", re.I), _actrn),
+    (re.compile(r"^TCTR\d+$", re.I), _tctr),
+    (re.compile(r"^DRKS\d+$", re.I), _drks),
+    (re.compile(r"^jRCT\w+$", re.I), _jrct),
+    (re.compile(r"^SLCTR/\S+$", re.I), _slctr),
+    (re.compile(r"^\d{4}-\d{6}-\d{2}$"), _eudract_classic),
+    (re.compile(r"^\d{4}-\d{6}-\d{2}-\d{2}$"), _euct),
+]
+
+# Registries whose public URL uses an internal database id that cannot be
+# derived from the registration number. The WHO ICTRP portal resolves any
+# primary-registry id, so we route these through it.
+_WHO_RESOLVER_REGISTRIES = [
+    re.compile(r"^ChiCTR\S+$", re.I),
+    re.compile(r"^CTRI/\S+$", re.I),
+    re.compile(r"^IRCT\w+$", re.I),
+    re.compile(r"^PACTR\d+$", re.I),
+    re.compile(r"^(?:NTR|NL)\d+$", re.I),
+]
+
+
+def _normalize(value) -> str:
+    """Trim a raw cell to a string; ``None``/pandas ``NaN`` become ``""``."""
+    if value is None or (isinstance(value, float) and pd.isna(value)):
+        return ""
+    return str(value).strip()
+
+
+def build_source_link(name, source):
+    """Return the canonical per-trial source URL, or ``None`` if unavailable.
+
+    Prefers a link derived from ``name`` (the trial's authoritative
+    registration id); ``source`` is used only when ``name`` is unrecognized.
+    """
+    candidate = _normalize(name)
+
+    if candidate:
+        # 1. Native registry URL (id embedded in the registry page).
+        for pattern, build in _NATIVE_REGISTRIES:
+            if pattern.match(candidate):
+                return build(candidate)
+        # 2. Recognized registry without a derivable deep link -> WHO ICTRP.
+        for pattern in _WHO_RESOLVER_REGISTRIES:
+            if pattern.match(candidate):
+                return (
+                    f"https://trialsearch.who.int/?TrialID={quote(candidate, safe='')}"
+                )
+
+    # 3. Unrecognized id: keep the raw source only if it is already a URL.
+    src = _normalize(source)
+    if src.lower().startswith(("http://", "https://")):
+        return src
+
+    # 4. Nothing usable.
+    return None
+
+
 def _synthesize_phase(val) -> str:
     """Standardize clinical trial phase values."""
     if pd.isna(val) or val == "None":
@@ -289,6 +410,21 @@ def transform_clinical_trials(
     """
     df = df.copy()
 
+    # Build a reliable per-trial source link from the authoritative
+    # registration id (vin_name), before the raw columns are renamed/dropped.
+    # vin_source is consulted only as a fallback inside build_source_link.
+    names = (
+        df["vin_name"]
+        if "vin_name" in df.columns
+        else pd.Series([None] * len(df), index=df.index)
+    )
+    sources = (
+        df["vin_source"]
+        if "vin_source" in df.columns
+        else pd.Series([None] * len(df), index=df.index)
+    )
+    df["source_link"] = [build_source_link(n, s) for n, s in zip(names, sources)]
+
     # Strip whitespace from age column before synthesis
     if "new_age" in df.columns:
         df["new_age"] = df["new_age"].str.strip()

src/igh_data_transform/transformations/silver_to_gold/config/schema_map.py

@@ -279,7 +279,7 @@
         "locations": "locations",
         "age_groups": "age",
         "study_type": "studytype",
-        "source_text": "vin_source",
+        "source_text": "source_link",
         "description": "description",
         "ct_results_status": "OPTIONSET:ctresultsstatus|vin_ctresultsstatus",
         "end_date_key": "FK:dim_date.full_date|EXTRACT_DATE:enddate",

tests/unit/test_clinical_trials.py

@@ -9,6 +9,7 @@
     _synthesize_age_groups,
     _synthesize_gender,
     _synthesize_phase,
+    build_source_link,
     transform_clinical_trials,
 )
 
@@ -564,3 +565,110 @@ def test_works_when_option_sets_is_none(self):
         result, cleaned = transform_clinical_trials(df, option_sets=None)
         assert isinstance(result, pd.DataFrame)
         assert len(cleaned) == 0
+
+
+class TestTransformAddsSourceLink:
+    def test_source_link_prefers_registration_id_over_stale_source(self):
+        df = pd.DataFrame(
+            {
+                "vin_name": ["NCT04406727", "CTRI/2020/02/023129", "N/A"],
+                "vin_source": [
+                    # Stale shared URL from a bulk import — must be overridden.
+                    "https://clinicaltrials.gov/study/NCT04882514",
+                    "http://www.ctri.nic.in/Clinicaltrials/pmaindet2.php?trialid=9948",
+                    "https://example.org/fallback",
+                ],
+            }
+        )
+
+        out, _ = transform_clinical_trials(df)
+
+        links = out["source_link"].tolist()
+        assert links[0] == "https://clinicaltrials.gov/study/NCT04406727"
+        assert (
+            links[1] == "https://trialsearch.who.int/?TrialID=CTRI%2F2020%2F02%2F023129"
+        )
+        assert links[2] == "https://example.org/fallback"
+
+
+class TestBuildSourceLink:
+    """Per-trial source link construction from the registration id."""
+
+    @pytest.mark.parametrize(
+        "name,expected",
+        [
+            # Native registry templates (id embedded in the registry page).
+            ("NCT04882514", "https://clinicaltrials.gov/study/NCT04882514"),
+            ("ISRCTN71619711", "https://www.isrctn.com/ISRCTN71619711"),
+            ("ACTRN12615000264583", "https://anzctr.org.au/ACTRN12615000264583.aspx"),
+            (
+                "TCTR20210826004",
+                "https://www.thaiclinicaltrials.org/show/TCTR20210826004",
+            ),
+            ("DRKS00033539", "https://drks.de/search/en/trial/DRKS00033539"),
+            (
+                "jRCTs021190020",
+                "https://jrct.niph.go.jp/en-latest-detail/jRCTs021190020",
+            ),
+            ("SLCTR/2016/015", "https://slctr.lk/trials/slctr-2016-015"),
+            (
+                "2018-000283-28",
+                "https://www.clinicaltrialsregister.eu/ctr-search/search"
+                "?query=eudract_number:2018-000283-28",
+            ),
+            (
+                "2024-518527-29-00",
+                "https://euclinicaltrials.eu/search-for-clinical-trials/"
+                "?lang=en&EUCT=2024-518527-29-00",
+            ),
+        ],
+    )
+    def test_native_registry_templates(self, name, expected):
+        # vin_source is deliberately wrong/stale; vin_name must win.
+        assert (
+            build_source_link(name, "https://clinicaltrials.gov/study/NCT00000000")
+            == expected
+        )
+
+    @pytest.mark.parametrize(
+        "name,expected_id",
+        [
+            ("ChiCTR2500096097", "ChiCTR2500096097"),
+            ("CTRI/2020/02/023129", "CTRI%2F2020%2F02%2F023129"),
+            ("IRCT20240912063018N1", "IRCT20240912063018N1"),
+            ("PACTR202408671139802", "PACTR202408671139802"),
+            ("NL8933", "NL8933"),
+            ("NTR4751", "NTR4751"),
+        ],
+    )
+    def test_who_ictrp_resolver_for_non_deep_linkable_registries(
+        self, name, expected_id
+    ):
+        assert (
+            build_source_link(name, None)
+            == f"https://trialsearch.who.int/?TrialID={expected_id}"
+        )
+
+    def test_unrecognized_name_falls_back_to_source_url(self):
+        assert (
+            build_source_link("N/A", "https://example.org/trial/123")
+            == "https://example.org/trial/123"
+        )
+
+    def test_trailing_whitespace_is_trimmed(self):
+        assert build_source_link("CTRI/2020/02/023129 ", None) == (
+            "https://trialsearch.who.int/?TrialID=CTRI%2F2020%2F02%2F023129"
+        )
+
+    @pytest.mark.parametrize(
+        "name,source",
+        [
+            ("Unknown", "CT.gov"),  # junk name, non-URL source
+            ("N/A", None),  # junk name, no source
+            ("", ""),  # both blank
+            (None, None),  # both missing
+            (np.nan, np.nan),  # pandas NaN cells
+        ],
+    )
+    def test_no_usable_link_returns_none(self, name, source):
+        assert build_source_link(name, source) is None