paper-atlas/paper_atlas/db/store.py

"""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
from paper_atlas.db.backend import require_sqlite_writable


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 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,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 = 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)
        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
        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
        existing=c.execute('SELECT id FROM venues WHERE acronym=? COLLATE NOCASE',(venue,)).fetchone()
        if existing:venue_id=existing[0]
        else:venue_id=c.execute("INSERT INTO venues(acronym,name,field,venue_type) VALUES(?,?,?,'conference')",(venue,venue,'unclassified')).lastrowid
        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 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')