#!/usr/bin/env python3 """WP44 — Stages S2–S3: deduplicate (preprint ↔ published), apply eligibility, define the mapping corpus, write PRISMA counts. Inputs: DuckDB views/tables made by ingestion/link_sources.py (oa_* views, dblp_keys, arxiv_keys, link_* tables). Outputs: table `corpus_works` (+ Parquet copy data/processed/corpus/corpus_works.parquet), `dblp_only_counts`, data/metadata/prisma_counts.json. Rules: protocol/PROTOCOL.md §P2; decisions logged in RESEARCH_PLAN §10. """ import argparse, json, os, time, datetime as dt import duckdb KEEP_TYPES = "('article','conference-paper','preprint','book-chapter','book','dissertation','report','review','software-paper')" # hard exclusions even for DBLP-linked records (DBLP-linked records of any other type are kept: DBLP is the CS gold standard) # generic deposit platforms (A-002): records whose primary source is one of these are kept only if DBLP- or arXiv-linked GENERIC_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')" HARD_EXCLUDE_TYPES = "('editorial','erratum','retraction','peer-review','paratext','conference-abstract','reference-entry','dataset','software','supplementary-materials','libguides','letter','grant','standard')" WINDOW_START, BOUNDARY, WINDOW_END = "2018-01-01", "2022-11-30", "2026-11-02" def main(): ap = argparse.ArgumentParser(); ap.add_argument("--root", required=True); a = ap.parse_args(); R = a.root con = duckdb.connect(os.path.join(R, "data/processed/wp44.duckdb")); con.execute("SET threads TO 24; SET memory_limit='60GB';") t0 = time.time(); q = lambda s: con.execute(s).fetchall(); n = lambda s: q(s)[0][0] log = lambda m: print(f"[{time.time()-t0:6.0f}s] {m}", flush=True) # ---- S0/S1: base table with membership flags ---- con.execute(f"""CREATE OR REPLACE TABLE cw0 AS SELECT w.*, regexp_replace(lower(w.title), '[^a-z0-9]+', '', 'g') AS ntitle, (w.id IN (SELECT oa_id FROM link_dblp_oa)) AS in_dblp, (w.id IN (SELECT oa_id FROM link_acl_oa)) AS in_acl, x.arxiv_id AS arxiv_id_linked, x.created AS arxiv_created, x.primary_category AS arxiv_primary, (x.primary_category LIKE 'cs.%') AS arxiv_cs_primary, (w.doi LIKE '10.48550/arxiv.%') AS is_arxiv_doi FROM oa_works w LEFT JOIN (SELECT l.oa_id, any_value(k.arxiv_id) arxiv_id, min(k.created) created, any_value(k.primary_category) primary_category FROM link_arxiv_oa l JOIN arxiv_keys k USING(arxiv_id) GROUP BY l.oa_id) x ON x.oa_id = w.id""") log(f"cw0 {n('SELECT count(*) FROM cw0'):,}") # ---- S2: preprint ↔ published duplicates (same normalised title, |year| <= 2, one preprint-like, one not) ---- con.execute("""CREATE OR REPLACE TABLE dup_pairs AS WITH pre AS (SELECT id, ntitle, publication_year y, publication_date d FROM cw0 WHERE (type = 'preprint' OR is_arxiv_doi) AND length(ntitle) >= 25), pub AS (SELECT id, ntitle, publication_year y, publication_date d FROM cw0 WHERE type <> 'preprint' AND NOT is_arxiv_doi AND length(ntitle) >= 25) SELECT p.id AS preprint_id, b.id AS published_id, p.d AS preprint_date, b.d AS published_date, row_number() OVER (PARTITION BY b.id ORDER BY p.d, p.id) AS rn_pub, row_number() OVER (PARTITION BY p.id ORDER BY b.d, b.id) AS rn_pre FROM pre p JOIN pub b ON p.ntitle = b.ntitle AND abs(p.y - b.y) <= 2""") con.execute("DELETE FROM dup_pairs WHERE rn_pub > 1 OR rn_pre > 1") log(f"duplicate preprint→published pairs: {n('SELECT count(*) FROM dup_pairs'):,}") # ---- S3: eligibility flags ---- con.execute(f"""CREATE OR REPLACE TABLE corpus_works AS SELECT c.*, dp.published_id AS duplicate_of, dp2.preprint_id AS preprint_work_id, least(c.publication_date, coalesce(dp2.preprint_date, c.publication_date), coalesce(c.arxiv_created, c.publication_date)) AS first_date, (c.publication_date IS NULL OR c.publication_date < '{WINDOW_START}' OR c.publication_date > '{WINDOW_END}') AS e1_outside_window, (c.type NOT IN {KEEP_TYPES} AND NOT (c.in_dblp AND c.type NOT IN {HARD_EXCLUDE_TYPES})) AS e2_type, (c.is_xpac AND NOT c.in_dblp) AS e3_xpac, (c.is_xpac AND c.in_dblp) AS xpac_kept_dblp, (c.source_name IN {GENERIC_DEPOSIT_SOURCES} AND NOT c.in_dblp AND c.arxiv_id_linked IS NULL) AS e7_generic_deposit, (CASE WHEN substr(c.publication_date, 6, 5) = '01-01' THEN 'year' ELSE 'month' END) AS date_precision, (c.title IS NULL OR length(trim(c.title)) = 0) AS e4_no_title, (dp.published_id IS NOT NULL) AS e5_duplicate, c.is_retracted AS e6_retracted, (c.in_dblp OR c.cs_primary OR (c.cs_any AND coalesce(c.arxiv_cs_primary, FALSE))) AS cs_core, (c.cs_any AND NOT (c.in_dblp OR c.cs_primary OR (c.cs_any AND coalesce(c.arxiv_cs_primary, FALSE)))) AS cs_penumbra FROM cw0 c LEFT JOIN dup_pairs dp ON dp.preprint_id = c.id LEFT JOIN dup_pairs dp2 ON dp2.published_id = c.id""") con.execute(f"""ALTER TABLE corpus_works ADD COLUMN eligible BOOLEAN; UPDATE corpus_works SET eligible = 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""") con.execute(f"""ALTER TABLE corpus_works ADD COLUMN period VARCHAR; UPDATE corpus_works SET period = CASE WHEN first_date < '{WINDOW_START}' THEN 'pre' WHEN first_date < '{BOUNDARY}' THEN 'baseline' WHEN first_date <= '{WINDOW_END}' THEN 'post' ELSE 'future' END""") con.execute("ALTER TABLE corpus_works ADD COLUMN first_month VARCHAR; UPDATE corpus_works SET first_month = substr(first_date, 1, 7)") log("corpus_works built") # ---- DBLP-only records (no OpenAlex record after both passes) ---- con.execute("""CREATE OR REPLACE TABLE dblp_only_counts AS SELECT year, rec_type, count(*) n, sum(CASE WHEN doi IS NOT NULL THEN 1 ELSE 0 END) with_doi FROM dblp_keys WHERE dblp_key NOT IN (SELECT dblp_key FROM link_dblp_oa) GROUP BY 1,2 ORDER BY 1,2""") # ---- PRISMA counts ---- P = {"generated": dt.datetime.now(dt.UTC).isoformat(), "window": [WINDOW_START, WINDOW_END], "boundary": BOUNDARY} P["identified"] = { "openalex_cs_topic_2018_2026_pass1": n("SELECT count(*) FROM oa_keys"), "openalex_total_records_loaded": n("SELECT count(*) FROM cw0"), "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"), "acl_2018_2026": n("SELECT count(*) FROM acl_keys WHERE year BETWEEN 2018 AND 2026")} 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"), "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")} 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"]} P["excluded_e2_by_type"] = dict(q("SELECT type, count(*) FROM corpus_works WHERE e2_type GROUP BY 1 ORDER BY 2 DESC")) P["excluded_sequential"] = {} cond = "TRUE" for k in ["e1_outside_window", "e2_type", "e3_xpac", "e4_no_title", "e5_duplicate", "e7_generic_deposit"]: P["excluded_sequential"][k] = n(f"SELECT count(*) FROM corpus_works WHERE {cond} AND {k}"); cond += f" AND NOT {k}" P["after_exclusions"] = n(f"SELECT count(*) FROM corpus_works WHERE {cond}") P["cs_core_eligible"] = n("SELECT count(*) FROM corpus_works WHERE eligible") P["cs_penumbra_eligible_otherwise"] = n(f"SELECT count(*) FROM corpus_works WHERE {cond} AND cs_penumbra") P["eligible_by_period"] = dict(q("SELECT period, count(*) FROM corpus_works WHERE eligible GROUP BY 1")) P["eligible_by_year"] = dict(q("SELECT publication_year, count(*) FROM corpus_works WHERE eligible GROUP BY 1 ORDER BY 1")) P["eligible_by_type"] = dict(q("SELECT type, count(*) FROM corpus_works WHERE eligible GROUP BY 1 ORDER BY 2 DESC")) 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"), "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"), "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"), "retracted_flagged": n("SELECT count(*) FROM corpus_works WHERE eligible AND e6_retracted")} P["eligible_date_precision"] = dict(q("SELECT date_precision, count(*) FROM corpus_works WHERE eligible GROUP BY 1")) P["strict_core_dblp_and_cs_primary"] = n("SELECT count(*) FROM corpus_works WHERE eligible AND in_dblp AND cs_primary") os.makedirs(os.path.join(R, "data/processed/corpus"), exist_ok=True); os.makedirs(os.path.join(R, "data/metadata"), exist_ok=True) con.execute(f"COPY corpus_works TO '{R}/data/processed/corpus/corpus_works.parquet' (FORMAT PARQUET, COMPRESSION ZSTD)") con.execute(f"COPY dblp_only_counts TO '{R}/data/metadata/dblp_only_counts.csv' (HEADER)") json.dump(P, open(os.path.join(R, "data/metadata/prisma_counts.json"), "w"), indent=1, default=str) print(json.dumps(P, indent=1, default=str)); log("done") if __name__ == "__main__": main()