152 lines
10 KiB
Python
152 lines
10 KiB
Python
"""Original series, exact edition boundaries, and complete candidate discovery."""
|
|
from contextlib import closing
|
|
from copy import deepcopy
|
|
import gzip
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import sqlite3
|
|
import tempfile
|
|
import unittest
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
from paper_atlas.collectors.dblp import catalogue, XML_SHA
|
|
from paper_atlas.collectors.dblp_records import collect, normalize_row
|
|
from paper_atlas.ingestion.records import ingest_file
|
|
|
|
FIXTURE = json.loads((Path(__file__).parent/'fixtures/historical_predecessors.json').read_text())
|
|
EXAMPLES = {row['key']: row for row in FIXTURE['volumes']}
|
|
|
|
|
|
def example(key, *, bibliographic_year=None):
|
|
fixed = EXAMPLES[key]
|
|
family = key.split('/')[1]
|
|
year = int(key.split('/')[-1])
|
|
volume = {'record_type': 'proceedings', 'key': key, 'fields': deepcopy(fixed['fields'])}
|
|
row = {'record_type': 'inproceedings', 'key': f'conf/{family}/SyntheticExample{year}', 'publtype': None,
|
|
'fields': {'title': [f'Synthetic historical research example {family} {year}'],
|
|
'author': ['Jane Doe 0001', 'John Smith'], 'booktitle': fixed['fields']['booktitle'],
|
|
'year': [str(bibliographic_year or year)], 'crossref': [key],
|
|
'ee': [f'https://doi.org/10.1234/synthetic.{family}.{year}']}}
|
|
return row, volume
|
|
|
|
|
|
class HistoricalPredecessorTests(unittest.TestCase):
|
|
def normalize(self, key, **kwargs):
|
|
row, volume = example(key, **kwargs)
|
|
fixed = EXAMPLES[key]
|
|
result, reason = normalize_row(row, {key: volume}, {fixed['expected_target_family']})
|
|
self.assertIsNone(reason)
|
|
self.assertIsNotNone(result)
|
|
return result
|
|
|
|
def test_ewsl_primary_series_retains_original_name_and_family_alias(self):
|
|
for year in [1987, 1988, 1989, 1991]:
|
|
with self.subTest(year=year):
|
|
record = self.normalize(f'conf/ecml/{year}')
|
|
self.assertEqual((record['venue'],record['venue_aliases'],record['year'],record['track']), ('EWSL',['ECML-PKDD'],year,'main'))
|
|
proof = record['metadata']['historical_predecessor_evidence']
|
|
self.assertEqual(proof['original_series'], 'EWSL')
|
|
self.assertEqual(proof['history_url'], 'https://ecmlpkdd2011.org/pastConferences.php')
|
|
self.assertEqual(record['metadata']['bibliography']['booktitle'], ['EWSL'])
|
|
self.assertEqual(record['metadata']['author_names_raw'], ['Jane Doe 0001','John Smith'])
|
|
self.assertEqual(record['authors'], ['Jane Doe','John Smith'])
|
|
self.assertEqual(EXAMPLES['conf/ecml/1989']['bibliographic_counts'], [])
|
|
|
|
def test_eps_ecp_ewsp_keep_original_series_instead_of_renaming_to_icaps(self):
|
|
for key in ['conf/aips/1990','conf/ecp/1997','conf/ecp/1999','conf/ewsp/1991']:
|
|
with self.subTest(key=key):
|
|
record = self.normalize(key)
|
|
fixed = EXAMPLES[key]
|
|
self.assertEqual(record['venue'], fixed['expected_original_series'])
|
|
self.assertEqual(record['venue_aliases'], ['ICAPS'])
|
|
self.assertEqual(record['track'], 'main')
|
|
proof = record['metadata']['historical_predecessor_evidence']
|
|
self.assertEqual(proof['history_url'], 'https://www.icaps-conference.org/conference-series/')
|
|
self.assertEqual(proof['proceedings_title'], fixed['fields']['title'][0])
|
|
self.assertEqual(proof['proceedings_year_fields'], fixed['fields']['year'])
|
|
|
|
def test_altered_titles_keys_suffixes_publication_years_and_unreviewed_editions_stay_held(self):
|
|
for key in ['conf/ecml/1988','conf/aips/1990','conf/ecp/1997','conf/ewsp/1991']:
|
|
row, volume = example(key)
|
|
for changed in ['title','key','suffix','publication_year']:
|
|
with self.subTest(key=key, changed=changed):
|
|
r, v = deepcopy(row), deepcopy(volume)
|
|
if changed == 'title': v['fields']['title'][0] += ', Affiliated Workshop'
|
|
elif changed == 'key': v['key'] += 'w'; r['fields']['crossref'] = [v['key']]
|
|
elif changed == 'suffix': r['fields']['booktitle'][0] += ' (Workshop)'
|
|
else: v['fields']['year'] = ['2001']
|
|
result, _ = normalize_row(r, {v['key']: v}, {EXAMPLES[key]['expected_target_family']})
|
|
self.assertIsNone(result)
|
|
row, volume = example('conf/ecp/1997')
|
|
row['fields']['year'] = ['2001']; row['fields']['crossref'] = ['conf/ecp/2001']
|
|
volume['key'] = 'conf/ecp/2001'; volume['fields']['year'] = ['2001']
|
|
self.assertIsNone(normalize_row(row, {volume['key']:volume}, {'ICAPS'})[0])
|
|
|
|
def test_ecp1999_event_and_bibliographic_years_are_independently_retained(self):
|
|
record = self.normalize('conf/ecp/1999', bibliographic_year=2000)
|
|
self.assertEqual(record['year'], 1999)
|
|
self.assertEqual(record['metadata']['bibliographic_year'], 2000)
|
|
self.assertEqual(record['metadata']['bibliography']['year'], ['2000'])
|
|
self.assertEqual(record['metadata']['historical_predecessor_evidence']['proceedings_year_fields'], ['2000'])
|
|
self.assertEqual(record['metadata']['historical_predecessor_evidence']['event_year'], 1999)
|
|
|
|
def test_old_candidates_require_new_families_only_when_requested_years_need_them(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary); row, volume = example('conf/aips/1990')
|
|
db = root/'dblp-catalogue.sqlite'
|
|
with closing(sqlite3.connect(db)) as connection, connection:
|
|
connection.execute('CREATE TABLE proceedings(key TEXT PRIMARY KEY,metadata_json TEXT)')
|
|
connection.execute('INSERT INTO proceedings VALUES(?,?)', (volume['key'],json.dumps(volume)))
|
|
candidate = root/'candidate.jsonl'; candidate.write_text(json.dumps(row)+'\n')
|
|
db.with_name('dblp-ai-candidates.manifest.json').write_text(json.dumps({
|
|
'snapshot_sha256': XML_SHA, 'candidate_sha256': hashlib.sha256(candidate.read_bytes()).hexdigest(),
|
|
'candidate_records': 1, 'candidate_family_counts': {'aips':1},
|
|
'record_types': {'proceedings':1,'inproceedings':1}}))
|
|
# Tiny CI fixtures may live on tmpfs; production reserve is tested separately.
|
|
with patch('paper_atlas.collectors.dblp_records.shutil.disk_usage', return_value=SimpleNamespace(free=16*1024**3)):
|
|
result = collect(db, root/'early', ['ICAPS'], candidates=candidate, through_year=1990)
|
|
self.assertEqual((result['normalized_records'],result['files'][0]['venue']), (1,'EPS'))
|
|
for year, missing in [(1991,'ICAPS/ewsp'),(1997,'ICAPS/ecp')]:
|
|
out = root/str(year); out.mkdir(); sentinel=out/'previous.txt';sentinel.write_text('accepted previous output')
|
|
with self.subTest(year=year), self.assertRaisesRegex(ValueError, missing):
|
|
collect(db, out, ['ICAPS'], candidates=candidate, through_year=year)
|
|
self.assertEqual(sentinel.read_text(), 'accepted previous output')
|
|
self.assertFalse((out/'dblp-normalization.json').exists())
|
|
|
|
def test_new_catalogue_includes_previously_missing_families_without_decision_inference(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary);dtd=root/'dblp.dtd';dtd.write_bytes(b'<!ENTITY eacute "é">')
|
|
archive=root/'dblp.xml.gz'
|
|
body=''.join(f'<inproceedings key="conf/{family}/Synthetic"><title>Synthetic example</title><year>{year}</year><booktitle>{book}</booktitle><crossref>conf/{family}/{year}</crossref></inproceedings>' for family,year,book in [('ecp',1997,'ECP'),('ewsp',1991,'EWSP')])
|
|
with gzip.open(archive,'wb') as stream:stream.write(('<?xml version="1.0"?><!DOCTYPE dblp SYSTEM "dblp.dtd"><dblp>'+body+'</dblp>').encode())
|
|
result=catalogue(archive,dtd,root/'out',hashlib.sha256(archive.read_bytes()).hexdigest(),hashlib.md5(dtd.read_bytes()).hexdigest())
|
|
rows=[json.loads(line)for line in Path(result['file']).read_text().splitlines()]
|
|
self.assertEqual({r['key'] for r in rows}, {'conf/ecp/Synthetic','conf/ewsp/Synthetic'})
|
|
self.assertEqual(result['candidate_records'], 2)
|
|
self.assertTrue(all('status' not in r and 'venue' not in r for r in rows))
|
|
|
|
def test_native_import_keeps_original_series_aliases_and_full_source_evidence(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root=Path(temporary);db=root/'sample.sqlite'
|
|
for key in ['conf/ecml/1988','conf/ecp/1999']:
|
|
record=self.normalize(key);path=root/(key.replace('/','-')+'.jsonl')
|
|
path.write_text(json.dumps(record,ensure_ascii=False)+'\n')
|
|
path.with_suffix('.manifest.json').write_text(json.dumps({'source':'dblp','venue':record['venue'],'year':record['year'],'track':'main','count':1,'list_url':'https://doi.org/10.4230/dblp.xml.2026-11-01','file_sha256':hashlib.sha256(path.read_bytes()).hexdigest(),'status':'partial','missing_abstract_count':1}))
|
|
ingest_file(db,path)
|
|
with closing(sqlite3.connect(db)) as connection:
|
|
rows=connection.execute('SELECT v.acronym,a.alias,e.year FROM editions e JOIN venues v ON v.id=e.venue_id JOIN venue_aliases a ON a.venue_id=v.id ORDER BY v.acronym').fetchall()
|
|
self.assertEqual(rows,[('ECP','ICAPS',1999),('EWSL','ECML-PKDD',1988)])
|
|
metadata=[json.loads(r[0])for r in connection.execute('SELECT metadata_json FROM source_records')]
|
|
self.assertEqual({r['historical_predecessor_evidence']['original_series'] for r in metadata}, {'ECP','EWSL'})
|
|
self.assertEqual(connection.execute('SELECT count(*) FROM decisions WHERE status=?',('accepted',)).fetchone()[0],2)
|
|
self.assertEqual(connection.execute('PRAGMA foreign_key_check').fetchall(),[])
|
|
|
|
def test_existing_aips_normalization_remains_original_aips(self):
|
|
row, volume=example('conf/aips/1990');row['fields']['booktitle']=['AIPS'];row['fields']['year']=['1994'];row['fields']['crossref']=['conf/aips/1994']
|
|
volume['key']='conf/aips/1994';volume['fields']['title']=['AIPS 1994 Proceedings'];volume['fields']['year']=['1994']
|
|
record,reason=normalize_row(row,{volume['key']:volume},{'ICAPS'})
|
|
self.assertIsNone(reason);self.assertEqual((record['venue'],record['venue_aliases']),('AIPS',['ICAPS']))
|
|
self.assertNotIn('historical_predecessor_evidence',record['metadata'])
|