paper-atlas/paper_atlas/collectors/dblp.py

"""Inventory a pinned DBLP snapshot and stage candidate metadata for scope review.

This does not label candidates accepted or import them into the paper database.
Only the checksum-verified local DTD is allowed; XML cannot fetch remote entities.
"""
import argparse
from collections import Counter,deque
import gzip
import json
from pathlib import Path
import sqlite3
from xml.parsers import expat

from paper_atlas.io.snapshots import digest_file
from paper_atlas.db.submissions import timestamp


XML_SHA='0376ad019b216553faff0a14e1757b2d4c7a3b84c1beacf0511ef83695211b18'
DTD_MD5='96e4283e71886dec2173718957293648'
DTD_NAMES={'dblp.dtd','dblp-2023-06-28.dtd'}
RELEASE='https://doi.org/10.4230/dblp.xml.2026-11-01'
# Broad discovery candidates, including historical names; not a venue mapping.
FAMILIES=set('nips icml iclr colm colt aistats uai clear acl emnlp naacl eacl coling conll cvpr iccv eccv aaai ijcai ecai kdd sigir www wsdm recsys cikm icdm sdm ecml pkdd ecmlpkdd icra iros rss corl aips icaps kr atal aamas icassp interspeech eurospeech icslp ismir miccai fat facct lrec mm mlsys'.split())


def records(archive,dtd_bytes):
    parser=expat.ParserCreate();parser.buffer_text=True
    pending=deque();depth=0;row=None;field=None;text=[];doctype=False

    def document_type(name,system_id,public_id,internal):
        nonlocal doctype
        if name!='dblp' or system_id not in DTD_NAMES or public_id or internal:
            raise ValueError('Only the pinned external DBLP DTD is allowed')
        doctype=True

    def external(context,base,system_id,public_id):
        if context is not None or system_id not in DTD_NAMES or public_id:
            raise ValueError('Unexpected external XML entity')
        child=parser.ExternalEntityParserCreate(context)
        child.ExternalEntityRefHandler=lambda *args: (_ for _ in ()).throw(ValueError('Nested external entity'))
        child.Parse(dtd_bytes,True);return 1

    def start(name,attrs):
        nonlocal depth,row,field,text
        depth+=1
        if depth==1 and name!='dblp':raise ValueError('Unexpected XML root')
        if depth==2 and name in ('inproceedings','proceedings'):
            row={'record_type':name,'key':attrs.get('key'),'publtype':attrs.get('publtype'),'fields':{}}
        elif depth==3 and row is not None:
            field=name;text=[]

    def content(value):
        if row is not None and field is not None:text.append(value)

    def end(name):
        nonlocal depth,row,field,text
        if depth==3 and row is not None and field is not None:
            row['fields'].setdefault(field,[]).append(' '.join(''.join(text).split()))
            field=None;text=[]
        if depth==2 and row is not None:
            pending.append(row);row=None
        depth-=1

    parser.StartDoctypeDeclHandler=document_type;parser.ExternalEntityRefHandler=external
    parser.SetParamEntityParsing(expat.XML_PARAM_ENTITY_PARSING_ALWAYS)
    parser.StartElementHandler=start;parser.EndElementHandler=end;parser.CharacterDataHandler=content
    with gzip.open(archive,'rb') as stream:
        while chunk:=stream.read(128*1024):
            parser.Parse(chunk,False)
            while pending:yield pending.popleft()
        parser.Parse(b'',True)
        while pending:yield pending.popleft()
    if not doctype:raise ValueError('DBLP DTD declaration missing')


def family(key):
    parts=(key or '').split('/')
    return parts[1] if len(parts)>=3 and parts[0]=='conf' else ''


def catalogue(archive,dtd,out,expected_sha=XML_SHA,expected_dtd_md5=DTD_MD5):
    archive=Path(archive);out=Path(out);dtd=Path(dtd)
    identity=digest_file(archive);dtd_identity=digest_file(dtd)
    if identity['sha256']!=expected_sha:raise ValueError('DBLP snapshot checksum mismatch')
    if dtd_identity['md5']!=expected_dtd_md5:raise ValueError('DBLP DTD checksum mismatch')
    if dtd.stat().st_size>128*1024:raise ValueError('Unexpectedly large DTD')
    out.mkdir(parents=True,exist_ok=True)
    target=out/'dblp-catalogue.sqlite';stage=out/'dblp-catalogue.sqlite.part'
    candidate=out/'dblp-ai-candidates.jsonl';part=candidate.with_suffix('.part')
    if stage.exists():raise ValueError('Unfinished catalogue exists; inspect it before another run')
    c=sqlite3.connect(stage)
    c.executescript('PRAGMA journal_mode=DELETE; PRAGMA synchronous=NORMAL; CREATE TABLE proceedings(key TEXT PRIMARY KEY,family TEXT,metadata_json TEXT); CREATE INDEX proceedings_family ON proceedings(family); CREATE TABLE volume_counts(crossref TEXT,year TEXT,booktitle TEXT,record_type TEXT,count INTEGER,PRIMARY KEY(crossref,year,booktitle,record_type)); CREATE TABLE family_counts(family TEXT,record_type TEXT,count INTEGER,PRIMARY KEY(family,record_type)); CREATE TABLE family_samples(family TEXT,seq INTEGER,metadata_json TEXT,PRIMARY KEY(family,seq)); CREATE TABLE volume_anomalies(key TEXT PRIMARY KEY,metadata_json TEXT);')
    totals=Counter();families=Counter();volumes=Counter();samples=Counter();staged=Counter();seen=0

    def flush():
        c.executemany('INSERT INTO volume_counts VALUES(?,?,?,?,?) ON CONFLICT(crossref,year,booktitle,record_type) DO UPDATE SET count=count+excluded.count',((*key,count) for key,count in volumes.items()))
        volumes.clear();c.commit()

    try:
        with part.open('w') as output:
            for row in records(archive,dtd.read_bytes()):
                seen+=1;kind=row['record_type'];key=row['key'];fields=row['fields'];prefix=family(key)
                if not key:raise ValueError('Bibliographic record lacks a key')
                totals[kind]+=1
                if kind=='proceedings':
                    c.execute('INSERT INTO proceedings VALUES(?,?,?)',(key,prefix,json.dumps(row,ensure_ascii=False)))
                if prefix:
                    families[(prefix,kind)]+=1
                    if kind=='inproceedings':
                        crossrefs=fields.get('crossref') or [''];years=fields.get('year') or [''];books=fields.get('booktitle') or ['']
                        if any(len(values)!=1 for values in (crossrefs,years,books)):
                            c.execute('INSERT INTO volume_anomalies VALUES(?,?)',(key,json.dumps(row,ensure_ascii=False)))
                            totals['multiple_volume_fields']+=1
                        group=tuple(values[0] if len(values)==1 else '@multiple:'+json.dumps(values,ensure_ascii=False) for values in (crossrefs,years,books))
                        volumes[(*group,kind)]+=1
                        if samples[prefix]<2:
                            c.execute('INSERT INTO family_samples VALUES(?,?,?)',(prefix,samples[prefix],json.dumps(row,ensure_ascii=False)));samples[prefix]+=1
                        if prefix in FAMILIES:
                            output.write(json.dumps(row,ensure_ascii=False)+'\n');staged[prefix]+=1
                if seen%50000==0:
                    flush();print(json.dumps({'source':'dblp','records_scanned':seen,'candidate_records':sum(staged.values())}),flush=True)
            flush()
        c.executemany('INSERT INTO family_counts VALUES(?,?,?)',((*key,count) for key,count in families.items()));c.commit()
        if c.execute('PRAGMA quick_check').fetchall()!=[('ok',)]:raise ValueError('Catalogue integrity failed')
        c.close();c=None
        manifest={'source':'dblp','status':'catalogued_candidates_pending_scope_review','source_release':RELEASE,'snapshot_sha256':identity['sha256'],
                  'dtd_md5':dtd_identity['md5'],'finished_at':timestamp(),'records_scanned':seen,'record_types':{kind:totals[kind] for kind in ('proceedings','inproceedings')},
                  'multiple_volume_field_records':totals['multiple_volume_fields'],
                  'candidate_records':sum(staged.values()),'candidate_family_counts':dict(staged),'candidate_sha256':digest_file(part)['sha256'],
                  'file':str(candidate),'catalogue':str(target),'note':'Candidate families include workshops and historical names. No acceptance labels or normalized venue assignments have been inferred.'}
        stage.replace(target);part.replace(candidate)
        manifest_path=candidate.with_suffix('.manifest.json');manifest_part=manifest_path.with_suffix('.part')
        manifest_part.write_text(json.dumps(manifest,indent=2));manifest_part.replace(manifest_path)
        print(json.dumps(manifest),flush=True);return manifest
    finally:
        if c is not None:c.close()


def main(argv=None, *, prog=None):
    parser=argparse.ArgumentParser(prog=prog);parser.add_argument('--archive',required=True);parser.add_argument('--dtd',required=True);parser.add_argument('--out',required=True)
    args=parser.parse_args(argv);catalogue(args.archive,args.dtd,args.out)


if __name__ == '__main__':
    main()