Verify ICLR 2020 legacy forum identities before importing submission history

Reviewed-on: #7

docs/data-sources.md

@@ -113,6 +113,25 @@ python3 -B -m paper_atlas catalogue dblp \
 
 `dblp-catalogue.sqlite` 保存论文集目录、按 crossref/年份/booktitle 的记录数量、各目录族数量及少量样本；`dblp-ai-candidates.jsonl` 保存候选会议族的原始索引元数据。存在多个出版卷字段的记录另存 `volume_anomalies`，保留全部值供归属核验；统计中仍只计一条，不任取第一个字段或展开成多篇。候选范围包含历史名称与 workshop，尚未赋予主会归属或录用状态。后续须依据具体论文集和届次核对目标范围、track 与出版标识，才能生成入库文件；不会把同一目录下的所有记录都视为主会。该盘点不下载 PDF，也不更改论文主库。
 
+目录盘点完成后，可按已核对的书目名称整理指定会议的分届文件。例如先补齐尚无主库记录的会议：
+
+```sh
+paper-atlas collect dblp \
+  --catalogue /data/paper-index/collected/dblp/dblp-catalogue.sqlite \
+  --archive /data/paper-index/raw/dblp-2026-11-01.xml.gz \
+  --dtd /data/paper-index/raw/dblp-2023-06-28.dtd \
+  --venue AAAI --venue IJCAI --venue KDD \
+  --out /data/paper-index/collected/dblp-normalized
+```
+
+也可用 `--candidates` 指定已通过清单 SHA-256 核验的候选文件；完整 XML 模式则可包含初次候选族之外的实际目录，例如 CLeaR 的 `clear2`。规则按书目名、crossref 论文集和会议年核对：不把另一个 `CLEAR` 或 Adaptive Agents 会议误标为 CLeaR/AAMAS；联合会议保留名称别名，出版年与会议年分别保存。workshop 默认单独留待审核，可显式使用 `--include-workshops`；其他已识别的官方附属 track 单列。
+
+输出按会议/年份/track 分文件，附 manifest、`dblp-normalization.json` 范围审查统计和带哈希的 `dblp-import-queue.json`。不能识别的书目、年份或归属会逐类登记，不默认算作主会。DBLP 不含摘要，因此输出摘要为空、清单为 partial；需要通过有证据的跨来源关联补齐论文实体摘要。书目发表证据与原始审稿决定分别保留，不依据书目缺席推断拒稿。已存在其他出版来源的届次，还须先回填其权威身份并核验镜像记录，不能直接累加投稿数。
+
+部分早期 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 历史副本，不等于最新官方全量数据：
@@ -204,6 +223,25 @@ python3 -B -m paper_atlas query --db /data/paper-index/papers.sqlite \
 
 只将 Conference 来源的论文视为 ICLR 主会录用，排除 Blog 和期刊转会展示，并在审计清单保存排除计数。2020/2021 旧数据允许依据官方主会 paper program 形成录用证据；无 forum ID 时保留 program 身份和匹配候选，不伪造标识。历史年度返回 404 会登记 unavailable，未来未发表届次登记 not_yet_published，其他访问或解析失败登记 blocked；这些状态都不能宣称该年已覆盖。
 
+ICLR 2020 的现代程序 JSON 没有真实 forum 链接，数字 event ID 也不能拼接为旧版海报地址。该年未核验身份的记录先使用官方 Papers 索引地址。旧版官方页面实际引用的 [papers.json](https://iclr.cc/virtual_2020/papers.json) 提供 687 个 forum；可用固定大小及 SHA-256 校验的副本进行独立身份核对：
+
+```sh
+paper-atlas download --url https://iclr.cc/virtual_2020/papers.json \
+  --output /data/paper-index/raw/iclr-2020-legacy-papers.json \
+  --expected-bytes 391687 \
+  --sha256 2405f2da6e2641fa600762f7d7cdf7c9d27f4d0ae65799168f3aee46036d30d2
+
+paper-atlas audit iclr-legacy \
+  --program /data/paper-index/collected/iclr-program-2020-main.jsonl \
+  --snapshot /data/paper-index/collected/openreview-snapshot-2020-main.jsonl \
+  --legacy /data/paper-index/raw/iclr-2020-legacy-papers.json \
+  --out /data/paper-index/collected/iclr-2020-verified
+```
+
+核对同时要求同届公开录用、唯一投稿编号和官方 forum 清单一致；编号只是候选关联，还必须有独立的标题/摘要/公开作者证据。旧版作者缩写或姓名格式差异不会放宽全库作者匹配规则。一篇 Explain Your Move 有三个来源中的三种标题，使用逐个核对的精确标题组合及至少两名共同公开作者作例外，不启用通用模糊标题合并。
+
+输出保留原 `2020/program-编号` 来源 ID、标题、摘要和作者，补充真实 forum、核验证据和旧版官方海报 URL。它不写数据库，也不修改输入文件。先在维护锁下使用输出的 `iclr-2020-backfill-queue.json` 给已有程序投稿回填身份，再导入核验后的程序文件与原始投稿快照；这样补充拒稿不会把已有录用事件再建一次。`iclr-2020-snapshot-queue.json` 和审计报告均记录待用文件的哈希，合并前仍须复核。
+
 主题、关键词使用 `iclr_program:topic`、`iclr_program:keyword` 命名空间。通用导入器按源记录刷新标签证据，保留同一论文其他源记录及 `custom:*` 标签。每年的 JSONL、manifest、`iclr-program-year-status.json` 和 `iclr-program-last-run.json` 保存实际范围、独立展示数、主会展示数、论文数、缺失摘要与缺失 forum 数。录用程序不能提供拒稿全集，公开投稿及拒稿补充仍需独立来源。
 
 ### CVF / ECVA 视觉会议元数据
@@ -215,8 +253,20 @@ python3 -B -m paper_atlas collect ecva \
   --cache /data/paper-index/raw/http-cache --out /data/paper-index/collected
 ```
 
-CVF 从官方根目录发现 CVPR、ICCV 主会与 Findings，分别保存 track；不把 workshop 或 WACV 算入这些会议。优先跟随页面明确提供的 All Papers 链接，避免只收录默认显示的第一天。ECVA 从官方总目录按论文 URL 的会议与年份分组，当前总目录列有 ECCV 2018、2020、2022、2024。不存在于来源的 `--venue`、`--year` 或二者组合会报错，不补造双年会议届次，也不能因此推断旧届次未举办。
+CVF 从官方根目录发现 CVPR、ICCV 主会与 Findings，分别保存 track；不把 workshop 或 WACV 算入这些会议。优先跟随页面明确提供的 All Papers 链接；只有逐日目录时遍历全部明确链接的日期，再按论文 URL 合并重复条目并检查标题、作者、DOI 一致性。`.py` 与无后缀的同届入口可关联，不跟随其他会议、年份或站点的日期链接。ECVA 从官方总目录按论文 URL 的会议与年份分组，当前总目录列有 ECCV 2018、2020、2022、2024。不存在于来源的 `--venue`、`--year` 或二者组合会报错，不补造双年会议届次，也不能因此推断旧届次未举办。
 
 只访问论文索引与摘要 HTML/PHP 页面。作者优先取 citation 元数据，旧页面尊重声明编码；无 citation 作者时取索引中的作者行，排除论文集出版信息。DOI 只取该论文条目所属的 DOI 链接或 citation 字段；冲突会失败，不从摘要引用中抽取当前论文 ID。不请求 PDF、补充材料、视频或代码。
 
 每份清单记录索引入口数、论文数、缺失摘要与作者数，缺失内容保留为空并标记 partial。存在未被标题条目解析的论文链接时会失败，防止静默漏项。逐来源的 discovery、index-status、last-run 文件分别保存发现目录、逐届结果和实际请求范围。可重复使用 `--year` 或 `--venue` 限定重试；其他届次状态会保留。官方开放目录之外的早期届次、尚未刊出的新届次和独立出版方历史数据仍需后续补齐，采集器完成不代表全历史覆盖已完成。
+
+单个摘要页面 404 时，仍保存官方目录提供的题目、作者和发表证据，摘要留空并记录失败 URL；整届文件标为 partial。403 等访问拒绝会终止该届采集。刷新缺失摘要时，存储层保留原来已知的论文摘要及其来源观察信息，后续页面恢复时可以更新。
+
+针对旧 catalogue 的 blocked 届次可单独补采到新目录，避免与现有采集任务同时改写状态文件：
+
+```sh
+paper-atlas collect cvf --cache /data/paper-index/raw/http-cache \
+  --retry-catalogue /data/paper-index/collected/cvf-index-status.json \
+  --out /data/paper-index/collected/cvf-repair
+```
+
+parser_version 2 的清单还保存实际遍历的 index_urls、入口错误及缺失摘要页面。补采结果经过清单与身份核验后才导入，不能因重试任务启动就标记缺口已解决。

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. 每次采集先核对来源、许可、可访问性、空闲空间和现有运行批次。保持低并发，遵守来源限流、重试与人机验证要求。
@@ -16,6 +34,16 @@
 
 2026-11-06 从目标服务器测试：Kaggle 元数据 GET 可下载；OpenReview 会议 group 接口可读，但 notes 接口返回 `ChallengeRequiredError`。这限制了直接补全；已使用来源可追溯的公开历史快照补充近期投稿与明确决定。仍不能用录用名单替代全部投稿，也不能声称覆盖了全部拒稿。
 
-已核验的采集文件计数：ICLR 2020–2026 主会 15,534 条、COLM 2024–2026 共 1,573 条，摘要均已取得；候选范围内的 PMLR 文件为 31,428 条，已剔除两份重复出版元数据。ACL Anthology 已选文件为 63,297 条，其中 15,147 条源记录缺摘要。这些是采集文件的来源记录数，不能相加当作去重论文数，也不能代替主库实时统计；实际入库量使用 paper-atlas query --stats 查看。
+## 查询批次与入库状态
 
-代码在 `codex/metadata-index` 工作分支交付，[Draft PR #1](https://gitea.entroia.com/Zayden-Jung/paper-atlas/pulls/1) 尚未合并，等待全量就绪后用户手动合并。各提交分别由远程 CI 验证，结果以对应提交的 Gitea Actions 检查为准。全量导入、其他会议适配器、去重复核和覆盖率核验继续进行，应随实际结果更新此说明。
+采集文件、已入库的来源记录和去重论文实体是不同的计数。来源可能覆盖同一篇论文，同一篇论文也可能有多次投稿；不能把各批文件数量相加作为独立论文总数。实时统计与覆盖报告使用：
+
+```sh
+paper-atlas query --db /data/paper-index/papers.sqlite --stats
+paper-atlas coverage --db /data/paper-index/papers.sqlite \
+  --out /data/paper-index/collected/venue-coverage-current.json
+```
+
+manifest 和采集器状态清单用于确认文件是否齐全及是否缺摘要；`ingest_runs` 和协调器日志用于确认主库导入是否成功。官方程序、书目目录和公开投稿快照分别保留，未出现在某份清单不等于被拒稿；第三方快照也不能代表平台当前全量状态。代码测试成功不能代替最终全库去重、完整性和逐届覆盖核验。
+
+当前代码与合并状态以 [Pull Requests](https://gitea.entroia.com/Zayden-Jung/paper-atlas/pulls) 为准；CI 结果以对应提交的 Actions 检查为准。工作分支交付后等待用户手动合并。合并代码不自动部署、不自动启动采集，也不代表数据批次已完成。

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/audits/iclr_legacy.py

@@ -0,0 +1,172 @@
+"""Reconcile the 2020 program with independently identified official papers."""
+import argparse
+from collections import Counter, defaultdict
+import copy
+import fcntl
+import hashlib
+import json
+from pathlib import Path
+import re
+import tempfile
+
+from paper_atlas.audits.iclr import load
+from paper_atlas.db.store import author_keys, normalized
+from paper_atlas.db.submissions import timestamp
+
+
+LEGACY_URL = 'https://iclr.cc/virtual_2020/papers.json'
+LEGACY_SHA256 = '2405f2da6e2641fa600762f7d7cdf7c9d27f4d0ae65799168f3aee46036d30d2'
+LEGACY_BYTES = 391687
+EXPECTED = 687
+# This one paper has three documented titles across the pinned official legacy
+# list, current official program, and public submission snapshot. The exception
+# also requires independent public-author agreement; it is not fuzzy matching.
+REVIEWED_TITLE_VARIANT = {
+    'forum': 'SJgzLkBKPB',
+    'program': 'Explain Your Move: Understanding Agent Actions Using Salient and Relevant Feature Attribution',
+    'legacy': 'Explain Your Move: Understanding Agent Actions Using Focused Feature Saliency',
+    'snapshot': 'Explain Your Move: Understanding Agent Actions Using Specific and Relevant Feature Attribution',
+}
+
+
+def reconcile(program, snapshot, legacy):
+    by_number = defaultdict(list)
+    for row in snapshot.values():
+        by_number[row['metadata'].get('source_submission_number')].append(row)
+    legacy_by_forum = {}
+    for note in legacy:
+        forum = note.get('forum')
+        content = note.get('content') or {}
+        if not isinstance(forum, str) or not re.fullmatch(r'[A-Za-z0-9_-]+', forum):
+            raise ValueError('Invalid legacy forum identity')
+        if note.get('id') != forum or content.get('iclr_id') != forum or forum in legacy_by_forum:
+            raise ValueError('Conflicting or repeated official legacy forum')
+        if not isinstance(content.get('title'), str) or not content['title'].strip() or not isinstance(content.get('authors'), list):
+            raise ValueError('Legacy paper lacks title or public authors')
+        legacy_by_forum[forum] = content
+    if len(program) != len(legacy_by_forum):
+        raise ValueError('Official program and legacy inventories differ')
+    title_counts = Counter(normalized(row['title']) for row in legacy_by_forum.values())
+    enriched = []
+    assigned = set()
+    numbers = set()
+    evidence_counts = Counter()
+    for original in program:
+        if any(original.get(k) != v for k, v in {'source': 'iclr_program', 'venue': 'ICLR', 'year': 2020, 'track': 'main', 'status': 'accepted'}.items()):
+            raise ValueError('Program record is outside the accepted 2020 main track')
+        metadata = original.get('metadata') or {}
+        number = metadata.get('program_source_id')
+        if metadata.get('program_identity_type') != 'submission' or type(number) is not int or number <= 0 or number in numbers:
+            raise ValueError('Missing or repeated program submission number')
+        numbers.add(number)
+        candidates = by_number[number]
+        if len(candidates) != 1:
+            raise ValueError('Submission number is absent or ambiguous in the public snapshot')
+        submission = candidates[0]
+        forum = submission['identifiers']['openreview']
+        if submission['status'] != 'accepted' or forum not in legacy_by_forum or forum in assigned:
+            raise ValueError('Snapshot acceptance and official forum inventory do not agree')
+        content = legacy_by_forum[forum]
+        shared_legacy = len(author_keys(original['authors']) & author_keys(content['authors']))
+        shared_snapshot = len(author_keys(original['authors']) & author_keys(submission['authors']))
+        checks = {'legacy_title': normalized(original['title']) == normalized(content['title']),
+                  'snapshot_title': normalized(original['title']) == normalized(submission['title']),
+                  'snapshot_abstract': bool(original.get('abstract')) and original['abstract'].strip() == (submission.get('abstract') or '').strip(),
+                  'two_snapshot_authors': shared_snapshot >= 2}
+        independent_snapshot = sum(checks[k] for k in ('snapshot_title', 'snapshot_abstract', 'two_snapshot_authors')) >= 2
+        checks['legacy_title_unique'] = title_counts[normalized(content['title'])] == 1
+        independent_legacy = checks['legacy_title'] and checks['legacy_title_unique'] and shared_legacy >= 1
+        reviewed_variant = (forum == REVIEWED_TITLE_VARIANT['forum']
+                            and normalized(original['title']) == normalized(REVIEWED_TITLE_VARIANT['program'])
+                            and normalized(content['title']) == normalized(REVIEWED_TITLE_VARIANT['legacy'])
+                            and normalized(submission['title']) == normalized(REVIEWED_TITLE_VARIANT['snapshot'])
+                            and shared_legacy >= 2 and shared_snapshot >= 2)
+        if not (independent_snapshot or independent_legacy or reviewed_variant):
+            raise ValueError('Submission number alone cannot establish forum identity: ' + original['external_id'])
+        own = original.get('identifiers') or {}
+        if own.get('openreview') not in (None, forum):
+            raise ValueError('Existing program forum conflicts with independent evidence')
+        row = copy.deepcopy(original)
+        row['identifiers'] = {**own, 'openreview': forum}
+        row['url'] = 'https://iclr.cc/virtual_2020/poster_' + forum + '.html'
+        row['metadata'].update(identity_state='forum', program_submission_number_verified=True,
+            program_identity_verification={'method': 'official_legacy_and_public_submission_crosscheck',
+                'legacy_url': LEGACY_URL, 'legacy_sha256': LEGACY_SHA256,
+                'original_program_url': original['url'], 'legacy_forum': forum,
+                'checks': checks, 'shared_legacy_authors': shared_legacy,
+                'shared_snapshot_authors': shared_snapshot, 'reviewed_title_variant': reviewed_variant})
+        enriched.append(row)
+        assigned.add(forum)
+        evidence_counts.update({'independent_snapshot': int(independent_snapshot),
+                                'independent_legacy': int(independent_legacy),
+                                'reviewed_title_variant': int(reviewed_variant)})
+    if assigned != legacy_by_forum.keys():
+        raise ValueError('Unmatched official legacy forum identities remain')
+    if assigned != {forum for forum, row in snapshot.items() if row['status'] == 'accepted'}:
+        raise ValueError('Accepted snapshot forum inventory differs from the official program')
+    return enriched, dict(evidence_counts)
+
+
+def audit(program_path, snapshot_path, legacy_path, out):
+    program_path, snapshot_path, legacy_path = map(lambda p: Path(p).resolve(), (program_path, snapshot_path, legacy_path))
+    raw = legacy_path.read_bytes()
+    if len(raw) != LEGACY_BYTES or hashlib.sha256(raw).hexdigest() != LEGACY_SHA256:
+        raise ValueError('Official legacy snapshot checksum mismatch')
+    legacy = json.loads(raw)
+    if not isinstance(legacy, list) or len(legacy) != EXPECTED:
+        raise ValueError('Official legacy count mismatch')
+    manifest = json.loads(program_path.with_suffix('.manifest.json').read_text())
+    if manifest.get('parser_version', 0) < 2 or any(manifest.get(k) != v for k, v in {'source': 'iclr_program', 'venue': 'ICLR', 'year': 2020, 'track': 'main', 'count': EXPECTED}.items()):
+        raise ValueError('Invalid collected program inventory')
+    program_bytes = program_path.read_bytes()
+    program = [json.loads(line) for line in program_bytes.splitlines()]
+    if len(program) != EXPECTED or len({r['external_id'] for r in program}) != EXPECTED:
+        raise ValueError('Program count or source identity mismatch')
+    snapshot, snapshot_manifest, snapshot_sha = load(snapshot_path, 'openreview_snapshot', 2020)
+    rows, checks = reconcile(program, snapshot, legacy)
+    out = Path(out).resolve()
+    if out / 'iclr-program-2020-main.jsonl' in (program_path, snapshot_path, legacy_path):
+        raise ValueError('Output must preserve the original input files')
+    out.mkdir(parents=True, exist_ok=True)
+    with (out / 'iclr-legacy.lock').open('a') as lock:
+        fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        if (out / 'iclr-2020-identity-audit.json').exists():
+            raise ValueError('Completed identity audit exists; choose a new output directory')
+        with tempfile.TemporaryDirectory(prefix='.iclr-legacy-', dir=out) as temporary:
+            stage = Path(temporary)
+            path = stage / 'iclr-program-2020-main.jsonl'
+            path.write_text(''.join(json.dumps(row, ensure_ascii=False) + '\n' for row in rows))
+            digest = hashlib.sha256(path.read_bytes()).hexdigest()
+            manifest.update(parser_version=3, file=str(out / path.name), file_sha256=digest,
+                            missing_forum_count=0, legacy_identity_sha256=LEGACY_SHA256)
+            path.with_suffix('.manifest.json').write_text(json.dumps(manifest, indent=2) + '\n')
+            backfill = {'files': [str(out / path.name)], 'file_sha256': {str(out / path.name): digest}}
+            snapshots = {'files': [str(snapshot_path)], 'file_sha256': {str(snapshot_path): snapshot_sha}}
+            result = {'year': 2020, 'checked_at': timestamp(), 'program_records': len(rows),
+                      'snapshot_records': len(snapshot), 'snapshot_status_counts': snapshot_manifest['submission_status_counts'],
+                      'legacy_url': LEGACY_URL, 'legacy_sha256': LEGACY_SHA256,
+                      'program_input_sha256': hashlib.sha256(program_bytes).hexdigest(),
+                      'corroboration_counts': checks, 'backfill_queue': backfill, 'snapshot_files': snapshots,
+                      'note': 'No database writes. Backfill existing program identities before importing the snapshot; source IDs, titles and abstracts are preserved.'}
+            (stage / 'iclr-2020-backfill-queue.json').write_text(json.dumps(backfill, indent=2) + '\n')
+            (stage / 'iclr-2020-snapshot-queue.json').write_text(json.dumps(snapshots, indent=2) + '\n')
+            report = stage / 'iclr-2020-identity-audit.json'
+            report.write_text(json.dumps(result, ensure_ascii=False, indent=2) + '\n')
+            for child in sorted(stage.iterdir(), key=lambda child: child == report):
+                child.replace(out / child.name)
+    print(json.dumps(result), flush=True)
+    return result
+
+
+def main(argv=None, *, prog=None):
+    parser = argparse.ArgumentParser(prog=prog)
+    parser.add_argument('--program', required=True)
+    parser.add_argument('--snapshot', required=True)
+    parser.add_argument('--legacy', required=True)
+    parser.add_argument('--out', required=True)
+    args = parser.parse_args(argv)
+    audit(args.program, args.snapshot, args.legacy, args.out)
+
+
+if __name__ == '__main__':
+    main()

paper_atlas/cli.py

@@ -18,12 +18,16 @@ COMMANDS = {
     ('collect','neurips'): ('paper_atlas.collectors.proceedings', {'source':'neurips'}, 'NeurIPS 官方论文集'),
     ('collect','cvf'): ('paper_atlas.collectors.vision', {'source':'cvf'}, 'CVPR / ICCV 官方开放目录'),
     ('collect','ecva'): ('paper_atlas.collectors.vision', {'source':'ecva'}, 'ECCV 官方开放目录'),
+    ('collect','dblp'): ('paper_atlas.collectors.dblp_records', {}, '核对并整理 DBLP 历史发表记录'),
     ('catalogue','dblp'): ('paper_atlas.collectors.dblp', {}, '盘点 DBLP 历史会议目录'),
     ('import','arxiv'): ('paper_atlas.ingestion.arxiv', {}, '导入 arXiv 元数据 ZIP'),
     ('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','iclr-legacy'): ('paper_atlas.audits.iclr_legacy', {}, '独立核验 ICLR 2020 旧目录投稿身份'),
+    ('audit','arxiv'): ('paper_atlas.audits.arxiv', {}, '核验 arXiv 原始编号、重复行与 ZIP 完整性'),
 }
 
 

paper_atlas/collectors/dblp_records.py

@@ -0,0 +1,186 @@
+"""Normalize pinned DBLP publication metadata into audited edition files."""
+import argparse
+from collections import Counter,OrderedDict
+from contextlib import closing
+import fcntl
+import hashlib
+import json
+from pathlib import Path
+import re
+import shutil
+import sqlite3
+import tempfile
+from urllib.parse import parse_qs,unquote,urlsplit
+
+from paper_atlas.config import DEFAULT_REGISTRY
+from paper_atlas.collectors.dblp import XML_SHA,DTD_MD5,RELEASE,records
+from paper_atlas.collectors.dblp_scope import scope
+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
+
+
+def identifiers(fields):
+    dois=set();arxiv=set();forums=set()
+    for link in fields.get('ee',[]):
+        parsed=urlsplit(link);path=unquote(parsed.path)
+        doi=None
+        if parsed.netloc in ('doi.org','dx.doi.org','doi.ieeecomputersociety.org'):doi=path.lstrip('/')
+        elif parsed.netloc=='link.springer.com' and path.startswith('/chapter/'):doi=path.removeprefix('/chapter/')
+        elif parsed.netloc=='dl.acm.org':
+            match=re.match(r'/doi/(?:abs/|full/|pdf/)?(10\..+)',path)
+            if match:doi=match[1]
+        if doi and re.fullmatch(r'10\.\d{4,9}/\S+',doi):dois.add(doi.lower())
+        if parsed.netloc in ('arxiv.org','www.arxiv.org'):
+            match=re.fullmatch(r'/(?:abs|pdf)/((?:\d{4}\.\d{4,5}|[a-zA-Z.-]+/\d{7})(?:v\d+)?)(?:\.pdf)?',path)
+            if match:arxiv.add(re.sub(r'v\d+$','',match[1]))
+        if parsed.netloc=='openreview.net' and parsed.path in ('/forum','/pdf'):
+            values=parse_qs(parsed.query).get('id',[])
+            if len(values)==1 and re.fullmatch(r'[A-Za-z0-9_-]+',values[0]) and not re.match(r'20\d{2}-Oral-',values[0]):forums.add(values[0])
+    for doi in dois:
+        match=re.fullmatch(r'10\.48550/arxiv\.((?:\d{4}\.\d{4,5}|[a-z.-]+/\d{7})(?:v\d+)?)',doi)
+        if match:arxiv.add(re.sub(r'v\d+$','',match[1]))
+    publications={doi for doi in dois if not doi.startswith('10.48550/arxiv.')}
+    own={}
+    for scheme,values in (('doi',publications),('arxiv',arxiv),('openreview',forums)):
+        if len(values)==1:own[scheme]=next(iter(values))
+    ambiguous={scheme:sorted(values) for scheme,values in (('doi',publications),('arxiv',arxiv),('openreview',forums)) if len(values)>1}
+    return own,ambiguous
+
+
+def normalize_row(row,volumes,targets,include_workshops=False,through_year=2026):
+    if row['record_type']!='inproceedings':return None,'not_a_paper'
+    fields=row['fields'];refs=fields.get('crossref',[])
+    if len(refs)!=1:return None,'ambiguous_or_missing_crossref'
+    edition,reason=scope(row,volumes.get(refs[0]),through_year)
+    if not edition:return None,reason
+    if not ({edition['venue']}|set(edition['venue_aliases']))&targets:return None,'outside_requested_venues'
+    if edition['track']=='workshop' and not include_workshops:return None,'separate_workshop'
+    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
+    own,ambiguous=identifiers(fields)
+    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'],
+              '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'}
+    flagged=bool(re.search(r'\b(?:withdrawn|retracted)\b',row.get('publtype') or '',re.I))
+    result={'source':'dblp','external_id':row['key'],'url':'https://dblp.org/rec/'+row['key']+'.html',
+            'title':title,'abstract':None,'authors':authors,'identifiers':own,
+            **{key:edition[key] for key in ('venue','venue_aliases','year','track')},
+            'status':'unknown' if flagged else 'accepted','observed_at':AS_OF,
+            'raw_decision':'listed in pinned DBLP inproceedings bibliography','metadata':metadata,
+            'tags':[{'namespace':'dblp:booktitle','value':fields['booktitle'][0]}]}
+    if flagged:result['decision_events']=[]
+    return result,None
+
+
+def collect(catalogue,out,venues=None,candidates=None,archive=None,dtd=None,include_workshops=False,through_year=2026):
+    if bool(candidates)==bool(archive):raise ValueError('Choose exactly one candidates file or original archive')
+    catalogue=Path(catalogue);manifest=json.loads(catalogue.with_name('dblp-ai-candidates.manifest.json').read_text())
+    if manifest['snapshot_sha256']!=XML_SHA:raise ValueError('Unsupported snapshot identity')
+    if candidates:
+        if digest_file(Path(candidates))['sha256']!=manifest['candidate_sha256']:raise ValueError('Candidate checksum mismatch')
+    else:
+        if not dtd or digest_file(Path(dtd))['md5']!=DTD_MD5:raise ValueError('DTD checksum mismatch')
+        if digest_file(Path(archive))['sha256']!=XML_SHA:raise ValueError('Archive checksum mismatch')
+    with closing(sqlite3.connect(catalogue.resolve().as_uri()+'?mode=ro',uri=True)) as c:
+        volumes={key:json.loads(payload) for key,payload in c.execute('SELECT key,metadata_json FROM proceedings')}
+    if len(volumes)!=manifest['record_types']['proceedings']:raise ValueError('Catalogue inventory mismatch')
+    allowed={row['acronym'].casefold():row['acronym'] for row in json.loads(DEFAULT_REGISTRY.read_text())['venues']}
+    if venues and set(name.casefold() for name in venues)-allowed.keys():raise ValueError('Unknown registry venue')
+    targets={allowed[name.casefold()] for name in venues} if venues else set(allowed.values())
+    out=Path(out);out.mkdir(parents=True,exist_ok=True)
+    with (out/'dblp.normalize.lock').open('a') as lock:
+        fcntl.flock(lock,fcntl.LOCK_EX|fcntl.LOCK_NB)
+        return _collect_rows(volumes,out,targets,candidates,archive,dtd,include_workshops,through_year,manifest)
+
+
+def _collect_rows(volumes,out,targets,candidates,archive,dtd,include_workshops,through_year,source_manifest):
+    counts=Counter();excluded=Counter();examples={};groups={};seen=set();handles=OrderedDict();scanned=0
+    def rows():
+        if candidates:
+            with Path(candidates).open() as stream:
+                for line in stream:yield json.loads(line)
+        else:yield from records(archive,Path(dtd).read_bytes())
+    if shutil.disk_usage(out).free<8*1024**3:raise RuntimeError('Less than 8 GiB free before normalization')
+    with tempfile.TemporaryDirectory(prefix='.dblp-normalize-',dir=out) as temporary:
+        stage=Path(temporary)
+        try:
+            for row in rows():
+                scanned+=1
+                record,reason=normalize_row(row,volumes,targets,include_workshops,through_year)
+                if record is None:
+                    excluded[reason]+=1
+                    label=reason+'|'+'/'.join(row.get('key','').split('/')[:2])+'|'+';'.join(row['fields'].get('booktitle',[]))
+                    if label not in examples:examples[label]={'count':0,'example_key':row.get('key')}
+                    examples[label]['count']+=1
+                else:
+                    key=record['external_id']
+                    if key in seen:raise ValueError('Repeated DBLP publication key')
+                    seen.add(key);group=(record['venue'],record['year'],record['track'])
+                    filename='dblp-'+re.sub(r'[^a-z0-9]+','-',group[0].casefold())+'-'+str(group[1])+'-'+group[2]+'.jsonl'
+                    if group not in groups:groups[group]={'filename':filename,'count':0,'unknown':0,'ambiguous_identifiers':0}
+                    info=groups[group];info['count']+=1;info['unknown']+=record['status']=='unknown';info['ambiguous_identifiers']+=bool(record['metadata']['ambiguous_identifiers'])
+                    if group not in handles:
+                        if len(handles)>=32:handles.popitem(last=False)[1].close()
+                        handles[group]=(stage/filename).open('a')
+                    handles.move_to_end(group);handles[group].write(json.dumps(record,ensure_ascii=False)+'\n');counts[record['venue']]+=1
+                if scanned%50000==0:
+                    if shutil.disk_usage(out).free<8*1024**3:raise RuntimeError('Less than 8 GiB free during normalization')
+                    print(json.dumps({'source':'dblp','scanned':scanned,'normalized':sum(counts.values())}),flush=True)
+        finally:
+            for handle in handles.values():handle.close()
+        expected=source_manifest['candidate_records'] if candidates else sum(source_manifest['record_types'].values())
+        if scanned!=expected:raise ValueError('Input count differs from verified catalogue')
+        if not seen:raise ValueError('No target publications were verified')
+        files=[]
+        for (venue,year,track),info in sorted(groups.items()):
+            path=out/info['filename'];checksum=digest_file(stage/info['filename'])['sha256']
+            item={'source':'dblp','parser_version':1,'venue':venue,'year':year,'track':track,'count':info['count'],
+                  'missing_abstract_count':info['count'],'unknown_status_count':info['unknown'],'ambiguous_identifier_count':info['ambiguous_identifiers'],
+                  'list_url':RELEASE,'status':'partial','file':str(path),'file_sha256':checksum,'snapshot_sha256':XML_SHA,
+                  'abstract_note':'DBLP provides no abstract; canonical abstracts must come from independently linked sources.'}
+            (stage/info['filename']).replace(path)
+            meta=path.with_suffix('.manifest.json');part=meta.with_suffix('.part');part.write_text(json.dumps(item,indent=2));part.replace(meta);files.append(item)
+        result={'source':'dblp','finished_at':timestamp(),'snapshot_sha256':XML_SHA,'input_mode':'candidates' if candidates else 'original_archive',
+                'requested_venues':sorted(targets),'include_workshops':include_workshops,'through_year':through_year,'scanned_records':scanned,
+                'normalized_records':len(seen),'records_by_venue':dict(counts),'excluded_by_reason':dict(excluded),'scope_review':examples,'files':files,
+                'input_candidate_sha256':source_manifest['candidate_sha256'] if candidates else None,
+                'note':'Classification covers requested recognizable bibliography entries, not an independent proof of complete venue histories. Unknown scopes remain in scope_review.'}
+        target=out/'dblp-normalization.json';part=target.with_suffix('.part');part.write_text(json.dumps(result,ensure_ascii=False,indent=2));part.replace(target)
+        queue={'files':[item['file'] for item in files],'file_sha256':{item['file']:item['file_sha256'] for item in files}}
+        target=out/'dblp-import-queue.json';part=target.with_suffix('.part');part.write_text(json.dumps(queue,indent=2));part.replace(target)
+        print(json.dumps({key:result[key] for key in ('scanned_records','normalized_records','records_by_venue','excluded_by_reason')}),flush=True)
+        return result
+
+
+def main(argv=None,*,prog=None):
+    parser=argparse.ArgumentParser(prog=prog);parser.add_argument('--catalogue',required=True);parser.add_argument('--out',required=True)
+    inputs=parser.add_mutually_exclusive_group(required=True);inputs.add_argument('--candidates');inputs.add_argument('--archive')
+    parser.add_argument('--dtd');parser.add_argument('--venue',action='append');parser.add_argument('--include-workshops',action='store_true')
+    parser.add_argument('--through-year',type=int,default=2026);args=parser.parse_args(argv)
+    collect(args.catalogue,args.out,args.venue,args.candidates,args.archive,args.dtd,args.include_workshops,args.through_year)
+
+
+if __name__=='__main__':main()

paper_atlas/collectors/dblp_scope.py

@@ -0,0 +1,96 @@
+"""Explicit bibliographic aliases; a directory family alone is not venue evidence."""
+import re
+
+
+# Ordered long/joint names precede their shorter forms.
+BOOKS = {
+    'aaai': [('AAAI/IAAI','AAAI+IAAI',['AAAI','IAAI']),('AAAI','AAAI',[])],
+    'ijcai': [('IJCAI/ECAI','ECAI+IJCAI',['ECAI','IJCAI']),('IJCAI','IJCAI',[])],
+    'ecai': [('ECAI','ECAI',[]),('AISB/GI (ECAI)','ECAI',[]),('AISB (ECAI)','ECAI',[])],
+    'nips': [('NeurIPS Datasets and Benchmarks','NeurIPS',[]),('NeurIPS','NeurIPS',[]),('NIPS','NeurIPS',[])],
+    'icml': [('ICML','ICML',[]),('ML','ICML',[])], 'iclr':[('ICLR','ICLR',[])],
+    'colm':[('COLM','COLM',[])], 'colt':[('COLT/EuroCOLT','COLT',[]),('COLT','COLT',[])],
+    'aistats':[('AISTATS','AISTATS',[])], 'uai':[('UAI','UAI',[])],
+    'clear2':[('CLeaR','CLeaR',[])],
+    'acl':[('COLING-ACL','ACL+COLING',['ACL','COLING']),('ACL/IJCNLP','ACL',[]),('ACL','ACL',[])],
+    'emnlp':[('EMNLP-CoNLL','CoNLL+EMNLP',['CoNLL','EMNLP']),('EMNLP/IJCNLP','EMNLP',[]),('EMNLP','EMNLP',[])],
+    'naacl':[('HLT/EMNLP','EMNLP',[]),('NAACL-HLT','NAACL',[]),('HLT-NAACL','NAACL',[]),('NAACL','NAACL',[]),('HLT','NAACL',[])],
+    'eacl':[('EACL','EACL',[])],
+    'coling':[('LREC/COLING','COLING+LREC',['COLING','LREC']),('COLING-ACL','ACL+COLING',['ACL','COLING']),('COLING','COLING',[])],
+    'lrec':[('LREC/COLING','COLING+LREC',['COLING','LREC'])],
+    'conll':[('CoNLL/LLL','CoNLL',[]),('CoNLL','CoNLL',[])],
+    'cvpr':[('CVPR','CVPR',[])], 'iccv':[('ICCVW','ICCV',[]),('ICCV','ICCV',[])],
+    'eccv':[('ECCV','ECCV',[])], 'kdd':[('KDD','KDD',[])], 'sigir':[('SIGIR','SIGIR',[])],
+    'www':[('WWW','WWW',[])], 'wsdm':[('WSDM','WSDM',[])], 'recsys':[('RecSys','RecSys',[])],
+    'cikm':[('CIKM','CIKM',[])], 'icdm':[('ICDM','ICDM',[])], 'sdm':[('SDM','SDM',[])],
+    'ecml':[('ECML','ECML',['ECML-PKDD']),('EWSL','EWSL',['ECML-PKDD'])],
+    'pkdd':[('ECML/PKDD','ECML-PKDD',[]),('ECML PKDD','ECML-PKDD',[]),('PKDD','PKDD',['ECML-PKDD'])],
+    'icra':[('ICRA','ICRA',[])], 'iros':[('IROS','IROS',[])],
+    'rss':[('Robotics: Science and Systems','RSS',[])], 'corl':[('CoRL','CoRL',[])],
+    'aips':[('ICAPS','ICAPS',[]),('AIPS','AIPS',['ICAPS'])], 'icaps':[('ICAPS','ICAPS',[])],
+    'kr':[('KR','KR',[])], 'atal':[('AAMAS','AAMAS',[])],
+    'icassp':[('ICASSP','ICASSP',[])],
+    'interspeech':[('INTERSPEECH','INTERSPEECH',[]),('EUROSPEECH','EUROSPEECH',['INTERSPEECH']),('ICSLP','ICSLP',['INTERSPEECH']),('ECST','ECST',['INTERSPEECH'])],
+    'ismir':[('ISMIR','ISMIR',[])], 'miccai':[('MICCAI','MICCAI',[])],
+    'fat':[('FAccT','FAccT',[]),('FAT*','FAccT',[]),('FAT','FAccT',[])],
+}
+
+
+def scope(row,volume,through_year=2026):
+    fields=row['fields'];books=fields.get('booktitle',[])
+    if len(books)!=1:return None,'ambiguous_booktitle'
+    book=books[0];key=row.get('key','');parts=key.split('/')
+    family=parts[1] if len(parts)>=3 and parts[0]=='conf' else ''
+    if family not in BOOKS:return None,'outside_target_family'
+    if not volume:return None,'missing_proceedings'
+    title=' '.join(volume['fields'].get('title',[]))
+    if family=='clear2' and 'causal learning and reasoning' not in title.casefold():return None,'ambiguous_venue_name'
+    matched=None
+    for base,venue,aliases in BOOKS[family]:
+        match=re.fullmatch(re.escape(base)+r'(?:\s+\((.*)\)|\s+(.*)|,\s*(Vol\.\s*\d+))?',book,re.I)
+        if match:
+            matched=(base,venue,aliases,' '.join(value for value in match.groups() if value));break
+    if not matched:return None,'unclassified_booktitle'
+    base,venue,aliases,suffix=matched
+    volume_family=volume['key'].split('/')[1] if volume['key'].startswith('conf/') else ''
+    if volume_family!=family and not any(item[1]==venue for item in BOOKS.get(volume_family,[])):
+        return None,'crossref_venue_conflict'
+    year_match=re.match(r'conf/[^/]+/((?:19|20)\d{2})',volume['key'])
+    year_basis='proceedings key, checked against bibliographic year or explicit event year in proceedings title'
+    if year_match:year=int(year_match[1])
+    else:
+        # Old SIGIR/CIKM/IJCAI keys use two digits. Never guess their century.
+        short=re.fullmatch(r'conf/[^/]+/(\d{2})',volume['key'])
+        years=set(re.findall(r'\b(?:19|20)\d{2}\b',title))
+        if not short or len(years)!=1:return None,'unverified_edition_year'
+        year=int(next(iter(years)))
+        if year%100!=int(short[1]):return None,'unverified_edition_year'
+        year_basis='unique explicit event year in proceedings title, corroborated by two-digit proceedings key'
+    if year>through_year:return None,'future_edition'
+    bibliographic=fields.get('year',[])
+    if len(bibliographic)!=1 or not re.fullmatch(r'\d{4}',bibliographic[0]):return None,'ambiguous_bibliographic_year'
+    if int(bibliographic[0])!=year and not re.search(r'\b'+str(year)+r'\b',title):return None,'unverified_year_difference'
+    text=(suffix+' '+title).casefold();track='main'
+    if 'findings' in text:track='findings'
+    elif 'shared task' in text:track='shared_task'
+    elif 'student research' in text or 'workshop' in text or 'wokshop' in text or base in ('ICCVW','EWSL'):track='workshop'
+    elif ('demo track' in text and ('industrial track' in text or 'applied data science track' in text)):track='proceedings_unspecified'
+    elif 'demonstration' in text or re.search(r'\bdemos?\b',text):track='demo'
+    elif 'tutorial' in text:track='tutorial'
+    elif 'industry' in text or 'industrial track' in text or 'applied data science' in text:track='industry'
+    elif 'position paper' in text:track='position'
+    elif 'companion' in text:track='companion'
+    elif 'late-breaking' in text or 'late breaking' in text:track='late_breaking'
+    elif 'doctoral' in text:track='doctoral'
+    elif 'challenge' in text or 'competition' in text:track='competition'
+    elif 'bridge program' in text:track='bridge'
+    elif 'special interest tracks' in text:track='special_interest'
+    elif 'alternate track' in text:track='alternate'
+    elif base=='NeurIPS Datasets and Benchmarks':track='datasets_benchmarks'
+    elif suffix and not (re.fullmatch(r'(?:\d+|Vol\.\s*\d+|Volume \d+: (?:Long|Short) Papers|Long Papers|Short Papers|Selected Papers|Fall)',suffix,re.I)
+                         or suffix.casefold() in ('poster','posters','oral','spotlight')):return None,'unclassified_track'
+    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'
+    return {'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},None

paper_atlas/collectors/iclr.py

@@ -141,6 +141,10 @@ def parse_program(data, abstracts, year, index_url):
         external = f'{year}/{forum}' if forum else f'{year}/program-{identity[1]}'
         missing_identity += not bool(forum)
         page = urljoin(ROOT, primary.get('virtualsite_url') or f'/virtual/{year}/{str(primary.get("eventtype", "Poster")).lower()}/{primary["id"]}')
+        if year == 2020 and not primary.get('virtualsite_url'):
+            # Numeric modern event IDs do not identify legacy poster pages.
+            # The independently verified forum bridge supplies the paper URL.
+            page = index_url
         topics = sorted({str(row['topic']).strip() for row in group if row.get('topic')})
         keywords = sorted({str(keyword).strip() for row in group for keyword in row.get('keywords', []) if keyword})
         records.append({'source': SOURCE, 'external_id': external, 'url': page,

paper_atlas/collectors/vision.py

@@ -5,6 +5,8 @@ import json
 from pathlib import Path
 import re
 import time
+from datetime import date
+from urllib.error import HTTPError
 from urllib.parse import parse_qs, unquote, urljoin, urlsplit
 
 from paper_atlas.io.http import fetch
@@ -98,6 +100,14 @@ def discover(source,cache):
     return sorted(indexes.values(),key=lambda row:(row['year'],row['venue'],row['track']!='main',row['track']))
 
 
+def author_names(raw):
+    raw=(raw or '').split(';',1)[0]
+    if ' and ' in raw and all(',' in name for name in raw.split(' and ')):
+        names=[' '.join(reversed([part.strip() for part in name.split(',',1)])) for name in raw.split(' and ')]
+    else:names=raw.split(',')
+    return [clean(name).removesuffix('*').strip() for name in names if clean(name).removesuffix('*').strip()]
+
+
 def parse_paper(html,entry,source,venue,year,track,index_url):
     nodes=list(Document(html).root.walk());meta=defaultdict(list)
     for node in nodes:
@@ -112,11 +122,7 @@ def parse_paper(html,entry,source,venue,year,track,index_url):
         if clean(author):authors.append(clean(author))
     if not authors:
         raw=next((clean(n.text()) for n in nodes if n.attrs.get('id')=='authors'),'') or entry.get('index_authors') or ''
-        raw=raw.split(';',1)[0]
-        if ' and ' in raw and all(',' in name for name in raw.split(' and ')):
-            names=[' '.join(reversed([part.strip() for part in name.split(',',1)])) for name in raw.split(' and ')]
-        else:names=raw.split(',')
-        authors=[clean(name).removesuffix('*').strip() for name in names if clean(name).removesuffix('*').strip()]
+        authors=author_names(raw)
     dois={value for doi in meta['citation_doi'] if doi and (value:=doi_link(doi) or doi.strip().lower()) and re.fullmatch(r'10\.\d{4,9}/\S+',value)}
     if entry.get('doi'):dois.add(entry['doi'])
     if len(dois)>1:raise ValueError('Citation DOI differs from the official index')
@@ -126,35 +132,95 @@ def parse_paper(html,entry,source,venue,year,track,index_url):
             'title':clean(title),'abstract':abstract,'authors':authors,'identifiers':identifiers,
             'venue':venue,'year':year,'track':track,'status':'accepted',
             'raw_decision':'listed in official proceedings','list_url':index_url,
-            'metadata':{'citation':dict(meta),'index_title':entry['index_title'],'index_authors':entry.get('index_authors')}}
+            'metadata':{'citation':dict(meta),'index_title':entry['index_title'],'index_authors':entry.get('index_authors'),
+                        'index_urls':entry.get('index_urls',[index_url])}}
+
+
+def index_identity(url):
+    parsed=urlsplit(url)
+    match=re.fullmatch(r'/(CVPR|ICCV)(20\d{2})(_findings)?(?:\.py)?/?',parsed.path)
+    return match.groups() if parsed.netloc=='openaccess.thecvf.com' and match else None
+
+
+def read_index(index,cache):
+    url=index['index_url'];html=decode_html(fetch(url,cache,32*1024**2,delay=1))
+    links=set()
+    for node in Document(html).root.walk():
+        if node.tag!='a':continue
+        link=urljoin(url,node.attrs.get('href',''));day=parse_qs(urlsplit(link).query).get('day',[])
+        if index_identity(link)!=index_identity(url) or index_identity(url) is None or len(day)!=1:continue
+        if day[0]!='all':
+            try:valid=date.fromisoformat(day[0]).year==index['year']
+            except ValueError:valid=False
+            if not valid:continue
+        links.add(link)
+    all_pages=sorted(link for link in links if parse_qs(urlsplit(link).query)['day']==['all'])
+    day_pages=sorted(links-set(all_pages));index_errors=[]
+    if len(all_pages)>1:raise ValueError('Ambiguous all-papers index')
+    pages=[]
+    if all_pages:
+        try:pages=[(all_pages[0],decode_html(fetch(all_pages[0],cache,32*1024**2,delay=1)))]
+        except HTTPError as error:
+            if error.code!=404 or not day_pages:
+                error.close();raise
+            index_errors.append({'url':all_pages[0],'status':404});error.close()
+    if not pages and day_pages:
+        pages=[(page,decode_html(fetch(page,cache,32*1024**2,delay=1))) for page in day_pages]
+    if not pages:pages=[(url,html)]
+    entries={};raw_count=0
+    for page,document in pages:
+        parsed,count=index_entries(document,page);raw_count+=count
+        for entry in parsed:
+            entry['index_urls']=[page];old=entries.get(entry['url'])
+            if old:
+                for key in ('index_title','index_authors','doi'):
+                    if old.get(key) and entry.get(key) and old[key]!=entry[key]:raise ValueError('Conflicting metadata across official day indexes')
+                    if not old.get(key):old[key]=entry.get(key)
+                old['index_urls'].append(page)
+            else:entries[entry['url']]=entry
+    return list(entries.values()),raw_count,[page for page,_ in pages],index_errors
+
+
+def index_only_record(entry,source,venue,year,track,index_url,error):
+    if not entry['index_title'].strip():raise ValueError('Missing title in official index')
+    return {'source':source,'external_id':urlsplit(entry['url']).path,'url':entry['url'],
+            'title':entry['index_title'],'abstract':None,'authors':author_names(entry.get('index_authors')),
+            'identifiers':{'doi':entry['doi']} if entry.get('doi') else {},'venue':venue,'year':year,'track':track,
+            'status':'accepted','raw_decision':'listed in official proceedings index',
+            'evidence_url':entry['index_urls'][0],'list_url':index_url,
+            'metadata':{'index_title':entry['index_title'],'index_authors':entry.get('index_authors'),
+                        'index_urls':entry['index_urls'],'abstract_fetch_error':{'status':error.code,'url':error.url,'message':str(error)}}}
 
 
 def collect_index(source,index,cache,out):
     url=index['index_url'];venue=index['venue'];year=index['year'];track=index['track']
     entries=index.get('entries')
     raw_count=len(entries) if entries is not None else None
+    index_urls=[url];index_errors=[];missing_pages=[]
     if entries is None:
-        html=decode_html(fetch(url,cache,32*1024**2,delay=1))
-        # Some defaults contain only day one; prefer the explicitly linked all-papers page.
-        all_pages={urljoin(url,n.attrs.get('href','')) for n in Document(html).root.walk() if n.tag=='a' and parse_qs(urlsplit(urljoin(url,n.attrs.get('href',''))).query).get('day')==['all'] and urlsplit(urljoin(url,n.attrs.get('href',''))).netloc==urlsplit(url).netloc and urlsplit(urljoin(url,n.attrs.get('href',''))).path==urlsplit(url).path}
-        if len(all_pages)>1:raise ValueError('Ambiguous all-papers index')
-        if all_pages:
-            url=all_pages.pop();html=decode_html(fetch(url,cache,32*1024**2,delay=1))
-        entries,raw_count=index_entries(html,url)
+        entries,raw_count,index_urls,index_errors=read_index(index,cache)
+        if len(index_urls)==1:url=index_urls[0]
     records=[]
     for number,entry in enumerate(entries,1):
         if publication(entry['url'])!=(venue,year):raise ValueError('Publication path differs from requested edition')
-        record=parse_paper(decode_html(fetch(entry['url'],cache,2*1024**2,delay=1)),entry,source,venue,year,track,url)
+        try:record=parse_paper(decode_html(fetch(entry['url'],cache,2*1024**2,delay=1)),entry,source,venue,year,track,url)
+        except HTTPError as error:
+            if error.code!=404:
+                error.close();raise
+            entry.setdefault('index_urls',[url])
+            record=index_only_record(entry,source,venue,year,track,url,error)
+            missing_pages.append(entry['url']);error.close()
         records.append(record)
         if number%25==0:print(json.dumps({'source':source,'venue':venue,'year':year,'track':track,'collected':number,'expected':len(entries)}),flush=True)
     if not records:raise ValueError('Empty publication list')
     path=Path(out)/f'{source}-{venue}{year}-{track}.jsonl';path.parent.mkdir(parents=True,exist_ok=True)
     part=path.with_suffix('.part');part.write_text(''.join(json.dumps(r,ensure_ascii=False)+'\n' for r in records));part.replace(path)
     missing=sum(not r['abstract'] for r in records);missing_authors=sum(not r['authors'] for r in records)
-    manifest={'source':source,'parser_version':1,'venue':venue,'year':year,'track':track,
+    manifest={'source':source,'parser_version':2,'venue':venue,'year':year,'track':track,
               'count':len(records),'index_link_count':len(entries),'raw_index_entry_count':raw_count,
               'missing_abstract_count':missing,'missing_authors_count':missing_authors,'file':str(path),
-              'list_url':url,'status':'partial' if missing or missing_authors else 'collected'}
+              'index_urls':index_urls,'index_errors':index_errors,'missing_abstract_pages':missing_pages,
+              'list_url':url,'status':'partial' if missing or missing_authors or index_errors else 'collected'}
     path.with_suffix('.manifest.json').write_text(json.dumps(manifest,indent=2));print(json.dumps(manifest),flush=True)
     return manifest
 
@@ -165,8 +231,13 @@ def main(argv=None, *, prog=None, source=None):
     else:parser.set_defaults(source=source)
     parser.add_argument('--cache',required=True);parser.add_argument('--out',required=True)
     parser.add_argument('--year',type=int,action='append');parser.add_argument('--venue',choices=['CVPR','ICCV','ECCV'],action='append')
+    parser.add_argument('--retry-catalogue',help='Select only blocked editions from an existing catalogue; write results to --out')
     args=parser.parse_args(argv);out=Path(args.out);out.mkdir(parents=True,exist_ok=True)
     indexes=discover(args.source,args.cache)
+    if args.retry_catalogue:
+        retry=json.loads(Path(args.retry_catalogue).read_text())
+        indexes=[index for index in indexes if retry.get(f'{index["venue"]}:{index["year"]}:{index["track"]}',{}).get('status')=='blocked']
+        if not indexes:parser.error('Retry catalogue has no blocked editions for this source')
     if args.year and set(args.year)-{r['year'] for r in indexes}:parser.error('Requested year is absent from the official source')
     if args.venue and set(args.venue)-{r['venue'] for r in indexes}:parser.error('Requested venue is absent from this source')
     if not any((not args.year or r['year'] in args.year) and (not args.venue or r['venue'] in args.venue) for r in indexes):parser.error('Requested venue/year combination is absent from the official source')
@@ -181,7 +252,7 @@ def main(argv=None, *, prog=None, source=None):
         try:
             manifest=collect_index(args.source,index,args.cache,args.out);catalogue[key]={'status':manifest['status'],'files':[manifest]}
         except Exception as exc:
-            catalogue[key]={'status':'blocked','year':index['year'],'venue':index['venue'],'track':index['track'],'index_url':index['index_url'],'error':str(exc)}
+            catalogue[key]={'status':'blocked','year':index['year'],'venue':index['venue'],'track':index['track'],'index_url':index['index_url'],'error':str(exc),'error_url':getattr(exc,'url',None)}
             print(json.dumps(catalogue[key]),flush=True)
         part=path.with_suffix('.part');part.write_text(json.dumps(catalogue,indent=2));part.replace(path)
     (out/f'{args.source}-last-run.json').write_text(json.dumps({'requested_indexes':requested,'year_filter':args.year,'venue_filter':args.venue},indent=2))

paper_atlas/db/store.py

@@ -88,7 +88,7 @@ class Store:
             found = c.execute('SELECT paper_id FROM identifiers WHERE scheme=? AND value=?',(scheme,value)).fetchone()
             if found:
                 matches.add(found[0])
-        old = c.execute('SELECT paper_id,title,abstract FROM source_records WHERE source_id=? AND external_id=?',(source_id:=self.source(source,url),external_id)).fetchone()
+        old = c.execute('SELECT paper_id,title,abstract,retrieved_at,metadata_json FROM source_records WHERE source_id=? AND external_id=?',(source_id:=self.source(source,url),external_id)).fetchone()
         if old:
             matches.add(old[0])
         key = normalized(title)
@@ -105,13 +105,17 @@ class Store:
             for other in sorted(matches-{paper}):
                 self.merge(paper,other,'explicit identifier or exact title corroborated by authors')
             if old:
-                c.execute("UPDATE papers SET title=?,abstract=?,title_key=?,authors_json=? WHERE id=? AND title=? AND COALESCE(abstract,'')=COALESCE(?,'')",(title,abstract,key,json.dumps(authors,ensure_ascii=False),paper,old[1],old[2]))
+                old_abstract=old[2] or json.loads(old[4]).get('last_known_abstract',{}).get('abstract')
+                c.execute("UPDATE papers SET title=?,abstract=COALESCE(NULLIF(?,''),abstract),title_key=?,authors_json=CASE WHEN ?='[]' THEN authors_json ELSE ? END WHERE id=? AND title=? AND COALESCE(abstract,'')=COALESCE(?,'')",(title,abstract,key,json.dumps(authors,ensure_ascii=False),json.dumps(authors,ensure_ascii=False),paper,old[1],old_abstract))
             c.execute("UPDATE papers SET abstract=CASE WHEN abstract IS NULL OR abstract='' THEN ? ELSE abstract END WHERE id=?",(abstract,paper))
         else:
             paper = c.execute('INSERT INTO papers(title,abstract,title_key,authors_json,first_published) VALUES(?,?,?,?,?)',(title,abstract,key,json.dumps(authors,ensure_ascii=False),published)).lastrowid
         for scheme,value in ids:
             c.execute('INSERT OR IGNORE INTO identifiers VALUES(?,?,?)',(scheme,value,paper))
         metadata=dict(metadata or {});metadata['record_identifiers']=dict(ids)
+        if old and not abstract:
+            previous={'abstract':old[2],'observed_at':old[3]} if old[2] else json.loads(old[4]).get('last_known_abstract')
+            if previous:metadata['last_known_abstract']=previous
         c.execute('INSERT INTO source_records(source_id,external_id,paper_id,url,title,abstract,authors_json,metadata_json) VALUES(?,?,?,?,?,?,?,?) ON CONFLICT(source_id,external_id) DO UPDATE SET paper_id=excluded.paper_id,url=excluded.url,title=excluded.title,abstract=excluded.abstract,authors_json=excluded.authors_json,metadata_json=excluded.metadata_json,retrieved_at=CURRENT_TIMESTAMP',(source_id,external_id,paper,url,title,abstract,json.dumps(authors,ensure_ascii=False),json.dumps(metadata,ensure_ascii=False,default=str)))
         record = c.execute('SELECT id FROM source_records WHERE source_id=? AND external_id=?',(source_id,external_id)).fetchone()[0]
         for other,_,_ in title_matches:
@@ -137,8 +141,9 @@ 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):
         c = self.conn
-        c.execute("INSERT OR IGNORE INTO venues(acronym,name,field,venue_type) VALUES(?,?,?,'conference')",(venue,venue,'unclassified'))
-        venue_id = c.execute('SELECT id FROM venues WHERE acronym=?',(venue,)).fetchone()[0]
+        existing=c.execute('SELECT id FROM venues WHERE acronym=? COLLATE NOCASE',(venue,)).fetchone()
+        if existing:venue_id=existing[0]
+        else:venue_id=c.execute("INSERT INTO venues(acronym,name,field,venue_type) VALUES(?,?,?,'conference')",(venue,venue,'unclassified')).lastrowid
         for alias in aliases or []:
             c.execute('INSERT OR IGNORE INTO venue_aliases VALUES(?,?)',(venue_id,alias))
         c.execute('INSERT OR IGNORE INTO editions(venue_id,year,track) VALUES(?,?,?)',(venue_id,int(year),track))

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()

paper_atlas/ingestion/backfill.py

@@ -36,7 +36,7 @@ def backfill(db,files):
                     if any(row.get(key)!=manifest.get(key) for key in ('source','venue','year','track')):
                         raise ValueError('Evidence row differs from edition 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 v.acronym=? AND e.year=? AND e.track=?',(row['venue'],row['year'],row['track'])).fetchone()
+                    edition=c.execute('SELECT e.id FROM editions e JOIN venues v ON v.id=e.venue_id WHERE v.acronym=? COLLATE NOCASE 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')

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

@@ -0,0 +1,145 @@
+import hashlib
+import json
+from contextlib import closing
+from pathlib import Path
+import sqlite3
+import tempfile
+from types import SimpleNamespace
+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_scope import scope
+from paper_atlas.db.store import Store
+from paper_atlas.ingestion.records import ingest_file
+
+
+class DblpRecordTests(unittest.TestCase):
+    def setUp(self):
+        # Tiny fixtures may live on a small tmpfs; production reserve stays 8 GiB.
+        budget=patch('paper_atlas.collectors.dblp_records.shutil.disk_usage',return_value=SimpleNamespace(free=16*1024**3))
+        budget.start();self.addCleanup(budget.stop)
+
+    def row(self,family='aaai',book='AAAI',year=2024,key='one'):
+        return {'record_type':'inproceedings','key':f'conf/{family}/{key}','publtype':None,
+                'fields':{'title':['A research publication'],'author':['Jane Doe 0001','John Smith'],
+                          'booktitle':[book],'year':[str(year)],'crossref':[f'conf/{family}/{year}'],'ee':['https://doi.org/10.1234/'+key]}}
+
+    def volume(self,family,year,title=None):
+        return {'record_type':'proceedings','key':f'conf/{family}/{year}','fields':{'title':[title or f'{family.upper()} {year}'],'year':[str(year)]}}
+
+    def test_real_directory_aliases_do_not_confuse_other_conferences(self):
+        row=self.row('clear2','CLeaR');volume=self.volume('clear2',2024,'Causal Learning and Reasoning, 2024')
+        self.assertEqual(scope(row,volume)[0]['venue'],'CLeaR')
+        self.assertIsNone(scope(self.row('clear','CLEAR'),self.volume('clear',2024))[0])
+        self.assertIsNone(scope(self.row('aamas','Adaptive Agents and Multi-Agent Systems'),self.volume('aamas',2024))[0])
+        self.assertEqual(scope(self.row('atal','AAMAS'),self.volume('atal',2024))[0]['venue'],'AAMAS')
+        self.assertEqual(scope(self.row('aips','ICAPS'),self.volume('aips',2024))[0]['venue'],'ICAPS')
+
+    def test_conference_year_is_distinct_from_publication_year(self):
+        row=self.row('nips','NIPS',1988);row['fields']['crossref']=['conf/nips/1987']
+        volume=self.volume('nips',1987,'NIPS Conference, Denver, 1987; published 1988')
+        result,_=scope(row,volume);self.assertEqual(result['year'],1987);self.assertEqual(result['bibliographic_year'],1988)
+        volume['fields']['title']=['NIPS collected papers']
+        self.assertEqual(scope(row,volume)[1],'unverified_year_difference')
+
+    def test_joint_events_and_tracks_are_explicit(self):
+        result,_=scope(self.row('ijcai','IJCAI/ECAI',2022),self.volume('ijcai',2022))
+        self.assertEqual(result['venue'],'ECAI+IJCAI');self.assertEqual(result['venue_aliases'],['ECAI','IJCAI'])
+        result,_=scope(self.row('coling','LREC/COLING'),self.volume('coling',2024))
+        self.assertEqual(result['venue'],'COLING+LREC')
+        for book,track in [('ACL (Findings)','findings'),('ACL (Student Research Workshop)','workshop'),('ACL (System Demonstrations)','demo')]:
+            self.assertEqual(scope(self.row('acl',book),self.volume('acl',2024))[0]['track'],track)
+        for suffix,track in [('Special interest tracks and posters','special_interest'),('Alternate Track Papers & Posters','alternate')]:
+            self.assertEqual(scope(self.row('www','WWW ('+suffix+')'),self.volume('www',2024))[0]['track'],track)
+        volume=self.volume('pkdd',2026,'ECML PKDD 2026, Applied Data Science Track, Demo Track and Industrial Track')
+        self.assertEqual(scope(self.row('pkdd','ECML PKDD (10)',2026),volume)[0]['track'],'proceedings_unspecified')
+
+    def test_short_year_keys_require_a_unique_explicit_year_not_a_guessed_century(self):
+        row=self.row('sigir','SIGIR',1980);volume=self.volume('sigir',1980,'Information Retrieval Research, Cambridge, June 1980')
+        volume['key']='conf/sigir/80';volume['fields']['year']=['1981'];row['fields']['crossref']=[volume['key']]
+        edition,_=scope(row,volume);self.assertEqual(edition['year'],1980)
+        self.assertIn('two-digit',edition['edition_year_basis'])
+        volume['fields']['title']=['SIGIR 1980, published 1981']
+        self.assertEqual(scope(row,volume)[1],'unverified_edition_year')
+        volume['fields']['title']=['SIGIR 1989']
+        self.assertEqual(scope(row,volume)[1],'unverified_edition_year')
+
+    def test_crossref_mismatch_and_future_editions_are_held_for_review(self):
+        self.assertEqual(scope(self.row(),self.volume('kdd',2024))[1],'crossref_venue_conflict')
+        self.assertEqual(scope(self.row(year=2027),self.volume('aaai',2027))[1],'future_edition')
+
+    def test_only_own_identifier_links_are_used_and_ambiguity_is_preserved(self):
+        own,ambiguous=identifiers({'ee':['https://doi.org/10.1234/published','https://doi.org/10.48550/arXiv.2401.00001','https://openreview.net/forum?id=realForum'],
+                                    'cite':['https://doi.org/10.9999/reference']})
+        self.assertEqual(own,{'doi':'10.1234/published','arxiv':'2401.00001','openreview':'realForum'});self.assertFalse(ambiguous)
+        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:
+            c.execute('CREATE TABLE proceedings(key TEXT PRIMARY KEY,metadata_json TEXT)')
+            c.execute('INSERT INTO proceedings VALUES(?,?)',(volume['key'],json.dumps(volume)))
+        candidates=root/'candidates.jsonl';candidates.write_text(''.join(json.dumps(row)+'\n' for row in rows))
+        manifest={'snapshot_sha256':XML_SHA,'candidate_sha256':hashlib.sha256(candidates.read_bytes()).hexdigest(),
+                  'candidate_records':len(rows),'record_types':{'proceedings':1,'inproceedings':len(rows)}}
+        catalogue.with_name('dblp-ai-candidates.manifest.json').write_text(json.dumps(manifest))
+        return catalogue,candidates
+
+    def test_normalized_files_import_with_unknown_abstracts_and_no_invented_retracted_acceptance(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root=Path(tmp);withdrawn=self.row(key='two');withdrawn['publtype']='withdrawn'
+            catalogue,candidates=self.fixture(root,[self.row(),withdrawn,self.row(book='AAAI Workshops',key='workshop')])
+            result=collect(catalogue,root/'out',['AAAI'],candidates=candidates)
+            self.assertEqual(result['normalized_records'],2);self.assertEqual(result['excluded_by_reason']['separate_workshop'],1)
+            manifest=result['files'][0];self.assertEqual(manifest['missing_abstract_count'],2)
+            rows=[json.loads(line) for line in Path(manifest['file']).read_text().splitlines()]
+            self.assertEqual(rows[0]['authors'],['Jane Doe','John Smith']);self.assertIsNone(rows[0]['abstract'])
+            self.assertEqual(rows[1]['status'],'unknown');self.assertEqual(rows[1]['decision_events'],[])
+            db=root/'papers.sqlite';ingest_file(db,manifest['file'])
+            with closing(sqlite3.connect(db)) as c:self.assertEqual(c.execute('SELECT count(*) FROM decisions').fetchone()[0],1)
+
+    def test_checksum_failure_and_duplicate_keys_do_not_publish_files(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root=Path(tmp);catalogue,candidates=self.fixture(root,[self.row(),self.row()])
+            with self.assertRaisesRegex(ValueError,'Repeated DBLP'):collect(catalogue,root/'out',['AAAI'],candidates=candidates)
+            self.assertFalse(list((root/'out').glob('*.jsonl')))
+            candidates.write_text('tampered')
+            with self.assertRaisesRegex(ValueError,'checksum mismatch'):collect(catalogue,root/'out',['AAAI'],candidates=candidates)
+
+    def test_case_variants_use_one_conference_edition_and_submission(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            store=Store(Path(tmp)/'papers.sqlite');self.addCleanup(store.conn.close)
+            paper,one=store.record('pmlr','one','https://example.org/one','A paper','Abstract',[],{'doi':'10.1234/one'})
+            store.decision(paper,one,'CORL',2024,'main','one','https://example.org/one','accepted','Accept',identities=[('publication_doi','10.1234/one')])
+            same,two=store.record('dblp','two','https://example.org/two','A paper',None,[],{'doi':'10.1234/one'})
+            store.decision(same,two,'CoRL',2024,'main','two','https://example.org/two','accepted','Published',identities=[('publication_doi','10.1234/one')])
+            self.assertEqual(store.conn.execute('SELECT count(*) FROM venues').fetchone()[0],1)
+            self.assertEqual(store.conn.execute('SELECT count(*) FROM submissions').fetchone()[0],1)
+
+    def test_production_disk_reserve_stops_before_publishing(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root=Path(tmp);catalogue,candidates=self.fixture(root,[self.row()])
+            with patch('paper_atlas.collectors.dblp_records.shutil.disk_usage',return_value=SimpleNamespace(free=1024)):
+                with self.assertRaisesRegex(RuntimeError,'8 GiB'):collect(catalogue,root/'out',['AAAI'],candidates=candidates)
+            self.assertFalse(list((root/'out').glob('*.jsonl')))
+
+
+if __name__=='__main__':unittest.main()

tests/test_iclr.py

@@ -64,6 +64,7 @@ class ICLRTests(unittest.TestCase):
         rows,_=self.parse([dict(row,paper_pdf_url=None)],year=2020)
         self.assertEqual(rows[0]['metadata']['identity_state'],'program_only')
         self.assertEqual(rows[0]['identifiers'],{})
+        self.assertEqual(rows[0]['url'],'https://iclr.cc/virtual/2020/papers.html')
 
     def test_only_referenced_official_assets_and_real_navigation_years(self):
         html='<a href="/archive/2013/">2013</a><a href="/Conferences/2026">2026</a><a href="https://other.example/2025">2025</a>'

tests/test_iclr_legacy.py

@@ -0,0 +1,143 @@
+import copy
+import hashlib
+import json
+from pathlib import Path
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from paper_atlas.audits import iclr_legacy
+from paper_atlas.ingestion.backfill import backfill, queue_files
+from paper_atlas.ingestion.records import ingest_file
+from paper_atlas.db.store import Store
+
+
+class ICLRLegacyTests(unittest.TestCase):
+    def program(self):
+        return {'source': 'iclr_program', 'external_id': '2020/program-42', 'venue': 'ICLR',
+                'year': 2020, 'track': 'main', 'status': 'accepted',
+                'url': 'https://iclr.cc/virtual/2020/papers.html', 'title': 'Historical paper title',
+                'abstract': 'Historical abstract', 'authors': ['Jane Doe', 'John Smith'],
+                'identifiers': {}, 'raw_decision': 'listed in official program',
+                'evidence_url': 'https://iclr.cc/virtual/2020/papers.html',
+                'metadata': {'program_identity_type': 'submission', 'program_source_id': 42}}
+
+    def snapshot(self):
+        return {'source': 'openreview_snapshot', 'external_id': '2020/realForum', 'venue': 'ICLR',
+                'year': 2020, 'track': 'main', 'status': 'accepted',
+                'url': 'https://openreview.net/forum?id=realForum', 'title': 'Historical paper title',
+                'abstract': 'Historical abstract', 'authors': ['Jane Doe', 'John Smith'],
+                'identifiers': {'openreview': 'realForum'},
+                'submission_identifiers': [{'scheme': 'openreview', 'value': 'realForum'},
+                                           {'scheme': 'iclr_number', 'value': '42'}],
+                'decision_events': [{'status': 'accepted', 'raw_value': 'Accept',
+                    'source_decision_id': 'decision1', 'evidence_url': 'https://openreview.net/forum?id=realForum&noteId=decision1',
+                    'authority': {'public_readers_verified': True}}],
+                'metadata': {'source_submission_number': 42, 'public_readers_verified': True}}
+
+    def legacy(self):
+        return {'id': 'realForum', 'forum': 'realForum', 'content': {
+            'iclr_id': 'realForum', 'title': 'Historical paper title', 'authors': ['Jane Doe', 'John Smith']}}
+
+    def test_independent_legacy_identity_preserves_existing_source_id_and_text(self):
+        program, submission = self.program(), self.snapshot()
+        submission.update(title='A later title', abstract='A later abstract', authors=['Jane Doe'])
+        rows, checks = iclr_legacy.reconcile([program], {'realForum': submission}, [self.legacy()])
+        self.assertEqual(rows[0]['external_id'], '2020/program-42')
+        self.assertEqual(rows[0]['url'], 'https://iclr.cc/virtual_2020/poster_realForum.html')
+        for field in ('title', 'abstract', 'authors'):
+            self.assertEqual(rows[0][field], program[field])
+        self.assertEqual(rows[0]['identifiers']['openreview'], 'realForum')
+        self.assertTrue(rows[0]['metadata']['program_submission_number_verified'])
+        self.assertEqual(checks['independent_legacy'], 1)
+        self.assertEqual(program['identifiers'], {})
+
+    def test_program_number_and_shared_authors_alone_do_not_assign_a_forum(self):
+        program = self.program()
+        program.update(title='A different paper', abstract='An unrelated abstract')
+        with self.assertRaisesRegex(ValueError, 'number alone'):
+            iclr_legacy.reconcile([program], {'realForum': self.snapshot()}, [self.legacy()])
+
+    def test_independent_snapshot_title_and_abstract_handle_legacy_name_variants(self):
+        legacy = self.legacy()
+        legacy['content']['authors'] = ['J. Doe', 'J. Smith']
+        rows, checks = iclr_legacy.reconcile([self.program()], {'realForum': self.snapshot()}, [legacy])
+        self.assertEqual(checks['independent_snapshot'], 1)
+        self.assertEqual(rows[0]['identifiers']['openreview'], 'realForum')
+
+    def test_reviewed_title_variant_is_exact_and_requires_public_author_corroboration(self):
+        variant = iclr_legacy.REVIEWED_TITLE_VARIANT
+        program, submission, legacy = self.program(), self.snapshot(), self.legacy()
+        program['title'] = variant['program']
+        submission.update(title=variant['snapshot'], abstract='A different revision')
+        submission['identifiers']['openreview'] = variant['forum']
+        legacy.update(id=variant['forum'], forum=variant['forum'])
+        legacy['content'].update(iclr_id=variant['forum'], title=variant['legacy'])
+        rows, checks = iclr_legacy.reconcile([program], {variant['forum']: submission}, [legacy])
+        self.assertEqual(checks['reviewed_title_variant'], 1)
+        self.assertTrue(rows[0]['metadata']['program_identity_verification']['reviewed_title_variant'])
+        program['authors'] = ['Jane Doe']
+        with self.assertRaisesRegex(ValueError, 'number alone'):
+            iclr_legacy.reconcile([program], {variant['forum']: submission}, [legacy])
+
+    def test_conflicting_forum_ambiguous_number_and_repeated_legacy_identity_fail(self):
+        p = self.program()
+        p['identifiers']['openreview'] = 'anotherForum'
+        with self.assertRaisesRegex(ValueError, 'conflicts'):
+            iclr_legacy.reconcile([p], {'realForum': self.snapshot()}, [self.legacy()])
+        duplicate = copy.deepcopy(self.snapshot())
+        duplicate['identifiers']['openreview'] = 'otherForum'
+        with self.assertRaisesRegex(ValueError, 'ambiguous'):
+            iclr_legacy.reconcile([self.program()], {'realForum': self.snapshot(), 'otherForum': duplicate}, [self.legacy()])
+        with self.assertRaisesRegex(ValueError, 'repeated'):
+            iclr_legacy.reconcile([self.program()], {'realForum': self.snapshot()}, [self.legacy(), self.legacy()])
+
+    def write_file(self, path, rows, source):
+        path.write_text(''.join(json.dumps(row) + '\n' for row in rows))
+        manifest = {'source': source, 'venue': 'ICLR', 'year': 2020, 'track': 'main',
+                    'parser_version': 2, 'count': len(rows), 'missing_abstract_count': 0,
+                    'list_url': 'https://iclr.cc/virtual/2020/papers.html', 'status': 'collected',
+                    'submission_status_counts': {status: sum(r['status'] == status for r in rows)
+                                                 for status in {r['status'] for r in rows}}}
+        path.with_suffix('.manifest.json').write_text(json.dumps(manifest))
+
+    def test_audited_bridge_then_snapshot_import_keeps_one_accepted_attempt(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            program_path, snapshot_path, legacy_path = (root / name for name in ('program.jsonl', 'snapshot.jsonl', 'legacy.json'))
+            self.write_file(program_path, [self.program()], 'iclr_program')
+            self.write_file(snapshot_path, [self.snapshot()], 'openreview_snapshot')
+            raw = json.dumps([self.legacy()]).encode()
+            legacy_path.write_bytes(raw)
+            with patch.multiple(iclr_legacy, LEGACY_BYTES=len(raw), LEGACY_SHA256=hashlib.sha256(raw).hexdigest(), EXPECTED=1):
+                report = iclr_legacy.audit(program_path, snapshot_path, legacy_path, root / 'out')
+            db = root / 'papers.sqlite'
+            ingest_file(db, program_path)
+            queue = root / 'out/iclr-2020-backfill-queue.json'
+            result = backfill(db, queue_files(queue))
+            self.assertEqual(result['submissions_before'], result['submissions_after'])
+            ingest_file(db, report['backfill_queue']['files'][0])
+            ingest_file(db, snapshot_path)
+            store = Store(db)
+            try:
+                self.assertEqual(store.conn.execute('SELECT count(*) FROM papers').fetchone()[0], 1)
+                self.assertEqual(store.conn.execute('SELECT count(*) FROM submissions').fetchone()[0], 1)
+                self.assertEqual(store.conn.execute('SELECT count(*) FROM submission_sources').fetchone()[0], 2)
+                self.assertEqual(store.conn.execute('SELECT url FROM source_records WHERE external_id=?', ('2020/program-42',)).fetchone()[0],
+                                 'https://iclr.cc/virtual_2020/poster_realForum.html')
+                self.assertEqual(store.conn.execute('PRAGMA foreign_key_check').fetchall(), [])
+            finally:
+                store.conn.close()
+
+    def test_unverified_legacy_bytes_publish_no_output(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            legacy = root / 'legacy.json'
+            legacy.write_text('[]')
+            with self.assertRaisesRegex(ValueError, 'checksum'):
+                iclr_legacy.audit(root / 'program', root / 'snapshot', legacy, root / 'out')
+            self.assertFalse((root / 'out').exists())
+
+
+if __name__ == '__main__':
+    unittest.main()

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()

tests/test_index_core.py

@@ -9,6 +9,17 @@ from paper_atlas.ingestion.records import ingest_file
 
 
 class StoreTests(unittest.TestCase):
+    def test_partial_source_refresh_preserves_known_abstract_and_can_later_restore_it(self):
+        paper,record=self.store.record('cvf','paper','https://example.org/paper','Known title','Known abstract',['Jane Doe'])
+        self.store.record('cvf','paper','https://example.org/paper','Known title',None,[],metadata={'abstract_fetch_error':{'status':404}})
+        self.store.record('cvf','paper','https://example.org/paper','Known title',None,[],metadata={'abstract_fetch_error':{'status':404}})
+        c=self.store.conn
+        self.assertEqual(c.execute('SELECT abstract,authors_json FROM papers').fetchone(),('Known abstract','["Jane Doe"]'))
+        source=c.execute('SELECT abstract,metadata_json FROM source_records WHERE id=?',(record,)).fetchone()
+        self.assertIsNone(source[0]);self.assertEqual(json.loads(source[1])['last_known_abstract']['abstract'],'Known abstract')
+        self.store.record('cvf','paper','https://example.org/paper','Updated title','Restored revised abstract',['Jane Doe'])
+        self.assertEqual(c.execute('SELECT title,abstract FROM papers').fetchone(),('Updated title','Restored revised abstract'))
+
     def setUp(self):
         self.tmp = tempfile.TemporaryDirectory()
         self.addCleanup(self.tmp.cleanup)

tests/test_vision.py

@@ -1,11 +1,85 @@
 import tempfile
+import io
+import json
+from pathlib import Path
 import unittest
 from unittest.mock import patch
+from urllib.error import HTTPError
 
-from paper_atlas.collectors.vision import collect_index, decode_html, discover, index_entries, parse_paper
+from paper_atlas.collectors.vision import collect_index, decode_html, discover, index_entries, parse_paper,main
 
 
 class VisionTests(unittest.TestCase):
+    def cvf_fixture(self,year=2018):
+        root=f'https://openaccess.thecvf.com/CVPR{year}.py'
+        papers=[f'https://openaccess.thecvf.com/content_cvpr_{year}/html/Paper{i}_CVPR_{year}_paper.html' for i in (1,2,3)]
+        def entry(url,title='Paper'):
+            return f'<dt><a href="{url}">{title}</a></dt><dd>Jane Doe, John Smith</dd>'
+        return root,papers,entry
+
+    def test_all_explicit_days_are_combined_and_overlapping_papers_are_not_duplicated(self):
+        root,papers,entry=self.cvf_fixture();day1=root+'?day=2018-06-19';day2=root+'?day=2018-06-20'
+        pages={root:f'<a href="{day1}">Day1</a><a href="{day2}">Day2</a><a href="https://evil.example/CVPR2018.py?day=all">Other</a><a href="/CVPR2019.py?day=all">Wrong year</a>',day1:'<dl>'+entry(papers[0])+entry(papers[1])+'</dl>',day2:'<dl>'+entry(papers[1])+entry(papers[2])+'</dl>'}
+        pages.update({paper:'<div id="papertitle">Paper</div><div id="abstract">Abstract</div>' for paper in papers})
+        with tempfile.TemporaryDirectory() as tmp,patch('paper_atlas.collectors.vision.fetch',side_effect=lambda url,*a,**kw:pages[url].encode()) as fetch:
+            manifest=collect_index('cvf',{'venue':'CVPR','year':2018,'track':'main','index_url':root},tmp,tmp)
+            rows=[json.loads(line) for line in Path(manifest['file']).read_text().splitlines()]
+        self.assertEqual(manifest['count'],3);self.assertEqual(manifest['raw_index_entry_count'],4)
+        self.assertEqual(manifest['index_urls'],[day1,day2]);self.assertEqual(manifest['status'],'collected')
+        self.assertEqual(rows[1]['metadata']['index_urls'],[day1,day2])
+        self.assertEqual({call.args[0] for call in fetch.call_args_list},set(pages))
+
+    def test_conflicting_copies_across_days_fail_before_publishing(self):
+        root,papers,entry=self.cvf_fixture();days=[root+'?day=2018-06-19',root+'?day=2018-06-20']
+        pages={root:''.join(f'<a href="{day}">Day</a>' for day in days),days[0]:'<dl>'+entry(papers[0],'First title')+'</dl>',days[1]:'<dl>'+entry(papers[0],'Different title')+'</dl>'}
+        with tempfile.TemporaryDirectory() as tmp,patch('paper_atlas.collectors.vision.fetch',side_effect=lambda url,*a,**kw:pages[url].encode()):
+            with self.assertRaisesRegex(ValueError,'Conflicting metadata'):
+                collect_index('cvf',{'venue':'CVPR','year':2018,'track':'main','index_url':root},tmp,tmp)
+            self.assertFalse(list(Path(tmp).glob('*.jsonl')))
+
+    def test_missing_abstract_page_keeps_official_index_metadata_and_marks_partial(self):
+        root,papers,entry=self.cvf_fixture();error=HTTPError(papers[0],404,'Not Found',{},io.BytesIO())
+        def fetch(url,*args,**kwargs):
+            if url==root:return ('<dl>'+entry(papers[0],'Known publication')+'</dl>').encode()
+            raise error
+        with tempfile.TemporaryDirectory() as tmp,patch('paper_atlas.collectors.vision.fetch',side_effect=fetch):
+            manifest=collect_index('cvf',{'venue':'CVPR','year':2018,'track':'main','index_url':root},tmp,tmp)
+            row=json.loads(Path(manifest['file']).read_text())
+        self.assertEqual(row['title'],'Known publication');self.assertIsNone(row['abstract'])
+        self.assertEqual(row['authors'],['Jane Doe','John Smith']);self.assertEqual(row['evidence_url'],root)
+        self.assertEqual(manifest['status'],'partial');self.assertEqual(manifest['missing_abstract_pages'],[papers[0]])
+        self.assertEqual(row['metadata']['abstract_fetch_error']['status'],404)
+
+    def test_forbidden_paper_is_not_replaced_with_a_fallback(self):
+        root,papers,entry=self.cvf_fixture()
+        def fetch(url,*args,**kwargs):
+            if url==root:return ('<dl>'+entry(papers[0])+'</dl>').encode()
+            raise HTTPError(url,403,'Forbidden',{},io.BytesIO())
+        with tempfile.TemporaryDirectory() as tmp,patch('paper_atlas.collectors.vision.fetch',side_effect=fetch):
+            with self.assertRaises(HTTPError):collect_index('cvf',{'venue':'CVPR','year':2018,'track':'main','index_url':root},tmp,tmp)
+            self.assertFalse(list(Path(tmp).glob('*.jsonl')))
+
+    def test_broken_all_papers_link_can_use_all_advertised_days(self):
+        root,papers,entry=self.cvf_fixture();day=root+'?day=2018-06-19';all_url=root+'?day=all'
+        pages={root:f'<a href="{day}">Day</a><a href="{all_url}">All</a>',day:'<dl>'+entry(papers[0])+'</dl>',papers[0]:'<div id="papertitle">Paper</div><div id="abstract">Abstract</div>'}
+        def fetch(url,*a,**kw):
+            if url==all_url:raise HTTPError(url,404,'Not Found',{},io.BytesIO())
+            return pages[url].encode()
+        with tempfile.TemporaryDirectory() as tmp,patch('paper_atlas.collectors.vision.fetch',side_effect=fetch):
+            manifest=collect_index('cvf',{'venue':'CVPR','year':2018,'track':'main','index_url':root},tmp,tmp)
+        self.assertEqual(manifest['index_urls'],[day]);self.assertEqual(manifest['index_errors'],[{'url':all_url,'status':404}])
+
+    def test_selective_retry_uses_separate_output_and_leaves_original_catalogue_intact(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            old=Path(tmp)/'old.json';payload={'CVPR:2018:main':{'status':'blocked'},'CVPR:2019:main':{'status':'collected'}};old.write_text(json.dumps(payload))
+            indexes=[{'venue':'CVPR','year':year,'track':'main','index_url':f'https://openaccess.thecvf.com/CVPR{year}'} for year in (2018,2019)]
+            output=Path(tmp)/'repaired'
+            with patch('paper_atlas.collectors.vision.discover',return_value=indexes),patch('paper_atlas.collectors.vision.collect_index',return_value={'status':'collected','count':1}) as collect:
+                main(['--cache',tmp,'--out',str(output),'--retry-catalogue',str(old)],source='cvf')
+            self.assertEqual(collect.call_count,1);self.assertEqual(collect.call_args.args[1]['year'],2018)
+            self.assertEqual(json.loads(old.read_text()),payload)
+            self.assertEqual(set(json.loads((output/'cvf-index-status.json').read_text())),{'CVPR:2018:main'})
+
     def entry(self):
         return {'url':'https://www.ecva.net/papers/eccv_2024/papers_ECCV/html/1_ECCV_2024_paper.php','index_title':'Paper','index_authors':'Jane Doe*, John Smith','doi':'10.1007/example_1'}