1#!/usr/bin/env python32"""WP44 — live verification of the landmark records against the OpenAlex API (single-record lookups are free):3for each landmark id, fetch the live record, compare the normalised title with the snapshot title, and record the live4cited_by_count and publication_year. Records whose live title differs (ids merged or re-assigned since the snapshot, or5snapshot data errors) are flagged live_match = False and excluded from Part IX landmark statistics (they stay in the file).6Output: literature_synthesis/landmarks/landmarks_verified.csv + verification_report.json. Politeness: 8 requests/s max."""7import argparse, json, os, re, time, datetime as dt, unicodedata, requests, pandas as pd89UA = "WP44-UQO-systematic-mapping/0.1 (mailto:spbou4@icloud.com)"10norm = lambda s: re.sub(r"[^a-z0-9]+", "", unicodedata.normalize("NFKD", str(s or "")).encode("ascii", "ignore").decode().lower())111213def main():14 ap = argparse.ArgumentParser(); ap.add_argument("--root", required=True); a = ap.parse_args(); R = a.root15 d = pd.read_csv(f"{R}/literature_synthesis/landmarks/landmarks.csv"); s = requests.Session(); rows = []16 for i, r in enumerate(d.itertuples()):17 for attempt in range(3):18 try:19 resp = s.get(f"https://api.openalex.org/works/W{int(r.work_id)}", params={"mailto": "spbou4@icloud.com"}, headers={"User-Agent": UA}, timeout=30); break20 except requests.RequestException:21 time.sleep(5)22 if resp.status_code == 200:23 j = resp.json(); lt = j.get("title") or j.get("display_name") or ""; lid = j.get("id", "").rsplit("W", 1)[-1]24 match = norm(lt) == norm(r.title) or (norm(lt) and norm(r.title) and (norm(lt) in norm(r.title) or norm(r.title) in norm(lt)))25 rows.append({"work_id": int(r.work_id), "live_status": 200, "live_id": lid, "live_title": lt, "live_year": j.get("publication_year"), "live_cited_by_count": j.get("cited_by_count"), "live_match": bool(match), "id_changed": lid != str(int(r.work_id))})26 else:27 rows.append({"work_id": int(r.work_id), "live_status": resp.status_code, "live_id": None, "live_title": None, "live_year": None, "live_cited_by_count": None, "live_match": False, "id_changed": None})28 time.sleep(0.13)29 if (i + 1) % 200 == 0:30 print(i + 1, "checked", flush=True)31 v = d.merge(pd.DataFrame(rows), on="work_id", how="left"); v.to_csv(f"{R}/literature_synthesis/landmarks/landmarks_verified.csv", index=False)32 rep = {"n": int(len(v)), "live_match": int(v.live_match.sum()), "mismatch": int((~v.live_match).sum()), "id_changed": int(v.id_changed.fillna(False).sum()), "http_not_200": int((v.live_status != 200).sum()),33 "when": dt.datetime.now(dt.UTC).isoformat(), "mismatch_examples": v[~v.live_match][["work_id", "title", "live_title", "c36"]].head(15).to_dict("records")}34 json.dump(rep, open(f"{R}/literature_synthesis/landmarks/verification_report.json", "w"), indent=1, default=str); print({k: rep[k] for k in ("n", "live_match", "mismatch", "id_changed", "http_not_200")})353637if __name__ == "__main__":38 main()39