91 lines
6.5 KiB
Python
91 lines
6.5 KiB
Python
"""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()
|