From 3f81734e6915df0978ecc47844c342622fa0f2d2 Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Sun, 8 Nov 2026 19:07:00 +0800 Subject: [PATCH 1/4] Repair verified joint edition membership without reimporting papers --- .gitea/workflows/check.yml | 1 + README.md | 2 + docs/joint-editions.md | 65 ++++++++ paper_atlas/cli.py | 1 + paper_atlas/collectors/dblp_records.py | 2 + paper_atlas/collectors/dblp_scope.py | 7 + paper_atlas/ingestion/joint_editions.py | 201 ++++++++++++++++++++++++ tests/test_joint_editions.py | 173 ++++++++++++++++++++ 8 files changed, 452 insertions(+) create mode 100644 docs/joint-editions.md create mode 100644 paper_atlas/ingestion/joint_editions.py create mode 100644 tests/test_joint_editions.py diff --git a/.gitea/workflows/check.yml b/.gitea/workflows/check.yml index 950c95a..d57a25b 100644 --- a/.gitea/workflows/check.yml +++ b/.gitea/workflows/check.yml @@ -49,6 +49,7 @@ jobs: subprocess.run([str(venv/'bin/paper-atlas'),'--help'],cwd=outside,check=True) subprocess.run([python,'-I','-B','-m','paper_atlas','collect','cvf','--help'],cwd=outside,check=True) subprocess.run([python,'-I','-B','-m','paper_atlas','maintain','postgres','--help'],cwd=outside,check=True) + subprocess.run([python,'-I','-B','-m','paper_atlas','maintain','joint-editions','--help'],cwd=outside,check=True) subprocess.run([python,'-I','-B','-m','paper_atlas','import','records','--help'],cwd=outside,check=True) subprocess.run([python,'-I','-B','-m','paper_atlas','import','arxiv','--help'],cwd=outside,check=True) subprocess.run([python,'-I','-B','-m','paper_atlas','maintain','backfill','--help'],cwd=outside,check=True) diff --git a/README.md b/README.md index 09dcf66..e9651fc 100644 --- a/README.md +++ b/README.md @@ -86,3 +86,5 @@ paper-atlas collect colm --cache /data/paper-index/raw/http-cache \ CI 在已授权的远程 runner 上运行测试、构建 wheel，并从源码目录外验证安装后的命令及随包资源。数据库全量去重和覆盖率是独立的数据验收步骤，不因代码测试通过而自动完成。 开发通过工作分支和 PR 交付，由用户手动合并。 + +历史联合会议的既有届次归属可通过 [`maintain joint-editions`](docs/joint-editions.md) 先只读规划，再按计划哈希执行；保留论文与投稿决定来源，不重导入。 diff --git a/docs/joint-editions.md b/docs/joint-editions.md new file mode 100644 index 0000000..a35775f --- /dev/null +++ b/docs/joint-editions.md @@ -0,0 +1,65 @@ +# Repairing joint conference membership + +EURAI's [official history](https://www.eurai.org/ecai/history) identifies ECAI 2018 +and 2022 as joint IJCAI/ECAI conferences. The pinned DBLP volumes +`conf/ijcai/2018` and `conf/ijcai/2022` use `IJCAI` as their booktitle and title. +Existing records can therefore already be present under IJCAI while an ECAI query +shows a missing year. Adding ECAI as a global alias to the ordinary IJCAI venue +would incorrectly include all other IJCAI years. + +The existing canonical `ECAI+IJCAI` venue has both member aliases. This maintenance +command changes only the venue reference of the two existing `main` edition rows. +Edition IDs, submission IDs, decisions, source text, authors, identifiers, tags, +source links and coverage records stay intact. It does not merge or create papers +or submissions, resolve per-paper tracks, or turn reviews into negative decisions. +The DBLP collector applies the joint mapping only to these two exact main volume +keys; original booktitle/title remain in source metadata with the membership evidence. +Other years, workshop volumes and additional tracks retain their existing rules. + +## Review a read-only plan + +Use a runtime containing this repair command. The configured PostgreSQL writer lock +is used only for the apply operation; do not replace a newer ingestion runtime with +an older main-based feature branch. + +```sh +PGSERVICEFILE=/etc/paper-atlas/pg_service.conf paper-atlas maintain joint-editions \ + --pg-service foundry-research --pg-schema research --out joint-plan.json +``` + +The preflight requires the existing joint venue and exactly its two member aliases, +one nonempty unambiguous edition per year, and matching pinned DBLP volume evidence +for every existing submission. A colliding target edition, missing provenance or +unexpected source scope fails closed. The plan binds current database/schema, IDs, +counts and full fingerprints of affected papers, identifiers, source records, tags, +submissions, decisions and their provenance, and coverage rows. Observation time is +recorded separately from conference year. These are scoped checks, not a final global +identity or conference completeness audit. + +## Apply the exact reviewed plan + +Compute the file's SHA256 after review and pass it explicitly: + +```sh +PGSERVICEFILE=/etc/paper-atlas/pg_service.conf paper-atlas maintain joint-editions \ + --pg-service foundry-research --pg-schema research --plan joint-plan.json \ + --plan-sha256 "$PLAN_SHA256" --out joint-applied.json +``` + +The plan hash is checked before opening a writer. Current evidence is checked again +under the dataset writer lock; both edition updates and an audit marker commit in one +transaction. Protected row hashes are checked again before commit. Changed evidence +requires a fresh reviewed plan. Existing output reports are preserved; choose a new +path for another run. SQLite samples use `--db sample.sqlite`; the frozen production +backup remains protected. + +If a response is lost after commit, rerun with the same plan and a new output path. +When protected evidence is unchanged and both editions already point to the joint +venue, the result is `already_joint` with zero updates. It does not replay imports or +regenerate source records. Keep the plan, apply report and independent postcheck +alongside collection evidence. The audit marker is `joint_ijcai_ecai_2018_2022_v1`. + +Publication counts can differ between the DBLP volume and the current publisher +page. This membership repair preserves all existing rows rather than deleting entries +to force an aggregate match; section-level main/invited/other classification and +publication completeness require separate per-paper evidence. diff --git a/paper_atlas/cli.py b/paper_atlas/cli.py index 549bfc0..ef89ce0 100644 --- a/paper_atlas/cli.py +++ b/paper_atlas/cli.py @@ -25,6 +25,7 @@ COMMANDS = { ('catalogue','dblp'): ('paper_atlas.collectors.dblp', {}, '盘点 DBLP 历史会议目录'), ('import','arxiv'): ('paper_atlas.ingestion.arxiv', {}, '导入 arXiv 元数据 ZIP'), ('import','records'): ('paper_atlas.ingestion.records', {}, '导入核验过的会议文件'), + ('maintain','joint-editions'): ('paper_atlas.ingestion.joint_editions', {}, '按已核验计划修正既有联合会议届次归属'), ('maintain','merge'): ('paper_atlas.ingestion.maintenance', {}, '协调单写入合并与恢复'), ('maintain','backfill'): ('paper_atlas.ingestion.backfill', {}, '回填既有投稿的来源身份'), ('maintain','arxiv-tags'): ('paper_atlas.ingestion.arxiv_tags', {}, '恢复逐来源记录的 arXiv 类别证据'), diff --git a/paper_atlas/collectors/dblp_records.py b/paper_atlas/collectors/dblp_records.py index 4f5685e..2eb24c3 100644 --- a/paper_atlas/collectors/dblp_records.py +++ b/paper_atlas/collectors/dblp_records.py @@ -88,6 +88,8 @@ def normalize_row(row,volumes,targets,include_workshops=False,through_year=2026) 'scope_rule':edition['rule'],'author_names_raw':fields.get('author',[]),'publtype':row.get('publtype'), 'bibliography':{key:value for key,value in fields.items() if key in ('booktitle','year','ee','url','pages','volume','number','publisher','series','isbn')}, 'ambiguous_identifiers':ambiguous,'publication_evidence':'inproceedings bibliography; not an original decision note'} + if edition.get('joint_edition_evidence'): + metadata['joint_edition_evidence'] = edition['joint_edition_evidence'] if edition.get('track_scope_note'): metadata['track_scope_note']=edition['track_scope_note'] flagged=bool(re.search(r'\b(?:withdrawn|retracted)\b',row.get('publtype') or '',re.I)) diff --git a/paper_atlas/collectors/dblp_scope.py b/paper_atlas/collectors/dblp_scope.py index 08a550d..ee09e6a 100644 --- a/paper_atlas/collectors/dblp_scope.py +++ b/paper_atlas/collectors/dblp_scope.py @@ -96,8 +96,15 @@ def scope(row,volume,through_year=2026): elif suffix.casefold() in ('poster','posters') and venue not in ('ICLR','ACL','EMNLP','NAACL','EACL','COLING','CoNLL'):track='poster' if venue=='AAAI+IAAI':track='proceedings_unspecified' if venue=='AAMAS' and year<2002:return None,'predecessor_not_aamas' + joint_evidence = None + if (family == 'ijcai' and base == 'IJCAI' and track == 'main' + and year in (2018, 2022) and volume['key'] == 'conf/ijcai/' + str(year)): + venue, aliases = 'ECAI+IJCAI', ['ECAI', 'IJCAI'] + joint_evidence = 'https://www.eurai.org/ecai/history' result = {'venue':venue,'venue_aliases':aliases,'year':year,'track':track,'bibliographic_year':int(bibliographic[0]), 'rule':family+':'+base,'proceedings_title':title,'edition_year_basis':year_basis} + if joint_evidence: + result['joint_edition_evidence'] = joint_evidence if venue=='MM' and track=='main': # A main MM volume also includes doctoral, tutorials, companion papers # and other sections. DBLP XML omits the per-paper TOC section. diff --git a/paper_atlas/ingestion/joint_editions.py b/paper_atlas/ingestion/joint_editions.py new file mode 100644 index 0000000..bd5a576 --- /dev/null +++ b/paper_atlas/ingestion/joint_editions.py @@ -0,0 +1,201 @@ +"""Evidence-bound repair of two existing IJCAI/ECAI edition memberships.""" +import argparse +from contextlib import closing +import hashlib +import json +from pathlib import Path + +from paper_atlas.db.backend import database_arguments, open_read_connection, open_store +from paper_atlas.db.submissions import timestamp +from paper_atlas.collectors.dblp import XML_SHA + +YEARS = (2018, 2022) +TARGET = 'ECAI+IJCAI' +MEMBERS = {'ECAI', 'IJCAI'} +EVIDENCE = 'https://www.eurai.org/ecai/history' +MARKER = 'joint_ijcai_ecai_2018_2022_v1' + + +def _digest(c, sql, params=()): + sha = hashlib.sha256(); count = 0 + for row in c.execute(sql, params): + sha.update((json.dumps([row[i] for i in range(len(row))], ensure_ascii=False, + separators=(',', ':')) + '\n').encode()) + count += 1 + if count > 100000: + raise ValueError('Joint repair evidence scope exceeded bounded row limit') + return {'rows': count, 'sha256': sha.hexdigest()} + + +def fingerprints(c, ids): + marks = ','.join('?' for _ in ids) + submissions = 'SELECT id FROM submissions WHERE edition_id IN (' + marks + ')' + papers = 'SELECT paper_id FROM submissions WHERE edition_id IN (' + marks + ')' + decisions = 'SELECT id FROM decisions WHERE submission_id IN (' + submissions + ')' + checks = { + 'papers': ('id IN (' + papers + ')', 'id'), + 'identifiers': ('paper_id IN (' + papers + ')', 'scheme,value'), + 'source_records': ('paper_id IN (' + papers + ')', 'id'), + 'paper_tags': ('paper_id IN (' + papers + ')', 'paper_id,tag_id,evidence_key'), + 'submissions': ('edition_id IN (' + marks + ')', 'id'), + 'decisions': ('submission_id IN (' + submissions + ')', 'id'), + 'submission_sources': ('submission_id IN (' + submissions + ')', 'submission_id,source_record_id,source_submission_id'), + 'submission_identifiers': ('edition_id IN (' + marks + ')', 'edition_id,scheme,value'), + 'decision_sources': ('decision_id IN (' + decisions + ')', 'decision_id,source_record_id,source_decision_id'), + 'edition_sources': ('edition_id IN (' + marks + ')', 'edition_id,source_id,list_url'), + } + return {table: _digest(c, 'SELECT * FROM ' + table + ' WHERE ' + where + ' ORDER BY ' + order, ids) + for table, (where, order) in checks.items()} + + +def inspect(c): + target = c.execute('SELECT id FROM venues WHERE acronym=?', (TARGET,)).fetchall() + if len(target) != 1: + raise ValueError('Exactly one existing canonical joint venue required') + target = target[0][0] + aliases = {r[0] for r in c.execute('SELECT alias FROM venue_aliases WHERE venue_id=?', (target,))} + if aliases != MEMBERS: + raise ValueError('Joint target must already have exactly ECAI and IJCAI aliases') + editions = [] + for year in YEARS: + rows = c.execute('''SELECT e.id,e.venue_id,v.acronym FROM editions e JOIN venues v ON v.id=e.venue_id + WHERE v.acronym IN ('IJCAI','ECAI+IJCAI') AND e.year=? AND e.track='main' ''', (year,)).fetchall() + if len(rows) != 1: + raise ValueError('Expected one existing edition per year; missing or colliding editions require separate review') + edition, venue, acronym = rows[0] + count = c.execute('SELECT count(*) FROM submissions WHERE edition_id=?', (edition,)).fetchone()[0] + if not 0 < count <= 5000: + raise ValueError('Existing edition must have a nonempty bounded submission scope') + # Every existing submission must be supported by its own pinned DBLP + # volume record. No title/author matching, new attempts or text writes. + evidence = c.execute('''SELECT s.id,r.external_id,r.metadata_json FROM submissions s + JOIN submission_sources ss ON ss.submission_id=s.id + JOIN source_records r ON r.id=ss.source_record_id JOIN sources src ON src.id=r.source_id + WHERE s.edition_id=? AND src.name='dblp' ORDER BY s.id,r.id''', (edition,)).fetchall() + seen = set() + for sid, external, metadata in evidence: + meta = json.loads(metadata) + bibliography = meta.get('bibliography') or {} + if (meta.get('snapshot_sha256') != XML_SHA + or meta.get('proceedings_key') != 'conf/ijcai/' + str(year) + or meta.get('bibliographic_year') != year + or bibliography.get('booktitle') not in (['IJCAI'], ['IJCAI/ECAI']) + or not external.startswith('conf/ijcai/')): + raise ValueError('Source evidence outside the exact verified IJCAI/ECAI volume') + seen.add(sid) + if len(seen) != count: + raise ValueError('Every submission must have matching original volume evidence') + editions.append({'id': edition, 'year': year, 'track': 'main', 'from_venue_id': venue, + 'from_venue': acronym, 'target_venue_id': target, 'submissions': count, + 'evidence_url': EVIDENCE, 'proceedings_key': 'conf/ijcai/' + str(year)}) + ids = tuple(e['id'] for e in editions) + return {'format_version': 1, 'kind': MARKER, 'checked_at': timestamp(), 'editions': editions, + 'content_fingerprints': fingerprints(c, ids), 'target_aliases': sorted(aliases), + 'status': 'already_joint' if all(e['from_venue_id'] == target for e in editions) else 'ready', + 'scope': 'Only these two existing edition venue references; no source text, paper, submission, decision, track, identifier or tag change.'} + + +def validate_plan(expected): + if not isinstance(expected, dict): + raise ValueError('Repair plan must be a JSON object') + if set(expected) != {'format_version','kind','checked_at','editions','content_fingerprints','target_aliases','status','scope','database'}: + raise ValueError('Unexpected repair plan fields') + if (type(expected['format_version']) is not int or expected['format_version'] != 1 + or expected['kind'] != MARKER or expected['status'] not in ('ready','already_joint')): + raise ValueError('Unsupported repair plan') + if (not isinstance(expected['database'],dict) or not isinstance(expected['content_fingerprints'],dict) + or not isinstance(expected['editions'],list) or len(expected['editions']) != 2 + or expected['target_aliases'] != sorted(MEMBERS)): + raise ValueError('Invalid repair plan structure') + for edition, year in zip(expected['editions'], YEARS): + if not isinstance(edition,dict) or edition.get('year') != year: + raise ValueError('Repair plan must name the two exact ordered years') + for key in ('id','year','from_venue_id','target_venue_id','submissions'): + if type(edition.get(key)) is not int or edition[key] <= 0: + raise ValueError('Repair plan IDs/counts must be positive integers') + + +def apply_plan(c, expected): + validate_plan(expected) + actual = inspect(c) + if actual['content_fingerprints'] != expected['content_fingerprints'] or actual['target_aliases'] != expected['target_aliases']: + raise ValueError('Stale repair plan: protected content or aliases changed') + if len(expected['editions']) != 2: + raise ValueError('Exactly two editions required') + for old, now in zip(expected['editions'], actual['editions']): + if set(old) != set(now): + raise ValueError('Unexpected edition fields') + for key in ('id','year','track','target_venue_id','submissions','evidence_url','proceedings_key'): + if old[key] != now[key]: + raise ValueError('Stale repair plan: edition differs') + if now['from_venue_id'] != old['from_venue_id'] and now['from_venue_id'] != now['target_venue_id']: + raise ValueError('Edition membership differs from plan') + if actual['status'] == 'already_joint': + return {**actual, 'status':'already_joint', 'updated_editions':0} + for edition in actual['editions']: + if edition['from_venue_id'] == edition['target_venue_id']: + continue + changed = c.execute('''UPDATE editions SET venue_id=? WHERE id=? AND venue_id=? AND year=? AND track=?''', + (edition['target_venue_id'],edition['id'],edition['from_venue_id'],edition['year'],edition['track'])) + if changed.rowcount != 1: + raise ValueError('Concurrent edition change; entire repair must roll back') + after = inspect(c) + if after['status'] != 'already_joint' or after['content_fingerprints'] != actual['content_fingerprints']: + raise ValueError('Postcheck changed protected content; entire repair must roll back') + c.execute('INSERT INTO schema_migrations(name) VALUES(?) ON CONFLICT DO NOTHING', (MARKER,)) + return {**after, 'status':'applied', 'updated_editions':sum(e['from_venue_id'] != e['target_venue_id'] for e in actual['editions'])} + + +def database_identity(c, db): + if getattr(c, 'postgres', False): + row = c.execute('SELECT current_database(),current_schema()').fetchone() + return {'engine':'postgresql','database':row[0],'schema':row[1]} + return {'engine':'sqlite','path':str(Path(db).resolve())} + + +def run(db, out, *, pg_service=None, pg_schema=None, plan=None, plan_sha256=None): + out = Path(out).resolve() + if not out.parent.is_dir(): + raise ValueError('Report parent directory must already exist') + if out.exists(): + raise ValueError('Preserve existing evidence report; choose a new output path') + expected = None + if plan: + raw = Path(plan).read_bytes() + if hashlib.sha256(raw).hexdigest() != plan_sha256: + raise ValueError('Repair plan hash differs before opening database') + expected = json.loads(raw) + validate_plan(expected) + elif plan_sha256: + raise ValueError('Plan hash requires an existing plan') + if db and out == Path(db).resolve(): + raise ValueError('Report must not overwrite database') + writer = open_store(db, pg_service=pg_service, pg_schema=pg_schema) if expected is not None else None + c = writer.conn if writer else open_read_connection(db, pg_service=pg_service, pg_schema=pg_schema) + with closing(c): + try: + if expected is not None and not getattr(c, 'postgres', False): + c.commit(); c.execute('BEGIN IMMEDIATE') + identity = database_identity(c, db) + if expected is not None and expected['database'] != identity: + raise ValueError('Plan belongs to another database/schema') + report = apply_plan(c, expected) if expected is not None else inspect(c) + report['database'] = identity + if expected is not None: + c.commit() + except BaseException: + if expected is not None: c.rollback() + raise + out.write_text(json.dumps(report,ensure_ascii=False,indent=2)+'\n') + print(json.dumps({'status':report['status'],'editions':len(report['editions']), + 'updated_editions':report.get('updated_editions',0),'report':str(out)})) + return report + + +def main(argv=None, *, prog=None): + parser = argparse.ArgumentParser(prog=prog); database_arguments(parser) + parser.add_argument('--out', required=True) + parser.add_argument('--plan', help='Apply an already reviewed plan; omitted means read-only preflight') + parser.add_argument('--plan-sha256', help='Required exact plan SHA256 when applying') + args = parser.parse_args(argv) + run(args.db,args.out,pg_service=args.pg_service,pg_schema=args.pg_schema,plan=args.plan,plan_sha256=args.plan_sha256) diff --git a/tests/test_joint_editions.py b/tests/test_joint_editions.py new file mode 100644 index 0000000..9a86a07 --- /dev/null +++ b/tests/test_joint_editions.py @@ -0,0 +1,173 @@ +from contextlib import closing, redirect_stdout +import hashlib +import io +import json +import os +from pathlib import Path +import tempfile +import unittest +from unittest.mock import patch +import uuid + +from paper_atlas.collectors.dblp import XML_SHA +from paper_atlas.collectors.dblp_scope import scope +from paper_atlas.collectors.dblp_records import normalize_row +from paper_atlas.db.store import Store +from paper_atlas.db.backend import open_store +from paper_atlas.ingestion.joint_editions import (TARGET, MARKER, inspect, apply_plan, run) +from paper_atlas.postgres.client import connect +from paper_atlas.postgres.migration import migrate +from paper_atlas.postgres.schema import quote +from paper_atlas import querying + + +def seed(store): + c = store.conn + c.execute("INSERT INTO venues(acronym,name,field,venue_type) VALUES(?,?,'reasoning','conference')", (TARGET,TARGET)) + target = c.execute('SELECT id FROM venues WHERE acronym=?',(TARGET,)).fetchone()[0] + for alias in ['ECAI','IJCAI']: + c.execute('INSERT INTO venue_aliases VALUES(?,?)',(target,alias)) + for year in [2018,2019,2022]: + metadata = {'snapshot_sha256':XML_SHA,'proceedings_key':'conf/ijcai/'+str(year), + 'bibliographic_year':year,'bibliography':{'booktitle':['IJCAI']}} + p,r=store.record('dblp','conf/ijcai/fixture'+str(year),'https://dblp.org/rec/fixture', + 'Original publication '+str(year),'Abstract\x00with\x01data',[], + {'doi':'10.24963/ijcai.'+str(year)+'/1'},metadata) + store.record_tags(p,r,'dblp',[{'namespace':'dblp:booktitle','value':'IJCAI'}]) + store.decision(p,r,'IJCAI',year,'main','fixture'+str(year),'https://example.org/decision','accepted','listed') + c.commit() + + +def plan(c): + return {**inspect(c),'database':{'engine':'sqlite','path':'fixture'}} + + +class JointEditionTests(unittest.TestCase): + def setUp(self): + self.tmp=tempfile.TemporaryDirectory(); self.addCleanup(self.tmp.cleanup) + self.db=Path(self.tmp.name)/'sample.sqlite' + self.store=Store(self.db); self.addCleanup(self.store.conn.close) + seed(self.store); self.c=self.store.conn + + def test_apply_preserves_every_protected_row_and_edition_id(self): + before=plan(self.c); ids=[r['id']for r in before['editions']] + result=apply_plan(self.c,before);self.c.commit() + self.assertEqual(result['updated_editions'],2) + self.assertEqual(result['content_fingerprints'],before['content_fingerprints']) + self.assertEqual([r['id']for r in result['editions']],ids) + untouched=self.c.execute('SELECT v.acronym FROM editions e JOIN venues v ON v.id=e.venue_id WHERE e.year=2019').fetchone()[0] + self.assertEqual(untouched,'IJCAI') + self.assertEqual(self.c.execute('PRAGMA foreign_key_check').fetchall(),[]) + self.assertEqual(self.c.execute('SELECT count(*) FROM schema_migrations WHERE name=?',(MARKER,)).fetchone()[0],1) + with redirect_stdout(io.StringIO())as out: + querying.main(['--db',str(self.db),'--venue','ECAI']) + self.assertEqual(len(json.loads(out.getvalue())),2) + self.assertEqual(apply_plan(self.c,before)['updated_editions'],0) + + def test_any_changed_source_text_rejects_stale_plan_before_membership_updates(self): + before=plan(self.c) + self.c.execute("UPDATE source_records SET title='New evidence' WHERE external_id='conf/ijcai/fixture2018'") + with self.assertRaisesRegex(ValueError,'Stale repair plan'): + apply_plan(self.c,before) + self.assertEqual(inspect(self.c)['status'],'ready') + + def test_conflicting_target_edition_is_never_merged_or_deleted(self): + target=self.c.execute('SELECT id FROM venues WHERE acronym=?',(TARGET,)).fetchone()[0] + self.c.execute("INSERT INTO editions(venue_id,year,track) VALUES(?,2018,'main')",(target,)) + with self.assertRaisesRegex(ValueError,'colliding'): + inspect(self.c) + self.assertEqual(self.c.execute('SELECT count(*) FROM editions WHERE year=2018').fetchone()[0],2) + + def test_wrong_source_volume_and_snapshot_fail_closed(self): + original=self.c.execute("SELECT metadata_json FROM source_records WHERE external_id='conf/ijcai/fixture2018'").fetchone()[0] + for key,value in [('proceedings_key','conf/ijcai/2018w'),('snapshot_sha256','wrong'),('bibliographic_year',2017),('bibliography',{'booktitle':['Other']})]: + m=json.loads(original);m[key]=value + self.c.execute("UPDATE source_records SET metadata_json=? WHERE external_id='conf/ijcai/fixture2018'",(json.dumps(m),)) + with self.subTest(key=key),self.assertRaisesRegex(ValueError,'Source evidence'): + inspect(self.c) + + def test_missing_source_support_and_wrong_aliases_are_not_guessed(self): + self.c.execute("DELETE FROM venue_aliases WHERE alias='ECAI'") + with self.assertRaisesRegex(ValueError,'aliases'): + inspect(self.c) + self.c.rollback() + self.c.execute('DELETE FROM submission_sources WHERE source_record_id=(SELECT id FROM source_records WHERE external_id=?)',('conf/ijcai/fixture2022',)) + with self.assertRaisesRegex(ValueError,'Every submission'): + inspect(self.c) + + def test_cli_plan_readonly_hash_binding_identity_and_repeat_apply(self): + out=Path(self.tmp.name)/'plan.json' + with redirect_stdout(io.StringIO()):run(self.db,out) + self.assertEqual(inspect(self.c)['status'],'ready') + raw=out.read_bytes();sha=hashlib.sha256(raw).hexdigest() + with patch('paper_atlas.ingestion.joint_editions.open_store')as writer: + with self.assertRaisesRegex(ValueError,'hash differs'): + run(self.db,Path(self.tmp.name)/'bad.json',plan=out,plan_sha256='0'*64) + writer.assert_not_called() + with redirect_stdout(io.StringIO()): + result=run(self.db,Path(self.tmp.name)/'applied.json',plan=out,plan_sha256=sha) + again=run(self.db,Path(self.tmp.name)/'retry.json',plan=out,plan_sha256=sha) + self.assertEqual(result['updated_editions'],2);self.assertEqual(again['updated_editions'],0) + altered=json.loads(raw);altered['database']['path']='another.sqlite' + bad=Path(self.tmp.name)/'wrong-db.json';bad.write_text(json.dumps(altered)) + with self.assertRaisesRegex(ValueError,'another database'): + run(self.db,Path(self.tmp.name)/'wrong-result.json',plan=bad,plan_sha256=hashlib.sha256(bad.read_bytes()).hexdigest()) + + def test_invalid_empty_plan_fails_before_opening_writer(self): + f=Path(self.tmp.name)/'empty.json';f.write_text('{}') + with patch('paper_atlas.ingestion.joint_editions.open_store')as writer: + with self.assertRaises(ValueError): + run(self.db,Path(self.tmp.name)/'empty-result.json',plan=f,plan_sha256=hashlib.sha256(f.read_bytes()).hexdigest()) + writer.assert_not_called() + + def test_collector_only_changes_the_two_exact_main_volume_keys(self): + for year in [2018,2019,2022]: + volume={'key':'conf/ijcai/'+str(year),'fields':{'title':['IJCAI '+str(year)],'year':[str(year)]}} + row={'record_type':'inproceedings','key':'conf/ijcai/One'+str(year),'fields':{'booktitle':['IJCAI'], + 'year':[str(year)],'crossref':[volume['key']],'title':['Original research title'], + 'author':['Jane Doe'],'ee':['https://doi.org/10.24963/ijcai.'+str(year)+'/1']}} + result,error=normalize_row(row,{volume['key']:volume},{'IJCAI','ECAI'}) + self.assertIsNone(error) + self.assertEqual(result['venue'],TARGET if year in(2018,2022)else'IJCAI') + self.assertEqual(result['metadata']['bibliography']['booktitle'],['IJCAI']) + self.assertEqual(result['title'],'Original research title') + if year in(2018,2022): + self.assertIn('joint_edition_evidence',result['metadata']) + row['fields']['booktitle']=['IJCAI (Workshop)'] + scoped,_=scope(row,volume) + self.assertEqual(scoped['venue'],'IJCAI') + self.assertEqual(scoped['track'],'workshop') + + +@unittest.skipUnless(os.environ.get('PAPER_ATLAS_TEST_PG_SERVICE'),'Requires approved remote PostgreSQL service') +class PostgreSQLJointTests(unittest.TestCase): + def setUp(self): + self.service=os.environ['PAPER_ATLAS_TEST_PG_SERVICE'];self.target='pa_joint_'+uuid.uuid4().hex[:12] + self.tmp=tempfile.TemporaryDirectory();self.addCleanup(self.tmp.cleanup) + self.db=Path(self.tmp.name)/'sample.sqlite';s=Store(self.db);seed(s);s.conn.close() + self.addCleanup(self.cleanup) + with redirect_stdout(io.StringIO()):migrate(self.db,self.service,Path(self.tmp.name)/'migration.json',schema=self.target,min_free_gib=0) + + def cleanup(self): + with connect(self.service)as c: + for name, in c.execute('SELECT nspname FROM pg_namespace WHERE starts_with(nspname,%s)',(self.target,)).fetchall(): + c.execute('DROP SCHEMA '+quote(name)+' CASCADE') + + def test_native_repair_is_idempotent_keeps_codec_and_provenance(self): + p=Path(self.tmp.name)/'plan.json' + with redirect_stdout(io.StringIO()): + before=run(None,p,pg_service=self.service,pg_schema=self.target) + after=run(None,Path(self.tmp.name)/'applied.json',pg_service=self.service,pg_schema=self.target,plan=p,plan_sha256=hashlib.sha256(p.read_bytes()).hexdigest()) + self.assertEqual(after['content_fingerprints'],before['content_fingerprints']) + with closing(open_store(pg_service=self.service,pg_schema=self.target).conn)as c: + self.assertEqual(apply_plan(c,before)['updated_editions'],0) + self.assertEqual(c.execute('SELECT abstract FROM papers ORDER BY id LIMIT 1').fetchone()[0],'Abstract\x00with\x01data') + + def test_native_stale_source_plan_preserves_both_original_memberships(self): + with closing(open_store(pg_service=self.service,pg_schema=self.target).conn)as c: + before=plan(c) + c.execute("UPDATE source_records SET title='Changed' WHERE external_id=?",('conf/ijcai/fixture2022',));c.commit() + with self.assertRaisesRegex(ValueError,'Stale repair plan'): + apply_plan(c,before) + c.rollback() + self.assertEqual(inspect(c)['status'],'ready') From c05b979b01af15cc19add27e81ce9e7ea8103a71 Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Sun, 8 Nov 2026 19:09:27 +0800 Subject: [PATCH 2/4] Bind repair execution receipts to the reviewed plan hash --- paper_atlas/ingestion/joint_editions.py | 2 ++ tests/test_joint_editions.py | 1 + 2 files changed, 3 insertions(+) diff --git a/paper_atlas/ingestion/joint_editions.py b/paper_atlas/ingestion/joint_editions.py index bd5a576..ff44857 100644 --- a/paper_atlas/ingestion/joint_editions.py +++ b/paper_atlas/ingestion/joint_editions.py @@ -181,6 +181,8 @@ def run(db, out, *, pg_service=None, pg_schema=None, plan=None, plan_sha256=None raise ValueError('Plan belongs to another database/schema') report = apply_plan(c, expected) if expected is not None else inspect(c) report['database'] = identity + if expected is not None: + report['plan_sha256'] = plan_sha256 if expected is not None: c.commit() except BaseException: diff --git a/tests/test_joint_editions.py b/tests/test_joint_editions.py index 9a86a07..1732589 100644 --- a/tests/test_joint_editions.py +++ b/tests/test_joint_editions.py @@ -108,6 +108,7 @@ class JointEditionTests(unittest.TestCase): result=run(self.db,Path(self.tmp.name)/'applied.json',plan=out,plan_sha256=sha) again=run(self.db,Path(self.tmp.name)/'retry.json',plan=out,plan_sha256=sha) self.assertEqual(result['updated_editions'],2);self.assertEqual(again['updated_editions'],0) + self.assertEqual(result['plan_sha256'],sha);self.assertEqual(again['plan_sha256'],sha) altered=json.loads(raw);altered['database']['path']='another.sqlite' bad=Path(self.tmp.name)/'wrong-db.json';bad.write_text(json.dumps(altered)) with self.assertRaisesRegex(ValueError,'another database'): From 866aeaf068f71677e688ea87305b889579eef7ea Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Sun, 8 Nov 2026 19:10:10 +0800 Subject: [PATCH 3/4] Read PostgreSQL mapping rows by position during joint preflight --- paper_atlas/ingestion/joint_editions.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/paper_atlas/ingestion/joint_editions.py b/paper_atlas/ingestion/joint_editions.py index ff44857..c09fa24 100644 --- a/paper_atlas/ingestion/joint_editions.py +++ b/paper_atlas/ingestion/joint_editions.py @@ -62,7 +62,7 @@ def inspect(c): WHERE v.acronym IN ('IJCAI','ECAI+IJCAI') AND e.year=? AND e.track='main' ''', (year,)).fetchall() if len(rows) != 1: raise ValueError('Expected one existing edition per year; missing or colliding editions require separate review') - edition, venue, acronym = rows[0] + edition, venue, acronym = (rows[0][i] for i in range(3)) count = c.execute('SELECT count(*) FROM submissions WHERE edition_id=?', (edition,)).fetchone()[0] if not 0 < count <= 5000: raise ValueError('Existing edition must have a nonempty bounded submission scope') @@ -73,7 +73,8 @@ def inspect(c): JOIN source_records r ON r.id=ss.source_record_id JOIN sources src ON src.id=r.source_id WHERE s.edition_id=? AND src.name='dblp' ORDER BY s.id,r.id''', (edition,)).fetchall() seen = set() - for sid, external, metadata in evidence: + for row in evidence: + sid, external, metadata = (row[i] for i in range(3)) meta = json.loads(metadata) bibliography = meta.get('bibliography') or {} if (meta.get('snapshot_sha256') != XML_SHA From 88c2819eddfcfa26c6ce311a14638ef964770eed Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Sun, 8 Nov 2026 19:15:26 +0800 Subject: [PATCH 4/4] Keep legacy source refreshes in their verified joint editions --- docs/joint-editions.md | 4 +++- paper_atlas/db/store.py | 14 ++++++++++++++ tests/test_joint_editions.py | 15 ++++++++++++++- 3 files changed, 31 insertions(+), 2 deletions(-) diff --git a/docs/joint-editions.md b/docs/joint-editions.md index a35775f..7c630fb 100644 --- a/docs/joint-editions.md +++ b/docs/joint-editions.md @@ -14,7 +14,9 @@ source links and coverage records stay intact. It does not merge or create paper or submissions, resolve per-paper tracks, or turn reviews into negative decisions. The DBLP collector applies the joint mapping only to these two exact main volume keys; original booktitle/title remain in source metadata with the membership evidence. -Other years, workshop volumes and additional tracks retain their existing rules. +The shared ingestion entry point also resolves these exact pinned source volumes before +creating an edition, so refreshing an older normalized file cannot recreate a separate +IJCAI edition. Other sources, years, workshop volumes and additional tracks retain their existing rules. ## Review a read-only plan diff --git a/paper_atlas/db/store.py b/paper_atlas/db/store.py index 7d57533..272ace4 100644 --- a/paper_atlas/db/store.py +++ b/paper_atlas/db/store.py @@ -144,6 +144,20 @@ class Store: def decision(self,paper,record,venue,year,track,external_id,url,status,raw_value,aliases=None,identities=None,observed_at=None,decision_date=None,source_decision_id='',submitted_at=None,guard_status=False): c = self.conn + # An already collected file may still call these exact pinned volumes + # IJCAI. Resolve membership before finding/creating an edition so a + # later refresh cannot split the repaired joint event again. + if venue == 'IJCAI' and year in (2018,2022) and track == 'main': + from paper_atlas.collectors.dblp import XML_SHA + evidence=c.execute('SELECT src.name,r.metadata_json FROM source_records r JOIN sources src ON src.id=r.source_id WHERE r.id=?',(record,)).fetchone() + if evidence and evidence[0] == 'dblp': + metadata=json.loads(evidence[1]) + if (metadata.get('snapshot_sha256') == XML_SHA + and metadata.get('proceedings_key') == 'conf/ijcai/'+str(year) + and metadata.get('bibliographic_year') == year + and (metadata.get('bibliography') or {}).get('booktitle') in (['IJCAI'],['IJCAI/ECAI'])): + venue='ECAI+IJCAI' + aliases=sorted(set(aliases or []) | {'ECAI','IJCAI'}) existing=c.execute('SELECT id FROM venues WHERE lower(acronym)=lower(?)',(venue,)).fetchone() if existing:venue_id=existing[0] else:venue_id=inserted_id(c,"INSERT INTO venues(acronym,name,field,venue_type) VALUES(?,?,?,'conference')",(venue,venue,'unclassified')) diff --git a/tests/test_joint_editions.py b/tests/test_joint_editions.py index 1732589..47246d0 100644 --- a/tests/test_joint_editions.py +++ b/tests/test_joint_editions.py @@ -32,9 +32,11 @@ def seed(store): 'bibliographic_year':year,'bibliography':{'booktitle':['IJCAI']}} p,r=store.record('dblp','conf/ijcai/fixture'+str(year),'https://dblp.org/rec/fixture', 'Original publication '+str(year),'Abstract\x00with\x01data',[], - {'doi':'10.24963/ijcai.'+str(year)+'/1'},metadata) + {'doi':'10.24963/ijcai.'+str(year)+'/1'}, {}) store.record_tags(p,r,'dblp',[{'namespace':'dblp:booktitle','value':'IJCAI'}]) store.decision(p,r,'IJCAI',year,'main','fixture'+str(year),'https://example.org/decision','accepted','listed') + stored=json.loads(c.execute('SELECT metadata_json FROM source_records WHERE id=?',(r,)).fetchone()[0]);stored.update(metadata) + c.execute('UPDATE source_records SET metadata_json=? WHERE id=?',(json.dumps(stored),r)) c.commit() @@ -64,6 +66,17 @@ class JointEditionTests(unittest.TestCase): self.assertEqual(len(json.loads(out.getvalue())),2) self.assertEqual(apply_plan(self.c,before)['updated_editions'],0) + def test_refresh_of_old_normalized_source_does_not_recreate_ordinary_edition(self): + before=plan(self.c);apply_plan(self.c,before) + for year in [2018,2022]: + row=self.c.execute('''SELECT r.paper_id,r.id,d.observed_at FROM source_records r + JOIN submissions s ON s.source_record_id=r.id JOIN decisions d ON d.submission_id=s.id + WHERE r.external_id=?''',('conf/ijcai/fixture'+str(year),)).fetchone() + self.store.decision(row[0],row[1],'IJCAI',year,'main','fixture'+str(year), + 'https://example.org/decision','accepted','listed',observed_at=row[2]) + self.assertEqual(inspect(self.c)['content_fingerprints'],before['content_fingerprints']) + self.assertEqual(self.c.execute('SELECT count(*) FROM editions').fetchone()[0],3) + def test_any_changed_source_text_rejects_stale_plan_before_membership_updates(self): before=plan(self.c) self.c.execute("UPDATE source_records SET title='New evidence' WHERE external_id='conf/ijcai/fixture2018'")