1#!/usr/bin/env python32"""WP44 — Stages S2–S3: deduplicate (preprint ↔ published), apply eligibility, define the mapping corpus, write PRISMA counts.34Inputs: DuckDB views/tables made by ingestion/link_sources.py (oa_* views, dblp_keys, arxiv_keys, link_* tables).5Outputs: table `corpus_works` (+ Parquet copy data/processed/corpus/corpus_works.parquet), `dblp_only_counts`,6data/metadata/prisma_counts.json. Rules: protocol/PROTOCOL.md §P2; decisions logged in RESEARCH_PLAN §10.7"""8import argparse, json, os, time, datetime as dt9import duckdb1011KEEP_TYPES = "('article','conference-paper','preprint','book-chapter','book','dissertation','report','review','software-paper')"12# hard exclusions even for DBLP-linked records (DBLP-linked records of any other type are kept: DBLP is the CS gold standard)13# generic deposit platforms (A-002): records whose primary source is one of these are kept only if DBLP- or arXiv-linked14GENERIC_DEPOSIT_SOURCES = "('Zenodo (CERN European Organization for Nuclear Research)','Figshare','Institutional Repositories DataBase (IRDB)','LA Referencia (Red Federada de Repositorios Institucionales de Publicaciones Científicas)','Dialnet (Universidad de la Rioja)','DOAJ (DOAJ: Directory of Open Access Journals)','OSTI OAI (U.S. Department of Energy Office of Scientific and Technical Information)','Zenodo')"15HARD_EXCLUDE_TYPES = "('editorial','erratum','retraction','peer-review','paratext','conference-abstract','reference-entry','dataset','software','supplementary-materials','libguides','letter','grant','standard')"16WINDOW_START, BOUNDARY, WINDOW_END = "2018-01-01", "2022-11-30", "2026-11-02"171819def main():20 ap = argparse.ArgumentParser(); ap.add_argument("--root", required=True); a = ap.parse_args(); R = a.root21 con = duckdb.connect(os.path.join(R, "data/processed/wp44.duckdb")); con.execute("SET threads TO 24; SET memory_limit='" + os.environ.get("WP44_DUCKDB_MEM", "60GB") + "';")22 t0 = time.time(); q = lambda s: con.execute(s).fetchall(); n = lambda s: q(s)[0][0]23 log = lambda m: print(f"[{time.time()-t0:6.0f}s] {m}", flush=True)24 # ---- S0/S1: base table with membership flags ----25 con.execute(f"""CREATE OR REPLACE TABLE cw0 AS26 SELECT w.*, regexp_replace(lower(w.title), '[^a-z0-9]+', '', 'g') AS ntitle,27 (w.id IN (SELECT oa_id FROM link_dblp_oa)) AS in_dblp,28 (w.id IN (SELECT oa_id FROM link_acl_oa)) AS in_acl,29 x.arxiv_id AS arxiv_id_linked, x.created AS arxiv_created, x.primary_category AS arxiv_primary,30 (x.primary_category LIKE 'cs.%') AS arxiv_cs_primary,31 coalesce(w.doi LIKE '10.48550/arxiv.%', FALSE) AS is_arxiv_doi32 FROM oa_works w33 LEFT JOIN (SELECT l.oa_id, any_value(k.arxiv_id) arxiv_id, min(k.created) created, any_value(k.primary_category) primary_category34 FROM link_arxiv_oa l JOIN arxiv_keys k USING(arxiv_id) GROUP BY l.oa_id) x ON x.oa_id = w.id""")35 log(f"cw0 {n('SELECT count(*) FROM cw0'):,}")36 # ---- S2: preprint ↔ published duplicates (same normalised title, |year| <= 2, one preprint-like, one not) ----37 # published twins must themselves pass E1–E4 (otherwise the preprint stays as the canonical record)38 con.execute(f"""CREATE OR REPLACE TABLE dup_pairs AS39 WITH pre AS (SELECT id, ntitle, publication_year y, publication_date d, coalesce(arxiv_id_linked, arxiv_id) AS ax FROM cw0 WHERE (coalesce(type,'') = 'preprint' OR is_arxiv_doi) AND length(ntitle) >= 25),40 pub AS (SELECT id, ntitle, publication_year y, publication_date d, coalesce(arxiv_id_linked, arxiv_id) AS ax FROM cw041 WHERE coalesce(type,'') <> 'preprint' AND NOT is_arxiv_doi AND length(ntitle) >= 2542 AND publication_date BETWEEN '{WINDOW_START}' AND '{WINDOW_END}' AND NOT is_xpac AND title IS NOT NULL43 AND (coalesce(type,'') IN {KEEP_TYPES} OR in_dblp)),44 cand AS (SELECT p.id AS preprint_id, b.id AS published_id, p.d AS preprint_date, b.d AS published_date, 'title_year' AS relation45 FROM pre p JOIN pub b ON p.ntitle = b.ntitle AND abs(p.y - b.y) <= 246 UNION47 SELECT p.id, b.id, p.d, b.d, 'arxiv_id' FROM pre p JOIN pub b ON p.ax = b.ax AND p.ax IS NOT NULL AND p.id <> b.id)48 SELECT *, row_number() OVER (PARTITION BY published_id ORDER BY preprint_date, preprint_id) AS rn_pub,49 row_number() OVER (PARTITION BY preprint_id ORDER BY published_date, published_id) AS rn_pre FROM cand""")50 con.execute("DELETE FROM dup_pairs WHERE rn_pub > 1 OR rn_pre > 1")51 log(f"duplicate preprint→published pairs: {n('SELECT count(*) FROM dup_pairs'):,}")52 # ---- S3: eligibility flags ----53 con.execute(f"""CREATE OR REPLACE TABLE corpus_works AS54 SELECT c.*,55 dp.published_id AS duplicate_of,56 dp2.preprint_id AS preprint_work_id,57 -- first public appearance: precise dates win over year-only (YYYY-01-01) publication dates58 CASE WHEN substr(c.publication_date, 6, 5) = '01-01' AND coalesce(dp2.preprint_date, c.arxiv_created) IS NOT NULL59 THEN least(coalesce(dp2.preprint_date, c.arxiv_created), coalesce(c.arxiv_created, dp2.preprint_date))60 ELSE least(c.publication_date, coalesce(dp2.preprint_date, c.publication_date), coalesce(c.arxiv_created, c.publication_date)) END AS first_date,61 dp2.relation AS version_relationship,62 (c.publication_date IS NULL OR c.publication_date < '{WINDOW_START}' OR c.publication_date > '{WINDOW_END}') AS e1_outside_window,63 (coalesce(c.type, '') NOT IN {KEEP_TYPES} AND NOT (c.in_dblp AND coalesce(c.type, '') NOT IN {HARD_EXCLUDE_TYPES})) AS e2_type,64 (c.is_xpac AND NOT c.in_dblp) AS e3_xpac,65 (c.is_xpac AND c.in_dblp) AS xpac_kept_dblp,66 (coalesce(c.source_name, '') IN {GENERIC_DEPOSIT_SOURCES} AND NOT c.in_dblp AND c.arxiv_id_linked IS NULL) AS e7_generic_deposit,67 (CASE WHEN substr(c.publication_date, 6, 5) = '01-01' AND coalesce(dp2.preprint_date, c.arxiv_created) IS NULL THEN 'year' ELSE 'month' END) AS date_precision,68 (c.title IS NULL OR length(trim(c.title)) = 0) AS e4_no_title,69 (dp.published_id IS NOT NULL) AS e5_duplicate,70 c.is_retracted AS e6_retracted,71 -- A-006: an arXiv primary category cs.* is itself a CS membership signal (works recovered by the second pass have no CS topic)72 (c.in_dblp OR c.cs_primary OR coalesce(c.arxiv_cs_primary, FALSE)) AS cs_core,73 (c.cs_any AND NOT (c.in_dblp OR c.cs_primary OR coalesce(c.arxiv_cs_primary, FALSE))) AS cs_penumbra74 FROM cw0 c75 LEFT JOIN dup_pairs dp ON dp.preprint_id = c.id76 LEFT JOIN dup_pairs dp2 ON dp2.published_id = c.id""")77 con.execute(f"""ALTER TABLE corpus_works ADD COLUMN eligible BOOLEAN; UPDATE corpus_works SET eligible =78 NOT e1_outside_window AND NOT e2_type AND NOT e3_xpac AND NOT e4_no_title AND NOT e5_duplicate AND NOT e7_generic_deposit AND cs_core""")79 con.execute(f"""ALTER TABLE corpus_works ADD COLUMN period VARCHAR; UPDATE corpus_works SET period = CASE80 WHEN first_date < '{WINDOW_START}' THEN 'pre' WHEN first_date < '{BOUNDARY}' THEN 'baseline' WHEN first_date <= '{WINDOW_END}' THEN 'post' ELSE 'future' END""")81 con.execute("ALTER TABLE corpus_works ADD COLUMN first_month VARCHAR; UPDATE corpus_works SET first_month = substr(first_date, 1, 7)")82 assert n("SELECT count(*) FROM corpus_works WHERE eligible IS NULL") == 0, "NULL eligibility flags"83 assert n("SELECT count(*) FROM corpus_works WHERE first_date IS NULL AND publication_date IS NOT NULL") == 0, "NULL first_date"84 log("corpus_works built")85 # ---- DBLP-only records (no OpenAlex record after both passes) ----86 con.execute("""CREATE OR REPLACE TABLE dblp_only_counts AS87 SELECT year, rec_type, count(*) n, sum(CASE WHEN doi IS NOT NULL THEN 1 ELSE 0 END) with_doi FROM dblp_keys88 WHERE dblp_key NOT IN (SELECT dblp_key FROM link_dblp_oa) GROUP BY 1,2 ORDER BY 1,2""")89 # ---- PRISMA counts ----90 P = {"generated": dt.datetime.now(dt.UTC).isoformat(), "window": [WINDOW_START, WINDOW_END], "boundary": BOUNDARY}91 P["identified"] = {92 "openalex_pass1_cs_topic_filter": n(f"SELECT count(*) FROM read_parquet('{R}/data/processed/compact/works.parquet')"),93 "openalex_pass2_dblp_arxiv_linked": n(f"SELECT count(*) FROM read_parquet('{R}/data/processed/compact/works_p2.parquet')") if os.path.exists(f"{R}/data/processed/compact/works_p2.parquet") else 0,94 "openalex_cs_topic_2018_2026_pass1": n("SELECT count(*) FROM oa_keys"),95 "openalex_total_records_loaded": n("SELECT count(*) FROM cw0"),96 "dblp_2018_2026": n("SELECT count(*) FROM dblp_keys"), "arxiv_cs_2018_2026": n("SELECT count(*) FROM arxiv_keys WHERE created_year BETWEEN 2018 AND 2026"),97 "acl_2018_2026": n("SELECT count(*) FROM acl_keys WHERE year BETWEEN 2018 AND 2026")}98 P["linked"] = {"dblp_to_openalex": n("SELECT count(DISTINCT dblp_key) FROM link_dblp_oa"), "arxiv_to_openalex": n("SELECT count(DISTINCT arxiv_id) FROM link_arxiv_oa"),99 "acl_to_openalex": n("SELECT count(DISTINCT acl_id) FROM link_acl_oa"), "dblp_only_no_openalex_record": n("SELECT sum(n) FROM dblp_only_counts")}100 P["excluded"] = {k: n(f"SELECT count(*) FROM corpus_works WHERE {k}") for k in ["e1_outside_window", "e2_type", "e3_xpac", "e4_no_title", "e5_duplicate", "e7_generic_deposit", "e6_retracted", "xpac_kept_dblp"]}101 P["excluded_e2_by_type"] = dict(q("SELECT type, count(*) FROM corpus_works WHERE e2_type GROUP BY 1 ORDER BY 2 DESC"))102 P["excluded_sequential"] = {}103 cond = "TRUE"104 for k in ["e1_outside_window", "e2_type", "e3_xpac", "e4_no_title", "e5_duplicate", "e7_generic_deposit"]:105 P["excluded_sequential"][k] = n(f"SELECT count(*) FROM corpus_works WHERE {cond} AND {k}"); cond += f" AND NOT {k}"106 P["after_exclusions"] = n(f"SELECT count(*) FROM corpus_works WHERE {cond}")107 assert P["identified"]["openalex_total_records_loaded"] == sum(P["excluded_sequential"].values()) + P["after_exclusions"], "PRISMA identity violated"108 P["version_relationship"] = dict(q("SELECT version_relationship, count(*) FROM corpus_works WHERE eligible AND version_relationship IS NOT NULL GROUP BY 1"))109 P["cs_core_eligible"] = n("SELECT count(*) FROM corpus_works WHERE eligible")110 P["cs_penumbra_eligible_otherwise"] = n(f"SELECT count(*) FROM corpus_works WHERE {cond} AND cs_penumbra")111 P["neither_core_nor_penumbra_after_exclusions"] = n(f"SELECT count(*) FROM corpus_works WHERE {cond} AND NOT cs_core AND NOT cs_penumbra")112 assert P["after_exclusions"] == P["cs_core_eligible"] + P["cs_penumbra_eligible_otherwise"] + P["neither_core_nor_penumbra_after_exclusions"], "core/penumbra partition violated"113 P["eligible_by_period"] = dict(q("SELECT period, count(*) FROM corpus_works WHERE eligible GROUP BY 1"))114 P["eligible_by_year"] = dict(q("SELECT publication_year, count(*) FROM corpus_works WHERE eligible GROUP BY 1 ORDER BY 1"))115 P["eligible_by_type"] = dict(q("SELECT type, count(*) FROM corpus_works WHERE eligible GROUP BY 1 ORDER BY 2 DESC"))116 P["eligible_membership"] = {"in_dblp": n("SELECT count(*) FROM corpus_works WHERE eligible AND in_dblp"), "cs_primary": n("SELECT count(*) FROM corpus_works WHERE eligible AND cs_primary"),117 "arxiv_linked": n("SELECT count(*) FROM corpus_works WHERE eligible AND arxiv_id_linked IS NOT NULL"), "in_acl": n("SELECT count(*) FROM corpus_works WHERE eligible AND in_acl"),118 "has_abstract": n("SELECT count(*) FROM corpus_works WHERE eligible AND has_abstract"), "has_references": n("SELECT count(*) FROM corpus_works WHERE eligible AND n_refs > 0"),119 "retracted_flagged": n("SELECT count(*) FROM corpus_works WHERE eligible AND e6_retracted")}120 P["eligible_date_precision"] = dict(q("SELECT date_precision, count(*) FROM corpus_works WHERE eligible GROUP BY 1"))121 P["strict_core_dblp_and_cs_primary"] = n("SELECT count(*) FROM corpus_works WHERE eligible AND in_dblp AND cs_primary")122 os.makedirs(os.path.join(R, "data/processed/corpus"), exist_ok=True); os.makedirs(os.path.join(R, "data/metadata"), exist_ok=True)123 con.execute(f"COPY corpus_works TO '{R}/data/processed/corpus/corpus_works.parquet' (FORMAT PARQUET, COMPRESSION ZSTD)")124 con.execute(f"COPY dblp_only_counts TO '{R}/data/metadata/dblp_only_counts.csv' (HEADER)")125 json.dump(P, open(os.path.join(R, "data/metadata/prisma_counts.json"), "w"), indent=1, default=str)126 print(json.dumps(P, indent=1, default=str)); log("done")127128129if __name__ == "__main__":130 main()131