"""Shared identity resolution and provenance-preserving metadata storage.""" import json from importlib.resources import files import re import sqlite3 import unicodedata from pathlib import Path from paper_atlas.db.submissions import attach_submission,migrate_sources,write_decision,latest_status from paper_atlas.db.backend import require_sqlite_writable, inserted_id 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): require_sqlite_writable(path) self.conn = sqlite3.connect(path,timeout=60) self.conn.executescript(files('paper_atlas.db').joinpath('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 INTO sources(name,homepage) VALUES(?,?) ON CONFLICT DO NOTHING',(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 INTO paper_tags SELECT ?,tag_id,source_record_id,evidence_key FROM paper_tags WHERE paper_id=? ON CONFLICT DO NOTHING',(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 INTO match_candidates VALUES(?,?,?,?,?,?) ON CONFLICT DO NOTHING',(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,retrieved_at,metadata_json 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: old_abstract=old[2] or json.loads(old[4]).get('last_known_abstract',{}).get('abstract') c.execute("UPDATE papers SET title=?,abstract=COALESCE(NULLIF(?,''),abstract),title_key=?,authors_json=CASE WHEN ?='[]' THEN authors_json ELSE ? END WHERE id=? AND title=? AND COALESCE(abstract,'')=COALESCE(?,'')",(title,abstract,key,json.dumps(authors,ensure_ascii=False),json.dumps(authors,ensure_ascii=False),paper,old[1],old_abstract)) c.execute("UPDATE papers SET abstract=CASE WHEN abstract IS NULL OR abstract='' THEN ? ELSE abstract END WHERE id=?",(abstract,paper)) else: paper = inserted_id(c,'INSERT INTO papers(title,abstract,title_key,authors_json,first_published) VALUES(?,?,?,?,?)',(title,abstract,key,json.dumps(authors,ensure_ascii=False),published)) for scheme,value in ids: c.execute('INSERT INTO identifiers VALUES(?,?,?) ON CONFLICT DO NOTHING',(scheme,value,paper)) metadata=dict(metadata or {});metadata['record_identifiers']=dict(ids) if old and not abstract: previous={'abstract':old[2],'observed_at':old[3]} if old[2] else json.loads(old[4]).get('last_known_abstract') if previous:metadata['last_known_abstract']=previous from paper_atlas.db.submissions import timestamp c.execute('INSERT INTO source_records(source_id,external_id,paper_id,url,title,abstract,authors_json,metadata_json,retrieved_at) 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=excluded.retrieved_at',(source_id,external_id,paper,url,title,abstract,json.dumps(authors,ensure_ascii=False),json.dumps(metadata,ensure_ascii=False,default=str),timestamp())) 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 INTO match_candidates VALUES(?,?,?,?,?,?) ON CONFLICT DO NOTHING',(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 INTO tags(namespace,value) VALUES(?,?) ON CONFLICT DO NOTHING',(namespace,value)) tag=c.execute('SELECT id FROM tags WHERE namespace=? AND value=?',(namespace,value)).fetchone()[0] c.execute('INSERT INTO paper_tags VALUES(?,?,?,?) ON CONFLICT DO NOTHING',(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,guard_status=False): c = self.conn existing=c.execute('SELECT id FROM venues WHERE lower(acronym)=lower(?)',(venue,)).fetchone() if existing:venue_id=existing[0] else:venue_id=inserted_id(c,"INSERT INTO venues(acronym,name,field,venue_type) VALUES(?,?,?,'conference')",(venue,venue,'unclassified')) for alias in aliases or []: c.execute('INSERT INTO venue_aliases VALUES(?,?) ON CONFLICT DO NOTHING',(venue_id,alias)) c.execute('INSERT INTO editions(venue_id,year,track) VALUES(?,?,?) ON CONFLICT DO NOTHING',(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 guard_status and status in ('rejected','desk_rejected','withdrawn'): current = latest_status(c, submission) if current and current not in ('unknown','under_review',status): raise ValueError('Curated decision conflicts with current canonical submission status: ' + current) if submitted_at: from paper_atlas.db.submissions 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')