158 lines
11 KiB
Python
158 lines
11 KiB
Python
"""Read-only command-line queries for the paper index."""
|
|
import argparse
|
|
import json
|
|
import sqlite3
|
|
from pathlib import Path
|
|
|
|
|
|
LATEST_DECISION = '''NOT EXISTS (
|
|
SELECT 1 FROM decisions newer WHERE newer.submission_id=d.submission_id
|
|
AND (newer.observed_at,COALESCE(newer.decision_date,newer.observed_at),newer.id)
|
|
> (d.observed_at,COALESCE(d.decision_date,d.observed_at),d.id)
|
|
)'''
|
|
ACCEPTED_PAPERS = "SELECT s.paper_id FROM submissions s JOIN decisions d ON d.submission_id=s.id WHERE d.status='accepted' AND "+LATEST_DECISION
|
|
|
|
|
|
def known_acceptances(conn,paper):
|
|
from paper_atlas.db.events import has_event_key
|
|
event = 'e.event_key' if has_event_key(conn) else "''"
|
|
return [dict(row) for row in conn.execute('''SELECT s.id AS submission_id,v.acronym AS venue,e.year,e.track,'''+event+''' AS event_key,
|
|
d.id AS decision_id,d.decision_date,d.observed_at,d.evidence_url
|
|
FROM submissions s JOIN editions e ON e.id=s.edition_id JOIN venues v ON v.id=e.venue_id
|
|
JOIN decisions d ON d.submission_id=s.id WHERE s.paper_id=? AND d.status='accepted' AND '''+LATEST_DECISION+
|
|
' ORDER BY e.year,v.acronym,e.track,s.id',(paper,))]
|
|
|
|
|
|
def main(argv=None, *, prog=None):
|
|
parser = argparse.ArgumentParser(prog=prog)
|
|
database = parser.add_mutually_exclusive_group(required=True)
|
|
database.add_argument('--db', help='SQLite file')
|
|
database.add_argument('--pg-service', help='Protected libpq service; credentials stay outside CLI arguments')
|
|
parser.add_argument('--pg-schema', help='Optional PostgreSQL schema override; otherwise use the configured search_path')
|
|
parser.add_argument('--query', help='Title/abstract search: FTS5 for SQLite, web-search syntax for PostgreSQL')
|
|
parser.add_argument('--paper',type=int,help='Paper ID; include sources, tags and submission history')
|
|
parser.add_argument('--research-type', help='Independent type filter as namespace=value; no type inferred from track')
|
|
parser.add_argument('--annotation-history', action='store_true', help='Include retraction evidence for --paper')
|
|
parser.add_argument('--tag', help='Filter as namespace=value, e.g. arxiv:category=cs.AI')
|
|
parser.add_argument('--venue', help='Conference acronym')
|
|
parser.add_argument('--year', type=int)
|
|
parser.add_argument('--track',help='Exact track, e.g. main, findings, workshop')
|
|
parser.add_argument('--event-key',help='Exact event within a venue/year/track')
|
|
parser.add_argument('--status', choices=['accepted','rejected','desk_rejected','withdrawn','under_review','unknown'])
|
|
parser.add_argument('--outcome',choices=['accepted','no_known_acceptance'],help='跨会议的已知录用情况；未查到录用不等于从未录用')
|
|
parser.add_argument('--limit', type=int, default=20)
|
|
parser.add_argument('--stats', action='store_true')
|
|
args = parser.parse_args(argv)
|
|
if args.pg_schema and not args.pg_service:
|
|
parser.error('--pg-schema requires --pg-service')
|
|
if args.pg_service:
|
|
from paper_atlas.postgres.client import ReadConnection
|
|
conn = ReadConnection(args.pg_service, schema=args.pg_schema) if args.pg_schema else ReadConnection(args.pg_service)
|
|
else:
|
|
conn = sqlite3.connect(Path(args.db).resolve().as_uri()+'?mode=ro', uri=True)
|
|
conn.row_factory = sqlite3.Row
|
|
conn.execute('BEGIN') # One consistent read snapshot while ingestion continues.
|
|
from paper_atlas.db.events import has_event_key
|
|
event = 'e.event_key' if has_event_key(conn) else "''"
|
|
if args.event_key is not None and event == "''":
|
|
conn.close()
|
|
raise ValueError('Event identity migration required for event-key queries')
|
|
from paper_atlas.curation.annotations import available, observations, ACTIVE
|
|
has_annotations = available(conn)
|
|
if args.stats:
|
|
result = {name:conn.execute('SELECT count(*) FROM '+name).fetchone()[0] for name in ('papers','source_records','submissions','decisions','tags','match_candidates')}
|
|
result['records_by_source'] = [dict(r) for r in conn.execute('SELECT s.name,count(r.id) AS imported_records FROM sources s LEFT JOIN source_records r ON r.source_id=s.id GROUP BY s.id ORDER BY s.name')]
|
|
result['decision_events_by_status'] = [dict(r) for r in conn.execute('SELECT status,count(*) AS events FROM decisions GROUP BY status')]
|
|
accepted=conn.execute('SELECT count(DISTINCT paper_id) FROM ('+ACCEPTED_PAPERS+')').fetchone()[0]
|
|
result['papers_with_known_acceptance']=accepted
|
|
result['papers_without_known_acceptance']=result['papers']-accepted
|
|
result['outcome_scope']='Only imported evidence is known. No known acceptance is not proof that a paper was never accepted.'
|
|
result['coverage'] = [dict(r) for r in conn.execute('SELECT status,count(*) AS sources FROM edition_sources GROUP BY status')]
|
|
result['runs'] = [dict(r) for r in conn.execute('SELECT * FROM ingest_runs ORDER BY id DESC LIMIT 10')]
|
|
else:
|
|
where = []; params = []
|
|
if args.paper is not None:
|
|
where.append('p.id=?'); params.append(args.paper)
|
|
if args.query:
|
|
if args.pg_service:
|
|
from paper_atlas.postgres.client import SEARCH_VECTOR
|
|
where.append('p.id IN (SELECT id FROM papers WHERE '+SEARCH_VECTOR+" @@ websearch_to_tsquery('english'::regconfig,?))")
|
|
else:
|
|
where.append('p.id IN (SELECT rowid FROM papers_fts WHERE papers_fts MATCH ?)')
|
|
params.append(args.query)
|
|
if args.tag:
|
|
if '=' not in args.tag:
|
|
parser.error('--tag requires namespace=value')
|
|
namespace,value = args.tag.split('=',1)
|
|
platform = 'EXISTS(SELECT 1 FROM paper_tags pt JOIN tags t ON t.id=pt.tag_id WHERE pt.paper_id=p.id AND t.namespace=? AND t.value=?)'
|
|
params.extend([namespace,value])
|
|
if has_annotations:
|
|
platform = '(' + platform + ' OR EXISTS(SELECT 1 FROM annotation_events a WHERE a.paper_id=p.id AND a.kind=\'tag\' AND a.namespace=? AND a.value=? AND ' + ACTIVE + '))'
|
|
params.extend([namespace,value])
|
|
where.append(platform)
|
|
if args.research_type:
|
|
if '=' not in args.research_type:
|
|
parser.error('--research-type requires namespace=value')
|
|
namespace,value = args.research_type.split('=',1)
|
|
if has_annotations:
|
|
where.append('EXISTS(SELECT 1 FROM annotation_events a WHERE a.paper_id=p.id AND a.kind=\'research_type\' AND a.namespace=? AND a.value=? AND ' + ACTIVE + ')')
|
|
params.extend([namespace,value])
|
|
else:
|
|
where.append('1=0')
|
|
if args.outcome:
|
|
where.append('p.id '+('IN' if args.outcome=='accepted' else 'NOT IN')+' ('+ACCEPTED_PAPERS+')')
|
|
if args.venue or args.year or args.status or args.track or args.event_key is not None:
|
|
clause = ['1=1']
|
|
if args.venue:
|
|
if conn.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name='venue_aliases'").fetchone():
|
|
clause.append('(lower(v.acronym)=lower(?) OR EXISTS(SELECT 1 FROM venue_aliases va WHERE va.venue_id=v.id AND lower(va.alias)=lower(?)))'); params.extend([args.venue,args.venue])
|
|
else:
|
|
clause.append('lower(v.acronym)=lower(?)');params.append(args.venue)
|
|
if args.event_key is not None:
|
|
clause.append('e.event_key=?');params.append(args.event_key)
|
|
if args.track:
|
|
clause.append('e.track=?');params.append(args.track)
|
|
if args.year:
|
|
clause.append('e.year=?'); params.append(args.year)
|
|
if args.status:
|
|
latest='EXISTS (SELECT 1 FROM decisions d WHERE d.submission_id=s.id AND d.status=? AND '+LATEST_DECISION+')'
|
|
clause.append('(NOT EXISTS(SELECT 1 FROM decisions d WHERE d.submission_id=s.id) OR '+latest+')' if args.status=='unknown' else latest); params.append(args.status)
|
|
where.append('p.id IN (SELECT s.paper_id FROM submissions s JOIN editions e ON e.id=s.edition_id JOIN venues v ON v.id=e.venue_id WHERE '+' AND '.join(clause)+')')
|
|
sql = 'SELECT p.* FROM papers p'+(' WHERE '+' AND '.join(where) if where else '')+' ORDER BY p.id LIMIT ?'
|
|
params.append(max(1,min(args.limit,200)))
|
|
result = []
|
|
for row in conn.execute(sql,params):
|
|
paper = dict(row)
|
|
paper['authors'] = json.loads(paper.pop('authors_json'))
|
|
paper['identifiers'] = [dict(r) for r in conn.execute('SELECT scheme,value FROM identifiers WHERE paper_id=?',(row['id'],))]
|
|
paper['known_acceptances']=known_acceptances(conn,row['id'])
|
|
paper['publication_outcome']='accepted' if paper['known_acceptances'] else 'no_known_acceptance'
|
|
if args.paper is not None:
|
|
paper['research_types'] = observations(conn, row['id'], kind='research_type')
|
|
paper['annotation_tags'] = observations(conn, row['id'], kind='tag')
|
|
if args.annotation_history:
|
|
paper['annotation_history'] = observations(conn, row['id'], history=True)
|
|
paper['tags'] = [dict(r) for r in conn.execute('SELECT t.namespace,t.value,pt.evidence_key FROM paper_tags pt JOIN tags t ON t.id=pt.tag_id WHERE pt.paper_id=?',(row['id'],))]
|
|
paper['sources'] = [dict(r) for r in conn.execute('SELECT s.name,r.external_id,r.url,r.title,r.abstract,r.retrieved_at FROM source_records r JOIN sources s ON s.id=r.source_id WHERE r.paper_id=?',(row['id'],))]
|
|
submissions = []
|
|
for sub in conn.execute('SELECT s.id,s.source_submission_id,s.submitted_at,v.acronym,e.year,e.track,'+event+' AS event_key FROM submissions s JOIN editions e ON e.id=s.edition_id JOIN venues v ON v.id=e.venue_id WHERE s.paper_id=? ORDER BY e.year,s.id',(row['id'],)):
|
|
submission = dict(sub)
|
|
submission['decisions'] = []
|
|
for d in conn.execute('SELECT id,status,raw_value,evidence_url,observed_at,decision_date FROM decisions WHERE submission_id=? ORDER BY observed_at,COALESCE(decision_date,observed_at),id',(sub['id'],)):
|
|
decision=dict(d)
|
|
if conn.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name='decision_sources'").fetchone():
|
|
decision['sources']=[dict(r) for r in conn.execute('SELECT s.name,r.external_id,ds.source_decision_id FROM decision_sources ds JOIN source_records r ON r.id=ds.source_record_id JOIN sources s ON s.id=r.source_id WHERE ds.decision_id=?',(d['id'],))]
|
|
submission['decisions'].append(decision)
|
|
submission['latest_status']=submission['decisions'][-1]['status'] if submission['decisions'] else 'unknown'
|
|
if conn.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name='submission_sources'").fetchone():
|
|
submission['sources']=[dict(r) for r in conn.execute('SELECT so.name,r.external_id,ss.source_submission_id FROM submission_sources ss JOIN source_records r ON r.id=ss.source_record_id JOIN sources so ON so.id=r.source_id WHERE ss.submission_id=?',(sub['id'],))]
|
|
submissions.append(submission)
|
|
paper['submissions'] = submissions
|
|
result.append(paper)
|
|
print(json.dumps(result,ensure_ascii=False,indent=2))
|
|
conn.close()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|