71 lines
4.6 KiB
Python
71 lines
4.6 KiB
Python
"""Attach authoritative identities to existing events without reimporting text."""
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
|
|
from paper_atlas.db.backend import database_arguments, open_store
|
|
from paper_atlas.db.submissions import attach_submission,identities_for_row
|
|
from paper_atlas.collectors.pmlr_tracks import require_track_manifest, require_track_record
|
|
|
|
|
|
def queue_files(path):
|
|
queue=json.loads(Path(path).read_text())
|
|
files=queue['files']
|
|
if len(files)!=len(set(files)):raise ValueError('Duplicate paths in evidence queue')
|
|
for filename in files:
|
|
if hashlib.sha256(Path(filename).read_bytes()).hexdigest()!=queue['file_sha256'][filename]:
|
|
raise ValueError('Evidence file differs from audited queue: '+filename)
|
|
return files
|
|
|
|
|
|
def backfill(db,files, *, pg_service=None, pg_schema=None):
|
|
for filename in files:
|
|
manifest=json.loads(Path(filename).with_suffix('.manifest.json').read_text())
|
|
require_track_manifest(manifest)
|
|
if manifest['source'] in ('colm','neurips','iclr_program') and manifest.get('parser_version',0)<2:
|
|
raise ValueError('Recollect old parser output before submission evidence backfill')
|
|
store=open_store(db,pg_service=pg_service,pg_schema=pg_schema);c=store.conn;count=0
|
|
try:
|
|
before=c.execute('SELECT count(*) FROM submissions').fetchone()[0]
|
|
for filename in files:
|
|
path=Path(filename);manifest=json.loads(path.with_suffix('.manifest.json').read_text());seen=set()
|
|
with path.open() as stream:
|
|
for line in stream:
|
|
row=json.loads(line)
|
|
if row['external_id'] in seen:raise ValueError('Repeated source ID in evidence file')
|
|
seen.add(row['external_id'])
|
|
if any(row.get(key)!=manifest.get(key) for key in ('source','venue','year','track')):
|
|
raise ValueError('Evidence row differs from edition manifest')
|
|
require_track_record(row, manifest)
|
|
record=c.execute('SELECT r.id,r.paper_id,r.metadata_json FROM source_records r JOIN sources s ON s.id=r.source_id WHERE s.name=? AND r.external_id=?',(row['source'],row['external_id'])).fetchone()
|
|
edition=c.execute('SELECT e.id FROM editions e JOIN venues v ON v.id=e.venue_id WHERE lower(v.acronym)=lower(?) AND e.year=? AND e.track=?',(row['venue'],row['year'],row['track'])).fetchone()
|
|
if not record or not edition:raise ValueError('Evidence backfill requires an already imported record and edition')
|
|
if not c.execute('SELECT 1 FROM submission_sources ss JOIN submissions s ON s.id=ss.submission_id WHERE ss.source_record_id=? AND ss.source_submission_id=? AND s.edition_id=?',(record[0],row['external_id'],edition[0])).fetchone():
|
|
raise ValueError('No existing submission to backfill')
|
|
attach_submission(store,edition[0],record[1],record[0],row['external_id'],identities_for_row(row))
|
|
metadata=json.loads(record[2]);metadata['record_identifiers_raw']=row.get('identifiers') or {}
|
|
if row['source']=='iclr_program':
|
|
for key in ('program_identity_type','program_submission_number_verified','program_identity_verification'):
|
|
if key in row['metadata']:metadata[key]=row['metadata'][key]
|
|
c.execute('UPDATE source_records SET metadata_json=? WHERE id=?',(json.dumps(metadata,ensure_ascii=False),record[0]))
|
|
count+=1
|
|
if len(seen)!=manifest['count']:raise ValueError('Evidence count differs from manifest')
|
|
conflict=c.execute('SELECT 1 FROM submission_sources ss JOIN submissions s ON s.id=ss.submission_id JOIN source_records r ON r.id=ss.source_record_id WHERE s.paper_id!=r.paper_id LIMIT 1').fetchone()
|
|
if conflict:raise ValueError('Submission evidence points to a different paper entity')
|
|
c.commit()
|
|
result={'status':'backfilled','records':count,'submissions_before':before,'submissions_after':c.execute('SELECT count(*) FROM submissions').fetchone()[0]}
|
|
print(json.dumps(result),flush=True);return result
|
|
except BaseException:
|
|
c.rollback();raise
|
|
finally:c.close()
|
|
|
|
|
|
def main(argv=None, *, prog=None):
|
|
parser=argparse.ArgumentParser(prog=prog);database_arguments(parser);parser.add_argument('--queue',required=True)
|
|
args=parser.parse_args(argv);backfill(args.db,queue_files(args.queue),pg_service=args.pg_service,pg_schema=args.pg_schema)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|