1#!/usr/bin/env python32"""WP44 — parse the DBLP XML dump (dblp.xml.gz + dblp.dtd) into Parquet.34Output: <out>/dblp_publications.parquet with one row per publication record5(article, inproceedings, proceedings, book, incollection, phdthesis, mastersthesis,6www is skipped) and <out>/dblp_parse_report.json (counts, md5 of input, timings).7All years are kept (the historical baseline needs 2018–2022; counts per year are8reported for the whole archive).9"""10import argparse, gzip, hashlib, json, os, time, datetime as dt11from lxml import etree12import pyarrow as pa, pyarrow.parquet as pq1314TYPES = {"article", "inproceedings", "proceedings", "book", "incollection", "phdthesis", "mastersthesis", "data"}15SCHEMA = pa.schema([16 ("key", pa.string()), ("rec_type", pa.string()), ("mdate", pa.string()), ("publtype", pa.string()),17 ("title", pa.string()), ("year", pa.int32()), ("authors", pa.list_(pa.string())), ("author_pids", pa.list_(pa.string())),18 ("editors", pa.list_(pa.string())), ("journal", pa.string()), ("booktitle", pa.string()), ("volume", pa.string()),19 ("number", pa.string()), ("pages", pa.string()), ("publisher", pa.string()), ("series", pa.string()),20 ("doi", pa.string()), ("ee", pa.list_(pa.string())), ("url", pa.string()), ("crossref", pa.string()),21 ("stream", pa.string()),22])232425def text(el):26 return "".join(el.itertext()).strip() if el is not None else None272829def main():30 ap = argparse.ArgumentParser(); ap.add_argument("--xml", required=True); ap.add_argument("--out", required=True)31 ap.add_argument("--min-year", type=int, default=0)32 a = ap.parse_args(); a.xml = os.path.abspath(a.xml); a.out = os.path.abspath(a.out); os.makedirs(a.out, exist_ok=True)33 t0 = time.time()34 md5 = hashlib.md5(open(a.xml, "rb").read(1 << 30)).hexdigest() # first GB hash (file is 1.1 GB) + size recorded35 size = os.path.getsize(a.xml)36 # lxml needs the DTD next to the XML for entity resolution; parse the gz stream37 parser = etree.XMLParser(load_dtd=True, resolve_entities=True, huge_tree=True)38 counts = {}; years = {}39 rows = {n: [] for n in SCHEMA.names}; n = 040 writer = pq.ParquetWriter(os.path.join(a.out, "dblp_publications.parquet"), SCHEMA, compression="zstd")4142 def flush():43 nonlocal rows44 if rows["key"]:45 writer.write_table(pa.Table.from_pydict(rows, schema=SCHEMA)); rows = {k: [] for k in SCHEMA.names}4647 # iterparse over the gzip stream; the DTD is referenced relatively (dblp.dtd) -> chdir48 os.chdir(os.path.dirname(os.path.abspath(a.xml)))49 with gzip.open(a.xml, "rb") as fh:50 ctx = etree.iterparse(fh, events=("end",), tag=list(TYPES | {"www"}), load_dtd=True, resolve_entities=True, huge_tree=True)51 for _, el in ctx:52 t = el.tag53 if t in TYPES:54 counts[t] = counts.get(t, 0) + 155 yr = el.findtext("year"); yr = int(yr) if yr and yr.isdigit() else None56 years[yr] = years.get(yr, 0) + 157 if yr is not None and yr >= a.min_year:58 ees = [e.text for e in el.findall("ee") if e.text]59 doi = next((e[16:].lower() for e in ees if e.startswith("https://doi.org/")), None)60 stream = el.get("key", "").rsplit("/", 1)[0] if "/" in el.get("key", "") else None61 vals = dict(key=el.get("key"), rec_type=t, mdate=el.get("mdate"), publtype=el.get("publtype"),62 title=text(el.find("title")), year=yr,63 authors=[text(x) for x in el.findall("author")], author_pids=[x.get("pid") for x in el.findall("author")],64 editors=[text(x) for x in el.findall("editor")], journal=el.findtext("journal"), booktitle=el.findtext("booktitle"),65 volume=el.findtext("volume"), number=el.findtext("number"), pages=el.findtext("pages"),66 publisher=el.findtext("publisher"), series=text(el.find("series")), doi=doi, ee=ees,67 url=el.findtext("url"), crossref=el.findtext("crossref"), stream=stream)68 for k, v in vals.items():69 rows[k].append(v)70 n += 171 if n % 200000 == 0:72 flush(); print(f"{n} rows, {time.time()-t0:.0f}s", flush=True)73 el.clear()74 while el.getprevious() is not None:75 del el.getparent()[0]76 flush(); writer.close()77 rep = {"input": os.path.abspath(a.xml), "input_bytes": size, "input_md5_first_GiB": md5, "parsed_at": dt.datetime.now(dt.UTC).isoformat(),78 "rows_written": n, "min_year": a.min_year, "counts_by_type": counts, "counts_by_year": {str(k): v for k, v in sorted(years.items(), key=lambda kv: (kv[0] is None, kv[0]))},79 "seconds": round(time.time() - t0)}80 json.dump(rep, open(os.path.join(a.out, "dblp_parse_report.json"), "w"), indent=1)81 print(json.dumps({k: v for k, v in rep.items() if k != "counts_by_year"}, indent=1))828384if __name__ == "__main__":85 main()86