75 lines
4.0 KiB
Python
75 lines
4.0 KiB
Python
"""Import normalized conference JSONL files with coverage evidence."""
|
|
import argparse
|
|
import json
|
|
import time
|
|
from pathlib import Path
|
|
|
|
from index_core import Store
|
|
|
|
|
|
def ingest_file(db,path):
|
|
path = Path(path)
|
|
manifest_path = path.with_suffix('.manifest.json')
|
|
manifest = json.loads(manifest_path.read_text()) if manifest_path.exists() else {}
|
|
if manifest.get('source') in ('colm','neurips') and manifest.get('parser_version',0)<2:
|
|
raise ValueError('Recollect proceedings with parser_version>=2 before identity matching')
|
|
store = Store(db)
|
|
c = store.conn
|
|
source_name = manifest.get('source')
|
|
if not source_name:
|
|
raise ValueError('A source manifest is required')
|
|
source_id = store.source(source_name,manifest['list_url'])
|
|
run = c.execute('INSERT INTO ingest_runs(source_id,checkpoint_json) VALUES(?,?)',(source_id,json.dumps({'file':str(path)}))).lastrowid
|
|
c.commit()
|
|
count = 0
|
|
seen = set()
|
|
editions = set()
|
|
try:
|
|
with path.open() as stream:
|
|
for line in stream:
|
|
row = json.loads(line)
|
|
if row['source'] != source_name:
|
|
raise ValueError('Source differs from manifest')
|
|
external = row['external_id']
|
|
if external in seen:
|
|
raise ValueError('Duplicate external ID in one collected file: '+external)
|
|
seen.add(external)
|
|
paper,record = store.record(source_name,external,row['url'],row['title'],row.get('abstract'),row.get('authors',[]),row.get('identifiers'),row.get('metadata'))
|
|
if 'tags' in row:
|
|
store.record_tags(paper,record,source_name,row['tags'])
|
|
if row.get('status') in ('accepted','rejected','desk_rejected','withdrawn','under_review','unknown'):
|
|
edition = store.decision(paper,record,row['venue'],row['year'],row['track'],external,row.get('evidence_url',row['url']),row['status'],row.get('raw_decision','listed in official proceedings' if source_name=='pmlr' else row['status']),row.get('venue_aliases'))
|
|
editions.add(edition)
|
|
else:
|
|
raise ValueError('This importer requires an explicit supported decision status')
|
|
count += 1
|
|
if not count:
|
|
raise ValueError('Empty metadata file')
|
|
if len(editions) != 1:
|
|
raise ValueError('A manifest must describe exactly one conference edition and track')
|
|
edition = next(iter(editions))
|
|
expected = manifest.get('count')
|
|
if expected is not None and expected != count:
|
|
raise ValueError(f'Manifest count {expected} differs from records {count}')
|
|
complete=expected==count and manifest.get('status') not in ('partial','blocked') and not manifest.get('missing_abstract_count',0)
|
|
c.execute('INSERT INTO edition_sources(edition_id,source_id,list_url,status,expected_count,imported_count,last_success_at) VALUES(?,?,?,?,?,?,?) ON CONFLICT(edition_id,source_id,list_url) DO UPDATE SET status=excluded.status,expected_count=excluded.expected_count,imported_count=excluded.imported_count,last_success_at=excluded.last_success_at,last_error=NULL',(edition,source_id,manifest['list_url'],'complete' if complete else 'partial',expected,count,time.strftime('%Y-%m-%dT%H:%M:%SZ',time.gmtime())))
|
|
c.execute("UPDATE ingest_runs SET status='complete',finished_at=CURRENT_TIMESTAMP,records_seen=?,records_written=? WHERE id=?",(count,count,run))
|
|
c.commit()
|
|
print(json.dumps({'file':str(path),'status':'imported','records':count}),flush=True)
|
|
except BaseException as exc:
|
|
c.rollback()
|
|
c.execute("UPDATE ingest_runs SET status='failed',finished_at=CURRENT_TIMESTAMP,error=? WHERE id=?",(str(exc),run))
|
|
c.commit()
|
|
raise
|
|
finally:
|
|
c.close()
|
|
|
|
|
|
if __name__=='__main__':
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument('--db',required=True)
|
|
parser.add_argument('files',nargs='+')
|
|
args = parser.parse_args()
|
|
for filename in args.files:
|
|
ingest_file(args.db,filename)
|