#!/usr/bin/env python3
"""WP44 — convert the OpenAlex API increment pages (works with a CS topic published 2026-10-01 → 2026-11-02, fetched after the
2026-10-14 snapshot) into the same JSONL.zst part format as the snapshot filter, so that build_corpus.py can normalise them:
data/raw/openalex/works_pass3/cs/updated_date=api-increment__part_0000.jsonl.zst. Works already present in pass 1 or pass 2
are skipped (ids read from the compact works tables). Provenance: api_increment/ledger.json (calls, cost, slices)."""
import argparse, glob, json, os, duckdb, orjson, zstandard as zstd


def main():
    ap = argparse.ArgumentParser(); ap.add_argument("--root", required=True); a = ap.parse_args(); R = a.root
    con = duckdb.connect(); known = set()
    for f in ("works", "works_p2"):
        p = f"{R}/data/processed/compact/{f}.parquet"
        if os.path.exists(p):
            known.update(x[0] for x in con.execute(f"SELECT id FROM read_parquet('{p}')").fetchall())
    out_dir = f"{R}/data/raw/openalex/works_pass3/cs"; os.makedirs(out_dir, exist_ok=True)
    seen = set(); n_pages = n_new = n_dup = 0
    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:
        for pf in sorted(glob.glob(f"{R}/data/raw/openalex/api_increment/pages/*/page_*.json")):
            n_pages += 1
            for work in json.load(open(pf)).get("results", []):
                wid = int(work["id"].rsplit("W", 1)[-1])
                if wid in known or wid in seen:
                    n_dup += 1; continue
                seen.add(wid); n_new += 1; w.write(orjson.dumps(work) + b"\n")
    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"))}
    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"})


if __name__ == "__main__":
    main()