Compare commits

..

4 Commits

Author SHA1 Message Date
codex-bot 88c2819edd Keep legacy source refreshes in their verified joint editions
Metadata integrity checks / metadata (push) Successful in 1m13s
2026-11-08 19:15:26 +08:00
codex-bot 866aeaf068 Read PostgreSQL mapping rows by position during joint preflight
Metadata integrity checks / metadata (push) Successful in 1m13s
2026-11-08 19:10:10 +08:00
codex-bot c05b979b01 Bind repair execution receipts to the reviewed plan hash 2026-11-08 19:09:27 +08:00
codex-bot 3f81734e69 Repair verified joint edition membership without reimporting papers
Metadata integrity checks / metadata (push) Failing after 1m5s
2026-11-08 19:07:00 +08:00
18 changed files with 490 additions and 607 deletions
+1 -2
View File
@@ -49,6 +49,7 @@ jobs:
subprocess.run([str(venv/'bin/paper-atlas'),'--help'],cwd=outside,check=True)
subprocess.run([python,'-I','-B','-m','paper_atlas','collect','cvf','--help'],cwd=outside,check=True)
subprocess.run([python,'-I','-B','-m','paper_atlas','maintain','postgres','--help'],cwd=outside,check=True)
subprocess.run([python,'-I','-B','-m','paper_atlas','maintain','joint-editions','--help'],cwd=outside,check=True)
subprocess.run([python,'-I','-B','-m','paper_atlas','import','records','--help'],cwd=outside,check=True)
subprocess.run([python,'-I','-B','-m','paper_atlas','import','arxiv','--help'],cwd=outside,check=True)
subprocess.run([python,'-I','-B','-m','paper_atlas','maintain','backfill','--help'],cwd=outside,check=True)
@@ -58,7 +59,5 @@ jobs:
smoke = "from paper_atlas.db.store import Store; from paper_atlas.config import DEFAULT_REGISTRY; import json; assert {'COLM','MM','MLSys'} <= {v['acronym'] for v in json.loads(DEFAULT_REGISTRY.read_text())['venues']}; s=Store('installed.sqlite'); s.conn.close()"
subprocess.run([python,'-I','-B','-c',smoke],cwd=outside,check=True)
subprocess.run([python,'-I','-B','-m','paper_atlas','query','--db','installed.sqlite','--stats'],cwd=outside,check=True)
subprocess.run([python,'-I','-B','-m','paper_atlas','annotate','init','--db','installed.sqlite'],cwd=outside,check=True)
subprocess.run([python,'-I','-B','-m','paper_atlas','annotate','list','--db','installed.sqlite','--paper','1'],cwd=outside,check=True)
subprocess.run([python,'-I','-B','-m','paper_atlas','audit','histories','--db','installed.sqlite','--examples','0'],cwd=outside,check=True)
PY
+1 -1
View File
@@ -87,4 +87,4 @@ CI 在已授权的远程 runner 上运行测试、构建 wheel,并从源码目
开发通过工作分支和 PR 交付,由用户手动合并。
研究类型与自定义标签可通过 [`annotate`](docs/annotations.md) 显式初始化、添加和撤回;两者独立查询,保留证据及完整更正历史,不从会议轨道推断研究类型。
历史联合会议的既有届次归属可通过 [`maintain joint-editions`](docs/joint-editions.md) 先只读规划,再按计划哈希执行;保留论文与投稿决定来源,不重导入。
-107
View File
@@ -1,107 +0,0 @@
# Research types and custom annotations
Research type describes a claimed characteristic of the work (for example a survey,
benchmark, empirical study or replication). It is independent of topic, paper length,
conference track, presentation format and acceptance status. No type is inferred from
those fields. Multiple source-specific or human classifications may coexist; no
cross-platform vocabulary mapping or automatic classifier is provided.
`annotation_events` stores both kinds with an explicit `kind` discriminator:
`research_type` and `tag` are separate domains, even when namespace/value strings
happen to coincide. Existing platform labels remain in `tags` / `paper_tags`.
Each assertion carries namespace, value, actor, reason, evidence key, optional source
record and evidence URL, and the time it was recorded. This is observation time, not
paper publication time or a conference decision date.
Values are not forced into a global taxonomy: `custom:method=empirical` and
`platform:declared_type=empirical` retain separate provenance. An explicit platform
assertion API requires a source record belonging to that paper and matching the
namespace prefix. The manual CLI only accepts `custom:*`, and cannot retract platform
assertions. No source type has been automatically assigned to existing papers.
## Initialize explicitly
On the published PostgreSQL dataset, using its protected service file:
```sh
PGSERVICEFILE=/etc/paper-atlas/pg_service.conf paper-atlas annotate init \
--pg-service foundry-research --pg-schema research
```
This creates one optional table, its indexes, and an upgrade marker in one transaction,
using the existing dataset writer lock. It does not copy data, change existing papers
or decisions, or create a PostgreSQL instance. Repeating it is safe; an unexplained
pre-existing table/marker mismatch fails closed. Readers and ordinary imports do not
install the extension. Legacy datasets remain readable. A SQLite **sample** uses
`--db sample.sqlite`; the cutover SQLite backup remains protected against writes.
## Add and retract evidence
Use the canonical paper ID returned by `query`. For a reviewed classification:
```sh
paper-atlas annotate assert --pg-service foundry-research --pg-schema research \
--paper "$PAPER_ID" --kind research_type --namespace custom:method \
--value empirical --evidence-key manual:abstract-review \
--actor reviewer --reason 'Classified from the explicitly described study design' \
--request-id "$REQUEST_ID"
```
`$REQUEST_ID` must be a new, stable identifier for this action (for example a UUID).
Retry with the same request ID and identical content after an interrupted response;
it returns the original event ID. Reusing that ID with changed content is rejected.
A topic annotation uses `--kind tag --namespace custom:topic --value robotics` instead.
Optional `--source-record` and `--evidence-url` attach additional evidence; the source
record must belong to this paper. URLs are retained as evidence and are not fetched.
To withdraw one particular assertion, use its returned event ID:
```sh
paper-atlas annotate retract --pg-service foundry-research --pg-schema research \
--event "$EVENT_ID" --actor reviewer --reason 'Classification corrected after review' \
--request-id "$RETRACTION_REQUEST_ID"
```
Retraction adds an event referencing that assertion, preserving its original content.
It does not remove another contributor's assertion of the same label, platform tags,
source metadata or submission/decision history. A retracted assertion can be asserted
again with a new request ID and evidence; it is a new observation. The API does not
update or delete past annotation events. Database administrators still control direct
SQL access. Work reviewed as the same paper may subsequently be merged by the existing
identity workflow: annotation IDs, original paper IDs, evidence and retraction links
are retained while the canonical paper reference follows the surviving entity.
## Query and analyze
```sh
paper-atlas query --pg-service foundry-research --pg-schema research \
--research-type custom:method=empirical
paper-atlas query --pg-service foundry-research --pg-schema research \
--tag custom:topic=robotics
paper-atlas query --pg-service foundry-research --pg-schema research \
--paper "$PAPER_ID" --annotation-history
paper-atlas annotate list --pg-service foundry-research --pg-schema research \
--paper "$PAPER_ID" --history
```
`--tag` includes original platform labels and active annotation tags. It never matches
research types. `--research-type` matches active type assertions only. Paper details
return separate `tags` (existing provenance), `annotation_tags`, and `research_types`;
`--annotation-history` additionally includes retractions. A missing extension yields
empty annotation results rather than changing the database.
For direct SQL analysis, active research-type evidence is:
```sql
SELECT a.* FROM research.annotation_events a
WHERE a.kind = 'research_type' AND a.retracts_id IS NULL
AND NOT EXISTS (SELECT 1 FROM research.annotation_events r WHERE r.retracts_id = a.id);
```
Several independent assertions can support one label; count distinct papers/labels
when appropriate rather than counting evidence rows as papers. Namespaces stay
separate. Standard paper queries decode the PostgreSQL text codec; direct SQL consumers
must honor the dataset's documented codec, including NUL/SOH values. SQLite-to-PostgreSQL
migration recognizes an installed annotation extension and verifies its full row hashes,
constraints and identity sequence along with the original business tables. An old
21-table snapshot stays a 21-table migration; no synthetic annotations are generated.
+67
View File
@@ -0,0 +1,67 @@
# Repairing joint conference membership
EURAI's [official history](https://www.eurai.org/ecai/history) identifies ECAI 2018
and 2022 as joint IJCAI/ECAI conferences. The pinned DBLP volumes
`conf/ijcai/2018` and `conf/ijcai/2022` use `IJCAI` as their booktitle and title.
Existing records can therefore already be present under IJCAI while an ECAI query
shows a missing year. Adding ECAI as a global alias to the ordinary IJCAI venue
would incorrectly include all other IJCAI years.
The existing canonical `ECAI+IJCAI` venue has both member aliases. This maintenance
command changes only the venue reference of the two existing `main` edition rows.
Edition IDs, submission IDs, decisions, source text, authors, identifiers, tags,
source links and coverage records stay intact. It does not merge or create papers
or submissions, resolve per-paper tracks, or turn reviews into negative decisions.
The DBLP collector applies the joint mapping only to these two exact main volume
keys; original booktitle/title remain in source metadata with the membership evidence.
The shared ingestion entry point also resolves these exact pinned source volumes before
creating an edition, so refreshing an older normalized file cannot recreate a separate
IJCAI edition. Other sources, years, workshop volumes and additional tracks retain their existing rules.
## Review a read-only plan
Use a runtime containing this repair command. The configured PostgreSQL writer lock
is used only for the apply operation; do not replace a newer ingestion runtime with
an older main-based feature branch.
```sh
PGSERVICEFILE=/etc/paper-atlas/pg_service.conf paper-atlas maintain joint-editions \
--pg-service foundry-research --pg-schema research --out joint-plan.json
```
The preflight requires the existing joint venue and exactly its two member aliases,
one nonempty unambiguous edition per year, and matching pinned DBLP volume evidence
for every existing submission. A colliding target edition, missing provenance or
unexpected source scope fails closed. The plan binds current database/schema, IDs,
counts and full fingerprints of affected papers, identifiers, source records, tags,
submissions, decisions and their provenance, and coverage rows. Observation time is
recorded separately from conference year. These are scoped checks, not a final global
identity or conference completeness audit.
## Apply the exact reviewed plan
Compute the file's SHA256 after review and pass it explicitly:
```sh
PGSERVICEFILE=/etc/paper-atlas/pg_service.conf paper-atlas maintain joint-editions \
--pg-service foundry-research --pg-schema research --plan joint-plan.json \
--plan-sha256 "$PLAN_SHA256" --out joint-applied.json
```
The plan hash is checked before opening a writer. Current evidence is checked again
under the dataset writer lock; both edition updates and an audit marker commit in one
transaction. Protected row hashes are checked again before commit. Changed evidence
requires a fresh reviewed plan. Existing output reports are preserved; choose a new
path for another run. SQLite samples use `--db sample.sqlite`; the frozen production
backup remains protected.
If a response is lost after commit, rerun with the same plan and a new output path.
When protected evidence is unchanged and both editions already point to the joint
venue, the result is `already_joint` with zero updates. It does not replay imports or
regenerate source records. Keep the plan, apply report and independent postcheck
alongside collection evidence. The audit marker is `joint_ijcai_ecai_2018_2022_v1`.
Publication counts can differ between the DBLP volume and the current publisher
page. This membership repair preserves all existing rows rather than deleting entries
to force an aggregate match; section-level main/invited/other classification and
publication completeness require separate per-paper evidence.
+1 -1
View File
@@ -6,7 +6,6 @@ from paper_atlas import __version__
COMMANDS = {
('annotate',): ('paper_atlas.curation.annotations', {}, '管理独立的研究类型与自定义标签证据'),
('query',): ('paper_atlas.querying', {}, '查询论文、标签和投稿历史'),
('coverage',): ('paper_atlas.audits.coverage', {}, '查看全部候选会议的覆盖情况'),
('download',): ('paper_atlas.io.snapshots', {}, '下载并校验固定版本快照'),
@@ -26,6 +25,7 @@ COMMANDS = {
('catalogue','dblp'): ('paper_atlas.collectors.dblp', {}, '盘点 DBLP 历史会议目录'),
('import','arxiv'): ('paper_atlas.ingestion.arxiv', {}, '导入 arXiv 元数据 ZIP'),
('import','records'): ('paper_atlas.ingestion.records', {}, '导入核验过的会议文件'),
('maintain','joint-editions'): ('paper_atlas.ingestion.joint_editions', {}, '按已核验计划修正既有联合会议届次归属'),
('maintain','merge'): ('paper_atlas.ingestion.maintenance', {}, '协调单写入合并与恢复'),
('maintain','backfill'): ('paper_atlas.ingestion.backfill', {}, '回填既有投稿的来源身份'),
('maintain','arxiv-tags'): ('paper_atlas.ingestion.arxiv_tags', {}, '恢复逐来源记录的 arXiv 类别证据'),
+2
View File
@@ -88,6 +88,8 @@ def normalize_row(row,volumes,targets,include_workshops=False,through_year=2026)
'scope_rule':edition['rule'],'author_names_raw':fields.get('author',[]),'publtype':row.get('publtype'),
'bibliography':{key:value for key,value in fields.items() if key in ('booktitle','year','ee','url','pages','volume','number','publisher','series','isbn')},
'ambiguous_identifiers':ambiguous,'publication_evidence':'inproceedings bibliography; not an original decision note'}
if edition.get('joint_edition_evidence'):
metadata['joint_edition_evidence'] = edition['joint_edition_evidence']
if edition.get('track_scope_note'):
metadata['track_scope_note']=edition['track_scope_note']
flagged=bool(re.search(r'\b(?:withdrawn|retracted)\b',row.get('publtype') or '',re.I))
+7
View File
@@ -96,8 +96,15 @@ def scope(row,volume,through_year=2026):
elif suffix.casefold() in ('poster','posters') and venue not in ('ICLR','ACL','EMNLP','NAACL','EACL','COLING','CoNLL'):track='poster'
if venue=='AAAI+IAAI':track='proceedings_unspecified'
if venue=='AAMAS' and year<2002:return None,'predecessor_not_aamas'
joint_evidence = None
if (family == 'ijcai' and base == 'IJCAI' and track == 'main'
and year in (2018, 2022) and volume['key'] == 'conf/ijcai/' + str(year)):
venue, aliases = 'ECAI+IJCAI', ['ECAI', 'IJCAI']
joint_evidence = 'https://www.eurai.org/ecai/history'
result = {'venue':venue,'venue_aliases':aliases,'year':year,'track':track,'bibliographic_year':int(bibliographic[0]),
'rule':family+':'+base,'proceedings_title':title,'edition_year_basis':year_basis}
if joint_evidence:
result['joint_edition_evidence'] = joint_evidence
if venue=='MM' and track=='main':
# A main MM volume also includes doctoral, tutorials, companion papers
# and other sections. DBLP XML omits the per-paper TOC section.
-1
View File
@@ -1 +0,0 @@
"""Evidence-bearing human and explicit source annotations, independent of outcomes."""
-218
View File
@@ -1,218 +0,0 @@
"""Append-only tag/type assertions and explicit retractions; no inferred labels."""
import argparse
from contextlib import closing
from importlib.resources import files
import json
import re
from urllib.parse import urlparse
from paper_atlas.db.backend import database_arguments, inserted_id, open_read_connection, open_store
from paper_atlas.db.submissions import timestamp
TABLE = 'annotation_events'
MIGRATION = 'annotation_events_v1'
COLUMNS = ('id', 'request_id', 'paper_id', 'original_paper_id', 'kind', 'namespace',
'value', 'source_record_id', 'evidence_key', 'evidence_url', 'reason',
'actor', 'observed_at', 'retracts_id')
ACTIVE = '''a.retracts_id IS NULL AND NOT EXISTS
(SELECT 1 FROM annotation_events removed WHERE removed.retracts_id=a.id)'''
def available(conn):
if getattr(conn, 'postgres', False):
if hasattr(conn, 'has_table'):
return conn.has_table(TABLE)
return bool(conn.execute("SELECT 1 FROM information_schema.tables WHERE table_schema=current_schema() AND table_name=?", (TABLE,)).fetchone())
return bool(conn.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (TABLE,)).fetchone())
def _begin_write(conn):
if not getattr(conn, 'postgres', False) and not conn.in_transaction:
conn.execute('BEGIN IMMEDIATE')
def initialize(conn):
"""Explicit transactional schema extension; does not rescan or rewrite papers."""
_begin_write(conn)
exists = available(conn)
marker = conn.execute('SELECT 1 FROM schema_migrations WHERE name=?', (MIGRATION,)).fetchone()
if exists != bool(marker):
raise RuntimeError('Annotation table/upgrade marker mismatch; inspect before changing schema')
if exists:
result = conn.execute('SELECT * FROM annotation_events LIMIT 0')
description = result.cursor.description if getattr(conn, 'postgres', False) else result.description
if tuple(d[0] for d in description) != COLUMNS:
raise RuntimeError('Unexpected annotation schema')
return False
ddl = files('paper_atlas.db').joinpath('annotations.sql').read_text()
if getattr(conn, 'postgres', False):
ddl = re.sub(r'\bINTEGER\b', 'BIGINT', ddl)
ddl = re.sub(r'\bTEXT\b', 'TEXT COLLATE "C"', ddl)
ddl = ddl.replace('id BIGINT PRIMARY KEY', 'id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY', 1)
for statement in ddl.split(';'):
if statement.strip():
conn.execute(statement)
conn.execute('INSERT INTO schema_migrations(name) VALUES(?)', (MIGRATION,))
return True
def _text(value, name, limit=512):
if not isinstance(value, str) or not value.strip() or len(value) > limit:
raise ValueError(name + ' must be nonempty text of at most ' + str(limit) + ' characters')
return value
def _id(value, name):
if type(value) is not int or value <= 0:
raise ValueError(name + ' must be a positive integer')
return value
def _url(value):
if value is not None:
_text(value, 'evidence_url', 4096)
parsed = urlparse(value)
if parsed.scheme not in ('https', 'http') or not parsed.hostname or parsed.username or parsed.password:
raise ValueError('evidence_url must be an HTTP(S) URL without credentials')
return value
def _require(conn):
if not available(conn) or not conn.execute('SELECT 1 FROM schema_migrations WHERE name=?', (MIGRATION,)).fetchone():
raise RuntimeError('Run paper-atlas annotate init explicitly before adding annotations')
def _row(row):
return dict(zip(COLUMNS, (row[i] for i in range(len(COLUMNS)))))
def _replay(conn, request_id, payload):
old = conn.execute('SELECT * FROM annotation_events WHERE request_id=?', (request_id,)).fetchone()
if old is None:
return None
old = _row(old)
if any(old[k] != value for k, value in payload.items()):
raise ValueError('request_id already identifies different annotation content')
return old['id']
def assert_annotation(conn, paper, kind, namespace, value, *, request_id, evidence_key,
reason, actor, evidence_url=None, source_record_id=None):
_begin_write(conn)
_require(conn)
paper = _id(paper, 'paper')
if kind not in ('tag', 'research_type'):
raise ValueError('kind must be tag or research_type')
_text(namespace, 'namespace', 128)
if not re.fullmatch(r'[a-z][a-z0-9_]*:[a-z][a-z0-9_.-]*', namespace):
raise ValueError('namespace must have an explicit source:scope')
payload = dict(original_paper_id=paper, kind=kind, namespace=namespace,
value=_text(value, 'value'), request_id=_text(request_id, 'request_id', 128),
source_record_id=source_record_id, evidence_key=_text(evidence_key, 'evidence_key', 256),
reason=_text(reason, 'reason', 8192), actor=_text(actor, 'actor', 256),
evidence_url=_url(evidence_url), retracts_id=None)
if source_record_id is not None:
_id(source_record_id, 'source_record_id')
replay = _replay(conn, request_id, payload)
if replay is not None:
return replay
if not conn.execute('SELECT 1 FROM papers WHERE id=?', (paper,)).fetchone():
raise ValueError('Existing canonical paper required; no automatic identity merge')
prefix = namespace.split(':', 1)[0]
if prefix != 'custom' and source_record_id is None:
raise ValueError('Platform annotations require their own source record')
if source_record_id is not None:
source = conn.execute('''SELECT r.paper_id,s.name FROM source_records r
JOIN sources s ON s.id=r.source_id WHERE r.id=?''', (source_record_id,)).fetchone()
if not source or source[0] != paper or (prefix != 'custom' and prefix != source[1]):
raise ValueError('Annotation source record/namespace must belong to this paper and platform')
payload.update(paper_id=paper, observed_at=timestamp())
fields = tuple(payload)
return inserted_id(conn, 'INSERT INTO annotation_events(' + ','.join(fields) + ') VALUES(' +
','.join('?' for _ in fields) + ')', tuple(payload[k] for k in fields))
def retract_annotation(conn, event, *, request_id, reason, actor, evidence_url=None, custom_only=False):
_begin_write(conn)
_require(conn)
event = _id(event, 'event')
payload = dict(request_id=_text(request_id, 'request_id', 128), retracts_id=event,
reason=_text(reason, 'reason', 8192), actor=_text(actor, 'actor', 256), evidence_url=_url(evidence_url))
target = conn.execute('SELECT * FROM annotation_events WHERE id=?', (event,)).fetchone()
if target is None or target[-1] is not None:
raise ValueError('Retraction must reference an existing assertion')
target = _row(target)
if custom_only and not target['namespace'].startswith('custom:'):
raise ValueError('Manual CLI cannot retract platform evidence')
replay = _replay(conn, request_id, payload)
if replay is not None:
return replay
if conn.execute('SELECT 1 FROM annotation_events WHERE retracts_id=?', (event,)).fetchone():
raise ValueError('Assertion already retracted; its history is retained')
payload.update({k: target[k] for k in ('paper_id', 'original_paper_id', 'kind', 'namespace', 'value', 'source_record_id', 'evidence_key')})
payload['observed_at'] = timestamp()
fields = tuple(payload)
return inserted_id(conn, 'INSERT INTO annotation_events(' + ','.join(fields) + ') VALUES(' +
','.join('?' for _ in fields) + ')', tuple(payload[k] for k in fields))
def observations(conn, paper, *, kind=None, history=False):
if not available(conn):
return []
where = ['a.paper_id=?']; params = [_id(paper, 'paper')]
if not history:
where.append(ACTIVE)
if kind is not None:
if kind not in ('tag', 'research_type'):
raise ValueError('Invalid annotation kind')
where.append('a.kind=?'); params.append(kind)
return [_row(r) for r in conn.execute('SELECT a.* FROM annotation_events a WHERE ' +
' AND '.join(where) + ' ORDER BY a.id', params)]
def main(argv=None, *, prog=None):
parser = argparse.ArgumentParser(prog=prog, description='显式初始化、追加或撤回带证据的自定义标签和研究类型')
actions = parser.add_subparsers(dest='action', required=True)
for action in ('init', 'assert', 'retract', 'list'):
p = actions.add_parser(action)
database_arguments(p)
if action in ('assert', 'retract'):
for field in ('request-id', 'reason', 'actor'):
p.add_argument('--' + field, required=True)
p.add_argument('--evidence-url')
if action in ('assert', 'list'):
p.add_argument('--paper', type=int, required=True)
p.add_argument('--kind', choices=('tag', 'research_type'), required=action == 'assert')
if action == 'assert':
for field in ('namespace', 'value', 'evidence-key'):
p.add_argument('--' + field, required=True)
p.add_argument('--source-record', type=int)
if action == 'retract':
p.add_argument('--event', type=int, required=True)
if action == 'list':
p.add_argument('--history', action='store_true')
args = parser.parse_args(argv)
if args.action == 'list':
with closing(open_read_connection(args.db, pg_service=args.pg_service, pg_schema=args.pg_schema)) as conn:
result = observations(conn, args.paper, kind=args.kind, history=args.history)
else:
if args.action == 'assert' and not args.namespace.startswith('custom:'):
parser.error('Manual CLI requires custom:* namespace; platform assertions use validated source APIs')
with closing(open_store(args.db, pg_service=args.pg_service, pg_schema=args.pg_schema).conn) as conn:
try:
if not getattr(conn, 'postgres', False):
conn.commit()
conn.execute('BEGIN IMMEDIATE')
if args.action == 'init':
result = {'initialized': initialize(conn)}
else:
kwargs = dict(request_id=args.request_id, reason=args.reason, actor=args.actor, evidence_url=args.evidence_url)
event = (assert_annotation(conn, args.paper, args.kind, args.namespace, args.value,
evidence_key=args.evidence_key, source_record_id=args.source_record, **kwargs)
if args.action == 'assert' else retract_annotation(conn, args.event, custom_only=True, **kwargs))
result = {'event_id': event}
conn.commit()
except BaseException:
conn.rollback()
raise
print(json.dumps(result, ensure_ascii=False, indent=2))
-15
View File
@@ -1,15 +0,0 @@
CREATE TABLE annotation_events (
id INTEGER PRIMARY KEY,
request_id TEXT NOT NULL UNIQUE,
paper_id INTEGER NOT NULL REFERENCES papers(id),
original_paper_id INTEGER NOT NULL,
kind TEXT NOT NULL CHECK(kind IN ('tag','research_type')),
namespace TEXT NOT NULL, value TEXT NOT NULL,
source_record_id INTEGER REFERENCES source_records(id),
evidence_key TEXT NOT NULL, evidence_url TEXT,
reason TEXT NOT NULL, actor TEXT NOT NULL, observed_at TEXT NOT NULL,
retracts_id INTEGER UNIQUE REFERENCES annotation_events(id),
CHECK(retracts_id IS NULL OR retracts_id<>id)
);
CREATE INDEX annotation_events_paper ON annotation_events(paper_id,kind,id);
CREATE INDEX annotation_events_label ON annotation_events(kind,namespace,value,paper_id);
+14 -3
View File
@@ -55,9 +55,6 @@ class Store:
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))
from paper_atlas.curation.annotations import available
if available(c):
c.execute('UPDATE annotation_events 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))
@@ -147,6 +144,20 @@ class Store:
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
# An already collected file may still call these exact pinned volumes
# IJCAI. Resolve membership before finding/creating an edition so a
# later refresh cannot split the repaired joint event again.
if venue == 'IJCAI' and year in (2018,2022) and track == 'main':
from paper_atlas.collectors.dblp import XML_SHA
evidence=c.execute('SELECT src.name,r.metadata_json FROM source_records r JOIN sources src ON src.id=r.source_id WHERE r.id=?',(record,)).fetchone()
if evidence and evidence[0] == 'dblp':
metadata=json.loads(evidence[1])
if (metadata.get('snapshot_sha256') == XML_SHA
and metadata.get('proceedings_key') == 'conf/ijcai/'+str(year)
and metadata.get('bibliographic_year') == year
and (metadata.get('bibliography') or {}).get('booktitle') in (['IJCAI'],['IJCAI/ECAI'])):
venue='ECAI+IJCAI'
aliases=sorted(set(aliases or []) | {'ECAI','IJCAI'})
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'))
+204
View File
@@ -0,0 +1,204 @@
"""Evidence-bound repair of two existing IJCAI/ECAI edition memberships."""
import argparse
from contextlib import closing
import hashlib
import json
from pathlib import Path
from paper_atlas.db.backend import database_arguments, open_read_connection, open_store
from paper_atlas.db.submissions import timestamp
from paper_atlas.collectors.dblp import XML_SHA
YEARS = (2018, 2022)
TARGET = 'ECAI+IJCAI'
MEMBERS = {'ECAI', 'IJCAI'}
EVIDENCE = 'https://www.eurai.org/ecai/history'
MARKER = 'joint_ijcai_ecai_2018_2022_v1'
def _digest(c, sql, params=()):
sha = hashlib.sha256(); count = 0
for row in c.execute(sql, params):
sha.update((json.dumps([row[i] for i in range(len(row))], ensure_ascii=False,
separators=(',', ':')) + '\n').encode())
count += 1
if count > 100000:
raise ValueError('Joint repair evidence scope exceeded bounded row limit')
return {'rows': count, 'sha256': sha.hexdigest()}
def fingerprints(c, ids):
marks = ','.join('?' for _ in ids)
submissions = 'SELECT id FROM submissions WHERE edition_id IN (' + marks + ')'
papers = 'SELECT paper_id FROM submissions WHERE edition_id IN (' + marks + ')'
decisions = 'SELECT id FROM decisions WHERE submission_id IN (' + submissions + ')'
checks = {
'papers': ('id IN (' + papers + ')', 'id'),
'identifiers': ('paper_id IN (' + papers + ')', 'scheme,value'),
'source_records': ('paper_id IN (' + papers + ')', 'id'),
'paper_tags': ('paper_id IN (' + papers + ')', 'paper_id,tag_id,evidence_key'),
'submissions': ('edition_id IN (' + marks + ')', 'id'),
'decisions': ('submission_id IN (' + submissions + ')', 'id'),
'submission_sources': ('submission_id IN (' + submissions + ')', 'submission_id,source_record_id,source_submission_id'),
'submission_identifiers': ('edition_id IN (' + marks + ')', 'edition_id,scheme,value'),
'decision_sources': ('decision_id IN (' + decisions + ')', 'decision_id,source_record_id,source_decision_id'),
'edition_sources': ('edition_id IN (' + marks + ')', 'edition_id,source_id,list_url'),
}
return {table: _digest(c, 'SELECT * FROM ' + table + ' WHERE ' + where + ' ORDER BY ' + order, ids)
for table, (where, order) in checks.items()}
def inspect(c):
target = c.execute('SELECT id FROM venues WHERE acronym=?', (TARGET,)).fetchall()
if len(target) != 1:
raise ValueError('Exactly one existing canonical joint venue required')
target = target[0][0]
aliases = {r[0] for r in c.execute('SELECT alias FROM venue_aliases WHERE venue_id=?', (target,))}
if aliases != MEMBERS:
raise ValueError('Joint target must already have exactly ECAI and IJCAI aliases')
editions = []
for year in YEARS:
rows = c.execute('''SELECT e.id,e.venue_id,v.acronym FROM editions e JOIN venues v ON v.id=e.venue_id
WHERE v.acronym IN ('IJCAI','ECAI+IJCAI') AND e.year=? AND e.track='main' ''', (year,)).fetchall()
if len(rows) != 1:
raise ValueError('Expected one existing edition per year; missing or colliding editions require separate review')
edition, venue, acronym = (rows[0][i] for i in range(3))
count = c.execute('SELECT count(*) FROM submissions WHERE edition_id=?', (edition,)).fetchone()[0]
if not 0 < count <= 5000:
raise ValueError('Existing edition must have a nonempty bounded submission scope')
# Every existing submission must be supported by its own pinned DBLP
# volume record. No title/author matching, new attempts or text writes.
evidence = c.execute('''SELECT s.id,r.external_id,r.metadata_json FROM submissions s
JOIN submission_sources ss ON ss.submission_id=s.id
JOIN source_records r ON r.id=ss.source_record_id JOIN sources src ON src.id=r.source_id
WHERE s.edition_id=? AND src.name='dblp' ORDER BY s.id,r.id''', (edition,)).fetchall()
seen = set()
for row in evidence:
sid, external, metadata = (row[i] for i in range(3))
meta = json.loads(metadata)
bibliography = meta.get('bibliography') or {}
if (meta.get('snapshot_sha256') != XML_SHA
or meta.get('proceedings_key') != 'conf/ijcai/' + str(year)
or meta.get('bibliographic_year') != year
or bibliography.get('booktitle') not in (['IJCAI'], ['IJCAI/ECAI'])
or not external.startswith('conf/ijcai/')):
raise ValueError('Source evidence outside the exact verified IJCAI/ECAI volume')
seen.add(sid)
if len(seen) != count:
raise ValueError('Every submission must have matching original volume evidence')
editions.append({'id': edition, 'year': year, 'track': 'main', 'from_venue_id': venue,
'from_venue': acronym, 'target_venue_id': target, 'submissions': count,
'evidence_url': EVIDENCE, 'proceedings_key': 'conf/ijcai/' + str(year)})
ids = tuple(e['id'] for e in editions)
return {'format_version': 1, 'kind': MARKER, 'checked_at': timestamp(), 'editions': editions,
'content_fingerprints': fingerprints(c, ids), 'target_aliases': sorted(aliases),
'status': 'already_joint' if all(e['from_venue_id'] == target for e in editions) else 'ready',
'scope': 'Only these two existing edition venue references; no source text, paper, submission, decision, track, identifier or tag change.'}
def validate_plan(expected):
if not isinstance(expected, dict):
raise ValueError('Repair plan must be a JSON object')
if set(expected) != {'format_version','kind','checked_at','editions','content_fingerprints','target_aliases','status','scope','database'}:
raise ValueError('Unexpected repair plan fields')
if (type(expected['format_version']) is not int or expected['format_version'] != 1
or expected['kind'] != MARKER or expected['status'] not in ('ready','already_joint')):
raise ValueError('Unsupported repair plan')
if (not isinstance(expected['database'],dict) or not isinstance(expected['content_fingerprints'],dict)
or not isinstance(expected['editions'],list) or len(expected['editions']) != 2
or expected['target_aliases'] != sorted(MEMBERS)):
raise ValueError('Invalid repair plan structure')
for edition, year in zip(expected['editions'], YEARS):
if not isinstance(edition,dict) or edition.get('year') != year:
raise ValueError('Repair plan must name the two exact ordered years')
for key in ('id','year','from_venue_id','target_venue_id','submissions'):
if type(edition.get(key)) is not int or edition[key] <= 0:
raise ValueError('Repair plan IDs/counts must be positive integers')
def apply_plan(c, expected):
validate_plan(expected)
actual = inspect(c)
if actual['content_fingerprints'] != expected['content_fingerprints'] or actual['target_aliases'] != expected['target_aliases']:
raise ValueError('Stale repair plan: protected content or aliases changed')
if len(expected['editions']) != 2:
raise ValueError('Exactly two editions required')
for old, now in zip(expected['editions'], actual['editions']):
if set(old) != set(now):
raise ValueError('Unexpected edition fields')
for key in ('id','year','track','target_venue_id','submissions','evidence_url','proceedings_key'):
if old[key] != now[key]:
raise ValueError('Stale repair plan: edition differs')
if now['from_venue_id'] != old['from_venue_id'] and now['from_venue_id'] != now['target_venue_id']:
raise ValueError('Edition membership differs from plan')
if actual['status'] == 'already_joint':
return {**actual, 'status':'already_joint', 'updated_editions':0}
for edition in actual['editions']:
if edition['from_venue_id'] == edition['target_venue_id']:
continue
changed = c.execute('''UPDATE editions SET venue_id=? WHERE id=? AND venue_id=? AND year=? AND track=?''',
(edition['target_venue_id'],edition['id'],edition['from_venue_id'],edition['year'],edition['track']))
if changed.rowcount != 1:
raise ValueError('Concurrent edition change; entire repair must roll back')
after = inspect(c)
if after['status'] != 'already_joint' or after['content_fingerprints'] != actual['content_fingerprints']:
raise ValueError('Postcheck changed protected content; entire repair must roll back')
c.execute('INSERT INTO schema_migrations(name) VALUES(?) ON CONFLICT DO NOTHING', (MARKER,))
return {**after, 'status':'applied', 'updated_editions':sum(e['from_venue_id'] != e['target_venue_id'] for e in actual['editions'])}
def database_identity(c, db):
if getattr(c, 'postgres', False):
row = c.execute('SELECT current_database(),current_schema()').fetchone()
return {'engine':'postgresql','database':row[0],'schema':row[1]}
return {'engine':'sqlite','path':str(Path(db).resolve())}
def run(db, out, *, pg_service=None, pg_schema=None, plan=None, plan_sha256=None):
out = Path(out).resolve()
if not out.parent.is_dir():
raise ValueError('Report parent directory must already exist')
if out.exists():
raise ValueError('Preserve existing evidence report; choose a new output path')
expected = None
if plan:
raw = Path(plan).read_bytes()
if hashlib.sha256(raw).hexdigest() != plan_sha256:
raise ValueError('Repair plan hash differs before opening database')
expected = json.loads(raw)
validate_plan(expected)
elif plan_sha256:
raise ValueError('Plan hash requires an existing plan')
if db and out == Path(db).resolve():
raise ValueError('Report must not overwrite database')
writer = open_store(db, pg_service=pg_service, pg_schema=pg_schema) if expected is not None else None
c = writer.conn if writer else open_read_connection(db, pg_service=pg_service, pg_schema=pg_schema)
with closing(c):
try:
if expected is not None and not getattr(c, 'postgres', False):
c.commit(); c.execute('BEGIN IMMEDIATE')
identity = database_identity(c, db)
if expected is not None and expected['database'] != identity:
raise ValueError('Plan belongs to another database/schema')
report = apply_plan(c, expected) if expected is not None else inspect(c)
report['database'] = identity
if expected is not None:
report['plan_sha256'] = plan_sha256
if expected is not None:
c.commit()
except BaseException:
if expected is not None: c.rollback()
raise
out.write_text(json.dumps(report,ensure_ascii=False,indent=2)+'\n')
print(json.dumps({'status':report['status'],'editions':len(report['editions']),
'updated_editions':report.get('updated_editions',0),'report':str(out)}))
return report
def main(argv=None, *, prog=None):
parser = argparse.ArgumentParser(prog=prog); database_arguments(parser)
parser.add_argument('--out', required=True)
parser.add_argument('--plan', help='Apply an already reviewed plan; omitted means read-only preflight')
parser.add_argument('--plan-sha256', help='Required exact plan SHA256 when applying')
args = parser.parse_args(argv)
run(args.db,args.out,pg_service=args.pg_service,pg_schema=args.pg_schema,plan=args.plan,plan_sha256=args.plan_sha256)
+2 -2
View File
@@ -143,7 +143,7 @@ def migrate(db, service, out, *, schema='research', storage_path=None, min_free_
free_space(storage_path, min_free_gib)
control = schema + '_migration'
report = {'status': 'preflight', 'source': str(db), 'service': service, 'schema': schema,
'started_at': timestamp(), 'tables': {}, 'scope': 'All baseline business tables and recognized installed extensions; FTS5 rebuilt as native PostgreSQL GIN.'}
'started_at': timestamp(), 'tables': {}, 'scope': 'All 21 business tables; FTS5 internals rebuilt as native PostgreSQL GIN.'}
with Path(str(db) + '.maintenance.lock').open('a') as lock:
fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
if Path(str(db) + '-wal').exists() and Path(str(db) + '-wal').stat().st_size:
@@ -240,7 +240,7 @@ def migrate(db, service, out, *, schema='research', storage_path=None, min_free_
JOIN pg_am am ON am.oid=ix.relam WHERE ns.nspname=%s AND ix.relname='papers_search_gin' ''', (stage,)).fetchone()
if search != ('gin', True, True, 'papers'):
raise ValueError('Native full-text index is not valid and ready')
for table in plan:
for table in TABLES:
pg.execute('ANALYZE ' + qualify(stage, table))
invalid = pg.execute('''SELECT count(*) FROM pg_constraint co JOIN pg_class cl ON cl.oid=co.conrelid
JOIN pg_namespace ns ON ns.oid=cl.relnamespace WHERE ns.nspname=%s AND NOT co.convalidated''', (stage,)).fetchone()[0]
+2 -4
View File
@@ -8,9 +8,7 @@ TABLES = ('sources', 'papers', 'identifiers', 'source_records', 'venues', 'editi
'submission_sources', 'submission_identifiers', 'decision_sources',
'submission_merges', 'decision_merges', 'schema_migrations', 'tags',
'paper_tags', 'match_candidates', 'paper_merges', 'ingest_runs')
OPTIONAL_TABLES = ('annotation_events',)
CHECKS = {
'annotation_events': ["kind IN ('tag','research_type')", 'retracts_id IS NULL OR retracts_id<>id'],
'venues': ["venue_type IN ('conference','journal','review_platform')"],
'edition_sources': ["status IN ('pending','running','complete','partial','blocked','unavailable','not_yet_published')"],
'decisions': ["status IN ('accepted','rejected','desk_rejected','withdrawn','under_review','unknown')"],
@@ -52,10 +50,10 @@ def quote(name):
def inspect(connection):
business = {name for schema, name, kind, *_ in connection.execute('PRAGMA table_list')
if schema == 'main' and kind == 'table' and not name.startswith('sqlite_')}
if not set(TABLES) <= business or business - set(TABLES) - set(OPTIONAL_TABLES):
if business != set(TABLES):
raise ValueError('Unexpected business tables: ' + repr(sorted(business ^ set(TABLES))))
result = {}
for table in (*TABLES, *(name for name in OPTIONAL_TABLES if name in business)):
for table in TABLES:
columns = [{'name': r[1], 'type': r[2].upper(), 'notnull': bool(r[3]),
'default': r[4], 'pk': r[5]}
for r in connection.execute('PRAGMA table_info(' + quote(table) + ')')]
+1 -25
View File
@@ -29,8 +29,6 @@ def main(argv=None, *, prog=None):
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)
@@ -49,8 +47,6 @@ def main(argv=None, *, prog=None):
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.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')]
@@ -73,24 +69,8 @@ def main(argv=None, *, prog=None):
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')
where.append('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 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:
@@ -118,10 +98,6 @@ def main(argv=None, *, prog=None):
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 = []
+1 -1
View File
@@ -21,5 +21,5 @@ include = ["paper_atlas*"]
exclude = ["tests*"]
[tool.setuptools.package-data]
"paper_atlas.db" = ["schema.sql", "annotations.sql"]
"paper_atlas.db" = ["schema.sql"]
"paper_atlas.resources" = ["venues.json"]
-227
View File
@@ -1,227 +0,0 @@
from contextlib import closing, redirect_stdout
import io
import json
import os
from pathlib import Path
import tempfile
import unittest
import uuid
from paper_atlas import querying
from paper_atlas.curation.annotations import (assert_annotation, available, initialize, main,
observations, retract_annotation)
from paper_atlas.db.backend import open_store
from paper_atlas.db.store import Store
from paper_atlas.postgres import schema
from paper_atlas.postgres.client import connect
from paper_atlas.postgres.migration import migrate
class AnnotationTests(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory(); self.addCleanup(self.tmp.cleanup)
self.db = Path(self.tmp.name) / 'sample.sqlite'
self.store = Store(self.db); self.addCleanup(self.store.conn.close)
self.c = self.store.conn
self.paper, self.record = self.store.record('fixture', 'one', 'https://example.org/one', 'First research', None, [])
self.other, self.other_record = self.store.record('fixture', 'two', 'https://example.org/two', 'Other research', None, [])
self.store.record_tags(self.paper, self.record, 'fixture', [{'namespace':'fixture:topic','value':'original'}])
self.c.commit()
def add(self, **changes):
args = dict(paper=self.paper, kind='tag', namespace='custom:topic', value='reading',
request_id='request-one', evidence_key='manual:reading', reason='Reviewed metadata', actor='reviewer')
args.update(changes)
return assert_annotation(self.c, **args)
def query(self, *args):
self.c.commit()
with redirect_stdout(io.StringIO()) as output:
querying.main(['--db', str(self.db), *map(str, args)])
return json.loads(output.getvalue())
def test_read_does_not_install_extension_and_write_requires_explicit_init(self):
self.assertFalse(available(self.c))
self.assertEqual(observations(self.c, self.paper), [])
with self.assertRaisesRegex(RuntimeError, 'init explicitly'):
self.add()
self.assertEqual(self.query('--research-type', 'custom:method=empirical'), [])
self.assertFalse(available(self.c))
self.assertEqual(len(schema.inspect(self.c)), 21)
def test_schema_upgrade_is_transactional_and_marker_mismatch_fails_closed(self):
self.assertTrue(initialize(self.c)); self.c.rollback()
self.assertFalse(available(self.c))
self.c.execute("INSERT INTO schema_migrations(name) VALUES('annotation_events_v1')")
with self.assertRaisesRegex(RuntimeError, 'mismatch'):
initialize(self.c)
self.c.rollback()
self.assertTrue(initialize(self.c)); self.assertFalse(initialize(self.c))
self.assertEqual(len(schema.inspect(self.c)), 22)
def test_unmarked_table_cannot_receive_assertions(self):
initialize(self.c)
self.c.execute("DELETE FROM schema_migrations WHERE name='annotation_events_v1'")
with self.assertRaisesRegex(RuntimeError, 'init explicitly'):
self.add()
self.assertEqual(self.c.execute('SELECT count(*) FROM annotation_events').fetchone()[0], 0)
def test_types_tags_and_platform_labels_are_independent_and_queryable(self):
initialize(self.c)
self.add()
self.add(kind='research_type', namespace='custom:topic', request_id='type', value='reading')
self.assertEqual(len(self.query('--tag', 'custom:topic=reading')), 1)
self.assertEqual(len(self.query('--research-type', 'custom:topic=reading')), 1)
self.assertEqual(len(self.query('--tag', 'fixture:topic=original')), 1)
detail = self.query('--paper', self.paper)[0]
self.assertEqual(len(detail['annotation_tags']), 1)
self.assertEqual(len(detail['research_types']), 1)
self.assertNotIn('annotation_history', detail)
self.assertEqual(detail['tags'][0]['namespace'], 'fixture:topic')
def test_repeated_request_is_idempotent_and_changed_content_fails(self):
initialize(self.c); event = self.add()
self.assertEqual(self.add(), event)
for change in ({'value':'different'}, {'kind':'research_type'}, {'reason':'changed'}, {'paper':self.other}):
with self.assertRaisesRegex(ValueError, 'different annotation'):
self.add(**change)
self.assertEqual(self.c.execute('SELECT count(*) FROM annotation_events').fetchone()[0], 1)
def test_retraction_keeps_original_and_other_assertions_of_same_label(self):
initialize(self.c); first = self.add(); second = self.add(request_id='second', evidence_key='independent')
kwargs = dict(request_id='retract', reason='No longer relevant', actor='reviewer')
removed = retract_annotation(self.c, first, **kwargs)
self.assertEqual(retract_annotation(self.c, first, **kwargs), removed)
self.assertEqual([r['id'] for r in observations(self.c, self.paper)], [second])
self.assertEqual(len(self.query('--paper', self.paper, '--annotation-history')[0]['annotation_history']), 3)
with self.assertRaisesRegex(ValueError, 'already retracted'):
retract_annotation(self.c, first, **{**kwargs,'request_id':'duplicate'})
with self.assertRaisesRegex(ValueError, 'existing assertion'):
retract_annotation(self.c, removed, **{**kwargs,'request_id':'nested'})
def test_source_assertions_require_matching_paper_and_platform(self):
initialize(self.c)
for change in ({'namespace':'fixture:type'}, {'namespace':'other:type','source_record_id':self.record},
{'namespace':'fixture:type','source_record_id':self.other_record}):
with self.assertRaises(ValueError):
self.add(**change)
event = self.add(kind='research_type', namespace='fixture:declared_type', source_record_id=self.record)
with self.assertRaisesRegex(ValueError, 'Manual CLI'):
retract_annotation(self.c, event, request_id='cannot', reason='x', actor='y', custom_only=True)
self.assertEqual(observations(self.c, self.paper)[0]['source_record_id'], self.record)
def test_platform_tag_refresh_preserves_both_custom_kinds(self):
initialize(self.c); self.add(); self.add(kind='research_type', request_id='type')
self.store.record_tags(self.paper, self.record, 'fixture', [{'namespace':'fixture:topic','value':'refreshed'}])
self.assertEqual(len(observations(self.c, self.paper)), 2)
self.assertEqual(len(self.query('--tag','fixture:topic=refreshed')), 1)
self.assertEqual(self.query('--tag','fixture:topic=original'), [])
def test_merge_preserves_retractions_evidence_and_original_paper_ids(self):
initialize(self.c); event = self.add(source_record_id=self.record)
removed = retract_annotation(self.c, event, request_id='withdraw', reason='corrected', actor='reviewer')
active = self.add(request_id='active', kind='research_type', value='survey')
self.store.merge(self.other, self.paper, 'Independently reviewed identity')
rows = observations(self.c, self.other, history=True)
self.assertEqual([r['id'] for r in rows], [event, removed, active])
self.assertEqual({r['original_paper_id'] for r in rows}, {self.paper})
self.assertEqual({r['paper_id'] for r in rows}, {self.other})
self.assertEqual([r['id'] for r in observations(self.c, self.other)], [active])
self.assertEqual(self.add(source_record_id=self.record), event) # exact retry survives canonical merge
self.assertEqual(self.c.execute('PRAGMA foreign_key_check').fetchall(), [])
def test_invalid_input_does_not_create_events(self):
initialize(self.c)
for change in ({'paper':True}, {'paper':999999}, {'kind':'main'}, {'namespace':'arxiv'},
{'namespace':'custom:topic;DROP'}, {'value':' '}, {'reason':''},
{'actor':''}, {'request_id':''}, {'evidence_key':''}, {'source_record_id':True},
{'evidence_url':'file:///etc/passwd'}, {'evidence_url':'https://user:secret@example.org/'}):
with self.subTest(change=change), self.assertRaises(ValueError):
self.add(**change)
self.assertEqual(self.c.execute('SELECT count(*) FROM annotation_events').fetchone()[0], 0)
def test_cli_manual_scope_and_complete_assert_retract_lifecycle(self):
with redirect_stdout(io.StringIO()):
main(['init','--db',str(self.db)])
main(['assert','--db',str(self.db),'--paper',str(self.paper),'--kind','research_type',
'--namespace','custom:method','--value','empirical','--evidence-key','manual:review',
'--request-id','cli-one','--reason','Read abstract','--actor','reviewer'])
self.assertEqual(len(self.query('--research-type','custom:method=empirical')), 1)
event = observations(self.c, self.paper)[0]['id']
with redirect_stdout(io.StringIO()):
main(['retract','--db',str(self.db),'--event',str(event),'--request-id','cli-two',
'--reason','Corrected classification','--actor','reviewer'])
self.assertEqual(self.query('--research-type','custom:method=empirical'), [])
with redirect_stdout(io.StringIO()) as out:
main(['list','--db',str(self.db),'--paper',str(self.paper),'--history'])
self.assertEqual(len(json.loads(out.getvalue())), 2)
@unittest.skipUnless(os.environ.get('PAPER_ATLAS_TEST_PG_SERVICE'), 'Requires approved remote PostgreSQL service')
class PostgreSQLAnnotationTests(unittest.TestCase):
def setUp(self):
self.service = os.environ['PAPER_ATLAS_TEST_PG_SERVICE']
self.target = 'pa_anno_' + uuid.uuid4().hex[:12]
self.tmp = tempfile.TemporaryDirectory(); self.addCleanup(self.tmp.cleanup)
self.db = Path(self.tmp.name) / 'source.sqlite'
self.s = Store(self.db)
self.paper, self.record = self.s.record('fixture','one','https://example.org','Paper',None,[])
self.s.conn.commit()
self.addCleanup(self.s.conn.close); self.addCleanup(self.cleanup)
def cleanup(self):
with connect(self.service) as c:
for name, in c.execute('SELECT nspname FROM pg_namespace WHERE starts_with(nspname,%s)', (self.target,)):
c.execute('DROP SCHEMA ' + schema.quote(name) + ' CASCADE')
def migrate(self):
self.s.conn.commit()
self.s.conn.close() # Freeze/checkpoint the fixture, just like a real source migration.
with redirect_stdout(io.StringIO()):
return migrate(self.db, self.service, Path(self.tmp.name)/'report.json', schema=self.target, min_free_gib=0)
def args(self, **changes):
a = dict(paper=self.paper,kind='research_type',namespace='custom:method',value='survey\x00and\x01',
request_id='native',evidence_key='manual',reason='Reviewed\x00evidence',actor='reviewer')
a.update(changes); return a
def test_native_schema_upgrade_roundtrip_retraction_and_queries(self):
self.migrate()
with closing(open_store(pg_service=self.service, pg_schema=self.target).conn) as c:
self.assertTrue(initialize(c)); self.assertFalse(initialize(c))
event = assert_annotation(c, **self.args()); c.commit()
self.assertEqual(assert_annotation(c, **self.args()), event)
self.assertEqual(observations(c,self.paper)[0]['value'], 'survey\x00and\x01')
c.commit()
with redirect_stdout(io.StringIO()) as out:
querying.main(['--pg-service',self.service,'--pg-schema',self.target,'--paper',str(self.paper)])
self.assertEqual(json.loads(out.getvalue())[0]['research_types'][0]['id'], event)
with closing(open_store(pg_service=self.service,pg_schema=self.target).conn) as c:
retract_annotation(c,event,request_id='removed',reason='Changed',actor='reviewer');c.commit()
self.assertEqual(observations(c,self.paper), [])
self.assertEqual(len(observations(c,self.paper,history=True)), 2)
def test_initialized_sqlite_extension_migrates_all_rows_hashes_and_identity(self):
initialize(self.s.conn)
event = assert_annotation(self.s.conn, **self.args())
retract_annotation(self.s.conn,event,request_id='retracted',reason='Corrected',actor='reviewer')
report = self.migrate()
self.assertEqual(len(report['tables']),22)
self.assertEqual(report['tables']['annotation_events']['source_rows'],2)
self.assertEqual(report['tables']['annotation_events']['source_sha256'],report['tables']['annotation_events']['target_sha256'])
with closing(open_store(pg_service=self.service,pg_schema=self.target).conn) as c:
new = assert_annotation(c, **self.args(request_id='after-migration'))
self.assertGreater(new,2); c.commit()
self.assertEqual(len(observations(c,self.paper,history=True)),3)
def test_native_upgrade_rollback_keeps_legacy_schema_usable(self):
self.migrate()
with closing(open_store(pg_service=self.service,pg_schema=self.target).conn) as c:
initialize(c); c.rollback()
self.assertFalse(available(c))
self.assertEqual(c.execute('SELECT count(*) FROM papers').fetchone()[0],1)
self.assertTrue(initialize(c))
with self.assertRaises(ValueError):
assert_annotation(c, **self.args(namespace='wrong:type',source_record_id=self.record))
self.assertEqual(c.execute('SELECT count(*) FROM annotation_events').fetchone()[0],0)
c.rollback()
+187
View File
@@ -0,0 +1,187 @@
from contextlib import closing, redirect_stdout
import hashlib
import io
import json
import os
from pathlib import Path
import tempfile
import unittest
from unittest.mock import patch
import uuid
from paper_atlas.collectors.dblp import XML_SHA
from paper_atlas.collectors.dblp_scope import scope
from paper_atlas.collectors.dblp_records import normalize_row
from paper_atlas.db.store import Store
from paper_atlas.db.backend import open_store
from paper_atlas.ingestion.joint_editions import (TARGET, MARKER, inspect, apply_plan, run)
from paper_atlas.postgres.client import connect
from paper_atlas.postgres.migration import migrate
from paper_atlas.postgres.schema import quote
from paper_atlas import querying
def seed(store):
c = store.conn
c.execute("INSERT INTO venues(acronym,name,field,venue_type) VALUES(?,?,'reasoning','conference')", (TARGET,TARGET))
target = c.execute('SELECT id FROM venues WHERE acronym=?',(TARGET,)).fetchone()[0]
for alias in ['ECAI','IJCAI']:
c.execute('INSERT INTO venue_aliases VALUES(?,?)',(target,alias))
for year in [2018,2019,2022]:
metadata = {'snapshot_sha256':XML_SHA,'proceedings_key':'conf/ijcai/'+str(year),
'bibliographic_year':year,'bibliography':{'booktitle':['IJCAI']}}
p,r=store.record('dblp','conf/ijcai/fixture'+str(year),'https://dblp.org/rec/fixture',
'Original publication '+str(year),'Abstract\x00with\x01data',[],
{'doi':'10.24963/ijcai.'+str(year)+'/1'}, {})
store.record_tags(p,r,'dblp',[{'namespace':'dblp:booktitle','value':'IJCAI'}])
store.decision(p,r,'IJCAI',year,'main','fixture'+str(year),'https://example.org/decision','accepted','listed')
stored=json.loads(c.execute('SELECT metadata_json FROM source_records WHERE id=?',(r,)).fetchone()[0]);stored.update(metadata)
c.execute('UPDATE source_records SET metadata_json=? WHERE id=?',(json.dumps(stored),r))
c.commit()
def plan(c):
return {**inspect(c),'database':{'engine':'sqlite','path':'fixture'}}
class JointEditionTests(unittest.TestCase):
def setUp(self):
self.tmp=tempfile.TemporaryDirectory(); self.addCleanup(self.tmp.cleanup)
self.db=Path(self.tmp.name)/'sample.sqlite'
self.store=Store(self.db); self.addCleanup(self.store.conn.close)
seed(self.store); self.c=self.store.conn
def test_apply_preserves_every_protected_row_and_edition_id(self):
before=plan(self.c); ids=[r['id']for r in before['editions']]
result=apply_plan(self.c,before);self.c.commit()
self.assertEqual(result['updated_editions'],2)
self.assertEqual(result['content_fingerprints'],before['content_fingerprints'])
self.assertEqual([r['id']for r in result['editions']],ids)
untouched=self.c.execute('SELECT v.acronym FROM editions e JOIN venues v ON v.id=e.venue_id WHERE e.year=2019').fetchone()[0]
self.assertEqual(untouched,'IJCAI')
self.assertEqual(self.c.execute('PRAGMA foreign_key_check').fetchall(),[])
self.assertEqual(self.c.execute('SELECT count(*) FROM schema_migrations WHERE name=?',(MARKER,)).fetchone()[0],1)
with redirect_stdout(io.StringIO())as out:
querying.main(['--db',str(self.db),'--venue','ECAI'])
self.assertEqual(len(json.loads(out.getvalue())),2)
self.assertEqual(apply_plan(self.c,before)['updated_editions'],0)
def test_refresh_of_old_normalized_source_does_not_recreate_ordinary_edition(self):
before=plan(self.c);apply_plan(self.c,before)
for year in [2018,2022]:
row=self.c.execute('''SELECT r.paper_id,r.id,d.observed_at FROM source_records r
JOIN submissions s ON s.source_record_id=r.id JOIN decisions d ON d.submission_id=s.id
WHERE r.external_id=?''',('conf/ijcai/fixture'+str(year),)).fetchone()
self.store.decision(row[0],row[1],'IJCAI',year,'main','fixture'+str(year),
'https://example.org/decision','accepted','listed',observed_at=row[2])
self.assertEqual(inspect(self.c)['content_fingerprints'],before['content_fingerprints'])
self.assertEqual(self.c.execute('SELECT count(*) FROM editions').fetchone()[0],3)
def test_any_changed_source_text_rejects_stale_plan_before_membership_updates(self):
before=plan(self.c)
self.c.execute("UPDATE source_records SET title='New evidence' WHERE external_id='conf/ijcai/fixture2018'")
with self.assertRaisesRegex(ValueError,'Stale repair plan'):
apply_plan(self.c,before)
self.assertEqual(inspect(self.c)['status'],'ready')
def test_conflicting_target_edition_is_never_merged_or_deleted(self):
target=self.c.execute('SELECT id FROM venues WHERE acronym=?',(TARGET,)).fetchone()[0]
self.c.execute("INSERT INTO editions(venue_id,year,track) VALUES(?,2018,'main')",(target,))
with self.assertRaisesRegex(ValueError,'colliding'):
inspect(self.c)
self.assertEqual(self.c.execute('SELECT count(*) FROM editions WHERE year=2018').fetchone()[0],2)
def test_wrong_source_volume_and_snapshot_fail_closed(self):
original=self.c.execute("SELECT metadata_json FROM source_records WHERE external_id='conf/ijcai/fixture2018'").fetchone()[0]
for key,value in [('proceedings_key','conf/ijcai/2018w'),('snapshot_sha256','wrong'),('bibliographic_year',2017),('bibliography',{'booktitle':['Other']})]:
m=json.loads(original);m[key]=value
self.c.execute("UPDATE source_records SET metadata_json=? WHERE external_id='conf/ijcai/fixture2018'",(json.dumps(m),))
with self.subTest(key=key),self.assertRaisesRegex(ValueError,'Source evidence'):
inspect(self.c)
def test_missing_source_support_and_wrong_aliases_are_not_guessed(self):
self.c.execute("DELETE FROM venue_aliases WHERE alias='ECAI'")
with self.assertRaisesRegex(ValueError,'aliases'):
inspect(self.c)
self.c.rollback()
self.c.execute('DELETE FROM submission_sources WHERE source_record_id=(SELECT id FROM source_records WHERE external_id=?)',('conf/ijcai/fixture2022',))
with self.assertRaisesRegex(ValueError,'Every submission'):
inspect(self.c)
def test_cli_plan_readonly_hash_binding_identity_and_repeat_apply(self):
out=Path(self.tmp.name)/'plan.json'
with redirect_stdout(io.StringIO()):run(self.db,out)
self.assertEqual(inspect(self.c)['status'],'ready')
raw=out.read_bytes();sha=hashlib.sha256(raw).hexdigest()
with patch('paper_atlas.ingestion.joint_editions.open_store')as writer:
with self.assertRaisesRegex(ValueError,'hash differs'):
run(self.db,Path(self.tmp.name)/'bad.json',plan=out,plan_sha256='0'*64)
writer.assert_not_called()
with redirect_stdout(io.StringIO()):
result=run(self.db,Path(self.tmp.name)/'applied.json',plan=out,plan_sha256=sha)
again=run(self.db,Path(self.tmp.name)/'retry.json',plan=out,plan_sha256=sha)
self.assertEqual(result['updated_editions'],2);self.assertEqual(again['updated_editions'],0)
self.assertEqual(result['plan_sha256'],sha);self.assertEqual(again['plan_sha256'],sha)
altered=json.loads(raw);altered['database']['path']='another.sqlite'
bad=Path(self.tmp.name)/'wrong-db.json';bad.write_text(json.dumps(altered))
with self.assertRaisesRegex(ValueError,'another database'):
run(self.db,Path(self.tmp.name)/'wrong-result.json',plan=bad,plan_sha256=hashlib.sha256(bad.read_bytes()).hexdigest())
def test_invalid_empty_plan_fails_before_opening_writer(self):
f=Path(self.tmp.name)/'empty.json';f.write_text('{}')
with patch('paper_atlas.ingestion.joint_editions.open_store')as writer:
with self.assertRaises(ValueError):
run(self.db,Path(self.tmp.name)/'empty-result.json',plan=f,plan_sha256=hashlib.sha256(f.read_bytes()).hexdigest())
writer.assert_not_called()
def test_collector_only_changes_the_two_exact_main_volume_keys(self):
for year in [2018,2019,2022]:
volume={'key':'conf/ijcai/'+str(year),'fields':{'title':['IJCAI '+str(year)],'year':[str(year)]}}
row={'record_type':'inproceedings','key':'conf/ijcai/One'+str(year),'fields':{'booktitle':['IJCAI'],
'year':[str(year)],'crossref':[volume['key']],'title':['Original research title'],
'author':['Jane Doe'],'ee':['https://doi.org/10.24963/ijcai.'+str(year)+'/1']}}
result,error=normalize_row(row,{volume['key']:volume},{'IJCAI','ECAI'})
self.assertIsNone(error)
self.assertEqual(result['venue'],TARGET if year in(2018,2022)else'IJCAI')
self.assertEqual(result['metadata']['bibliography']['booktitle'],['IJCAI'])
self.assertEqual(result['title'],'Original research title')
if year in(2018,2022):
self.assertIn('joint_edition_evidence',result['metadata'])
row['fields']['booktitle']=['IJCAI (Workshop)']
scoped,_=scope(row,volume)
self.assertEqual(scoped['venue'],'IJCAI')
self.assertEqual(scoped['track'],'workshop')
@unittest.skipUnless(os.environ.get('PAPER_ATLAS_TEST_PG_SERVICE'),'Requires approved remote PostgreSQL service')
class PostgreSQLJointTests(unittest.TestCase):
def setUp(self):
self.service=os.environ['PAPER_ATLAS_TEST_PG_SERVICE'];self.target='pa_joint_'+uuid.uuid4().hex[:12]
self.tmp=tempfile.TemporaryDirectory();self.addCleanup(self.tmp.cleanup)
self.db=Path(self.tmp.name)/'sample.sqlite';s=Store(self.db);seed(s);s.conn.close()
self.addCleanup(self.cleanup)
with redirect_stdout(io.StringIO()):migrate(self.db,self.service,Path(self.tmp.name)/'migration.json',schema=self.target,min_free_gib=0)
def cleanup(self):
with connect(self.service)as c:
for name, in c.execute('SELECT nspname FROM pg_namespace WHERE starts_with(nspname,%s)',(self.target,)).fetchall():
c.execute('DROP SCHEMA '+quote(name)+' CASCADE')
def test_native_repair_is_idempotent_keeps_codec_and_provenance(self):
p=Path(self.tmp.name)/'plan.json'
with redirect_stdout(io.StringIO()):
before=run(None,p,pg_service=self.service,pg_schema=self.target)
after=run(None,Path(self.tmp.name)/'applied.json',pg_service=self.service,pg_schema=self.target,plan=p,plan_sha256=hashlib.sha256(p.read_bytes()).hexdigest())
self.assertEqual(after['content_fingerprints'],before['content_fingerprints'])
with closing(open_store(pg_service=self.service,pg_schema=self.target).conn)as c:
self.assertEqual(apply_plan(c,before)['updated_editions'],0)
self.assertEqual(c.execute('SELECT abstract FROM papers ORDER BY id LIMIT 1').fetchone()[0],'Abstract\x00with\x01data')
def test_native_stale_source_plan_preserves_both_original_memberships(self):
with closing(open_store(pg_service=self.service,pg_schema=self.target).conn)as c:
before=plan(c)
c.execute("UPDATE source_records SET title='Changed' WHERE external_id=?",('conf/ijcai/fixture2022',));c.commit()
with self.assertRaisesRegex(ValueError,'Stale repair plan'):
apply_plan(c,before)
c.rollback()
self.assertEqual(inspect(c)['status'],'ready')