diff --git a/docs/operations.md b/docs/operations.md index a455d06..91e73fe 100644 --- a/docs/operations.md +++ b/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. 每次采集先核对来源、许可、可访问性、空闲空间和现有运行批次。保持低并发，遵守来源限流、重试与人机验证要求。 diff --git a/paper_atlas/audits/arxiv_tags.py b/paper_atlas/audits/arxiv_tags.py new file mode 100644 index 0000000..688c651 --- /dev/null +++ b/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() diff --git a/paper_atlas/cli.py b/paper_atlas/cli.py index b2d6a7f..b084666 100644 --- a/paper_atlas/cli.py +++ b/paper_atlas/cli.py @@ -31,6 +31,7 @@ COMMANDS = { ('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', {}, '核对公开负面决定与已有投稿历史'), } diff --git a/tests/test_arxiv_tag_audit.py b/tests/test_arxiv_tag_audit.py new file mode 100644 index 0000000..adb0127 --- /dev/null +++ b/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()