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 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("""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.ntitle60 WHERE abs(d.year - o.year) <= 1""")61 log("dblp linked")62 # --- arXiv ↔ OpenAlex ---63 con.execute("""CREATE OR REPLACE TABLE link_arxiv_oa AS64 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""")65 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.doi66 WHERE x.doi IS NOT NULL AND x.arxiv_id NOT IN (SELECT arxiv_id FROM link_arxiv_oa)""")67 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_keys68 WHERE length(ntitle) >= 25 AND arxiv_id NOT IN (SELECT arxiv_id FROM link_arxiv_oa) GROUP BY ntitle HAVING count(*) = 1""")69 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_keys70 WHERE tlen >= 25 GROUP BY ntitle HAVING count(*) = 1""")71 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.ntitle72 WHERE abs(x.year - o.year) <= 2 AND o.oa_id NOT IN (SELECT oa_id FROM link_arxiv_oa)""")73 log("arxiv linked")74 # --- ACL ↔ OpenAlex ---75 con.execute("""CREATE OR REPLACE TABLE link_acl_oa AS76 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""")77 con.execute("""INSERT INTO link_acl_oa SELECT c.acl_id, l.work_id, 'url' FROM acl_keys c78 JOIN (SELECT DISTINCT work_id, regexp_extract(landing_page_url, 'aclanthology\\.org/([^/ ]+)', 1) AS aid FROM oa_locations79 WHERE landing_page_url LIKE '%aclanthology.org/%') l ON l.aid = c.acl_id OR l.aid = c.acl_id || '/'80 WHERE c.acl_id NOT IN (SELECT acl_id FROM link_acl_oa)""")81 con.execute("""INSERT INTO link_acl_oa SELECT c.acl_id, o.oa_id, 'title_year' FROM82 (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) c83 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)""")84 log("acl linked")85 # --- report ---86 q = lambda s: con.execute(s).fetchall()87 rep["oa_cs_records"] = q("SELECT count(*) FROM oa_keys")[0][0]88 rep["oa_cs_distinct_ids"] = q("SELECT count(DISTINCT oa_id) FROM oa_keys")[0][0]89 rep["dblp_2018_2026"] = q("SELECT count(*) FROM dblp_keys")[0][0]90 rep["dblp_with_doi"] = q("SELECT count(*) FROM dblp_keys WHERE doi IS NOT NULL")[0][0]91 rep["dblp_with_arxiv"] = q("SELECT count(*) FROM dblp_keys WHERE arxiv_id IS NOT NULL")[0][0]92 rep["link_dblp_oa"] = dict(q("SELECT method, count(*) FROM link_dblp_oa GROUP BY 1"))93 rep["dblp_linked_distinct"] = q("SELECT count(DISTINCT dblp_key) FROM link_dblp_oa")[0][0]94 rep["dblp_unlinked"] = rep["dblp_2018_2026"] - rep["dblp_linked_distinct"]95 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"))96 rep["dblp_unlinked_with_doi_in_slim"] = q("""SELECT count(*) FROM dblp_keys d JOIN oa_slim s ON d.doi = s.doi97 WHERE d.dblp_key NOT IN (SELECT dblp_key FROM link_dblp_oa)""")[0][0]98 rep["arxiv_cs_records"] = q("SELECT count(*) FROM arxiv_keys")[0][0]99 rep["arxiv_created_2018_2026"] = q("SELECT count(*) FROM arxiv_keys WHERE created_year BETWEEN 2018 AND 2026")[0][0]100 rep["link_arxiv_oa"] = dict(q("SELECT method, count(*) FROM link_arxiv_oa GROUP BY 1"))101 rep["arxiv_linked_distinct"] = q("SELECT count(DISTINCT arxiv_id) FROM link_arxiv_oa")[0][0]102 rep["acl_records"] = q("SELECT count(*) FROM acl_keys")[0][0]103 rep["acl_2018_2026"] = q("SELECT count(*) FROM acl_keys WHERE year BETWEEN 2018 AND 2026")[0][0]104 rep["link_acl_oa"] = dict(q("SELECT method, count(*) FROM link_acl_oa GROUP BY 1"))105 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]106 # 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 subset107 con.execute("""CREATE OR REPLACE TABLE needed_oa_ids AS108 SELECT DISTINCT s.id AS oa_id, 'dblp_doi' AS why FROM dblp_keys d JOIN oa_slim s ON d.doi = s.doi109 WHERE d.dblp_key NOT IN (SELECT dblp_key FROM link_dblp_oa) AND s.id NOT IN (SELECT oa_id FROM oa_keys)110 UNION111 SELECT DISTINCT s.id, 'arxiv_doi' FROM arxiv_keys x JOIN oa_slim s ON x.doi = s.doi112 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)""")113 rep["needed_oa_ids_second_pass"] = dict(q("SELECT why, count(*) FROM needed_oa_ids GROUP BY 1"))114 rep["slim_total"] = q("SELECT count(*) FROM oa_slim")[0][0]115 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]116 rep["seconds"] = round(time.time() - t0)117 os.makedirs(os.path.join(R, "data/metadata"), exist_ok=True)118 json.dump(rep, open(os.path.join(R, "data/metadata/linkage_report.json"), "w"), indent=1, default=str)119 con.execute(f"COPY (SELECT oa_id FROM needed_oa_ids) TO '{R}/data/metadata/needed_oa_ids.csv' (HEADER)")120 print(json.dumps(rep, indent=1, default=str))121122123if __name__ == "__main__":124 main()125