1#!/usr/bin/env python32"""WP44 — Stage S1 record linkage: OpenAlex CS records ↔ DBLP ↔ arXiv ↔ ACL Anthology, in DuckDB.34Creates/refreshes <db> with views over the Parquet tables and the linkage tables5 link_dblp_oa(dblp_key, oa_id, method), link_arxiv_oa(arxiv_id, oa_id, method), link_acl_oa(acl_id, oa_id, method)6and writes data/metadata/linkage_report.json. Also writes the list of OpenAlex ids of DBLP-indexed works (2018–2026)7that are in the slim table but NOT in the CS-topic subset (needed for the second snapshot pass).8"""9import argparse, json, os, time, datetime as dt10import duckdb111213def main():14 ap = argparse.ArgumentParser(); ap.add_argument("--root", required=True); ap.add_argument("--db", default=None)15 a = ap.parse_args(); R = a.root; db = a.db or os.path.join(R, "data/processed/wp44.duckdb")16 con = duckdb.connect(db); con.execute("SET threads TO 24; SET memory_limit='60GB'; SET preserve_insertion_order=false;")17 T = os.path.join(R, "data/processed/tables"); t0 = time.time(); rep = {"created": dt.datetime.now(dt.UTC).isoformat()}18 def log(m): print(f"[{time.time()-t0:7.0f}s] {m}", flush=True)19 for t in ["works", "authorships", "work_topics", "work_keywords", "references", "abstracts", "locations", "grants", "counts_by_year"]:20 con.execute(f"CREATE OR REPLACE VIEW oa_{t} AS SELECT * FROM read_parquet('{T}/{t}/*.parquet')")21 con.execute(f"CREATE OR REPLACE VIEW oa_slim AS SELECT * FROM read_parquet('{R}/data/raw/openalex/works/slim/*.parquet')")22 con.execute(f"CREATE OR REPLACE VIEW arxiv_raw AS SELECT * FROM read_parquet('{R}/data/raw/arxiv/arxiv_cs.parquet')")23 con.execute(f"CREATE OR REPLACE VIEW acl_raw AS SELECT * FROM read_parquet('{R}/data/raw/acl/acl_anthology.parquet')")24 con.execute(f"CREATE OR REPLACE VIEW dblp_raw AS SELECT * FROM read_parquet('{R}/data/raw/dblp/dblp_publications.parquet')")25 log("views created")26 # --- normalised keys ---27 con.execute("""CREATE OR REPLACE TABLE oa_keys AS28 SELECT id AS oa_id, doi, lower(regexp_replace(arxiv_id, 'v[0-9]+$', '')) AS arxiv_id, publication_year AS year,29 regexp_replace(lower(title), '[^a-z0-9]+', '', 'g') AS ntitle, length(title) AS tlen, type30 FROM oa_works""")31 con.execute("""CREATE OR REPLACE TABLE dblp_keys AS32 SELECT key AS dblp_key, rec_type, year, doi,33 lower(regexp_replace(regexp_extract(array_to_string(ee, ' '), 'arxiv\\.org/abs/([^ ]+)', 1), 'v[0-9]+$', '')) AS arxiv_id,34 regexp_replace(lower(title), '[^a-z0-9]+', '', 'g') AS ntitle, length(title) AS tlen, stream35 FROM dblp_raw WHERE year BETWEEN 2018 AND 2026 AND rec_type IN ('article','inproceedings','incollection','book','phdthesis','data')""")36 con.execute("UPDATE dblp_keys SET arxiv_id = NULL WHERE arxiv_id = ''")37 con.execute("""CREATE OR REPLACE TABLE arxiv_keys AS38 SELECT lower(arxiv_id) AS arxiv_id, created, substr(created,1,4)::INT AS created_year, doi, primary_category, categories,39 regexp_replace(lower(title), '[^a-z0-9]+', '', 'g') AS ntitle FROM arxiv_raw""")40 con.execute("""CREATE OR REPLACE TABLE acl_keys AS41 SELECT acl_id, try_cast(year AS INT) AS year, lower(doi) AS doi, url, regexp_replace(lower(title), '[^a-z0-9]+', '', 'g') AS ntitle,42 entry_type FROM acl_raw WHERE entry_type IN ('inproceedings','article')""")43 log("keys built")44 # --- DBLP ↔ OpenAlex ---45 con.execute("""CREATE OR REPLACE TABLE link_dblp_oa AS46 SELECT d.dblp_key, o.oa_id, 'doi' AS method FROM dblp_keys d JOIN oa_keys o ON d.doi = o.doi WHERE d.doi IS NOT NULL""")47 con.execute("""INSERT INTO link_dblp_oa48 SELECT d.dblp_key, o.oa_id, 'arxiv' FROM dblp_keys d JOIN oa_keys o ON d.arxiv_id = o.arxiv_id49 WHERE d.arxiv_id IS NOT NULL AND o.arxiv_id IS NOT NULL AND d.dblp_key NOT IN (SELECT dblp_key FROM link_dblp_oa)50 AND o.oa_id NOT IN (SELECT oa_id FROM link_dblp_oa)""")51 # title+year: unique normalised titles on both sides, length >= 25, |year diff| <= 152 con.execute("""CREATE OR REPLACE TEMP TABLE d_un AS SELECT ntitle, any_value(dblp_key) dblp_key, any_value(year) year FROM dblp_keys53 WHERE tlen >= 25 AND dblp_key NOT IN (SELECT dblp_key FROM link_dblp_oa) GROUP BY ntitle HAVING count(*) = 1""")54 con.execute("""CREATE OR REPLACE TEMP TABLE o_un AS SELECT ntitle, any_value(oa_id) oa_id, any_value(year) year FROM oa_keys55 WHERE tlen >= 25 AND oa_id NOT IN (SELECT oa_id FROM link_dblp_oa) GROUP BY ntitle HAVING count(*) = 1""")56 con.execute("""INSERT INTO link_dblp_oa SELECT d.dblp_key, o.oa_id, 'title_year' FROM d_un d JOIN o_un o ON d.ntitle = o.ntitle57 WHERE abs(d.year - o.year) <= 1""")58 log("dblp linked")59 # --- arXiv ↔ OpenAlex ---60 con.execute("""CREATE OR REPLACE TABLE link_arxiv_oa AS61 SELECT x.arxiv_id, o.oa_id, 'arxiv_id' AS method FROM arxiv_keys x JOIN oa_keys o ON x.arxiv_id = o.arxiv_id""")62 con.execute("""INSERT INTO link_arxiv_oa SELECT x.arxiv_id, o.oa_id, 'doi' FROM arxiv_keys x JOIN oa_keys o ON x.doi = o.doi63 WHERE x.doi IS NOT NULL AND x.arxiv_id NOT IN (SELECT arxiv_id FROM link_arxiv_oa)""")64 con.execute("""CREATE OR REPLACE TEMP TABLE x_un AS SELECT ntitle, any_value(arxiv_id) arxiv_id, any_value(created_year) year FROM arxiv_keys65 WHERE length(ntitle) >= 25 AND arxiv_id NOT IN (SELECT arxiv_id FROM link_arxiv_oa) GROUP BY ntitle HAVING count(*) = 1""")66 con.execute("""CREATE OR REPLACE TEMP TABLE o_un2 AS SELECT ntitle, any_value(oa_id) oa_id, any_value(year) year FROM oa_keys67 WHERE tlen >= 25 GROUP BY ntitle HAVING count(*) = 1""")68 con.execute("""INSERT INTO link_arxiv_oa SELECT x.arxiv_id, o.oa_id, 'title_year' FROM x_un x JOIN o_un2 o ON x.ntitle = o.ntitle69 WHERE abs(x.year - o.year) <= 2 AND o.oa_id NOT IN (SELECT oa_id FROM link_arxiv_oa)""")70 log("arxiv linked")71 # --- ACL ↔ OpenAlex ---72 con.execute("""CREATE OR REPLACE TABLE link_acl_oa AS73 SELECT c.acl_id, o.oa_id, 'doi' AS method FROM acl_keys c JOIN oa_keys o ON c.doi = o.doi WHERE c.doi IS NOT NULL""")74 con.execute("""INSERT INTO link_acl_oa SELECT c.acl_id, l.work_id, 'url' FROM acl_keys c75 JOIN (SELECT DISTINCT work_id, regexp_extract(landing_page_url, 'aclanthology\\.org/([^/ ]+)', 1) AS aid FROM oa_locations76 WHERE landing_page_url LIKE '%aclanthology.org/%') l ON l.aid = c.acl_id OR l.aid = c.acl_id || '/'77 WHERE c.acl_id NOT IN (SELECT acl_id FROM link_acl_oa)""")78 con.execute("""INSERT INTO link_acl_oa SELECT c.acl_id, o.oa_id, 'title_year' FROM79 (SELECT ntitle, any_value(acl_id) acl_id, any_value(year) year FROM acl_keys WHERE length(ntitle) >= 25 AND acl_id NOT IN (SELECT acl_id FROM link_acl_oa) GROUP BY ntitle HAVING count(*)=1) c80 JOIN o_un2 o ON c.ntitle = o.ntitle WHERE abs(c.year - o.year) <= 1 AND o.oa_id NOT IN (SELECT oa_id FROM link_acl_oa)""")81 log("acl linked")82 # --- report ---83 q = lambda s: con.execute(s).fetchall()84 rep["oa_cs_records"] = q("SELECT count(*) FROM oa_keys")[0][0]85 rep["oa_cs_distinct_ids"] = q("SELECT count(DISTINCT oa_id) FROM oa_keys")[0][0]86 rep["dblp_2018_2026"] = q("SELECT count(*) FROM dblp_keys")[0][0]87 rep["dblp_with_doi"] = q("SELECT count(*) FROM dblp_keys WHERE doi IS NOT NULL")[0][0]88 rep["dblp_with_arxiv"] = q("SELECT count(*) FROM dblp_keys WHERE arxiv_id IS NOT NULL")[0][0]89 rep["link_dblp_oa"] = dict(q("SELECT method, count(*) FROM link_dblp_oa GROUP BY 1"))90 rep["dblp_linked_distinct"] = q("SELECT count(DISTINCT dblp_key) FROM link_dblp_oa")[0][0]91 rep["dblp_unlinked"] = rep["dblp_2018_2026"] - rep["dblp_linked_distinct"]92 rep["dblp_unlinked_by_type"] = dict(q("SELECT rec_type, count(*) FROM dblp_keys WHERE dblp_key NOT IN (SELECT dblp_key FROM link_dblp_oa) GROUP BY 1"))93 rep["dblp_unlinked_with_doi_in_slim"] = q("""SELECT count(*) FROM dblp_keys d JOIN oa_slim s ON d.doi = s.doi94 WHERE d.dblp_key NOT IN (SELECT dblp_key FROM link_dblp_oa)""")[0][0]95 rep["arxiv_cs_records"] = q("SELECT count(*) FROM arxiv_keys")[0][0]96 rep["arxiv_created_2018_2026"] = q("SELECT count(*) FROM arxiv_keys WHERE created_year BETWEEN 2018 AND 2026")[0][0]97 rep["link_arxiv_oa"] = dict(q("SELECT method, count(*) FROM link_arxiv_oa GROUP BY 1"))98 rep["arxiv_linked_distinct"] = q("SELECT count(DISTINCT arxiv_id) FROM link_arxiv_oa")[0][0]99 rep["acl_records"] = q("SELECT count(*) FROM acl_keys")[0][0]100 rep["acl_2018_2026"] = q("SELECT count(*) FROM acl_keys WHERE year BETWEEN 2018 AND 2026")[0][0]101 rep["link_acl_oa"] = dict(q("SELECT method, count(*) FROM link_acl_oa GROUP BY 1"))102 rep["acl_linked_distinct_2018_2026"] = q("SELECT count(DISTINCT l.acl_id) FROM link_acl_oa l JOIN acl_keys c USING(acl_id) WHERE c.year BETWEEN 2018 AND 2026")[0][0]103 # ids needed for the second snapshot pass: DBLP-indexed (2018–2026) works in slim but not in the CS subset; plus arXiv cs works in slim via DOI not in subset104 con.execute("""CREATE OR REPLACE TABLE needed_oa_ids AS105 SELECT DISTINCT s.id AS oa_id, 'dblp_doi' AS why FROM dblp_keys d JOIN oa_slim s ON d.doi = s.doi106 WHERE d.dblp_key NOT IN (SELECT dblp_key FROM link_dblp_oa) AND s.id NOT IN (SELECT oa_id FROM oa_keys)107 UNION108 SELECT DISTINCT s.id, 'arxiv_doi' FROM arxiv_keys x JOIN oa_slim s ON x.doi = s.doi109 WHERE x.doi IS NOT NULL AND x.arxiv_id NOT IN (SELECT arxiv_id FROM link_arxiv_oa) AND s.id NOT IN (SELECT oa_id FROM oa_keys)""")110 rep["needed_oa_ids_second_pass"] = dict(q("SELECT why, count(*) FROM needed_oa_ids GROUP BY 1"))111 rep["slim_total"] = q("SELECT count(*) FROM oa_slim")[0][0]112 rep["slim_cs_any_2018_2026"] = q("SELECT count(*) FROM oa_slim WHERE cs_any_topic AND publication_year BETWEEN 2018 AND 2026")[0][0]113 rep["seconds"] = round(time.time() - t0)114 os.makedirs(os.path.join(R, "data/metadata"), exist_ok=True)115 json.dump(rep, open(os.path.join(R, "data/metadata/linkage_report.json"), "w"), indent=1, default=str)116 con.execute(f"COPY (SELECT oa_id FROM needed_oa_ids) TO '{R}/data/metadata/needed_oa_ids.csv' (HEADER)")117 print(json.dumps(rep, indent=1, default=str))118119120if __name__ == "__main__":121 main()122