Validate publication identities and completeness before bulk imports

README.md

@@ -114,6 +114,10 @@ python3 -B import_records.py --db /data/paper-index/papers.sqlite \
 
 采集输出 JSONL 和对应 manifest，只有明确来源标签才产生相应决定：官方论文集记录为 accepted；PeerRead 使用其明确的 accepted 布尔标签，缺失标签保留 unknown。清单数目不一致时整个文件导入回滚，记录失败批次，不能标为覆盖完成。多个历史卷的全量覆盖仍需单独核验。旧卷可能把 PDF 放入归档；采集器对前 50 卷使用 Git partial clone 和 sparse checkout，仅取 `_posts` 和卷配置，不执行出版方项目代码。较新归档若超出 64 MiB 限额，也转为此方式。已对原本超限的 AISTATS 2009（第 5 卷）验证采集 84 条且工作区无 PDF。旧 YAML 的 `!binary` 文本、空文档尾标记会安全解析；带年份的 AISTATS/COLT 等历史 shortname 会规范化，避免名称别名漏届；缺失 year 时仅从官方会议 start 日期提取年份。`pmlr-volume-status.json` 逐卷保存结果和候选范围归属，`pmlr-last-run.json` 记录本批实际请求范围，局部重试不能证明全历史完成。缓存默认有效 24 小时；同一次批次可复用缓存，后续更新会重新读取动态入口。
 
+PMLR 的同一出版 ID 可能对应多份 Git 元数据，例如 CoRL 2023 的两篇论文分别存在不同日期的副本。只有论文内容与标识完全一致、差异限于元数据日期时才合为一条，并保留源文件路径和日期版本；冲突副本会使该卷采集失败，保留上一份有效文件。manifest 分别记录 metadata_post_count、duplicate_post_count 和论文 count。
+
+会议导入器还会独立检查每条记录的会议、年份、track 与 manifest 是否一致，并从实际记录计算缺失摘要数；旧 manifest 即使声称摘要完整，也不能让缺失摘要的来源被标为 complete。缺失的源摘要与合并后论文实体是否从其他来源补得摘要分别统计。
+
 PeerRead 采集仅选择作者公开发布的 ICLR 2017 投稿元数据，固定到源仓库 commit。只依据明确的布尔 `accepted` 标签形成 accepted/rejected；空标签为 unknown，不使用其他数据分区中的 probably-rejected 推断。不会保存 reviews 或 parsed_pdfs 正文。
 
 ```sh

collect_pmlr.py

@@ -127,8 +127,8 @@ def collect_volume(volume,cache,out,allowed=None):
     path = Path(out)/f'pmlr-v{volume}.jsonl'
     path.parent.mkdir(parents=True,exist_ok=True)
     part = path.with_suffix('.part')
-    count = 0
-    missing_abstracts = 0
+    records = {}
+    duplicate_posts = 0
     with part.open('w') as output:
         for member_name,content in metadata_texts(volume,cache):
             parts = re.split(r'(?m)^---\s*$',content,maxsplit=2)
@@ -139,9 +139,9 @@ def collect_volume(volume,cache,out,allowed=None):
                 continue
             if not row.get('title'):
                 raise ValueError('Paper metadata has no title: '+member_name)
-            if not (row.get('abstract') or '').strip():
-                missing_abstracts+=1
-            external = f"v{volume}/{row.get('id') or Path(member_name).stem}"
+            if not row.get('id'):
+                raise ValueError('Paper metadata has no publication ID: '+member_name)
+            external = f"v{volume}/{row['id']}"
             ids = {}
             for key in ('doi','openreview','arxiv'):
                 value = row.get(key)
@@ -153,13 +153,28 @@ def collect_volume(volume,cache,out,allowed=None):
                         value = value.split('id=')[-1].split('&')[0]
                     ids[key] = value
             record = {'source':'pmlr','external_id':external,'url':f"https://proceedings.mlr.press/v{volume}/{row.get('id')}.html",'title':row['title'],'abstract':row.get('abstract'),'authors':row.get('author',[]),'identifiers':ids,'venue':venue,'year':year,'track':track,'status':'accepted','list_url':f'https://proceedings.mlr.press/v{volume}/','metadata':{key:row[key] for key in ('id','volume','date','page','doi','openreview','arxiv','publisher') if key in row}}
+            version={'path':member_name,'date':row.get('date')}
+            if external in records:
+                previous=records[external]
+                if ({k:v for k,v in previous.items() if k!='metadata'} != {k:v for k,v in record.items() if k!='metadata'}
+                    or {k:v for k,v in previous['metadata'].items() if k not in ('date','source_versions')} != {k:v for k,v in record['metadata'].items() if k!='date'}):
+                    raise ValueError('Conflicting metadata posts share a publication ID: '+external)
+                previous['metadata']['source_versions'].append(version)
+                if str(row.get('date') or '')>str(previous['metadata'].get('date') or ''):
+                    previous['metadata']['date']=row['date']
+                duplicate_posts+=1
+                continue
+            record['metadata']['source_versions']=[version]
+            records[external]=record
+        for record in records.values():
             output.write(json.dumps(record,ensure_ascii=False,default=str)+'\n')
-            count += 1
+    count=len(records)
+    missing_abstracts=sum(not (record.get('abstract') or '').strip() for record in records.values())
     if count==0:
         part.unlink(missing_ok=True)
         raise RuntimeError('No papers parsed; refusing an empty successful collection')
     part.replace(path)
-    result = {'source':'pmlr','venue':venue,'year':year,'track':track,'volume':volume,'count':count,'missing_abstract_count':missing_abstracts,'list_url':f'https://proceedings.mlr.press/v{volume}/','file':str(path),'status':'partial' if missing_abstracts else 'collected'}
+    result = {'source':'pmlr','parser_version':2,'venue':venue,'year':year,'track':track,'volume':volume,'count':count,'metadata_post_count':count+duplicate_posts,'duplicate_post_count':duplicate_posts,'missing_abstract_count':missing_abstracts,'list_url':f'https://proceedings.mlr.press/v{volume}/','file':str(path),'status':'partial' if missing_abstracts else 'collected'}
     path.with_suffix('.manifest.json').write_text(json.dumps(result,indent=2))
     print(json.dumps(result),flush=True)
     return result

import_records.py

@@ -22,6 +22,7 @@ def ingest_file(db,path):
     run = c.execute('INSERT INTO ingest_runs(source_id,checkpoint_json) VALUES(?,?)',(source_id,json.dumps({'file':str(path)}))).lastrowid
     c.commit()
     count = 0
+    missing_abstracts = 0
     seen = set()
     editions = set()
     try:
@@ -30,6 +31,8 @@ def ingest_file(db,path):
                 row = json.loads(line)
                 if row['source'] != source_name:
                     raise ValueError('Source differs from manifest')
+                if any(key in manifest and manifest[key]!=row[key] for key in ('venue','year','track')):
+                    raise ValueError('Conference edition differs from manifest')
                 external = row['external_id']
                 if external in seen:
                     raise ValueError('Duplicate external ID in one collected file: '+external)
@@ -43,6 +46,7 @@ def ingest_file(db,path):
                 else:
                     raise ValueError('This importer requires an explicit supported decision status')
                 count += 1
+                missing_abstracts += not bool((row.get('abstract') or '').strip())
         if not count:
             raise ValueError('Empty metadata file')
         if len(editions) != 1:
@@ -51,11 +55,11 @@ def ingest_file(db,path):
         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)
+        complete=expected==count and manifest.get('status') not in ('partial','blocked') and not manifest.get('missing_abstract_count',0) and not missing_abstracts
         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)
+        print(json.dumps({'file':str(path),'status':'imported','records':count,'missing_abstract_count':missing_abstracts}),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))

test_index_core.py

@@ -86,6 +86,24 @@ class StoreTests(unittest.TestCase):
             ingest_file(self.db,path)
         self.assertEqual(self.store.conn.execute('SELECT count(*) FROM source_records').fetchone()[0],0)
 
+    def test_missing_abstract_is_checked_even_when_manifest_claims_complete(self):
+        path=Path(self.tmp.name)/'incomplete.jsonl'
+        row={'source':'pmlr','external_id':'v1/paper','url':'https://example.org/paper','title':'Paper','abstract':None,'authors':[],'venue':'ICML','year':2025,'track':'main','status':'accepted'}
+        path.write_text(json.dumps(row)+'\n')
+        path.with_suffix('.manifest.json').write_text(json.dumps({'source':'pmlr','list_url':'https://example.org/list','count':1,'status':'collected','missing_abstract_count':0}))
+        ingest_file(self.db,path)
+        self.assertEqual(self.store.conn.execute('SELECT status FROM edition_sources').fetchone()[0],'partial')
+
+    def test_manifest_edition_mismatch_rolls_back_instead_of_marking_coverage(self):
+        path=Path(self.tmp.name)/'wrong-edition.jsonl'
+        row={'source':'pmlr','external_id':'v1/paper','url':'https://example.org/paper','title':'Paper','abstract':'Abstract','authors':[],'venue':'ICML','year':2025,'track':'main','status':'accepted'}
+        path.write_text(json.dumps(row)+'\n')
+        path.with_suffix('.manifest.json').write_text(json.dumps({'source':'pmlr','list_url':'https://example.org/list','count':1,'year':2024}))
+        with self.assertRaisesRegex(ValueError,'edition differs'):
+            ingest_file(self.db,path)
+        self.assertEqual(self.store.conn.execute('SELECT count(*) FROM source_records').fetchone()[0],0)
+        self.assertEqual(self.store.conn.execute('SELECT count(*) FROM edition_sources').fetchone()[0],0)
+
 
 if __name__=='__main__':
     unittest.main()

test_pmlr.py

@@ -1,5 +1,7 @@
 import unittest
 import tempfile
+import json
+from pathlib import Path
 from unittest.mock import patch
 
 from collect_pmlr import load_metadata,venue_key,collect_volume
@@ -31,6 +33,28 @@ class MetadataTests(unittest.TestCase):
         self.assertEqual(result['missing_abstract_count'],1)
         self.assertEqual(result['status'],'partial')
 
+    def test_duplicate_publication_posts_preserve_date_evidence_once(self):
+        config=b'shortname: CoRL\nyear: 2023\nbooktitle: CoRL\n'
+        paper='---\nid: paper1\nlayout: inproceedings\ntitle: Paper\nabstract: Real abstract\ndate: {date}\n---\n'
+        posts=[('2023-11-13-paper1.md',paper.format(date='2023-11-13')),('2023-12-02-paper1.md',paper.format(date='2023-12-02'))]
+        with tempfile.TemporaryDirectory() as tmp,patch('collect_pmlr.fetch',return_value=config),patch('collect_pmlr.metadata_texts',return_value=iter(posts)):
+            result=collect_volume(229,tmp,tmp)
+            rows=[json.loads(line) for line in Path(result['file']).read_text().splitlines()]
+        self.assertEqual(result['count'],1)
+        self.assertEqual(result['metadata_post_count'],2)
+        self.assertEqual(result['duplicate_post_count'],1)
+        self.assertEqual(len(rows[0]['metadata']['source_versions']),2)
+        self.assertEqual(rows[0]['metadata']['date'],'2023-12-02')
+
+    def test_conflicting_duplicate_posts_do_not_replace_previous_collection(self):
+        config=b'shortname: CoRL\nyear: 2023\nbooktitle: CoRL\n'
+        paper='---\nid: paper1\nlayout: inproceedings\ntitle: {title}\nabstract: Abstract\n---\n'
+        with tempfile.TemporaryDirectory() as tmp,patch('collect_pmlr.fetch',return_value=config),patch('collect_pmlr.metadata_texts',return_value=iter([('a.md',paper.format(title='First paper')),('b.md',paper.format(title='Other paper'))])):
+            previous=Path(tmp)/'pmlr-v229.jsonl';previous.write_text('previous valid collection')
+            with self.assertRaisesRegex(ValueError,'Conflicting metadata posts'):
+                collect_volume(229,tmp,tmp)
+            self.assertEqual(previous.read_text(),'previous valid collection')
+
 
 if __name__=='__main__':
     unittest.main()