138 lines
7.8 KiB
Python
138 lines
7.8 KiB
Python
"""Describe imported venue coverage without implying a complete census."""
|
|
import argparse
|
|
from collections import defaultdict
|
|
import json
|
|
from pathlib import Path
|
|
import sqlite3
|
|
|
|
from paper_atlas.config import DEFAULT_REGISTRY
|
|
from paper_atlas.db.backend import database_arguments, open_read_connection
|
|
from paper_atlas.db.submissions import timestamp
|
|
from paper_atlas.querying import LATEST_DECISION
|
|
|
|
|
|
STATUSES = ('accepted', 'rejected', 'desk_rejected', 'withdrawn', 'under_review', 'unknown')
|
|
|
|
|
|
def _status_counts(connection, *, latest):
|
|
if latest:
|
|
query = '''SELECT s.edition_id,COALESCE(d.status,'unknown') AS status,count(*) AS count
|
|
FROM submissions s LEFT JOIN decisions d ON d.submission_id=s.id AND ''' + LATEST_DECISION
|
|
else:
|
|
query = '''SELECT s.edition_id,COALESCE(d.status,'unknown') AS status,count(*) AS count
|
|
FROM submissions s JOIN decisions d ON d.submission_id=s.id'''
|
|
counts = defaultdict(dict)
|
|
for row in connection.execute(query + " GROUP BY s.edition_id,COALESCE(d.status,'unknown')"):
|
|
counts[row[0]][row[1]] = row[2]
|
|
return counts
|
|
|
|
|
|
def report(db, registry, *, pg_service=None, pg_schema=None):
|
|
started_at = timestamp()
|
|
registry = json.loads(Path(registry).read_text())
|
|
targets = registry['venues']
|
|
names = [row['acronym'].casefold() for row in targets]
|
|
if len(names) != len(set(names)):
|
|
raise ValueError('Duplicate target venue')
|
|
scope = '''(lower(v.acronym)=lower(?) OR EXISTS(
|
|
SELECT 1 FROM venue_aliases a WHERE a.venue_id=v.id AND lower(a.alias)=lower(?)))'''
|
|
connection = open_read_connection(db, pg_service=pg_service, pg_schema=pg_schema)
|
|
if not pg_service:
|
|
connection.row_factory = sqlite3.Row
|
|
try:
|
|
if pg_service:
|
|
identity = connection.execute('SELECT current_database(),current_schema()').fetchone()
|
|
database, schema = identity[0], identity[1]
|
|
else:
|
|
database, schema = str(Path(db).resolve()), None
|
|
latest = _status_counts(connection, latest=True)
|
|
observations = _status_counts(connection, latest=False)
|
|
# Aggregate the paper join once for the whole snapshot. Repeating it
|
|
# for each registry target caused repeated scans on the production DB.
|
|
edition_counts = {}
|
|
for row in connection.execute('''SELECT s.edition_id,count(s.id) AS submissions,
|
|
count(DISTINCT s.paper_id) AS papers,
|
|
count(DISTINCT CASE WHEN p.abstract IS NULL OR p.abstract='' THEN s.paper_id END)
|
|
AS papers_without_abstract_text
|
|
FROM submissions s LEFT JOIN papers p ON p.id=s.paper_id GROUP BY s.edition_id'''):
|
|
entry = dict(row)
|
|
edition_counts[entry.pop('edition_id')] = entry
|
|
source_counts = defaultdict(list)
|
|
for row in connection.execute('''SELECT s.edition_id,so.name AS source,
|
|
count(DISTINCT sr.id) AS source_records,
|
|
count(DISTINCT CASE WHEN sr.abstract IS NULL OR sr.abstract='' THEN sr.id END)
|
|
AS source_records_without_abstract_text
|
|
FROM submission_sources ss JOIN submissions s ON s.id=ss.submission_id
|
|
JOIN source_records sr ON sr.id=ss.source_record_id JOIN sources so ON so.id=sr.source_id
|
|
GROUP BY s.edition_id,so.name ORDER BY s.edition_id,so.name'''):
|
|
entry = dict(row)
|
|
source_counts[entry.pop('edition_id')].append(entry)
|
|
venues, covered_venue_ids = [], set()
|
|
for target in targets:
|
|
acronym = target['acronym']
|
|
params = (acronym, acronym)
|
|
editions = []
|
|
for row in connection.execute('''SELECT e.id,v.id AS venue_id,v.acronym AS canonical_venue,
|
|
e.year,e.track FROM editions e JOIN venues v ON v.id=e.venue_id
|
|
WHERE ''' + scope + '''
|
|
ORDER BY e.year,e.track,v.acronym''', params):
|
|
edition = dict(row)
|
|
edition.update(edition_counts.get(row['id'], {
|
|
'submissions': 0, 'papers': 0, 'papers_without_abstract_text': 0}))
|
|
covered_venue_ids.add(edition['venue_id'])
|
|
edition['latest_observed_status_counts'] = {status: latest[row['id']].get(status, 0) for status in STATUSES}
|
|
edition['decision_observation_counts'] = {status: observations[row['id']].get(status, 0) for status in STATUSES}
|
|
edition['source_record_counts'] = source_counts[row['id']]
|
|
edition['sources'] = [dict(source) for source in connection.execute('''SELECT so.name,
|
|
es.status,es.expected_count,es.imported_count,es.list_url,es.last_success_at,es.last_error
|
|
FROM edition_sources es JOIN sources so ON so.id=es.source_id
|
|
WHERE es.edition_id=? ORDER BY so.name,es.list_url''', (row['id'],))]
|
|
editions.append(edition)
|
|
totals = connection.execute('''SELECT count(s.id),count(DISTINCT 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 ''' + scope, params).fetchone()
|
|
venues.append({'acronym': acronym, 'field': target['field'],
|
|
'imported_submissions': totals[0], 'distinct_papers': totals[1],
|
|
'history_status': 'partial_history_pending_independent_audit' if totals[0] else 'no_imported_records',
|
|
'discovery_sources': target.get('discovery_sources', []), 'editions': editions})
|
|
outside_registry = [dict(row) for row in connection.execute('''SELECT v.id AS venue_id,
|
|
v.acronym,count(s.id) AS submissions,count(DISTINCT s.paper_id) AS papers
|
|
FROM submissions s JOIN editions e ON e.id=s.edition_id JOIN venues v ON v.id=e.venue_id
|
|
GROUP BY v.id,v.acronym ORDER BY v.acronym''') if row[0] not in covered_venue_ids]
|
|
return {'report_started_at': started_at, 'checked_at': timestamp(),
|
|
'engine': 'postgresql' if pg_service else 'sqlite', 'database': database, 'schema': schema,
|
|
'pg_service': pg_service, 'target_venue_count': len(targets),
|
|
'venues_with_imported_records': sum(bool(v['imported_submissions']) for v in venues),
|
|
'scope_note': registry['scope_note'],
|
|
'interpretation': 'All registry targets are listed, including zero imports. '
|
|
'Latest status counts use the latest observed decision for each attempt; decision observations retain history. '
|
|
'Source record totals are per source and edition, not per list URL; multiple list entries may describe overlapping records. '
|
|
'Missing abstract text means NULL or empty string; nonempty text does not prove a complete abstract. '
|
|
'An edition source marked complete is not a proof of complete conference history or public rejection coverage. '
|
|
'Zero imported rejections does not establish that no rejections are public. '
|
|
'Papers can appear in several venues/years; do not sum venue paper counts as a global total.',
|
|
'venues': venues, 'imported_venues_outside_registry': outside_registry}
|
|
finally:
|
|
connection.close()
|
|
|
|
|
|
def main(argv=None, *, prog=None):
|
|
parser = argparse.ArgumentParser(prog=prog)
|
|
database_arguments(parser)
|
|
parser.add_argument('--registry', default=str(DEFAULT_REGISTRY))
|
|
parser.add_argument('--out')
|
|
args = parser.parse_args(argv)
|
|
result = report(args.db, args.registry, pg_service=args.pg_service, pg_schema=args.pg_schema)
|
|
text = json.dumps(result, ensure_ascii=False, indent=2)
|
|
if args.out:
|
|
path = Path(args.out)
|
|
stage = path.with_suffix('.part')
|
|
stage.write_text(text + '\n')
|
|
stage.replace(path)
|
|
else:
|
|
print(text)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|