From 503908198e28b0abe567d76900bf97992bd19bfe Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Fri, 6 Nov 2026 19:11:02 +0800 Subject: [PATCH] Audit raw arXiv IDs and preserve per-record category evidence --- docs/operations.md | 18 ++++++ paper_atlas/audits/arxiv.py | 90 +++++++++++++++++++++++++++++ paper_atlas/cli.py | 2 + paper_atlas/ingestion/arxiv.py | 13 +++-- paper_atlas/ingestion/arxiv_tags.py | 74 ++++++++++++++++++++++++ tests/test_arxiv_audit.py | 44 ++++++++++++++ tests/test_arxiv_tags.py | 49 ++++++++++++++++ tests/test_import_arxiv.py | 24 ++++++++ 8 files changed, 309 insertions(+), 5 deletions(-) create mode 100644 paper_atlas/audits/arxiv.py create mode 100644 paper_atlas/ingestion/arxiv_tags.py create mode 100644 tests/test_arxiv_audit.py create mode 100644 tests/test_arxiv_tags.py diff --git a/docs/operations.md b/docs/operations.md index c121906..030827a 100644 --- a/docs/operations.md +++ b/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. 每次采集先核对来源、许可、可访问性、空闲空间和现有运行批次。保持低并发，遵守来源限流、重试与人机验证要求。 diff --git a/paper_atlas/audits/arxiv.py b/paper_atlas/audits/arxiv.py new file mode 100644 index 0000000..c84d0c2 --- /dev/null +++ b/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() diff --git a/paper_atlas/cli.py b/paper_atlas/cli.py index 4f1ce89..4bc2c12 100644 --- a/paper_atlas/cli.py +++ b/paper_atlas/cli.py @@ -24,7 +24,9 @@ COMMANDS = { ('import','records'): ('paper_atlas.ingestion.records', {}, '导入核验过的会议文件'), ('maintain','merge'): ('paper_atlas.ingestion.maintenance', {}, '协调单写入合并与恢复'), ('maintain','backfill'): ('paper_atlas.ingestion.backfill', {}, '回填既有投稿的来源身份'), + ('maintain','arxiv-tags'): ('paper_atlas.ingestion.arxiv_tags', {}, '恢复逐来源记录的 arXiv 类别证据'), ('audit','iclr'): ('paper_atlas.audits.iclr', {}, '核对 ICLR 快照与同届官方程序'), + ('audit','arxiv'): ('paper_atlas.audits.arxiv', {}, '核验 arXiv 原始编号、重复行与 ZIP 完整性'), } diff --git a/paper_atlas/ingestion/arxiv.py b/paper_atlas/ingestion/arxiv.py index 8ca497d..78d88a7 100644 --- a/paper_atlas/ingestion/arxiv.py +++ b/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') diff --git a/paper_atlas/ingestion/arxiv_tags.py b/paper_atlas/ingestion/arxiv_tags.py new file mode 100644 index 0000000..2e5cdc9 --- /dev/null +++ b/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