Bound arXiv tag audit batches by primary key instead of sorting the source

# Conflicts:
#	paper_atlas/cli.py

docs/data-sources.md

@@ -253,6 +253,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 视觉会议元数据

docs/operations.md

@@ -20,6 +20,20 @@ paper-atlas maintain arxiv-tags --db /data/paper-index/papers.sqlite \
 
 修复使用同一维护锁，分批提交并保存断点，保留 `custom:*` 和其他来源证据；失败后确认原因已解决，再增加 `--resume` 从上次提交位置继续。完成报告不能作为重新扫描的断点，重新核验应指定新报告。源审计和标签修复均保留 8 GiB 空闲空间，不能代替所有来源入库后的实体、投稿去重与完整性检查。
 
+修复结束后还须核对每条 arXiv 来源记录的类别集合，不能只看旧证据键是否清零。下面的只读审计比较来源 metadata 中的类别、所属论文实体和逐记录证据键，同时检查没有合法 arXiv 来源的类别行：
+
+```sh
+paper-atlas audit arxiv-tags --db /data/paper-index/papers.sqlite \
+  --repair-report /data/paper-index/collected/arxiv-tag-repair.json \
+  --out /data/paper-index/collected/arxiv-tag-coverage-audit.json
+```
+
+审计持有共享维护锁；修复或其他协调写入尚未结束时会拒绝运行。提供修复报告时，还会核对其已完成状态、数据库路径和固定来源范围。审计以一致的只读快照分批读取，记录完整扫描数量、缺失/多余证据、错误论文归属、残留旧键及少量异常样例，不修改主库。
+
+来源记录按整数主键区间读取，避免 SQLite 误选按 external_id 排列的来源索引、每批重新排序整个来源；其他来源交错在编号区间内时仍按 source_id 过滤。类别证据继续使用 source_record_id 索引批量读取。完整核验范围不因查询优化缩小。
+
+只有全量记录核对完毕且无差异时才标记 `passed`；存在差异为 `failed` 并返回非零退出码，解析或执行中断为 `error`。运行中的报告不能作为完成证明。重新核验须使用新输出路径，不能覆盖已有报告；该检查不替代最后的实体去重、外键、SQLite 与全文索引核验。
+
 ## 长期维护与更新
 
 1. 每次采集先核对来源、许可、可访问性、空闲空间和现有运行批次。保持低并发，遵守来源限流、重试与人机验证要求。

paper_atlas/audits/arxiv_tags.py

@@ -0,0 +1,143 @@
+"""Compare every arXiv record's category evidence with retained source metadata."""
+import argparse
+from collections import defaultdict
+from contextlib import closing
+import fcntl
+import json
+from pathlib import Path
+import sqlite3
+
+from paper_atlas.db.submissions import timestamp
+
+
+def save(path, state):
+    part = path.with_suffix('.part')
+    part.write_text(json.dumps(state, ensure_ascii=False, indent=2) + '\n')
+    part.replace(path)
+
+
+def audit(db, out, repair_report=None, batch_size=500):
+    db, out = Path(db).resolve(), Path(out).resolve()
+    if not db.is_file():
+        raise ValueError('Existing database required')
+    if not 1 <= batch_size <= 500:
+        raise ValueError('Batch size must be between 1 and 500')
+    if out == db or repair_report and out == Path(repair_report).resolve():
+        raise ValueError('Audit output must preserve its inputs')
+    out.parent.mkdir(parents=True, exist_ok=True)
+    with Path(str(db) + '.maintenance.lock').open('a') as maintenance, Path(str(out) + '.lock').open('a') as output_lock:
+        # A partial repair is not a useful full-category audit. This also keeps
+        # cooperating maintenance writers out for the entire read-only snapshot.
+        fcntl.flock(maintenance, fcntl.LOCK_SH | fcntl.LOCK_NB)
+        fcntl.flock(output_lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        if out.exists():
+            raise ValueError('Audit report exists; select a new output path')
+        with closing(sqlite3.connect(db.as_uri() + '?mode=ro', uri=True)) as c:
+            c.execute('PRAGMA cache_size=-32768')
+            c.execute('BEGIN')
+            source = c.execute("SELECT id FROM sources WHERE name='arxiv'").fetchone()
+            if source is None:
+                raise ValueError('No imported arXiv source')
+            total, maximum = c.execute('SELECT count(*),COALESCE(max(id),0) FROM source_records WHERE source_id=?', source).fetchone()
+            if not total:
+                raise ValueError('No imported arXiv records')
+            scope = {'db': str(db), 'source_id': source[0], 'source_record_count': total,
+                     'maximum_source_record_id': maximum}
+            if repair_report:
+                repair = json.loads(Path(repair_report).read_text())
+                if repair.get('status') != 'complete' or repair.get('records_refreshed') != total:
+                    raise ValueError('Category repair has not completed this source inventory')
+                if any(repair.get(key) != value for key, value in scope.items()):
+                    raise ValueError('Repair report belongs to a different source inventory')
+            state = {**scope, 'started_at': timestamp(), 'status': 'running', 'records_checked': 0,
+                     'last_record_id': 0, 'expected_evidence': 0, 'actual_evidence': 0,
+                     'records_with_mismatches': 0, 'missing_evidence': 0, 'unexpected_evidence': 0,
+                     'legacy_evidence': 0, 'examples': [], 'repair_report': str(Path(repair_report).resolve()) if repair_report else None}
+            save(out, state)
+            try:
+                while True:
+                    # The (source_id,external_id) index orders by external ID,
+                    # so choosing it sorts the entire source again per batch.
+                    # This fixed ID interval must use the integer primary key.
+                    records = c.execute('''SELECT id,external_id,paper_id,metadata_json FROM source_records NOT INDEXED
+                        WHERE source_id=? AND id>? AND id<=? ORDER BY id LIMIT ?''',
+                        (source[0], state['last_record_id'], maximum, batch_size)).fetchall()
+                    if not records:
+                        break
+                    ids = [row[0] for row in records]
+                    actual = defaultdict(set)
+                    for record, paper, category, evidence in c.execute('''SELECT pt.source_record_id,pt.paper_id,t.value,pt.evidence_key
+                        FROM paper_tags pt JOIN tags t ON t.id=pt.tag_id WHERE t.namespace='arxiv:category'
+                        AND pt.source_record_id IN (''' + ','.join('?' for _ in ids) + ')', ids):
+                        actual[record].add((category, paper, evidence))
+                        state['legacy_evidence'] += evidence == 'arxiv:categories'
+                    for record, external, paper, payload in records:
+                        metadata = json.loads(payload)
+                        if not isinstance(metadata, dict):
+                            raise ValueError('arXiv source metadata must be an object')
+                        categories = metadata.get('categories')
+                        if categories is None:
+                            categories = ''
+                        if not isinstance(categories, str):
+                            raise ValueError('arXiv source categories must be text')
+                        evidence_key = f'arxiv:record:{record}:tags'
+                        expected = {(category, paper, evidence_key) for category in categories.split()}
+                        found = actual[record]
+                        missing, unexpected = expected - found, found - expected
+                        state['expected_evidence'] += len(expected)
+                        state['actual_evidence'] += len(found)
+                        state['missing_evidence'] += len(missing)
+                        state['unexpected_evidence'] += len(unexpected)
+                        if missing or unexpected:
+                            state['records_with_mismatches'] += 1
+                            if len(state['examples']) < 20:
+                                state['examples'].append({'record_id': record, 'external_id': external,
+                                    'paper_id': paper, 'missing': sorted(missing), 'unexpected': sorted(unexpected)})
+                        state['records_checked'] += 1
+                    state['last_record_id'] = ids[-1]
+                    if state['records_checked'] % 10000 < len(records):
+                        save(out, state)
+                        print(json.dumps({k: state[k] for k in ('status', 'records_checked', 'source_record_count', 'records_with_mismatches')}), flush=True)
+                if state['records_checked'] != total:
+                    raise ValueError('Audited record count differs from the fixed source inventory')
+                # Every category row must also have an arXiv source-record owner.
+                # The full namespace total catches orphan and wrong-source evidence
+                # that cannot appear in the per-record traversal above.
+                all_evidence = c.execute("SELECT count(*) FROM paper_tags pt JOIN tags t ON t.id=pt.tag_id WHERE t.namespace='arxiv:category'").fetchone()[0]
+                unscoped = all_evidence - state['actual_evidence']
+                if unscoped < 0:
+                    raise ValueError('Category evidence totals are inconsistent')
+                state.update(namespace_evidence=all_evidence, evidence_without_arxiv_record=unscoped)
+                if unscoped:
+                    state['unscoped_examples'] = [dict(zip(('paper_id', 'category', 'source_record_id', 'evidence_key'), row))
+                        for row in c.execute('''SELECT pt.paper_id,t.value,pt.source_record_id,pt.evidence_key
+                            FROM paper_tags pt JOIN tags t ON t.id=pt.tag_id
+                            LEFT JOIN source_records r ON r.id=pt.source_record_id
+                            WHERE t.namespace='arxiv:category' AND (r.source_id IS NULL OR r.source_id<>?) LIMIT 20''', source)]
+                passed = not state['records_with_mismatches'] and not unscoped
+                state.update(status='passed' if passed else 'failed', finished_at=timestamp(),
+                             note='Full per-record category, canonical paper and evidence-key equality. '
+                                  'Read-only; not a global paper deduplication or full database-integrity audit.')
+                save(out, state)
+            except BaseException as error:
+                state.update(status='error', error=str(error), finished_at=timestamp())
+                save(out, state)
+                raise
+    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('--repair-report')
+    parser.add_argument('--batch-size', type=int, default=500)
+    args = parser.parse_args(argv)
+    result = audit(args.db, args.out, args.repair_report, args.batch_size)
+    if result['status'] != 'passed':
+        raise SystemExit('arXiv category evidence audit failed; inspect the report')
+
+
+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

@@ -28,7 +28,9 @@ COMMANDS = {
     ('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 完整性'),
+    ('audit','arxiv-tags'): ('paper_atlas.audits.arxiv_tags', {}, '逐记录核验 arXiv 类别及出处完整性'),
     ('audit','public-openreview'): ('paper_atlas.audits.public_openreview', {}, '核对公开负面决定与已有投稿历史'),
 }
 

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,

tests/test_arxiv_tag_audit.py

@@ -0,0 +1,138 @@
+import fcntl
+import json
+from pathlib import Path
+import tempfile
+import unittest
+import sqlite3
+from unittest.mock import patch
+
+from paper_atlas.audits.arxiv_tags import audit, main
+from paper_atlas.db.store import Store
+from paper_atlas.ingestion.arxiv import refresh_categories
+
+
+class ArxivTagAuditTests(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 / 'audit.json'
+        self.store = Store(self.db)
+        self.addCleanup(self.store.conn.close)
+        self.records = []
+        for external in ('one', 'two'):
+            paper, record = self.store.record('arxiv', external, 'https://arxiv.org/abs/' + external,
+                'Same research work', 'Abstract', [], {'doi': '10.1234/same'},
+                {'categories': 'cs.AI cs.LG cs.AI'})
+            refresh_categories(self.store, paper, record, 'cs.AI cs.LG cs.AI')
+            self.records.append(record)
+        self.paper = paper
+        tag = self.store.conn.execute("INSERT INTO tags(namespace,value) VALUES('custom:topic','cs.AI')").lastrowid
+        self.store.conn.execute("INSERT INTO paper_tags VALUES(?,?,NULL,'manual:1')", (paper, tag))
+        self.store.conn.commit()
+
+    def test_all_merged_records_keep_their_own_categories_and_custom_tags_are_untouched(self):
+        before = self.store.conn.execute('SELECT * FROM paper_tags ORDER BY evidence_key,tag_id').fetchall()
+        result = audit(self.db, self.out, batch_size=1)
+        self.assertEqual(result['status'], 'passed')
+        self.assertEqual(result['records_checked'], 2)
+        self.assertEqual(result['expected_evidence'], 4)
+        self.assertEqual(result['actual_evidence'], 4)
+        self.assertEqual(result['evidence_without_arxiv_record'], 0)
+        self.assertEqual(self.store.conn.execute('SELECT * FROM paper_tags ORDER BY evidence_key,tag_id').fetchall(), before)
+
+    def test_zero_legacy_keys_does_not_hide_missing_categories_and_cli_fails(self):
+        self.store.conn.execute("DELETE FROM paper_tags WHERE source_record_id IS NOT NULL")
+        self.store.conn.commit()
+        with self.assertRaises(SystemExit):
+            main(['--db', str(self.db), '--out', str(self.out)])
+        result = json.loads(self.out.read_text())
+        self.assertEqual(result['status'], 'failed')
+        self.assertEqual(result['legacy_evidence'], 0)
+        self.assertEqual(result['missing_evidence'], 4)
+        self.assertEqual(result['records_with_mismatches'], 2)
+
+    def test_wrong_canonical_paper_or_stale_key_is_detected(self):
+        other, _ = self.store.record('example', 'other', 'https://example.org', 'Different work', 'AB', [])
+        self.store.conn.execute("UPDATE paper_tags SET paper_id=?,evidence_key='arxiv:categories' WHERE source_record_id=?", (other, self.records[0]))
+        self.store.conn.commit()
+        result = audit(self.db, self.out)
+        self.assertEqual(result['status'], 'failed')
+        self.assertEqual(result['records_with_mismatches'], 1)
+        self.assertEqual(result['legacy_evidence'], 2)
+        self.assertEqual(result['missing_evidence'], 2)
+        self.assertEqual(result['unexpected_evidence'], 2)
+
+    def test_wrong_source_and_unowned_category_rows_cannot_escape_the_full_audit(self):
+        paper, record = self.store.record('example', 'other', 'https://example.org', 'Other work', 'AB', [])
+        tag = self.store.conn.execute("SELECT id FROM tags WHERE namespace='arxiv:category' AND value='cs.AI'").fetchone()[0]
+        self.store.conn.execute('INSERT INTO paper_tags VALUES(?,?,?,?)', (paper, tag, record, 'wrong-source'))
+        self.store.conn.execute('INSERT INTO paper_tags VALUES(?,?,NULL,?)', (paper, tag, 'unowned'))
+        self.store.conn.commit()
+        result = audit(self.db, self.out)
+        self.assertEqual(result['status'], 'failed')
+        self.assertEqual(result['records_with_mismatches'], 0)
+        self.assertEqual(result['evidence_without_arxiv_record'], 2)
+        self.assertEqual(len(result['unscoped_examples']), 2)
+
+    def test_active_writer_and_incomplete_repair_are_rejected_before_publishing(self):
+        with Path(str(self.db) + '.maintenance.lock').open('a') as lock:
+            fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+            with self.assertRaises(BlockingIOError):
+                audit(self.db, self.out)
+        self.assertFalse(self.out.exists())
+        repair = self.root / 'repair.json'
+        repair.write_text(json.dumps({'status': 'running'}))
+        with self.assertRaisesRegex(ValueError, 'not completed'):
+            audit(self.db, self.out, repair)
+        self.assertFalse(self.out.exists())
+
+    def test_repair_report_scope_and_completed_audit_are_not_silently_reused(self):
+        repair = self.root / 'repair.json'
+        repair.write_text(json.dumps({'status': 'complete', 'records_refreshed': 2, 'db': str(self.db),
+            'source_id': 1, 'source_record_count': 2, 'maximum_source_record_id': 999}))
+        with self.assertRaisesRegex(ValueError, 'different source inventory'):
+            audit(self.db, self.out, repair)
+        result = audit(self.db, self.out)
+        before = self.out.read_bytes()
+        with self.assertRaisesRegex(ValueError, 'report exists'):
+            audit(self.db, self.out)
+        self.assertEqual(self.out.read_bytes(), before)
+        self.assertEqual(result['status'], 'passed')
+
+    def test_invalid_source_metadata_is_an_error_not_a_partial_pass(self):
+        self.store.conn.execute('UPDATE source_records SET metadata_json=? WHERE id=?',
+                                (json.dumps({'categories': ['invalid']}), self.records[0]))
+        self.store.conn.commit()
+        with self.assertRaisesRegex(ValueError, 'categories must be text'):
+            audit(self.db, self.out)
+        self.assertEqual(json.loads(self.out.read_text())['status'], 'error')
+
+    def test_interleaved_sources_use_bounded_primary_key_batches_without_sorting(self):
+        self.store.record('example', 'interleaved', 'https://example.org', 'Other source', 'AB', [])
+        paper, record = self.store.record('arxiv', 'three', 'https://arxiv.org/abs/three',
+                                          'A third record', 'AB', [], metadata={'categories': 'cs.AI'})
+        refresh_categories(self.store, paper, record, 'cs.AI')
+        self.store.conn.commit()
+        queries = []
+        connect = sqlite3.connect
+        def traced(*args, **kwargs):
+            connection = connect(*args, **kwargs)
+            connection.set_trace_callback(queries.append)
+            return connection
+        with patch('paper_atlas.audits.arxiv_tags.sqlite3.connect', side_effect=traced):
+            result = audit(self.db, self.out, batch_size=1)
+        self.assertEqual(result['status'], 'passed')
+        self.assertEqual(result['records_checked'], 3)
+        self.assertEqual(result['last_record_id'], record)
+        batches = [sql for sql in queries if sql.startswith('SELECT id,external_id,paper_id,metadata_json')]
+        self.assertEqual(len(batches), 4)  # Includes the bounded final empty read.
+        for sql in batches:
+            plan = ' '.join(row[3] for row in self.store.conn.execute('EXPLAIN QUERY PLAN ' + sql))
+            self.assertIn('INTEGER PRIMARY KEY', plan)
+            self.assertNotIn('TEMP B-TREE', plan)
+
+
+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()