156 lines
10 KiB
Python
156 lines
10 KiB
Python
"""Shared identity resolution and provenance-preserving metadata storage."""
|
|
import json
|
|
import re
|
|
import sqlite3
|
|
import unicodedata
|
|
from pathlib import Path
|
|
from submission_evidence import attach_submission,migrate_sources,write_decision
|
|
|
|
|
|
def normalized(value):
|
|
return ''.join(c for c in unicodedata.normalize('NFKD',value).casefold() if c.isalnum())
|
|
|
|
|
|
def author_keys(authors):
|
|
if not isinstance(authors,list):
|
|
return set()
|
|
result = set()
|
|
for author in authors:
|
|
if isinstance(author,list):
|
|
author = ' '.join(str(x) for x in author[1:2]+author[:1])
|
|
elif isinstance(author,dict):
|
|
author = ' '.join(str(author.get(k,'')) for k in ('given','family'))
|
|
key = normalized(str(author))
|
|
if key:
|
|
result.add(key)
|
|
return result
|
|
|
|
|
|
class Store:
|
|
def __init__(self,path):
|
|
self.conn = sqlite3.connect(path,timeout=60)
|
|
self.conn.executescript(Path(__file__).with_name('schema.sql').read_text())
|
|
self.conn.execute('PRAGMA synchronous=NORMAL')
|
|
self.conn.execute('PRAGMA cache_size=-32768')
|
|
migrate_sources(self.conn)
|
|
|
|
def source(self,name,url):
|
|
self.conn.execute('INSERT OR IGNORE INTO sources(name,homepage) VALUES(?,?)',(name,url))
|
|
return self.conn.execute('SELECT id FROM sources WHERE name=?',(name,)).fetchone()[0]
|
|
|
|
def merge(self,target,other,reason):
|
|
c = self.conn
|
|
if target == other:
|
|
return
|
|
columns=('abstract','authors_json','first_published')
|
|
target_values=c.execute('SELECT abstract,authors_json,first_published FROM papers WHERE id=?',(target,)).fetchone()
|
|
other_values=c.execute('SELECT abstract,authors_json,first_published FROM papers WHERE id=?',(other,)).fetchone()
|
|
if not target_values or not other_values:raise ValueError('Missing paper entity during merge')
|
|
fill=[(column,value) for column,current,value in zip(columns,target_values,other_values) if current in (None,'','[]') and value not in (None,'','[]')]
|
|
if fill:c.execute('UPDATE papers SET '+','.join(column+'=?' for column,_ in fill)+' WHERE id=?',(*[value for _,value in fill],target))
|
|
c.execute('INSERT INTO paper_merges(former_paper_id,surviving_paper_id,reason) VALUES(?,?,?)',(other,target,reason))
|
|
c.execute('UPDATE identifiers SET paper_id=? WHERE paper_id=?',(target,other))
|
|
c.execute('UPDATE source_records SET paper_id=? WHERE paper_id=?',(target,other))
|
|
c.execute('UPDATE submissions SET paper_id=? WHERE paper_id=?',(target,other))
|
|
c.execute('INSERT OR IGNORE INTO paper_tags SELECT ?,tag_id,source_record_id,evidence_key FROM paper_tags WHERE paper_id=?',(target,other))
|
|
c.execute('DELETE FROM paper_tags WHERE paper_id=?',(other,))
|
|
c.execute('UPDATE paper_merges SET surviving_paper_id=? WHERE surviving_paper_id=?',(target,other))
|
|
# Reattach pending candidate evidence without losing colliding rows.
|
|
candidates = c.execute('SELECT left_paper_id,right_paper_id,method,score,evidence_json,status FROM match_candidates WHERE left_paper_id=? OR right_paper_id=?',(other,other)).fetchall()
|
|
c.execute('DELETE FROM match_candidates WHERE left_paper_id=? OR right_paper_id=?',(other,other))
|
|
for left,right,method,score,evidence,status in candidates:
|
|
left = target if left == other else left
|
|
right = target if right == other else right
|
|
if left != right:
|
|
c.execute('INSERT OR IGNORE INTO match_candidates VALUES(?,?,?,?,?,?)',(min(left,right),max(left,right),method,score,evidence,status))
|
|
c.execute('DELETE FROM papers WHERE id=?',(other,))
|
|
|
|
def record(self,source,external_id,url,title,abstract,authors,identifiers=None,metadata=None,published=None):
|
|
c = self.conn
|
|
identifiers = dict(identifiers or {})
|
|
# External IDs are scoped to platforms; an OpenReview ID is a forum ID.
|
|
identifiers[source] = external_id
|
|
ids = []
|
|
for scheme,value in identifiers.items():
|
|
if not value:
|
|
continue
|
|
value = str(value).strip()
|
|
if scheme == 'doi':
|
|
value = re.sub(r'^https?://(?:dx\.)?doi\.org/','',value,flags=re.I).lower()
|
|
if not re.fullmatch(r'10\.\d{4,9}/\S+',value):
|
|
continue
|
|
if scheme == 'arxiv':
|
|
value = re.sub(r'v\d+$','',value)
|
|
ids.append((scheme,value))
|
|
matches = set()
|
|
for scheme,value in ids:
|
|
found = c.execute('SELECT paper_id FROM identifiers WHERE scheme=? AND value=?',(scheme,value)).fetchone()
|
|
if found:
|
|
matches.add(found[0])
|
|
old = c.execute('SELECT paper_id,title,abstract FROM source_records WHERE source_id=? AND external_id=?',(source_id:=self.source(source,url),external_id)).fetchone()
|
|
if old:
|
|
matches.add(old[0])
|
|
key = normalized(title)
|
|
title_matches = c.execute('SELECT id,authors_json,abstract FROM papers WHERE title_key=?',(key,)).fetchall()
|
|
authors_set = author_keys(authors)
|
|
for paper,old_authors,old_abstract in title_matches:
|
|
overlap = authors_set & author_keys(json.loads(old_authors))
|
|
# Cross-platform title matching needs independent author corroboration.
|
|
same_source = c.execute('SELECT 1 FROM source_records WHERE paper_id=? AND source_id=? LIMIT 1',(paper,source_id)).fetchone()
|
|
if not same_source and len(key)>=25 and (len(overlap)>=2 or (overlap and abstract and abstract.strip()==(old_abstract or '').strip())):
|
|
matches.add(paper)
|
|
if matches:
|
|
paper = min(matches)
|
|
for other in sorted(matches-{paper}):
|
|
self.merge(paper,other,'explicit identifier or exact title corroborated by authors')
|
|
if old:
|
|
c.execute("UPDATE papers SET title=?,abstract=?,title_key=?,authors_json=? WHERE id=? AND title=? AND COALESCE(abstract,'')=COALESCE(?,'')",(title,abstract,key,json.dumps(authors,ensure_ascii=False),paper,old[1],old[2]))
|
|
c.execute("UPDATE papers SET abstract=CASE WHEN abstract IS NULL OR abstract='' THEN ? ELSE abstract END WHERE id=?",(abstract,paper))
|
|
else:
|
|
paper = c.execute('INSERT INTO papers(title,abstract,title_key,authors_json,first_published) VALUES(?,?,?,?,?)',(title,abstract,key,json.dumps(authors,ensure_ascii=False),published)).lastrowid
|
|
for scheme,value in ids:
|
|
c.execute('INSERT OR IGNORE INTO identifiers VALUES(?,?,?)',(scheme,value,paper))
|
|
metadata=dict(metadata or {});metadata['record_identifiers']=dict(ids)
|
|
c.execute('INSERT INTO source_records(source_id,external_id,paper_id,url,title,abstract,authors_json,metadata_json) VALUES(?,?,?,?,?,?,?,?) ON CONFLICT(source_id,external_id) DO UPDATE SET paper_id=excluded.paper_id,url=excluded.url,title=excluded.title,abstract=excluded.abstract,authors_json=excluded.authors_json,metadata_json=excluded.metadata_json,retrieved_at=CURRENT_TIMESTAMP',(source_id,external_id,paper,url,title,abstract,json.dumps(authors,ensure_ascii=False),json.dumps(metadata,ensure_ascii=False,default=str)))
|
|
record = c.execute('SELECT id FROM source_records WHERE source_id=? AND external_id=?',(source_id,external_id)).fetchone()[0]
|
|
for other,_,_ in title_matches:
|
|
if other != paper and c.execute('SELECT 1 FROM papers WHERE id=?',(other,)).fetchone():
|
|
c.execute('INSERT OR IGNORE INTO match_candidates VALUES(?,?,?,?,?,?)',(min(paper,other),max(paper,other),'exact_title_unconfirmed',None,json.dumps({'title':title}), 'pending'))
|
|
return paper,record
|
|
|
|
def record_tags(self,paper,record,source,tags):
|
|
"""Refresh this record's platform tags without removing other evidence."""
|
|
values=set()
|
|
for tag in tags:
|
|
namespace,value=tag['namespace'],tag['value']
|
|
if not isinstance(namespace,str) or not namespace.startswith(source+':') or not isinstance(value,str) or not value.strip():
|
|
raise ValueError('Platform tag must have its own source namespace and a nonempty text value')
|
|
values.add((namespace,value.strip()))
|
|
evidence=f'{source}:record:{record}:tags'
|
|
c=self.conn
|
|
c.execute('DELETE FROM paper_tags WHERE source_record_id=? AND evidence_key=?',(record,evidence))
|
|
for namespace,value in sorted(values):
|
|
c.execute('INSERT OR IGNORE INTO tags(namespace,value) VALUES(?,?)',(namespace,value))
|
|
tag=c.execute('SELECT id FROM tags WHERE namespace=? AND value=?',(namespace,value)).fetchone()[0]
|
|
c.execute('INSERT OR IGNORE INTO paper_tags VALUES(?,?,?,?)',(paper,tag,record,evidence))
|
|
|
|
def decision(self,paper,record,venue,year,track,external_id,url,status,raw_value,aliases=None,identities=None,observed_at=None,decision_date=None,source_decision_id='',submitted_at=None):
|
|
c = self.conn
|
|
c.execute("INSERT OR IGNORE INTO venues(acronym,name,field,venue_type) VALUES(?,?,?,'conference')",(venue,venue,'unclassified'))
|
|
venue_id = c.execute('SELECT id FROM venues WHERE acronym=?',(venue,)).fetchone()[0]
|
|
for alias in aliases or []:
|
|
c.execute('INSERT OR IGNORE INTO venue_aliases VALUES(?,?)',(venue_id,alias))
|
|
c.execute('INSERT OR IGNORE INTO editions(venue_id,year,track) VALUES(?,?,?)',(venue_id,int(year),track))
|
|
edition = c.execute('SELECT id FROM editions WHERE venue_id=? AND year=? AND track=?',(venue_id,int(year),track)).fetchone()[0]
|
|
# A previous event in this row may already have merged the input paper.
|
|
paper=c.execute('SELECT paper_id FROM source_records WHERE id=?',(record,)).fetchone()[0]
|
|
submission=attach_submission(self,edition,paper,record,external_id,identities or [])
|
|
if submitted_at:
|
|
from submission_evidence import timestamp
|
|
c.execute('UPDATE submissions SET submitted_at=COALESCE(submitted_at,?) WHERE id=?',(timestamp(submitted_at),submission))
|
|
if status is not None:write_decision(self,submission,record,status,raw_value,url,observed_at,decision_date,source_decision_id)
|
|
return edition
|
|
|
|
def accepted(self,paper,record,venue,year,track,external_id,url):
|
|
return self.decision(paper,record,venue,year,track,external_id,url,'accepted','listed in official proceedings')
|