1#!/usr/bin/env python32"""WP44 — select the synthesis populations (protocol P2/S5):3 high-priority set (target >= 10,000): per (primary category × publication year) the top works by NCS36 (field-normalised4 36-month citations; for 2025–2026 by c12 then cited_by_count), proportional allocation with a floor of 25 per cell, eligible5 works only, retracted excluded;6 landmark set (target >= 1,000): top by NCS36 within category (floor 15 per category) plus the top 200 overall by c36,7 deduplicated; flagged for PI review (H-3).8Outputs: literature_synthesis/landmarks/{high_priority.parquet,landmarks.csv,<area>_landmarks.csv,<area>_abstracts.jsonl,selection_report.json}."""9import argparse, json, os, duckdb, yaml, pandas as pd101112def main():13 ap = argparse.ArgumentParser(); ap.add_argument("--root", required=True); ap.add_argument("--ontology", required=True)14 ap.add_argument("--n-high", type=int, default=10000); ap.add_argument("--n-land", type=int, default=1000); a = ap.parse_args(); R = a.root15 out = f"{R}/literature_synthesis/landmarks"; os.makedirs(out, exist_ok=True)16 ont = yaml.safe_load(open(a.ontology)); area = {c["id"]: c["area"] for c in ont["categories"]}; cats = list(area)17 con = duckdb.connect(f"{R}/data/processed/wp44.duckdb", read_only=True); con.execute("SET threads TO 16;")18 con.execute(f"CREATE OR REPLACE VIEW work_primary AS SELECT * FROM read_parquet('{R}/data/processed/classification/primary/*.parquet')")19 con.execute(f"CREATE OR REPLACE VIEW wcw AS SELECT * FROM read_parquet('{R}/data/processed/corpus/work_citation_windows.parquet')")20 df = con.execute("""SELECT c.id AS work_id, c.doi, c.arxiv_id_linked AS arxiv_id, c.title, c.source_name, c.publication_year AS year, c.type, p.primary_category, p.n_categories,21 w.c12, w.c36, w.ncs36, c.cited_by_count, c.fwci, c.has_abstract, c.is_oa, c.period22 FROM corpus_works c JOIN work_primary p ON p.work_id = c.id JOIN wcw w ON w.work_id = c.id WHERE c.eligible AND NOT c.e6_retracted""").df()23 df["score"] = df.ncs36.fillna(0); recent = df.year >= 2025; df.loc[recent, "score"] = df.loc[recent, "c12"].fillna(0) / (df.loc[recent].groupby("year").c12.transform("mean") + 1e-9)24 cell = df.groupby(["primary_category", "year"]).size(); alloc = (cell / cell.sum() * a.n_high).clip(lower=25).round().astype(int)25 parts = []26 for (c, y), k in alloc.items():27 g = df[(df.primary_category == c) & (df.year == y)].nlargest(int(k), ["score", "cited_by_count"]); parts.append(g)28 high = pd.concat(parts).drop_duplicates("work_id"); high["set"] = "high_priority"29 land_parts = [df[df.primary_category == c].nlargest(max(15, a.n_land // len(cats)), ["score", "cited_by_count"]) for c in cats]30 land = pd.concat(land_parts + [df.nlargest(200, ["c36", "cited_by_count"])]).drop_duplicates("work_id"); land["set"] = "landmark"; land["pi_review"] = True31 high.to_parquet(f"{out}/high_priority.parquet"); land.to_csv(f"{out}/landmarks.csv", index=False)32 con.execute(f"CREATE OR REPLACE TEMP TABLE L AS SELECT * FROM read_csv_auto('{out}/landmarks.csv')")33 for ar in sorted(set(area.values())):34 ids = [c for c in cats if area[c] == ar]; sub = land[land.primary_category.isin(ids)]35 sub.to_csv(f"{out}/{ar}_landmarks.csv", index=False)36 ab = con.execute(f"SELECT l.work_id, l.title, a.abstract FROM L l LEFT JOIN oa_abstracts a ON a.work_id = l.work_id WHERE l.primary_category IN ({','.join(repr(i) for i in ids)})").fetchall()37 with open(f"{out}/{ar}_abstracts.jsonl", "w") as fh:38 for w, t, abst in ab:39 fh.write(json.dumps({"work_id": int(w), "title": t, "abstract": abst}) + "\n")40 rep = {"high_priority": int(len(high)), "landmarks": int(len(land)), "by_area": land.primary_category.map(area).value_counts().to_dict(), "by_year_landmarks": land.year.value_counts().sort_index().to_dict(),41 "share_with_abstract_landmarks": float(land.has_abstract.mean()), "share_oa_landmarks": float(land.is_oa.mean())}42 json.dump(rep, open(f"{out}/selection_report.json", "w"), indent=1); print(rep)434445if __name__ == "__main__":46 main()47