1#!/usr/bin/env python32"""WP44 — convert the OpenAlex API increment pages (works with a CS topic published 2026-10-01 → 2026-11-02, fetched after the32026-10-14 snapshot) into the same JSONL.zst part format as the snapshot filter, so that build_corpus.py can normalise them:4data/raw/openalex/works_pass3/cs/updated_date=api-increment__part_0000.jsonl.zst. Works already present in pass 1 or pass 25are skipped (ids read from the compact works tables). Provenance: api_increment/ledger.json (calls, cost, slices)."""6import argparse, glob, json, os, duckdb, orjson, zstandard as zstd789def main():10 ap = argparse.ArgumentParser(); ap.add_argument("--root", required=True); a = ap.parse_args(); R = a.root11 con = duckdb.connect(); known = set()12 for f in ("works", "works_p2"):13 p = f"{R}/data/processed/compact/{f}.parquet"14 if os.path.exists(p):15 known.update(x[0] for x in con.execute(f"SELECT id FROM read_parquet('{p}')").fetchall())16 out_dir = f"{R}/data/raw/openalex/works_pass3/cs"; os.makedirs(out_dir, exist_ok=True)17 seen = set(); n_pages = n_new = n_dup = 018 with open(f"{out_dir}/updated_date=api-increment__part_0000.jsonl.zst", "wb") as fh, zstd.ZstdCompressor(level=6).stream_writer(fh) as w:19 for pf in sorted(glob.glob(f"{R}/data/raw/openalex/api_increment/pages/*/page_*.json")):20 n_pages += 121 for work in json.load(open(pf)).get("results", []):22 wid = int(work["id"].rsplit("W", 1)[-1])23 if wid in known or wid in seen:24 n_dup += 1; continue25 seen.add(wid); n_new += 1; w.write(orjson.dumps(work) + b"\n")26 rep = {"pages": n_pages, "new_works": n_new, "already_in_snapshot": n_dup, "ledger": json.load(open(f"{R}/data/raw/openalex/api_increment/ledger.json"))}27 json.dump(rep, open(f"{R}/data/raw/openalex/works_pass3/increment_report.json", "w"), indent=1, default=str); print({k: v for k, v in rep.items() if k != "ledger"})282930if __name__ == "__main__":31 main()32