99 lines
5.0 KiB
Python
99 lines
5.0 KiB
Python
"""One-shot server operation: merge collected files at an arXiv checkpoint.
|
|
|
|
Run as a transient systemd job so an SSH disconnect cannot strand the importer.
|
|
Only the Paper Atlas arXiv unit is controlled; business services are untouched.
|
|
"""
|
|
import argparse
|
|
from contextlib import closing
|
|
import json
|
|
import fcntl
|
|
import sqlite3
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from paper_atlas.ingestion.records import ingest_file
|
|
from paper_atlas.db.store import Store
|
|
from paper_atlas.ingestion.backfill import backfill,queue_files
|
|
from paper_atlas.db.backend import require_sqlite_writable
|
|
|
|
|
|
UNIT='paper-atlas-arxiv-import.service'
|
|
|
|
|
|
def property_value(name):
|
|
return subprocess.check_output(['systemctl','show',UNIT,'-p',name,'--value'],text=True).strip()
|
|
|
|
|
|
def merge_locked(db,archive,files,backfill_queue=None):
|
|
require_sqlite_writable(db)
|
|
# Validate inputs before stopping the existing writer.
|
|
for filename in files:
|
|
path=Path(filename)
|
|
if not path.is_file() or not path.with_suffix('.manifest.json').is_file():
|
|
raise ValueError('Missing collected file or manifest: '+filename)
|
|
evidence_files=queue_files(backfill_queue) if backfill_queue else []
|
|
if not Path(archive).is_file():
|
|
raise ValueError('Missing arXiv archive')
|
|
if not Path(db).is_file():
|
|
raise ValueError('Missing existing database')
|
|
resume_arxiv=property_value('ActiveState')=='active'
|
|
if not resume_arxiv:
|
|
with closing(sqlite3.connect(Path(db).resolve().as_uri()+'?mode=ro',uri=True)) as c:
|
|
last=c.execute("SELECT status FROM ingest_runs WHERE source_id=(SELECT id FROM sources WHERE name='arxiv') ORDER BY id DESC LIMIT 1").fetchone()
|
|
if not last or last[0]!='complete':
|
|
raise RuntimeError('arXiv is neither active nor verified complete; inspect its state before maintenance')
|
|
try:
|
|
if resume_arxiv:subprocess.run(['systemctl','stop',UNIT],check=True)
|
|
with closing(sqlite3.connect(db)) as c, c:
|
|
checkpoint=c.execute("SELECT records_written,status FROM ingest_runs WHERE source_id=(SELECT id FROM sources WHERE name='arxiv') ORDER BY id DESC LIMIT 1").fetchone()
|
|
print('Saved arxiv checkpoint:',checkpoint,flush=True)
|
|
if checkpoint and checkpoint[1]=='complete':resume_arxiv=False
|
|
schema_store=Store(db)
|
|
schema_store.conn.close()
|
|
if evidence_files:backfill(db,evidence_files)
|
|
for filename in files:
|
|
ingest_file(db,filename)
|
|
with closing(sqlite3.connect(db)) as c, c:
|
|
c.execute('PRAGMA cache_size=-32768')
|
|
if files or evidence_files:
|
|
errors=c.execute('PRAGMA foreign_key_check').fetchall()
|
|
if errors:
|
|
raise RuntimeError('Foreign key violations after merge')
|
|
else:
|
|
print('Tag refresh plan:',c.execute('EXPLAIN QUERY PLAN DELETE FROM paper_tags WHERE source_record_id=? AND evidence_key=?',(1,'arxiv:categories')).fetchall(),flush=True)
|
|
print('Decision totals:',c.execute('SELECT status,count(*) FROM decisions GROUP BY status').fetchall(),flush=True)
|
|
finally:
|
|
if resume_arxiv:
|
|
if property_value('LoadState')=='loaded':
|
|
subprocess.run(['systemctl','start',UNIT],check=True)
|
|
else:
|
|
subprocess.run(['systemd-run','--unit='+UNIT,'--property=WorkingDirectory='+str(Path(__file__).resolve().parents[2]),'--property=CPUQuota=50%','--property=MemoryMax=512M','--property=Nice=15','--property=IOWeight=10','--property=StandardOutput=append:/data/paper-index/arxiv-import.log','--property=StandardError=append:/data/paper-index/arxiv-import.log','/usr/bin/env','PYTHONDONTWRITEBYTECODE=1',sys.executable,'-B','-m','paper_atlas','import','arxiv','--db',db,'--archive',archive],check=True)
|
|
if property_value('ActiveState')!='active':
|
|
raise RuntimeError('arXiv unit did not resume; inspect the logs')
|
|
print('ARXIV_RESUMED',flush=True)
|
|
else:print('ARXIV_ALREADY_COMPLETE',flush=True)
|
|
|
|
|
|
def merge(db,archive,files,backfill_queue=None):
|
|
with Path(str(db)+'.maintenance.lock').open('a') as lock:
|
|
fcntl.flock(lock,fcntl.LOCK_EX|fcntl.LOCK_NB)
|
|
return merge_locked(db,archive,files,backfill_queue)
|
|
|
|
|
|
def main(argv=None, *, prog=None):
|
|
parser=argparse.ArgumentParser(prog=prog)
|
|
parser.add_argument('--db',required=True)
|
|
parser.add_argument('--archive',required=True)
|
|
parser.add_argument('--schema-only',action='store_true',help='Apply idempotent schema/index updates at a saved checkpoint')
|
|
parser.add_argument('--backfill-queue',help='Audited existing-source queue for canonical submission evidence')
|
|
parser.add_argument('files',nargs='*')
|
|
args=parser.parse_args(argv)
|
|
if not args.files and not args.schema_only and not args.backfill_queue:
|
|
parser.error('Provide collected files or --schema-only')
|
|
merge(args.db,args.archive,args.files,args.backfill_queue)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|