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='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 (w.doi LIKE '10.48550/arxiv.%') 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 con.execute("""CREATE OR REPLACE TABLE dup_pairs AS38 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),39 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)40 SELECT p.id AS preprint_id, b.id AS published_id, p.d AS preprint_date, b.d AS published_date,41 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_pre42 FROM pre p JOIN pub b ON p.ntitle = b.ntitle AND abs(p.y - b.y) <= 2""")43 con.execute("DELETE FROM dup_pairs WHERE rn_pub > 1 OR rn_pre > 1")44 log(f"duplicate preprint→published pairs: {n('SELECT count(*) FROM dup_pairs'):,}")45 # ---- S3: eligibility flags ----46 con.execute(f"""CREATE OR REPLACE TABLE corpus_works AS47 SELECT c.*,48 dp.published_id AS duplicate_of,49 dp2.preprint_id AS preprint_work_id,50 least(c.publication_date, coalesce(dp2.preprint_date, c.publication_date), coalesce(c.arxiv_created, c.publication_date)) AS first_date,51 (c.publication_date IS NULL OR c.publication_date < '{WINDOW_START}' OR c.publication_date > '{WINDOW_END}') AS e1_outside_window,52 (c.type NOT IN {KEEP_TYPES} AND NOT (c.in_dblp AND c.type NOT IN {HARD_EXCLUDE_TYPES})) AS e2_type,53 (c.is_xpac AND NOT c.in_dblp) AS e3_xpac,54 (c.is_xpac AND c.in_dblp) AS xpac_kept_dblp,55 (c.source_name IN {GENERIC_DEPOSIT_SOURCES} AND NOT c.in_dblp AND c.arxiv_id_linked IS NULL) AS e7_generic_deposit,56 (CASE WHEN substr(c.publication_date, 6, 5) = '01-01' THEN 'year' ELSE 'month' END) AS date_precision,57 (c.title IS NULL OR length(trim(c.title)) = 0) AS e4_no_title,58 (dp.published_id IS NOT NULL) AS e5_duplicate,59 c.is_retracted AS e6_retracted,60 (c.in_dblp OR c.cs_primary OR (c.cs_any AND coalesce(c.arxiv_cs_primary, FALSE))) AS cs_core,61 (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_penumbra62 FROM cw0 c63 LEFT JOIN dup_pairs dp ON dp.preprint_id = c.id64 LEFT JOIN dup_pairs dp2 ON dp2.published_id = c.id""")65 con.execute(f"""ALTER TABLE corpus_works ADD COLUMN eligible BOOLEAN; UPDATE corpus_works SET eligible =66 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""")67 con.execute(f"""ALTER TABLE corpus_works ADD COLUMN period VARCHAR; UPDATE corpus_works SET period = CASE68 WHEN first_date < '{WINDOW_START}' THEN 'pre' WHEN first_date < '{BOUNDARY}' THEN 'baseline' WHEN first_date <= '{WINDOW_END}' THEN 'post' ELSE 'future' END""")69 con.execute("ALTER TABLE corpus_works ADD COLUMN first_month VARCHAR; UPDATE corpus_works SET first_month = substr(first_date, 1, 7)")70 log("corpus_works built")71 # ---- DBLP-only records (no OpenAlex record after both passes) ----72 con.execute("""CREATE OR REPLACE TABLE dblp_only_counts AS73 SELECT year, rec_type, count(*) n, sum(CASE WHEN doi IS NOT NULL THEN 1 ELSE 0 END) with_doi FROM dblp_keys74 WHERE dblp_key NOT IN (SELECT dblp_key FROM link_dblp_oa) GROUP BY 1,2 ORDER BY 1,2""")75 # ---- PRISMA counts ----76 P = {"generated": dt.datetime.now(dt.UTC).isoformat(), "window": [WINDOW_START, WINDOW_END], "boundary": BOUNDARY}77 P["identified"] = {78 "openalex_cs_topic_2018_2026_pass1": n("SELECT count(*) FROM oa_keys"),79 "openalex_total_records_loaded": n("SELECT count(*) FROM cw0"),80 "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"),81 "acl_2018_2026": n("SELECT count(*) FROM acl_keys WHERE year BETWEEN 2018 AND 2026")}82 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"),83 "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")}84 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"]}85 P["excluded_e2_by_type"] = dict(q("SELECT type, count(*) FROM corpus_works WHERE e2_type GROUP BY 1 ORDER BY 2 DESC"))86 P["excluded_sequential"] = {}87 cond = "TRUE"88 for k in ["e1_outside_window", "e2_type", "e3_xpac", "e4_no_title", "e5_duplicate", "e7_generic_deposit"]:89 P["excluded_sequential"][k] = n(f"SELECT count(*) FROM corpus_works WHERE {cond} AND {k}"); cond += f" AND NOT {k}"90 P["after_exclusions"] = n(f"SELECT count(*) FROM corpus_works WHERE {cond}")91 P["cs_core_eligible"] = n("SELECT count(*) FROM corpus_works WHERE eligible")92 P["cs_penumbra_eligible_otherwise"] = n(f"SELECT count(*) FROM corpus_works WHERE {cond} AND cs_penumbra")93 P["eligible_by_period"] = dict(q("SELECT period, count(*) FROM corpus_works WHERE eligible GROUP BY 1"))94 P["eligible_by_year"] = dict(q("SELECT publication_year, count(*) FROM corpus_works WHERE eligible GROUP BY 1 ORDER BY 1"))95 P["eligible_by_type"] = dict(q("SELECT type, count(*) FROM corpus_works WHERE eligible GROUP BY 1 ORDER BY 2 DESC"))96 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"),97 "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"),98 "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"),99 "retracted_flagged": n("SELECT count(*) FROM corpus_works WHERE eligible AND e6_retracted")}100 P["eligible_date_precision"] = dict(q("SELECT date_precision, count(*) FROM corpus_works WHERE eligible GROUP BY 1"))101 P["strict_core_dblp_and_cs_primary"] = n("SELECT count(*) FROM corpus_works WHERE eligible AND in_dblp AND cs_primary")102 os.makedirs(os.path.join(R, "data/processed/corpus"), exist_ok=True); os.makedirs(os.path.join(R, "data/metadata"), exist_ok=True)103 con.execute(f"COPY corpus_works TO '{R}/data/processed/corpus/corpus_works.parquet' (FORMAT PARQUET, COMPRESSION ZSTD)")104 con.execute(f"COPY dblp_only_counts TO '{R}/data/metadata/dblp_only_counts.csv' (HEADER)")105 json.dump(P, open(os.path.join(R, "data/metadata/prisma_counts.json"), "w"), indent=1, default=str)106 print(json.dumps(P, indent=1, default=str)); log("done")107108109if __name__ == "__main__":110 main()111