Audit raw arXiv IDs and preserve per-record category evidence

docs/data-sources.md

@@ -130,8 +130,6 @@ paper-atlas collect dblp \
 
 部分早期 SIGIR、CIKM、IJCAI 出版卷使用两位年份键；只有论文集标题明确给出唯一完整年份且与短键相符时才采用，不推测世纪。WWW 特殊兴趣、Alternate track 单列；同卷混合 Applied Data Science、Demo、Industrial 且不能逐篇区分时使用 `proceedings_unspecified`，不把整卷归为某一个具体 track。
 
-书目中的前后页、目录组合和明确会务标题不产生论文录用记录；缺作者的赞助、委员会、活动汇总等标题单独排除。WWW 2019 companion 的匿名合成活动条目归入待核查材料，原始快照仍保留。范围报告分别统计这些原因；单纯缺作者并不排除真实研究论文，也不补造作者。
-
 XML 只允许校验过的本地 DTD；官方通用名 `dblp.dtd` 与月度快照使用的 `dblp-2023-06-28.dtd` 均解析到同一份已校验文件。拒绝其他外部实体和内部 DTD 子集，不访问 XML 声明的任意网络地址，并遵循源 XML 的字符编码。临时目录库或 gzip 检查失败时不发布正式候选文件，需先调查残留的 `.part` 状态，再决定是否重新运行。
 
 公开 ICLR 投稿快照的固定 Release 供后续筛选投稿和明确决定；它是第三方保存的 OpenReview 历史副本，不等于最新官方全量数据：

docs/operations.md

@@ -2,6 +2,24 @@
 
 代码以固定 commit 部署，已经启动的任务继续使用其原版本。目录重构不要求迁移数据库，也不更改已有 JSONL 和 manifest 格式。
 
+## arXiv 原始编号与类别证据
+
+原始行数、唯一 arXiv 编号数和合并后的论文实体数分别核验。以下只读原始压缩包，流式处理到 ZIP EOF/CRC 检查完成；输出审计 JSON 与编号 SQLite 索引，不修改论文主库：
+
+```sh
+paper-atlas audit arxiv --archive /data/paper-index/raw/arxiv-metadata.zip \
+  --out /data/paper-index/collected/arxiv-source-audit.json
+```
+
+同编号的重复行会记录出现位置、次数和内容哈希版本，不直接视作不同论文；同篇论文的多个 arXiv 编号则保留各自来源及类别证据。旧导入器使用共享类别证据键，可能在论文实体合并时丢失某个编号的相同类别出处。新导入使用逐记录证据键；已有数据须在没有其他主库写入任务时按保留的来源 metadata 修复：
+
+```sh
+paper-atlas maintain arxiv-tags --db /data/paper-index/papers.sqlite \
+  --out /data/paper-index/collected/arxiv-tag-repair.json
+```
+
+修复使用同一维护锁，分批提交并保存断点，保留 `custom:*` 和其他来源证据；失败后确认原因已解决，再增加 `--resume` 从上次提交位置继续。完成报告不能作为重新扫描的断点，重新核验应指定新报告。源审计和标签修复均保留 8 GiB 空闲空间，不能代替所有来源入库后的实体、投稿去重与完整性检查。
+
 ## 长期维护与更新
 
 1. 每次采集先核对来源、许可、可访问性、空闲空间和现有运行批次。保持低并发，遵守来源限流、重试与人机验证要求。

paper_atlas/audits/arxiv.py

@@ -0,0 +1,90 @@
+"""Independently count raw arXiv IDs and their payload variants to ZIP EOF."""
+import argparse
+from contextlib import closing
+import fcntl
+import hashlib
+import json
+from pathlib import Path
+import shutil
+import sqlite3
+import tempfile
+import zipfile
+
+from paper_atlas.db.submissions import timestamp
+from paper_atlas.io.snapshots import digest_file
+
+MEMBER='arxiv-metadata-oai-snapshot.json'
+
+
+def fingerprint(path):
+    info=path.stat()
+    return {'bytes':info.st_size,'mtime_ns':info.st_mtime_ns}
+
+
+def audit(archive,out,expected_sha256=None):
+    out=Path(out).resolve();out.parent.mkdir(parents=True,exist_ok=True)
+    with out.with_suffix('.lock').open('a') as lock:
+        fcntl.flock(lock,fcntl.LOCK_EX|fcntl.LOCK_NB)
+        return _audit(archive,out,expected_sha256)
+
+
+def _audit(archive,out,expected_sha256=None):
+    archive=Path(archive).resolve();out=Path(out).resolve()
+    ids=out.with_suffix('.ids.sqlite')
+    if out.exists() or ids.exists():raise ValueError('Audit output exists; inspect it before another run')
+    out.parent.mkdir(parents=True,exist_ok=True)
+    before=fingerprint(archive);identity=digest_file(archive)
+    if expected_sha256 and identity['sha256']!=expected_sha256:raise ValueError('Archive checksum mismatch')
+    if shutil.disk_usage(out.parent).free<8*1024**3:raise RuntimeError('Less than 8 GiB free before source audit')
+    count=0;missing_title=0;missing_abstract=0
+    with tempfile.TemporaryDirectory(prefix='.arxiv-audit-',dir=out.parent) as temporary:
+        stage=Path(temporary)/ids.name
+        with closing(sqlite3.connect(stage)) as c:
+            c.executescript('PRAGMA cache_size=-16384; CREATE TABLE raw_ids(external_id TEXT PRIMARY KEY,first_line INTEGER,last_line INTEGER,occurrences INTEGER,first_sha256 BLOB,has_payload_variants INTEGER) WITHOUT ROWID; CREATE TABLE duplicate_variants(external_id TEXT,payload_sha256 BLOB,first_line INTEGER,last_line INTEGER,occurrences INTEGER,PRIMARY KEY(external_id,payload_sha256)) WITHOUT ROWID;')
+            with zipfile.ZipFile(archive) as zipped:
+                info=zipped.getinfo(MEMBER)
+                with zipped.open(info) as stream:
+                    for count,line in enumerate(stream,1):
+                        row=json.loads(line);external=row.get('id')
+                        if not isinstance(external,str) or not external or external!=external.strip():raise ValueError('Missing or malformed arXiv ID')
+                        payload=hashlib.sha256(json.dumps(row,ensure_ascii=False,sort_keys=True,separators=(',',':')).encode()).digest()
+                        inserted=c.execute('INSERT OR IGNORE INTO raw_ids VALUES(?,?,?,1,?,0)',(external,count,count,payload)).rowcount
+                        if not inserted:
+                            first_line,first=c.execute('SELECT first_line,first_sha256 FROM raw_ids WHERE external_id=?',(external,)).fetchone()
+                            c.execute('INSERT OR IGNORE INTO duplicate_variants VALUES(?,?,?,?,1)',(external,first,first_line,first_line))
+                            c.execute('INSERT INTO duplicate_variants VALUES(?,?,?,?,1) ON CONFLICT(external_id,payload_sha256) DO UPDATE SET last_line=excluded.last_line,occurrences=duplicate_variants.occurrences+1',(external,payload,count,count))
+                            c.execute('UPDATE raw_ids SET last_line=?,occurrences=occurrences+1,has_payload_variants=has_payload_variants OR first_sha256!=? WHERE external_id=?',(count,payload,external))
+                        missing_title+=not bool((row.get('title') or '').strip())
+                        missing_abstract+=not bool((row.get('abstract') or '').strip())
+                        if count%10000==0:
+                            c.commit()
+                            if shutil.disk_usage(out.parent).free<8*1024**3:raise RuntimeError('Less than 8 GiB free during source audit')
+                        if count%100000==0:print(json.dumps({'source':'arxiv_audit','raw_rows':count}),flush=True)
+                # zipfile verifies the member CRC only when its stream reaches EOF.
+            if not count:raise ValueError('Empty arXiv metadata member')
+            if fingerprint(archive)!=before:raise ValueError('Archive changed during audit')
+            c.commit()
+            if c.execute('PRAGMA quick_check').fetchone()[0]!='ok':raise ValueError('Audit index failed SQLite quick_check')
+            unique=c.execute('SELECT count(*) FROM raw_ids').fetchone()[0]
+            duplicate_ids=[]
+            for key,first,last,occurrences,variants in c.execute('SELECT external_id,first_line,last_line,occurrences,has_payload_variants FROM raw_ids WHERE occurrences>1 ORDER BY external_id'):
+                duplicate_ids.append({'external_id':key,'first_line':first,'last_line':last,'occurrences':occurrences,'has_payload_variants':bool(variants),
+                                      'variants':[{'sha256':digest.hex(),'first_line':a,'last_line':b,'occurrences':n} for digest,a,b,n in c.execute('SELECT payload_sha256,first_line,last_line,occurrences FROM duplicate_variants WHERE external_id=? ORDER BY first_line',(key,))]})
+            result={'checked_at':timestamp(),'archive':str(archive),'archive_identity':identity,'member':MEMBER,'member_crc32':f'{info.CRC:08x}',
+                    'uncompressed_bytes':info.file_size,'zip_eof_crc_verified':True,'raw_rows':count,'unique_external_ids':unique,'extra_rows_for_repeated_ids':count-unique,
+                    'repeated_id_count':len(duplicate_ids),'repeated_ids_with_payload_variants':sum(item['has_payload_variants'] for item in duplicate_ids),
+                    'missing_title_rows':missing_title,'missing_abstract_rows':missing_abstract,'duplicate_ids':duplicate_ids,'id_index':str(ids),
+                    'interpretation':'Raw ID/payload audit, not a count of unique research works. Payload variants are retained as evidence; latest source import uses the last row for each ID. Cross-source entity/submission dedup requires a separate database audit.'}
+        stage.replace(ids)
+        part=out.with_suffix('.part');part.write_text(json.dumps(result,ensure_ascii=False,indent=2)+'\n');part.replace(out)
+    print(json.dumps({key:value for key,value in result.items() if key!='duplicate_ids'}),flush=True)
+    return result
+
+
+def main(argv=None,*,prog=None):
+    parser=argparse.ArgumentParser(prog=prog);parser.add_argument('--archive',required=True);parser.add_argument('--out',required=True)
+    parser.add_argument('--expected-sha256');args=parser.parse_args(argv)
+    audit(args.archive,args.out,args.expected_sha256)
+
+
+if __name__=='__main__':main()

paper_atlas/cli.py

@@ -24,7 +24,9 @@ COMMANDS = {
     ('import','records'): ('paper_atlas.ingestion.records', {}, '导入核验过的会议文件'),
     ('maintain','merge'): ('paper_atlas.ingestion.maintenance', {}, '协调单写入合并与恢复'),
     ('maintain','backfill'): ('paper_atlas.ingestion.backfill', {}, '回填既有投稿的来源身份'),
+    ('maintain','arxiv-tags'): ('paper_atlas.ingestion.arxiv_tags', {}, '恢复逐来源记录的 arXiv 类别证据'),
     ('audit','iclr'): ('paper_atlas.audits.iclr', {}, '核对 ICLR 快照与同届官方程序'),
+    ('audit','arxiv'): ('paper_atlas.audits.arxiv', {}, '核验 arXiv 原始编号、重复行与 ZIP 完整性'),
 }
 
 

paper_atlas/collectors/dblp_records.py

@@ -19,20 +19,7 @@ from paper_atlas.db.submissions import timestamp
 from paper_atlas.io.snapshots import digest_file
 
 AS_OF='2026-11-01T00:00:00.000Z'
-MATERIAL_PART=r'(?:title(?: page)?|preface|foreword|editorial|front\s?matter|back\s?matter|(?:table of )?contents|(?:author |subject )?index|contributors|(?:conference |program |organizing |steering )?committee|conference organization|reviewers)'
-FRONTMATTER=re.compile(MATERIAL_PART+r'(?:\s*[,;&]\s*'+MATERIAL_PART+r')*[.!:]?',re.I)
-HEADINGS=re.compile(r'(?:(?:IJCAI|AAAI)(?:[- ]\d{2,4})? |(?:19|20)\d{2} )?(?:conference )?(?:organization|organizers and sponsors|sponsorship|sponsors|(?:non-PC )?reviewers|awards and distinguished papers|keynote and invited speakers|area chairs and program committee|tutorials|panel session|invited talks|workshops|workshop summaries|special track on continuous planning)[.!:]?',re.I)
-
-
-def material_reason(title,authors,key,year,track):
-    if FRONTMATTER.fullmatch(title):return 'frontmatter'
-    if re.search(r"\bchairs['’]? welcome(?:\s*&\s*organization listing)?[.!:]?$",title,re.I):return 'frontmatter'
-    if not authors:
-        if HEADINGS.fullmatch(title) or re.fullmatch(r'(?:past IJCAI conferences|events colocated with IJCAI-\d{2}|welcome message from the conference chairs|welcome to (?:ICDM|IJCAI|AAAI) \d{4})[.!:]?',title,re.I):return 'proceedings_heading'
-        # These anonymous synthetic entries name whole workshop events, rather
-        # than identifiable papers. Hold them without discarding the raw source.
-        if year==2019 and track=='companion' and re.fullmatch(r'conf/www/X19[a-z]?',key):return 'unverified_proceedings_material'
-    return None
+FRONTMATTER=re.compile(r'(?:preface|foreword|editorial|welcome(?: message)?|organizing committee|program committee|title page|table of contents)[.!:]?',re.I)
 
 
 def identifiers(fields):
@@ -74,10 +61,9 @@ def normalize_row(row,volumes,targets,include_workshops=False,through_year=2026)
     titles=fields.get('title',[])
     if len(titles)!=1 or not titles[0].strip():return None,'missing_or_multiple_titles'
     title=' '.join(titles[0].split())
-    authors=[re.sub(r'\s+\d{4}$','',name).strip() for name in fields.get('author',[]) if name.strip()]
-    reason=material_reason(title,authors,row['key'],edition['year'],edition['track'])
-    if reason:return None,reason
+    if FRONTMATTER.fullmatch(title):return None,'frontmatter'
     own,ambiguous=identifiers(fields)
+    authors=[re.sub(r'\s+\d{4}$','',name).strip() for name in fields.get('author',[]) if name.strip()]
     metadata={'snapshot_sha256':XML_SHA,'snapshot_release':RELEASE,'observation_date_precision':'day',
               'proceedings_key':refs[0],'proceedings_title':edition['proceedings_title'],'bibliographic_year':edition['bibliographic_year'],
               'edition_year_basis':edition['edition_year_basis'],

paper_atlas/ingestion/arxiv.py

@@ -10,6 +10,13 @@ from pathlib import Path
 from paper_atlas.db.store import Store
 
 
+def refresh_categories(store,paper,record,categories):
+    """Keep category evidence for each arXiv record, including merged papers."""
+    if not isinstance(categories,str):raise ValueError('arXiv categories must be text')
+    store.record_tags(paper,record,'arxiv',[{'namespace':'arxiv:category','value':value} for value in categories.split()])
+    store.conn.execute("DELETE FROM paper_tags WHERE source_record_id=? AND evidence_key='arxiv:categories' AND tag_id IN (SELECT id FROM tags WHERE namespace='arxiv:category')",(record,))
+
+
 def ingest(db, archive):
     store = Store(db)
     conn = store.conn
@@ -45,11 +52,7 @@ def ingest(db, archive):
                     authors = row.get('authors_parsed') or row.get('authors') or []
                     metadata = {k:v for k,v in row.items() if k not in ('title','abstract','authors','authors_parsed')}
                     paper,record = store.record('arxiv',external_id,'https://arxiv.org/abs/'+external_id,title,abstract,authors,{'arxiv':external_id,'doi':row.get('doi')},metadata,(row.get('versions') or [{}])[0].get('created'))
-                    conn.execute("DELETE FROM paper_tags WHERE source_record_id=? AND evidence_key='arxiv:categories'", (record,))
-                    for category in (row.get('categories') or '').split():
-                        conn.execute("INSERT OR IGNORE INTO tags(namespace,value) VALUES('arxiv:category',?)",(category,))
-                        tag = conn.execute("SELECT id FROM tags WHERE namespace='arxiv:category' AND value=?",(category,)).fetchone()[0]
-                        conn.execute('INSERT OR IGNORE INTO paper_tags VALUES(?,?,?,?)',(paper,tag,record,'arxiv:categories'))
+                    refresh_categories(store,paper,record,row.get('categories') or '')
                     if line_number % 500 == 0:
                         if shutil.disk_usage(Path(db).parent).free < 4*1024**3:
                             raise RuntimeError('Stopped before exhausting disk: less than 4 GiB free')

paper_atlas/ingestion/arxiv_tags.py

@@ -0,0 +1,74 @@
+"""Rebuild per-record arXiv category evidence from retained source metadata."""
+import argparse
+from contextlib import closing
+import fcntl
+import json
+from pathlib import Path
+import shutil
+import sqlite3
+
+from paper_atlas.db.store import Store
+from paper_atlas.db.submissions import timestamp
+from paper_atlas.ingestion.arxiv import refresh_categories
+
+
+def save(path,state):
+    part=path.with_suffix('.part');part.write_text(json.dumps(state,indent=2)+'\n');part.replace(path)
+
+
+def repair(db,out,resume=False,batch_size=1000):
+    db=Path(db).resolve();out=Path(out).resolve()
+    if not db.is_file():raise ValueError('Existing database required')
+    if batch_size<1 or batch_size>10000:raise ValueError('Batch size must be between 1 and 10000')
+    out.parent.mkdir(parents=True,exist_ok=True)
+    with Path(str(db)+'.maintenance.lock').open('a') as lock:
+        fcntl.flock(lock,fcntl.LOCK_EX|fcntl.LOCK_NB)
+        with closing(sqlite3.connect(db.as_uri()+'?mode=ro',uri=True)) as c:
+            source=c.execute("SELECT id FROM sources WHERE name='arxiv'").fetchone()
+            if not source:raise ValueError('No imported arXiv source')
+            total,maximum=c.execute('SELECT count(*),COALESCE(max(id),0) FROM source_records WHERE source_id=?',source).fetchone()
+        scope={'db':str(db),'source_id':source[0],'source_record_count':total,'maximum_source_record_id':maximum}
+        if resume:
+            state=json.loads(out.read_text())
+            if any(state.get(key)!=value for key,value in scope.items()):raise ValueError('Repair scope changed since checkpoint')
+            if state['status']=='complete':raise ValueError('Completed report exists; use a new report for a fresh repair')
+        else:
+            if out.exists():raise ValueError('Repair report exists; inspect it or explicitly resume')
+            state={**scope,'started_at':timestamp(),'last_record_id':0,'records_refreshed':0}
+        if shutil.disk_usage(db.parent).free<8*1024**3:raise RuntimeError('Less than 8 GiB free before repair')
+        state.update(status='running',error=None);save(out,state)
+        c=None
+        try:
+            store=Store(db);c=store.conn
+            while True:
+                rows=c.execute('SELECT id,paper_id,metadata_json FROM source_records WHERE source_id=? AND id>? AND id<=? ORDER BY id LIMIT ?',(source[0],state['last_record_id'],maximum,batch_size)).fetchall()
+                if not rows:break
+                for record,paper,payload in rows:
+                    categories=json.loads(payload).get('categories','')
+                    if categories is None:categories=''
+                    refresh_categories(store,paper,record,categories)
+                c.commit()
+                state.update(last_record_id=rows[-1][0],records_refreshed=state['records_refreshed']+len(rows));save(out,state)
+                if state['records_refreshed']%10000<len(rows):print(json.dumps(state),flush=True)
+                if shutil.disk_usage(db.parent).free<8*1024**3:raise RuntimeError('Less than 8 GiB free during repair')
+            if state['records_refreshed']!=total:raise ValueError('Repaired count differs from fixed source inventory')
+            current=c.execute('SELECT count(*),COALESCE(max(id),0) FROM source_records WHERE source_id=?',source).fetchone()
+            if current!=(total,maximum):raise ValueError('arXiv source inventory changed during repair')
+            legacy=c.execute("SELECT count(*) FROM paper_tags pt JOIN source_records r ON r.id=pt.source_record_id JOIN tags t ON t.id=pt.tag_id WHERE r.source_id=? AND t.namespace='arxiv:category' AND pt.evidence_key='arxiv:categories'",source).fetchone()[0]
+            if legacy:raise ValueError('Legacy category evidence remains after repair')
+            state.update(status='complete',finished_at=timestamp(),legacy_category_evidence_remaining=legacy);save(out,state)
+        except BaseException as error:
+            if c is not None:c.rollback()
+            state.update(status='failed',error=str(error));save(out,state);raise
+        finally:
+            if c is not None:c.close()
+    print(json.dumps(state),flush=True);return state
+
+
+def main(argv=None,*,prog=None):
+    parser=argparse.ArgumentParser(prog=prog);parser.add_argument('--db',required=True);parser.add_argument('--out',required=True)
+    parser.add_argument('--resume',action='store_true');parser.add_argument('--batch-size',type=int,default=1000);args=parser.parse_args(argv)
+    repair(args.db,args.out,args.resume,args.batch_size)
+
+
+if __name__=='__main__':main()

tests/test_arxiv_audit.py

@@ -0,0 +1,44 @@
+from types import SimpleNamespace
+from unittest.mock import patch
+import json
+from pathlib import Path
+import sqlite3
+from contextlib import closing
+import tempfile
+import unittest
+import zipfile
+
+from paper_atlas.audits.arxiv import audit,MEMBER
+
+
+class ArxivSourceAuditTests(unittest.TestCase):
+    def setUp(self):
+        self.temp=tempfile.TemporaryDirectory();self.addCleanup(self.temp.cleanup);self.root=Path(self.temp.name)
+        self.archive=self.root/'metadata.zip';self.out=self.root/'audit.json'
+        disk=patch('paper_atlas.audits.arxiv.shutil.disk_usage',return_value=SimpleNamespace(free=16*1024**3));disk.start();self.addCleanup(disk.stop)
+
+    def write(self,body):
+        with zipfile.ZipFile(self.archive,'w',compression=zipfile.ZIP_DEFLATED) as z:z.writestr(MEMBER,body)
+
+    def test_repeated_ids_and_payload_changes_are_distinct_from_unique_papers(self):
+        one={'id':'2601.00001','title':'A paper','abstract':'First text'};updated={**one,'abstract':'Revised text'}
+        two={'id':'2601.00002','title':'Another paper','abstract':''}
+        self.write('\n'.join(json.dumps(row) for row in (one,updated,updated,two)))
+        result=audit(self.archive,self.out)
+        self.assertEqual((result['raw_rows'],result['unique_external_ids'],result['extra_rows_for_repeated_ids']),(4,2,2))
+        self.assertTrue(result['zip_eof_crc_verified']);self.assertEqual(result['missing_abstract_rows'],1)
+        duplicate=result['duplicate_ids'][0];self.assertEqual(duplicate['occurrences'],3)
+        self.assertTrue(duplicate['has_payload_variants']);self.assertEqual([v['occurrences'] for v in duplicate['variants']],[1,2])
+        with closing(sqlite3.connect(result['id_index'])) as c:self.assertEqual(c.execute('SELECT count(*) FROM raw_ids').fetchone()[0],2)
+        with self.assertRaisesRegex(ValueError,'output exists'):audit(self.archive,self.out)
+
+    def test_invalid_or_unverified_input_never_publishes_audit_outputs(self):
+        self.write('{"id":"2601.00001","title":"Valid"}\nnot JSON')
+        with self.assertRaises(json.JSONDecodeError):audit(self.archive,self.out)
+        self.assertFalse(self.out.exists());self.assertFalse(self.out.with_suffix('.ids.sqlite').exists())
+        self.write('{"id":"2601.00001","title":"Valid"}')
+        with self.assertRaisesRegex(ValueError,'checksum mismatch'):audit(self.archive,self.out,'0'*64)
+        self.assertFalse(self.out.exists())
+
+
+if __name__=='__main__':unittest.main()

tests/test_arxiv_tags.py

@@ -0,0 +1,49 @@
+from contextlib import closing
+import json
+from pathlib import Path
+import sqlite3
+import tempfile
+from types import SimpleNamespace
+import unittest
+from unittest.mock import patch
+
+from paper_atlas.db.store import Store
+from paper_atlas.ingestion.arxiv_tags import repair
+
+
+class ArxivTagRepairTests(unittest.TestCase):
+    def setUp(self):
+        self.temp=tempfile.TemporaryDirectory();self.addCleanup(self.temp.cleanup);self.root=Path(self.temp.name)
+        self.db=self.root/'papers.sqlite';self.out=self.root/'repair.json'
+        disk=patch('paper_atlas.ingestion.arxiv_tags.shutil.disk_usage',return_value=SimpleNamespace(free=16*1024**3));disk.start();self.addCleanup(disk.stop)
+        store=Store(self.db)
+        for key in ('one','two'):
+            paper,record=store.record('arxiv',key,'https://arxiv.org/abs/'+key,'Same title','AB',[],{'doi':'10.1234/same'},{'categories':'cs.AI cs.LG'})
+            for category in ('cs.AI','cs.LG'):
+                store.conn.execute("INSERT OR IGNORE INTO tags(namespace,value) VALUES('arxiv:category',?)",(category,))
+                tag=store.conn.execute("SELECT id FROM tags WHERE namespace='arxiv:category' AND value=?",(category,)).fetchone()[0]
+                store.conn.execute("INSERT OR IGNORE INTO paper_tags VALUES(?,?,?,'arxiv:categories')",(paper,tag,record))
+        tag=store.conn.execute("INSERT INTO tags(namespace,value) VALUES('custom:topic','cs.AI')").lastrowid
+        store.conn.execute("INSERT INTO paper_tags VALUES(?,?,NULL,'manual:1')",(paper,tag));store.conn.commit();store.conn.close()
+
+    def test_restore_collided_evidence_and_preserve_custom_tags_idempotently(self):
+        result=repair(self.db,self.out,batch_size=1)
+        self.assertEqual(result['records_refreshed'],2);self.assertEqual(result['status'],'complete')
+        with closing(sqlite3.connect(self.db)) as c:
+            self.assertEqual(c.execute('SELECT count(*) FROM paper_tags').fetchone()[0],5)
+            self.assertEqual(c.execute('SELECT count(DISTINCT source_record_id) FROM paper_tags').fetchone()[0],2)
+            self.assertEqual(c.execute("SELECT count(*) FROM paper_tags WHERE evidence_key='manual:1'").fetchone()[0],1)
+        repair(self.db,self.root/'again.json')
+        with closing(sqlite3.connect(self.db)) as c:self.assertEqual(c.execute('SELECT count(*) FROM paper_tags').fetchone()[0],5)
+
+    def test_failed_batch_resumes_only_after_last_committed_record(self):
+        with closing(sqlite3.connect(self.db)) as c,c:c.execute("UPDATE source_records SET metadata_json=? WHERE external_id='two'",(json.dumps({'categories':['invalid']}),))
+        with self.assertRaisesRegex(ValueError,'categories must be text'):repair(self.db,self.out,batch_size=1)
+        state=json.loads(self.out.read_text());self.assertEqual(state['records_refreshed'],1);self.assertEqual(state['status'],'failed')
+        with closing(sqlite3.connect(self.db)) as c,c:c.execute("UPDATE source_records SET metadata_json=? WHERE external_id='two'",(json.dumps({'categories':'cs.AI cs.LG'}),))
+        result=repair(self.db,self.out,resume=True,batch_size=1)
+        self.assertEqual(result['records_refreshed'],2);self.assertEqual(result['status'],'complete')
+        with closing(sqlite3.connect(self.db)) as c:self.assertEqual(c.execute('SELECT count(*) FROM paper_tags').fetchone()[0],5)
+
+
+if __name__=='__main__':unittest.main()

tests/test_dblp_records.py

@@ -9,7 +9,7 @@ import unittest
 from unittest.mock import patch
 
 from paper_atlas.collectors.dblp import XML_SHA
-from paper_atlas.collectors.dblp_records import collect,identifiers,normalize_row,material_reason
+from paper_atlas.collectors.dblp_records import collect,identifiers,normalize_row
 from paper_atlas.collectors.dblp_scope import scope
 from paper_atlas.db.store import Store
 from paper_atlas.ingestion.records import ingest_file
@@ -77,21 +77,6 @@ class DblpRecordTests(unittest.TestCase):
         own,ambiguous=identifiers({'ee':['https://doi.org/10.1234/a','https://doi.org/10.1234/b','https://openreview.net/forum?id=2026-Oral--99-placeholder']})
         self.assertEqual(own,{});self.assertEqual(len(ambiguous['doi']),2)
 
-    def test_frontmatter_and_anonymous_event_headers_are_not_paper_acceptances(self):
-        for title in ('Title, Preface.','Title, Conference Committee, Table of Contents, Author Index.','Front Matter.','Backmatter.','Foreword.'):
-            self.assertEqual(material_reason(title,['An Editor'],'conf/sigir/C78',1978,'main'),'frontmatter')
-        for title in ('Sponsors.','IJCAI-09 Awards and Distinguished Papers.','2013 Conference Organization.','Area Chairs and Program Committee.','Welcome to ICDM 2005.','Workshop Summaries.'):
-            self.assertIsNotNone(material_reason(title,[],'conf/ijcai/X09',2009,'main'))
-        self.assertEqual(material_reason("Wiki Workshop Chairs' Welcome.",['An Organizer'],'conf/www/X19t',2019,'companion'),'frontmatter')
-        self.assertEqual(material_reason('Attention - from Neuroscience to the Web and Wellbeing.',[],'conf/www/X19b',2019,'companion'),'unverified_proceedings_material')
-        row=self.row();row['fields']['title']=['Sponsors.'];row['fields']['author']=[]
-        self.assertEqual(normalize_row(row,{'conf/aaai/2024':self.volume('aaai',2024)},{'AAAI'})[1],'proceedings_heading')
-
-    def test_missing_authors_alone_do_not_remove_real_research_records(self):
-        self.assertIsNone(material_reason('The Onomastica interlanguage pronunciation lexicon.',[],'conf/interspeech/X95',1995,'main'))
-        self.assertIsNone(material_reason('Learning Models of Relational Stochastic Processes.',[],'conf/ecml/SanghaiDW05',2005,'main'))
-        self.assertIsNone(material_reason('Sponsors.', ['A Researcher'],'conf/aaai/research',2024,'main'))
-
     def fixture(self,root,rows):
         catalogue=root/'dblp-catalogue.sqlite';volume=self.volume('aaai',2024)
         with closing(sqlite3.connect(catalogue)) as c, c:

tests/test_import_arxiv.py

@@ -7,6 +7,8 @@ import zipfile
 from pathlib import Path
 
 from paper_atlas.ingestion.arxiv import ingest
+from paper_atlas.db.store import Store
+from paper_atlas.ingestion.arxiv import refresh_categories
 
 
 class ImportTests(unittest.TestCase):
@@ -53,6 +55,28 @@ class ImportTests(unittest.TestCase):
         with closing(sqlite3.connect(self.db)) as c, c:
             self.assertEqual(c.execute("SELECT count(*) FROM paper_tags pt JOIN tags t ON pt.tag_id=t.id WHERE t.value='cs.AI'").fetchone()[0],2)
 
+    def test_merged_records_retain_separate_category_evidence_across_refresh(self):
+        one=self.row('2601.00001','10.0000/example');two=self.row('2601.00002','10.0000/example')
+        self.write([one,two]);ingest(self.db,self.archive)
+        with closing(sqlite3.connect(self.db)) as c:
+            self.assertEqual(c.execute('SELECT count(*) FROM paper_tags').fetchone()[0],4)
+            self.assertEqual(c.execute('SELECT count(DISTINCT source_record_id) FROM paper_tags').fetchone()[0],2)
+        one['categories']='cs.AI';two['categories']='cs.LG';self.write([one,two]);ingest(self.db,self.archive)
+        with closing(sqlite3.connect(self.db)) as c:
+            actual=c.execute('SELECT r.external_id,t.value FROM paper_tags pt JOIN source_records r ON r.id=pt.source_record_id JOIN tags t ON t.id=pt.tag_id ORDER BY r.external_id').fetchall()
+            self.assertEqual(actual,[('2601.00001','cs.AI'),('2601.00002','cs.LG')])
+
+    def test_migrate_legacy_categories_without_removing_custom_evidence(self):
+        store=Store(self.db);self.addCleanup(store.conn.close)
+        paper,record=store.record('arxiv','2601.00001','https://arxiv.org/abs/2601.00001','A title','AB',[])
+        for namespace,key in [('arxiv:category','arxiv:categories'),('custom:topic','manual:1')]:
+            tag=store.conn.execute('INSERT INTO tags(namespace,value) VALUES(?,?)',(namespace,'cs.AI')).lastrowid
+            store.conn.execute('INSERT INTO paper_tags VALUES(?,?,?,?)',(paper,tag,record,key))
+        refresh_categories(store,paper,record,'cs.AI cs.LG');refresh_categories(store,paper,record,'cs.AI cs.LG')
+        self.assertEqual(store.conn.execute('SELECT count(*) FROM paper_tags').fetchone()[0],3)
+        self.assertEqual(store.conn.execute("SELECT count(*) FROM paper_tags WHERE evidence_key='arxiv:categories'").fetchone()[0],0)
+        self.assertEqual(store.conn.execute("SELECT count(*) FROM paper_tags WHERE evidence_key='manual:1'").fetchone()[0],1)
+
 
 if __name__ == '__main__':
     unittest.main()