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='" + os.environ.get("WP44_DUCKDB_MEM", "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 srcs = [f"{R}/data/processed/compact/{t}.parquet"]21 if os.path.exists(f"{R}/data/processed/compact/{t}_p2.parquet"):22 srcs.append(f"{R}/data/processed/compact/{t}_p2.parquet")23 con.execute(f"CREATE OR REPLACE VIEW oa_{t} AS SELECT * FROM read_parquet({srcs!r}, union_by_name=true)")24 con.execute(f"CREATE OR REPLACE VIEW oa_slim AS SELECT * FROM read_parquet('{R}/data/processed/compact/slim_all_works.parquet')")25 con.execute(f"CREATE OR REPLACE VIEW arxiv_raw AS SELECT * FROM read_parquet('{R}/data/raw/arxiv/arxiv_cs.parquet')")26 con.execute(f"CREATE OR REPLACE VIEW acl_raw AS SELECT * FROM read_parquet('{R}/data/raw/acl/acl_anthology.parquet')")27 con.execute(f"CREATE OR REPLACE VIEW dblp_raw AS SELECT * FROM read_parquet('{R}/data/raw/dblp/dblp_publications.parquet')")28 log("views created")29 # --- normalised keys ---30 con.execute("""CREATE OR REPLACE TABLE oa_keys AS31 SELECT id AS oa_id, doi, lower(regexp_replace(arxiv_id, 'v[0-9]+$', '')) AS arxiv_id, publication_year AS year,32 regexp_replace(lower(title), '[^a-z0-9]+', '', 'g') AS ntitle, length(title) AS tlen, type33 FROM oa_works""")34 con.execute("""CREATE OR REPLACE TABLE dblp_keys AS35 SELECT key AS dblp_key, rec_type, year, doi,36 lower(regexp_replace(regexp_extract(array_to_string(ee, ' '), 'arxiv\\.org/abs/([^ ]+)', 1), 'v[0-9]+$', '')) AS arxiv_id,37 regexp_replace(lower(title), '[^a-z0-9]+', '', 'g') AS ntitle, length(title) AS tlen, stream38 FROM dblp_raw WHERE year BETWEEN 2018 AND 2026 AND rec_type IN ('article','inproceedings','incollection','book','phdthesis','data')""")39 con.execute("UPDATE dblp_keys SET arxiv_id = NULL WHERE arxiv_id = ''")40 con.execute("""CREATE OR REPLACE TABLE arxiv_keys AS41 SELECT lower(arxiv_id) AS arxiv_id, created, substr(created,1,4)::INT AS created_year, doi, primary_category, categories,42 regexp_replace(lower(title), '[^a-z0-9]+', '', 'g') AS ntitle FROM arxiv_raw""")43 con.execute("""CREATE OR REPLACE TABLE acl_keys AS44 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,45 entry_type FROM acl_raw WHERE entry_type IN ('inproceedings','article')""")46 log("keys built")47 # --- DBLP ↔ OpenAlex ---48 con.execute("""CREATE OR REPLACE TABLE link_dblp_oa AS49 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""")50 con.execute("""INSERT INTO link_dblp_oa51 SELECT d.dblp_key, o.oa_id, 'arxiv' FROM dblp_keys d JOIN oa_keys o ON d.arxiv_id = o.arxiv_id52 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)53 AND o.oa_id NOT IN (SELECT oa_id FROM link_dblp_oa)""")54 # title+year: unique normalised titles on both sides, length >= 25, |year diff| <= 155 con.execute("""CREATE OR REPLACE TEMP TABLE d_un AS SELECT ntitle, any_value(dblp_key) AS dblp_key, any_value(year) AS year FROM dblp_keys56 WHERE tlen >= 25 AND dblp_key NOT IN (SELECT dblp_key FROM link_dblp_oa) GROUP BY ntitle HAVING count(*) = 1""")57 con.execute("""CREATE OR REPLACE TEMP TABLE o_un AS SELECT ntitle, any_value(oa_id) AS oa_id, any_value(year) AS year FROM oa_keys58 WHERE tlen >= 25 AND oa_id NOT IN (SELECT oa_id FROM link_dblp_oa) GROUP BY ntitle HAVING count(*) = 1""")59 con.execute("""CREATE OR REPLACE TEMP TABLE d_fa AS SELECT key AS dblp_key, lower(regexp_extract(authors[1], '([^ ]+)$', 1)) AS fa FROM dblp_raw WHERE len(authors) > 0""")60 con.execute("""CREATE OR REPLACE TEMP TABLE o_fa AS SELECT work_id AS oa_id, lower(regexp_extract(author_name, '([^ ]+)$', 1)) AS fa FROM oa_authorships WHERE position = 0""")61 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.ntitle62 LEFT JOIN d_fa ON d_fa.dblp_key = d.dblp_key LEFT JOIN o_fa ON o_fa.oa_id = o.oa_id63 WHERE abs(d.year - o.year) <= 1 AND (d_fa.fa IS NULL OR o_fa.fa IS NULL OR d_fa.fa = o_fa.fa)""")64 log("dblp linked")65 # --- arXiv ↔ OpenAlex ---66 con.execute("""CREATE OR REPLACE TABLE link_arxiv_oa AS67 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""")68 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.doi69 WHERE x.doi IS NOT NULL AND x.arxiv_id NOT IN (SELECT arxiv_id FROM link_arxiv_oa)""")70 con.execute("""CREATE OR REPLACE TEMP TABLE x_un AS SELECT ntitle, any_value(arxiv_id) AS arxiv_id, any_value(created_year) AS year FROM arxiv_keys71 WHERE length(ntitle) >= 25 AND arxiv_id NOT IN (SELECT arxiv_id FROM link_arxiv_oa) GROUP BY ntitle HAVING count(*) = 1""")72 con.execute("""CREATE OR REPLACE TEMP TABLE o_un2 AS SELECT ntitle, any_value(oa_id) AS oa_id, any_value(year) AS year FROM oa_keys73 WHERE tlen >= 25 GROUP BY ntitle HAVING count(*) = 1""")74 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.ntitle75 WHERE abs(x.year - o.year) <= 2 AND o.oa_id NOT IN (SELECT oa_id FROM link_arxiv_oa)""")76 log("arxiv linked")77 # --- ACL ↔ OpenAlex ---78 con.execute("""CREATE OR REPLACE TABLE link_acl_oa AS79 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""")80 con.execute("""INSERT INTO link_acl_oa SELECT c.acl_id, l.work_id, 'url' FROM acl_keys c81 JOIN (SELECT DISTINCT work_id, regexp_extract(landing_page_url, 'aclanthology\\.org/([^/ ]+)', 1) AS aid FROM oa_locations82 WHERE landing_page_url LIKE '%aclanthology.org/%') l ON l.aid = c.acl_id OR l.aid = c.acl_id || '/'83 WHERE c.acl_id NOT IN (SELECT acl_id FROM link_acl_oa)""")84 con.execute("""INSERT INTO link_acl_oa SELECT c.acl_id, o.oa_id, 'title_year' FROM85 (SELECT ntitle, any_value(acl_id) AS acl_id, any_value(year) AS 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) c86 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)""")87 log("acl linked")88 # --- report ---89 q = lambda s: con.execute(s).fetchall()90 rep["oa_cs_records"] = q("SELECT count(*) FROM oa_keys")[0][0]91 rep["oa_cs_distinct_ids"] = q("SELECT count(DISTINCT oa_id) FROM oa_keys")[0][0]92 rep["dblp_2018_2026"] = q("SELECT count(*) FROM dblp_keys")[0][0]93 rep["dblp_with_doi"] = q("SELECT count(*) FROM dblp_keys WHERE doi IS NOT NULL")[0][0]94 rep["dblp_with_arxiv"] = q("SELECT count(*) FROM dblp_keys WHERE arxiv_id IS NOT NULL")[0][0]95 rep["link_dblp_oa"] = {"doi": 0, "arxiv": 0, "title_year": 0, **dict(q("SELECT method, count(*) FROM link_dblp_oa GROUP BY 1"))}96 rep["dblp_linked_distinct"] = q("SELECT count(DISTINCT dblp_key) FROM link_dblp_oa")[0][0]97 rep["dblp_unlinked"] = rep["dblp_2018_2026"] - rep["dblp_linked_distinct"]98 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"))99 rep["dblp_unlinked_with_doi_in_slim"] = q("""SELECT count(*) FROM dblp_keys d JOIN oa_slim s ON d.doi = s.doi100 WHERE d.dblp_key NOT IN (SELECT dblp_key FROM link_dblp_oa)""")[0][0]101 rep["arxiv_cs_records"] = q("SELECT count(*) FROM arxiv_keys")[0][0]102 rep["arxiv_created_2018_2026"] = q("SELECT count(*) FROM arxiv_keys WHERE created_year BETWEEN 2018 AND 2026")[0][0]103 rep["link_arxiv_oa"] = {"arxiv_id": 0, "doi": 0, "title_year": 0, **dict(q("SELECT method, count(*) FROM link_arxiv_oa GROUP BY 1"))}104 rep["arxiv_linked_distinct"] = q("SELECT count(DISTINCT arxiv_id) FROM link_arxiv_oa")[0][0]105 rep["acl_records"] = q("SELECT count(*) FROM acl_keys")[0][0]106 rep["acl_2018_2026"] = q("SELECT count(*) FROM acl_keys WHERE year BETWEEN 2018 AND 2026")[0][0]107 rep["link_acl_oa"] = dict(q("SELECT method, count(*) FROM link_acl_oa GROUP BY 1"))108 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]109 # 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 subset110 con.execute("""CREATE OR REPLACE TABLE needed_oa_ids AS111 SELECT DISTINCT s.id AS oa_id, 'dblp_doi' AS why FROM dblp_keys d JOIN oa_slim s ON d.doi = s.doi112 WHERE d.dblp_key NOT IN (SELECT dblp_key FROM link_dblp_oa) AND s.id NOT IN (SELECT oa_id FROM oa_keys)113 UNION114 SELECT DISTINCT s.id, 'arxiv_doi' FROM arxiv_keys x JOIN oa_slim s ON x.doi = s.doi115 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)""")116 rep["needed_oa_ids_second_pass"] = {"dblp_doi": 0, "arxiv_doi": 0, **dict(q("SELECT why, count(*) FROM needed_oa_ids GROUP BY 1"))}117 rep["slim_total"] = q("SELECT count(*) FROM oa_slim")[0][0]118 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]119 rep["seconds"] = round(time.time() - t0)120 os.makedirs(os.path.join(R, "data/metadata"), exist_ok=True)121 json.dump(rep, open(os.path.join(R, "data/metadata/linkage_report.json"), "w"), indent=1, default=str)122 con.execute(f"COPY (SELECT oa_id FROM needed_oa_ids) TO '{R}/data/metadata/needed_oa_ids.csv' (HEADER)")123 print(json.dumps(rep, indent=1, default=str))124125126if __name__ == "__main__":127 main()128