From 27e336eee9d9ac3a01b5190a2f892368bd0bab11 Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Sun, 8 Nov 2026 18:38:48 +0800 Subject: [PATCH 01/11] Add evidence-bearing research types and retractable custom annotations --- .gitea/workflows/check.yml | 2 + README.md | 2 + docs/annotations.md | 107 ++++++++++++++ paper_atlas/cli.py | 1 + paper_atlas/curation/__init__.py | 1 + paper_atlas/curation/annotations.py | 218 +++++++++++++++++++++++++++ paper_atlas/db/annotations.sql | 15 ++ paper_atlas/db/store.py | 3 + paper_atlas/postgres/migration.py | 4 +- paper_atlas/postgres/schema.py | 6 +- paper_atlas/querying.py | 26 +++- pyproject.toml | 2 +- tests/test_annotations.py | 219 ++++++++++++++++++++++++++++ 13 files changed, 600 insertions(+), 6 deletions(-) create mode 100644 docs/annotations.md create mode 100644 paper_atlas/curation/__init__.py create mode 100644 paper_atlas/curation/annotations.py create mode 100644 paper_atlas/db/annotations.sql create mode 100644 tests/test_annotations.py diff --git a/.gitea/workflows/check.yml b/.gitea/workflows/check.yml index 950c95a..be24fb9 100644 --- a/.gitea/workflows/check.yml +++ b/.gitea/workflows/check.yml @@ -58,5 +58,7 @@ 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 diff --git a/README.md b/README.md index 09dcf66..f12719a 100644 --- a/README.md +++ b/README.md @@ -86,3 +86,5 @@ paper-atlas collect colm --cache /data/paper-index/raw/http-cache \ CI 在已授权的远程 runner 上运行测试、构建 wheel，并从源码目录外验证安装后的命令及随包资源。数据库全量去重和覆盖率是独立的数据验收步骤，不因代码测试通过而自动完成。 开发通过工作分支和 PR 交付，由用户手动合并。 + +研究类型与自定义标签可通过 [`annotate`](docs/annotations.md) 显式初始化、添加和撤回；两者独立查询，保留证据及完整更正历史，不从会议轨道推断研究类型。 diff --git a/docs/annotations.md b/docs/annotations.md new file mode 100644 index 0000000..fbb10b8 --- /dev/null +++ b/docs/annotations.md @@ -0,0 +1,107 @@ +# 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. diff --git a/paper_atlas/cli.py b/paper_atlas/cli.py index 549bfc0..eca3325 100644 --- a/paper_atlas/cli.py +++ b/paper_atlas/cli.py @@ -6,6 +6,7 @@ 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', {}, '下载并校验固定版本快照'), diff --git a/paper_atlas/curation/__init__.py b/paper_atlas/curation/__init__.py new file mode 100644 index 0000000..27d2716 --- /dev/null +++ b/paper_atlas/curation/__init__.py @@ -0,0 +1 @@ +"""Evidence-bearing human and explicit source annotations, independent of outcomes.""" diff --git a/paper_atlas/curation/annotations.py b/paper_atlas/curation/annotations.py new file mode 100644 index 0000000..a8fa0d8 --- /dev/null +++ b/paper_atlas/curation/annotations.py @@ -0,0 +1,218 @@ +"""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): + 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)) diff --git a/paper_atlas/db/annotations.sql b/paper_atlas/db/annotations.sql new file mode 100644 index 0000000..eabb873 --- /dev/null +++ b/paper_atlas/db/annotations.sql @@ -0,0 +1,15 @@ +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); diff --git a/paper_atlas/db/store.py b/paper_atlas/db/store.py index 7d57533..e45ef4f 100644 --- a/paper_atlas/db/store.py +++ b/paper_atlas/db/store.py @@ -55,6 +55,9 @@ 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)) diff --git a/paper_atlas/postgres/migration.py b/paper_atlas/postgres/migration.py index 0a051ed..2865851 100644 --- a/paper_atlas/postgres/migration.py +++ b/paper_atlas/postgres/migration.py @@ -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 21 business tables; FTS5 internals rebuilt as native PostgreSQL GIN.'} + 'started_at': timestamp(), 'tables': {}, 'scope': 'All baseline business tables and recognized installed extensions; FTS5 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 TABLES: + for table in plan: 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] diff --git a/paper_atlas/postgres/schema.py b/paper_atlas/postgres/schema.py index 90a97de..0120746 100644 --- a/paper_atlas/postgres/schema.py +++ b/paper_atlas/postgres/schema.py @@ -8,7 +8,9 @@ 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')"], @@ -50,10 +52,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 business != set(TABLES): + if not set(TABLES) <= business or business - set(TABLES) - set(OPTIONAL_TABLES): raise ValueError('Unexpected business tables: ' + repr(sorted(business ^ set(TABLES)))) result = {} - for table in TABLES: + for table in (*TABLES, *(name for name in OPTIONAL_TABLES if name in business)): 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) + ')')] diff --git a/paper_atlas/querying.py b/paper_atlas/querying.py index cc82979..a0f60c9 100644 --- a/paper_atlas/querying.py +++ b/paper_atlas/querying.py @@ -29,6 +29,8 @@ 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) @@ -47,6 +49,8 @@ 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')] @@ -69,8 +73,24 @@ 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) - 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]) + 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') 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: @@ -98,6 +118,10 @@ 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 = [] diff --git a/pyproject.toml b/pyproject.toml index 0357a1c..9655ba8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -21,5 +21,5 @@ include = ["paper_atlas*"] exclude = ["tests*"] [tool.setuptools.package-data] -"paper_atlas.db" = ["schema.sql"] +"paper_atlas.db" = ["schema.sql", "annotations.sql"] "paper_atlas.resources" = ["venues.json"] diff --git a/tests/test_annotations.py b/tests/test_annotations.py new file mode 100644 index 0000000..2f4dcb7 --- /dev/null +++ b/tests/test_annotations.py @@ -0,0 +1,219 @@ +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_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() + 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() -- 2.52.0 From a13c5359194cf48e66e6000924a6826ac013b907 Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Sun, 8 Nov 2026 18:40:51 +0800 Subject: [PATCH 02/11] Require completed annotation upgrade before accepting evidence --- paper_atlas/curation/annotations.py | 2 +- tests/test_annotations.py | 7 +++++++ 2 files changed, 8 insertions(+), 1 deletion(-) diff --git a/paper_atlas/curation/annotations.py b/paper_atlas/curation/annotations.py index a8fa0d8..f6e593c 100644 --- a/paper_atlas/curation/annotations.py +++ b/paper_atlas/curation/annotations.py @@ -78,7 +78,7 @@ def _url(value): def _require(conn): - if not available(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') diff --git a/tests/test_annotations.py b/tests/test_annotations.py index 2f4dcb7..eba7a54 100644 --- a/tests/test_annotations.py +++ b/tests/test_annotations.py @@ -59,6 +59,13 @@ class AnnotationTests(unittest.TestCase): 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() -- 2.52.0 From adcbbb66a224320f236fb28ad88405b44ddeb5b2 Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Sun, 8 Nov 2026 18:41:38 +0800 Subject: [PATCH 03/11] Freeze SQLite annotation fixtures before PostgreSQL migration --- tests/test_annotations.py | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/test_annotations.py b/tests/test_annotations.py index eba7a54..c7b9f0a 100644 --- a/tests/test_annotations.py +++ b/tests/test_annotations.py @@ -176,6 +176,7 @@ class PostgreSQLAnnotationTests(unittest.TestCase): 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) -- 2.52.0 From 3f81734e6915df0978ecc47844c342622fa0f2d2 Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Sun, 8 Nov 2026 19:07:00 +0800 Subject: [PATCH 04/11] Repair verified joint edition membership without reimporting papers --- .gitea/workflows/check.yml | 1 + README.md | 2 + docs/joint-editions.md | 65 ++++++++ paper_atlas/cli.py | 1 + paper_atlas/collectors/dblp_records.py | 2 + paper_atlas/collectors/dblp_scope.py | 7 + paper_atlas/ingestion/joint_editions.py | 201 ++++++++++++++++++++++++ tests/test_joint_editions.py | 173 ++++++++++++++++++++ 8 files changed, 452 insertions(+) create mode 100644 docs/joint-editions.md create mode 100644 paper_atlas/ingestion/joint_editions.py create mode 100644 tests/test_joint_editions.py diff --git a/.gitea/workflows/check.yml b/.gitea/workflows/check.yml index 950c95a..d57a25b 100644 --- a/.gitea/workflows/check.yml +++ b/.gitea/workflows/check.yml @@ -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) diff --git a/README.md b/README.md index 09dcf66..e9651fc 100644 --- a/README.md +++ b/README.md @@ -86,3 +86,5 @@ paper-atlas collect colm --cache /data/paper-index/raw/http-cache \ CI 在已授权的远程 runner 上运行测试、构建 wheel，并从源码目录外验证安装后的命令及随包资源。数据库全量去重和覆盖率是独立的数据验收步骤，不因代码测试通过而自动完成。 开发通过工作分支和 PR 交付，由用户手动合并。 + +历史联合会议的既有届次归属可通过 [`maintain joint-editions`](docs/joint-editions.md) 先只读规划，再按计划哈希执行；保留论文与投稿决定来源，不重导入。 diff --git a/docs/joint-editions.md b/docs/joint-editions.md new file mode 100644 index 0000000..a35775f --- /dev/null +++ b/docs/joint-editions.md @@ -0,0 +1,65 @@ +# 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. +Other 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. diff --git a/paper_atlas/cli.py b/paper_atlas/cli.py index 549bfc0..ef89ce0 100644 --- a/paper_atlas/cli.py +++ b/paper_atlas/cli.py @@ -25,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 类别证据'), diff --git a/paper_atlas/collectors/dblp_records.py b/paper_atlas/collectors/dblp_records.py index 4f5685e..2eb24c3 100644 --- a/paper_atlas/collectors/dblp_records.py +++ b/paper_atlas/collectors/dblp_records.py @@ -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)) diff --git a/paper_atlas/collectors/dblp_scope.py b/paper_atlas/collectors/dblp_scope.py index 08a550d..ee09e6a 100644 --- a/paper_atlas/collectors/dblp_scope.py +++ b/paper_atlas/collectors/dblp_scope.py @@ -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. diff --git a/paper_atlas/ingestion/joint_editions.py b/paper_atlas/ingestion/joint_editions.py new file mode 100644 index 0000000..bd5a576 --- /dev/null +++ b/paper_atlas/ingestion/joint_editions.py @@ -0,0 +1,201 @@ +"""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] + 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 sid, external, metadata in evidence: + 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: + 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) diff --git a/tests/test_joint_editions.py b/tests/test_joint_editions.py new file mode 100644 index 0000000..9a86a07 --- /dev/null +++ b/tests/test_joint_editions.py @@ -0,0 +1,173 @@ +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'},metadata) + 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') + 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_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) + 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') -- 2.52.0 From c05b979b01af15cc19add27e81ce9e7ea8103a71 Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Sun, 8 Nov 2026 19:09:27 +0800 Subject: [PATCH 05/11] Bind repair execution receipts to the reviewed plan hash --- paper_atlas/ingestion/joint_editions.py | 2 ++ tests/test_joint_editions.py | 1 + 2 files changed, 3 insertions(+) diff --git a/paper_atlas/ingestion/joint_editions.py b/paper_atlas/ingestion/joint_editions.py index bd5a576..ff44857 100644 --- a/paper_atlas/ingestion/joint_editions.py +++ b/paper_atlas/ingestion/joint_editions.py @@ -181,6 +181,8 @@ def run(db, out, *, pg_service=None, pg_schema=None, plan=None, plan_sha256=None 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: diff --git a/tests/test_joint_editions.py b/tests/test_joint_editions.py index 9a86a07..1732589 100644 --- a/tests/test_joint_editions.py +++ b/tests/test_joint_editions.py @@ -108,6 +108,7 @@ class JointEditionTests(unittest.TestCase): 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'): -- 2.52.0 From 866aeaf068f71677e688ea87305b889579eef7ea Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Sun, 8 Nov 2026 19:10:10 +0800 Subject: [PATCH 06/11] Read PostgreSQL mapping rows by position during joint preflight --- paper_atlas/ingestion/joint_editions.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/paper_atlas/ingestion/joint_editions.py b/paper_atlas/ingestion/joint_editions.py index ff44857..c09fa24 100644 --- a/paper_atlas/ingestion/joint_editions.py +++ b/paper_atlas/ingestion/joint_editions.py @@ -62,7 +62,7 @@ def inspect(c): 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] + 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') @@ -73,7 +73,8 @@ def inspect(c): 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 sid, external, metadata in evidence: + 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 -- 2.52.0 From 88c2819eddfcfa26c6ce311a14638ef964770eed Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Sun, 8 Nov 2026 19:15:26 +0800 Subject: [PATCH 07/11] Keep legacy source refreshes in their verified joint editions --- docs/joint-editions.md | 4 +++- paper_atlas/db/store.py | 14 ++++++++++++++ tests/test_joint_editions.py | 15 ++++++++++++++- 3 files changed, 31 insertions(+), 2 deletions(-) diff --git a/docs/joint-editions.md b/docs/joint-editions.md index a35775f..7c630fb 100644 --- a/docs/joint-editions.md +++ b/docs/joint-editions.md @@ -14,7 +14,9 @@ source links and coverage records stay intact. It does not merge or create paper 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. -Other years, workshop volumes and additional tracks retain their existing rules. +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 diff --git a/paper_atlas/db/store.py b/paper_atlas/db/store.py index 7d57533..272ace4 100644 --- a/paper_atlas/db/store.py +++ b/paper_atlas/db/store.py @@ -144,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')) diff --git a/tests/test_joint_editions.py b/tests/test_joint_editions.py index 1732589..47246d0 100644 --- a/tests/test_joint_editions.py +++ b/tests/test_joint_editions.py @@ -32,9 +32,11 @@ def seed(store): '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'},metadata) + {'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() @@ -64,6 +66,17 @@ class JointEditionTests(unittest.TestCase): 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'") -- 2.52.0 From 4182070e76dfa13ef7dbb526a0b6a4d155f19cd6 Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Sun, 8 Nov 2026 21:04:05 +0800 Subject: [PATCH 08/11] Scope incremental parents to events and verify integrated write safeguards --- README.md | 2 + docs/event-identity.md | 2 +- docs/research-runtime.md | 44 +++++++ paper_atlas/ingestion/event_identity.py | 8 +- paper_atlas/ingestion/records.py | 20 ++- tests/test_research_runtime.py | 154 ++++++++++++++++++++++++ 6 files changed, 224 insertions(+), 6 deletions(-) create mode 100644 docs/research-runtime.md create mode 100644 tests/test_research_runtime.py diff --git a/README.md b/README.md index fc4b55d..dd61ba5 100644 --- a/README.md +++ b/README.md @@ -90,3 +90,5 @@ CI 在已授权的远程 runner 上运行测试、构建 wheel，并从源码目 历史联合会议的既有届次归属可通过 [`maintain joint-editions`](docs/joint-editions.md) 先只读规划，再按计划哈希执行；保留论文与投稿决定来源，不重导入。 同年多届的 `event_key`、显式迁移、旧文件刷新与查询边界见 [会议事件身份](docs/event-identity.md)。 研究类型与自定义标签可通过 [`annotate`](docs/annotations.md) 显式初始化、添加和撤回；两者独立查询，保留证据及完整更正历史，不从会议轨道推断研究类型。 + +整合后的运行版本依赖、交互约束和生产切换顺序见 [运行版本整合](docs/research-runtime.md)。 diff --git a/docs/event-identity.md b/docs/event-identity.md index 7cf6eba..8f65575 100644 --- a/docs/event-identity.md +++ b/docs/event-identity.md @@ -17,7 +17,7 @@ PGSERVICEFILE=/etc/paper-atlas/pg_service.conf paper-atlas maintain event-identi --pg-service foundry-research --pg-schema research --out /data/paper-index/event-plan.json ``` -检查计划的数据库身份、原 edition IDs、业务表数量、目标来源与历史指纹。精确计数在同一快照中按主键分页，每页最多5000行，只读取主键；不使用估计行数。CLI向stderr持续报告计数进度，避免大库长时间无输出。已有未标记 WWW1995 edition 必须逐投稿都有固定 DBLP Fall 卷证据，才会只为原 edition 填入 `www-1995-fall`；缺证或混合卷会拒绝计划。本文命令不表示生产已经迁移。 +检查计划的数据库身份、原 edition IDs、业务表数量、目标来源与历史指纹。计数也包含已显式启用的可选注解表。精确计数在同一快照中按主键分页，每页最多5000行，只读取主键；不使用估计行数。CLI向stderr持续报告计数进度，避免大库长时间无输出。已有未标记 WWW1995 edition 必须逐投稿都有固定 DBLP Fall 卷证据，才会只为原 edition 填入 `www-1995-fall`；缺证或混合卷会拒绝计划。本文命令不表示生产已经迁移。 应用时传入同一文件及命令输出的文件 SHA256： diff --git a/docs/research-runtime.md b/docs/research-runtime.md new file mode 100644 index 0000000..16bfac2 --- /dev/null +++ b/docs/research-runtime.md @@ -0,0 +1,44 @@ +# Integrated research runtime + +This branch composes the independently reviewed identity guards (PR25), incremental +coverage (PR27), joint-edition refresh guard (PR29), event identity (PR30), and optional +annotation support (PR28). It starts from main 1b81d1a and preserves their commits; +it is an explicit integration dependency, not evidence that those PRs have merged. +Users merge through Gitea manually. Review this combined diff against main, including +its conflict resolutions, before using it as the maintained runtime. + +The shared Store preserves stable source/identifier anchors before title matching, +rejects ambiguous unanchored matches, resolves pinned IJCAI/ECAI volumes before event +lookup, and selects same-year events by event_key. Incremental manifests for a named +or ambiguous event must state event_key; their parent count and optimistic update +are checked against that exact event, not the first venue/year/track match. + +Query filters compose event_key, status, tags and research type. An earlier rejected +attempt and a later known acceptance remain distinct; neither annotations nor track +names infer a research type. Installing this code does not initialize annotation_events. +Event plans count that optional table when present, alongside the original business +rows. Annotation initialization and event identity upgrade remain explicit commands +with their own transactional checks. Neither command imports historical records. + +## Production sequencing + +1. Verify the exact integrated commit on the approved ces-control runner, including + the real PostgreSQL interaction tests and wheel/installed CLI checks. +2. Stage that exact source in a separate release directory and verify package hashes; + leave the current runtime and frozen SQLite backup intact. +3. Inspect live writers and coordinate through the existing database writer lock. + Generate a fresh read-only event plan using that integrated release. Check its + database/schema identity, all business counts and preserved history hashes. A plan + from before further data or optional-schema changes must not be reused. +4. Apply only that reviewed file with its actual SHA256, in one transaction. Independently + verify the constraint, original edition IDs and fields, assigned Fall source, source + and history bytes, ordinary queries, full-text queries and foundry health. +5. Use the verified integrated release for subsequent writes. Older writers lacking + event identity support must not resume against the upgraded schema. Annotation + initialization requires a separate explicit operation and acceptance. + +These are operational requirements, not claims of completed production deployment. +The production event-key migration and optional annotation activation have not occurred +at preparation time. Existing migration/import/repair acceptance remains valid; do not +rerun those jobs to install source code. Cross-feature tests use disposable fixtures +and PostgreSQL schemas, never the production research tables. diff --git a/paper_atlas/ingestion/event_identity.py b/paper_atlas/ingestion/event_identity.py index b5f4020..71e082d 100644 --- a/paper_atlas/ingestion/event_identity.py +++ b/paper_atlas/ingestion/event_identity.py @@ -9,7 +9,7 @@ import sys from paper_atlas.db.backend import database_arguments, open_read_connection, require_sqlite_writable from paper_atlas.db.events import has_event_key, pinned_www_fall -from paper_atlas.postgres.schema import TABLES, quote +from paper_atlas.postgres.schema import TABLES, OPTIONAL_TABLES, quote MARKER = 'edition_event_identity_v1:' @@ -27,7 +27,11 @@ def exact_counts(c, *, page_size=5000, progress=None): if not isinstance(page_size, int) or page_size < 1: raise ValueError('Positive count page size required') result = {} - for table in TABLES: + if getattr(c, 'postgres', False): + optional = [table for table in OPTIONAL_TABLES if c.execute("SELECT 1 FROM information_schema.tables WHERE table_schema=current_schema() AND table_name=?", (table,)).fetchone()] + else: + optional = [table for table in OPTIONAL_TABLES if c.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (table,)).fetchone()] + for table in (*TABLES, *optional): if table == 'schema_migrations': continue if getattr(c, 'postgres', False): diff --git a/paper_atlas/ingestion/records.py b/paper_atlas/ingestion/records.py index 50baa40..63898ca 100644 --- a/paper_atlas/ingestion/records.py +++ b/paper_atlas/ingestion/records.py @@ -42,9 +42,23 @@ def source_record_count(c, source, edition): def require_incremental_parent(c, source, manifest, options): - edition = c.execute('''SELECT e.id FROM editions e JOIN venues v ON v.id=e.venue_id - WHERE lower(v.acronym)=lower(?) AND e.year=? AND e.track=?''', - (manifest['venue'], manifest['year'], manifest['track'])).fetchone() + from paper_atlas.db.events import has_event_key, validate_event_key + key = validate_event_key(manifest.get('event_key')) + enabled = has_event_key(c) + if key and not enabled: + raise ValueError('Event identity migration required for incremental coverage') + field = 'e.event_key' if enabled else "''" + rows = c.execute('SELECT e.id,'+field+""" FROM editions e JOIN venues v ON v.id=e.venue_id + WHERE lower(v.acronym)=lower(?) AND e.year=? AND e.track=?""", + (manifest['venue'], manifest['year'], manifest['track'])).fetchall() + if key is None: + if len(rows) > 1 or any(r[1] for r in rows) or (manifest['venue'] == 'WWW' and manifest['year'] in (1994,1995)): + raise ValueError('Explicit manifest event_key required for incremental coverage') + else: + rows = [r for r in rows if r[1] == key] + if len(rows) > 1: + raise ValueError('Ambiguous incremental coverage event') + edition = rows[0] if rows else None entries = c.execute('''SELECT list_url,status,expected_count,imported_count FROM edition_sources WHERE edition_id=? AND source_id=?''', (edition[0], source)).fetchall() if edition else [] diff --git a/tests/test_research_runtime.py b/tests/test_research_runtime.py new file mode 100644 index 0000000..c451278 --- /dev/null +++ b/tests/test_research_runtime.py @@ -0,0 +1,154 @@ +"""Interactions between event identity and previously independent write safeguards.""" +from contextlib import contextmanager, redirect_stdout +import hashlib +import io +import json +import os +from pathlib import Path +import tempfile +import unittest +import uuid + +from paper_atlas import querying +from paper_atlas.collectors.dblp import XML_SHA +from paper_atlas.curation.annotations import initialize, assert_annotation, retract_annotation, observations +from paper_atlas.db.backend import open_store +from paper_atlas.ingestion.event_identity import plan, upgrade +from paper_atlas.ingestion.records import ingest_file +from paper_atlas.postgres.client import connect +from paper_atlas.postgres.migration import migrate +from paper_atlas.postgres.schema import quote +from tests.test_event_identity import legacy, add + + +class RuntimeCases: + @contextmanager + def store(self): + s = open_store(**self.target) + try: + yield s + finally: + s.conn.close() + + def query(self, *args): + target = (['--db', str(self.target['db'])] if 'db' in self.target else + ['--pg-service', self.target['pg_service'], '--pg-schema', self.target['pg_schema']]) + with redirect_stdout(io.StringIO()) as out: + querying.main([*target, *map(str,args)]) + return json.loads(out.getvalue()) + + def assertion(self, c, paper, request, **changes): + args = dict(paper=paper, kind='research_type', namespace='custom:method', value='empirical', + request_id=request, evidence_key='review:'+request, actor='reviewer', reason='Reviewed metadata') + args.update(changes) + return assert_annotation(c, **args) + + def test_event_upgrade_preserves_annotation_history_and_combined_queries(self): + with self.store() as s: + initialize(s.conn) + event = self.assertion(s.conn,1,'first') + retract_annotation(s.conn,event,request_id='retract',actor='reviewer',reason='Corrected') + self.assertion(s.conn,1,'second') + original = observations(s.conn,1,history=True) + s.conn.commit() + proposal = plan(**self.target) + self.assertEqual(proposal['before']['counts']['annotation_events'],3) + result = upgrade(proposal, **self.target) + self.assertEqual(result['after']['counts'],proposal['before']['counts']) + with self.store() as s: + self.assertEqual(observations(s.conn,1,history=True),original) + paper,_,_ = add(s,'spring','www-1995-spring','rejected') + self.assertEqual(paper,1) + s.conn.commit() + row = self.query('--venue','WWW','--event-key','www-1995-spring','--status','rejected', + '--research-type','custom:method=empirical','--paper',1,'--annotation-history')[0] + self.assertEqual([a['event_key'] for a in row['known_acceptances']],['www-1995-fall']) + self.assertEqual({a['latest_status'] for a in row['submissions']},{'accepted','rejected'}) + self.assertEqual(len(row['annotation_history']),3) + self.assertEqual(self.query('--event-key','www-1995-fall','--status','rejected'),[]) + + def collected(self, name, event, external, *, incremental=False, unscoped=False): + row = dict(source='fixture',external_id=external,url='https://example.org/'+external, + title='Independent contribution '+external,authors=[],abstract='Known abstract', + identifiers={'doi':'10.1234/'+external},venue='WWW',year=1995,track='main', + event_key=event,status='accepted') + p = self.root/(name+'.jsonl');p.write_text(json.dumps(row)+'\n') + manifest = {k:row[k] for k in ('source','venue','year','track','event_key')} + manifest.update(list_url='https://example.org/'+event,count=1,status='partial', + file_sha256=hashlib.sha256(p.read_bytes()).hexdigest()) + if incremental: + manifest.update(coverage_mode='incremental',coverage_previous_count=1, + coverage_previous_expected_count=1,coverage_expected_count=2) + if unscoped: + manifest.pop('event_key') + p.with_suffix('.manifest.json').write_text(json.dumps(manifest)) + return p + + def ingest(self, path): + with redirect_stdout(io.StringIO()): + ingest_file(self.target.get('db'),path,**{k:v for k,v in self.target.items() if k != 'db'}) + + def test_incremental_parent_is_scoped_to_exact_event(self): + upgrade(plan(**self.target), **self.target) + self.ingest(self.collected('spring','www-1995-spring','spring-one')) + self.ingest(self.collected('fall','www-1995-fall','fall-one')) + self.ingest(self.collected('append','www-1995-fall','fall-two',incremental=True)) + with self.store() as s: + rows = s.conn.execute('''SELECT e.event_key,es.expected_count,es.imported_count FROM edition_sources es + JOIN editions e ON e.id=es.edition_id JOIN sources src ON src.id=es.source_id + WHERE src.name='fixture' ORDER BY e.event_key''').fetchall() + self.assertEqual([tuple(r) for r in rows],[('www-1995-fall',2,2),('www-1995-spring',1,1)]) + with self.assertRaisesRegex(ValueError,'event_key'): + self.ingest(self.collected('unscoped','www-1995-spring','unwanted',incremental=True,unscoped=True)) + with self.assertRaisesRegex(ValueError,'parent changed'): + self.ingest(self.collected('stale','www-1995-fall','stale',incremental=True)) + with self.store() as s: + self.assertEqual(s.conn.execute("SELECT count(*) FROM source_records r JOIN sources src ON src.id=r.source_id WHERE src.name='fixture'").fetchone()[0],3) + + def test_joint_refresh_keeps_identity_annotations_and_event_query(self): + upgrade(plan(**self.target), **self.target) + metadata = dict(snapshot_sha256=XML_SHA,proceedings_key='conf/ijcai/2022', + bibliographic_year=2022,bibliography={'booktitle':['IJCAI']}) + with self.store() as s: + initialize(s.conn) + paper,record = s.record('dblp','conf/ijcai/Joint22','https://example.org/joint', + 'Joint paper',None,[],{'doi':'10.1234/joint'},metadata) + eid = s.decision(paper,record,'IJCAI',2022,'main','conf/ijcai/Joint22','https://example.org/joint','accepted','Accept') + self.assertion(s.conn,paper,'joint') + s.conn.commit() + with self.store() as s: + again,r = s.record('dblp','conf/ijcai/Joint22','https://example.org/joint', + 'Corrected joint title',None,[],{'doi':'10.1234/joint'},metadata) + self.assertEqual(again,paper) + self.assertEqual(s.decision(again,r,'IJCAI',2022,'main','conf/ijcai/Joint22','https://example.org/joint','accepted','Accept'),eid) + self.assertEqual(s.conn.execute("SELECT count(*) FROM editions e JOIN venues v ON v.id=e.venue_id WHERE v.acronym='IJCAI' AND e.year=2022").fetchone()[0],0) + s.conn.commit() + for venue in ['IJCAI','ECAI']: + rows = self.query('--venue',venue,'--year',2022,'--event-key','', + '--research-type','custom:method=empirical','--paper',paper) + self.assertEqual(len(rows),1) + self.assertEqual(len(rows[0]['submissions']),1) + self.assertEqual(rows[0]['submissions'][0]['acronym'],'ECAI+IJCAI') + + +class SQLiteRuntimeTests(RuntimeCases, unittest.TestCase): + def setUp(self): + t=tempfile.TemporaryDirectory();self.addCleanup(t.cleanup);self.root=Path(t.name) + db=self.root/'source.sqlite';legacy(db);self.target={'db':db} + + +@unittest.skipUnless(os.environ.get('PAPER_ATLAS_TEST_PG_SERVICE'),'Requires approved remote PostgreSQL service') +class PostgreSQLRuntimeTests(RuntimeCases, unittest.TestCase): + def setUp(self): + t=tempfile.TemporaryDirectory();self.addCleanup(t.cleanup);self.root=Path(t.name) + db=self.root/'source.sqlite';legacy(db) + self.service=os.environ['PAPER_ATLAS_TEST_PG_SERVICE'];self.schema='pa_runtime_'+uuid.uuid4().hex[:12] + self.addCleanup(self.cleanup) + with redirect_stdout(io.StringIO()): + migrate(db,self.service,self.root/'migration.json',schema=self.schema,min_free_gib=0) + self.target={'pg_service':self.service,'pg_schema':self.schema} + + 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.schema,)): + c.execute('DROP SCHEMA '+quote(name)+' CASCADE') -- 2.52.0 From f9f4d263af703cb1bd58d9f3a504226e8f1f15c8 Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Sun, 8 Nov 2026 23:17:55 +0800 Subject: [PATCH 09/11] Expose independent event keys in integrated coverage reports --- docs/research-runtime.md | 5 +++++ paper_atlas/audits/coverage.py | 6 ++++-- tests/test_research_runtime.py | 18 ++++++++++++++++++ 3 files changed, 27 insertions(+), 2 deletions(-) diff --git a/docs/research-runtime.md b/docs/research-runtime.md index 16bfac2..0959f7e 100644 --- a/docs/research-runtime.md +++ b/docs/research-runtime.md @@ -42,3 +42,8 @@ The production event-key migration and optional annotation activation have not o at preparation time. Existing migration/import/repair acceptance remains valid; do not rerun those jobs to install source code. Cross-feature tests use disposable fixtures and PostgreSQL schemas, never the production research tables. + +Coverage output also returns event_key for each independent edition ID. Two same-year +main events keep their own status/source counts; venue paper totals remain distinct +work counts rather than sums across events. Legacy datasets without the event column +return an empty event key and remain read-only compatible. diff --git a/paper_atlas/audits/coverage.py b/paper_atlas/audits/coverage.py index 42db2b8..0f7bf8d 100644 --- a/paper_atlas/audits/coverage.py +++ b/paper_atlas/audits/coverage.py @@ -45,6 +45,8 @@ def report(db, registry, *, pg_service=None, pg_schema=None): database, schema = identity[0], identity[1] else: database, schema = str(Path(db).resolve()), None + from paper_atlas.db.events import has_event_key + event = 'e.event_key' if has_event_key(connection) else "''" latest = _status_counts(connection, latest=True) observations = _status_counts(connection, latest=False) # Aggregate the paper join once for the whole snapshot. Repeating it @@ -73,9 +75,9 @@ def report(db, registry, *, pg_service=None, pg_schema=None): params = (acronym, acronym) editions = [] for row in connection.execute('''SELECT e.id,v.id AS venue_id,v.acronym AS canonical_venue, - e.year,e.track FROM editions e JOIN venues v ON v.id=e.venue_id + e.year,e.track,''' + event + ''' AS event_key FROM editions e JOIN venues v ON v.id=e.venue_id WHERE ''' + scope + ''' - ORDER BY e.year,e.track,v.acronym''', params): + ORDER BY e.year,e.track,event_key,v.acronym,e.id''', params): edition = dict(row) edition.update(edition_counts.get(row['id'], { 'submissions': 0, 'papers': 0, 'papers_without_abstract_text': 0})) diff --git a/tests/test_research_runtime.py b/tests/test_research_runtime.py index c451278..b5b96fc 100644 --- a/tests/test_research_runtime.py +++ b/tests/test_research_runtime.py @@ -10,6 +10,7 @@ import unittest import uuid from paper_atlas import querying +from paper_atlas.audits.coverage import report as coverage_report from paper_atlas.collectors.dblp import XML_SHA from paper_atlas.curation.annotations import initialize, assert_annotation, retract_annotation, observations from paper_atlas.db.backend import open_store @@ -88,6 +89,23 @@ class RuntimeCases: with redirect_stdout(io.StringIO()): ingest_file(self.target.get('db'),path,**{k:v for k,v in self.target.items() if k != 'db'}) + def test_coverage_labels_both_same_year_events_and_keeps_legacy_readable(self): + registry=self.root/'registry.json' + registry.write_text(json.dumps({'scope_note':'Fixture only','venues':[{'acronym':'WWW','field':'web'}]})) + def snapshot(): + return coverage_report(self.target.get('db'),registry,**{k:v for k,v in self.target.items() if k!='db'})['venues'][0] + self.assertEqual([e['event_key']for e in snapshot()['editions']],['']) + upgrade(plan(**self.target),**self.target) + with self.store() as s: + add(s,'spring','www-1995-spring','rejected');s.conn.commit() + result=snapshot();events={e['event_key']:e for e in result['editions']} + self.assertEqual(set(events),{'www-1995-fall','www-1995-spring'}) + self.assertEqual((result['imported_submissions'],result['distinct_papers']),(2,1)) + self.assertEqual(events['www-1995-fall']['latest_observed_status_counts']['accepted'],1) + self.assertEqual(events['www-1995-spring']['latest_observed_status_counts']['rejected'],1) + self.assertEqual(events['www-1995-fall']['source_record_counts'][0]['source'],'dblp') + self.assertEqual(events['www-1995-spring']['source_record_counts'][0]['source'],'fixture') + def test_incremental_parent_is_scoped_to_exact_event(self): upgrade(plan(**self.target), **self.target) self.ingest(self.collected('spring','www-1995-spring','spring-one')) -- 2.52.0 From 7035cf33b9cabf13184703d8b3318d646b67c5fe Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Sun, 8 Nov 2026 23:46:37 +0800 Subject: [PATCH 10/11] Correct existing attempt tracks with bound evidence and refresh guards --- .gitea/workflows/check.yml | 1 + README.md | 2 + docs/edition-tracks.md | 45 +++++ paper_atlas/cli.py | 1 + paper_atlas/db/edition_scopes.py | 38 ++++ paper_atlas/db/store.py | 7 +- paper_atlas/ingestion/edition_tracks.py | 243 ++++++++++++++++++++++++ paper_atlas/ingestion/records.py | 6 + tests/test_edition_tracks.py | 135 +++++++++++++ 9 files changed, 477 insertions(+), 1 deletion(-) create mode 100644 docs/edition-tracks.md create mode 100644 paper_atlas/db/edition_scopes.py create mode 100644 paper_atlas/ingestion/edition_tracks.py create mode 100644 tests/test_edition_tracks.py diff --git a/.gitea/workflows/check.yml b/.gitea/workflows/check.yml index 66daccd..204e40a 100644 --- a/.gitea/workflows/check.yml +++ b/.gitea/workflows/check.yml @@ -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','edition-tracks','--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','maintain','event-identity','--help'],cwd=outside,check=True) subprocess.run([python,'-I','-B','-m','paper_atlas','import','records','--help'],cwd=outside,check=True) diff --git a/README.md b/README.md index 920965a..f6d8487 100644 --- a/README.md +++ b/README.md @@ -93,3 +93,5 @@ CI 在已授权的远程 runner 上运行测试、构建 wheel，并从源码目 研究类型与自定义标签可通过 [`annotate`](docs/annotations.md) 显式初始化、添加和撤回；两者独立查询，保留证据及完整更正历史，不从会议轨道推断研究类型。 整合后的运行版本依赖、交互约束和生产切换顺序见 [运行版本整合](docs/research-runtime.md)。 + +已导入整卷的逐条轨道修正见 [edition-tracks 维护流程](docs/edition-tracks.md)：按完整证据生成只读计划，保留原投稿和决定 ID，事务性修正 edition、标识与覆盖，并防止旧整卷刷新覆盖分类。 diff --git a/docs/edition-tracks.md b/docs/edition-tracks.md new file mode 100644 index 0000000..769538b --- /dev/null +++ b/docs/edition-tracks.md @@ -0,0 +1,45 @@ +# 按证据修正既有投稿的轨道 + +`paper-atlas maintain edition-tracks` 用于已导入整卷论文集、后来取得逐条官方分区证据的情形。它更正原投稿的 edition 归属，保留投稿 ID、所有决定和来源关系；不会重新导入论文、合并实体、生成研究方法标签或推断拒稿。支持独立 SQLite 样本和共享 PostgreSQL；不得写入已冻结的生产 SQLite 备份。 + +## 证据与完整范围 + +先核实官方目录、稳定 DOI、原始来源记录和分区，保存原始页面及 SHA256。输入 JSON 为 `{"format":1,"records":[...]}`，每条必须包含： + +| 字段 | 含义 | +| --- | --- | +| `submission_id`、`source_record_id`、`external_id` | 原有投稿、主来源记录及主来源投稿标识 | +| `source`、`doi` | 原来源名称及其 metadata.record_identifiers 中的 DOI | +| `from_edition_id`、`year`、`venue` | 原 edition 及其真实年份和规范会议名 | +| `venue_names` | 规范会议名加全部已有别名；必须与库中一致 | +| `track` | 经核实的目标轨道，使用小写字母、数字、下划线；范围未明确可用 proceedings_unspecified | +| `section` | 官方分区原文或明确的未确定说明，不推断研究方法 | +| `evidence_url`、`evidence_sha256` | 已人工/独立核实的 HTTPS 原始证据和内容哈希 | + +输入必须列出每个原 edition 的**全部**投稿，包含留在原轨道的记录。工具核对身份、DOI、单来源发表观察、现有别名、完整范围以及目标冲突；不自动解释 HTML，也不把填写了 URL/哈希当成已证明内容。核查者应先独立证明每条分区对应关系。 + +多来源共同拥有的投稿、有非 accepted 决定历史的投稿、已有分类修订、非空目标 edition、目标已有覆盖或标识等情形会停止，需要专门核对。每个原 edition 须有一个明确来源清单；其他来源或多清单覆盖需先另行协调。本功能不是任意批量移动或跨会合并工具。 + +## 只读计划、事务执行和验收 + +```sh +paper-atlas maintain edition-tracks \ + --pg-service foundry-research --pg-schema research \ + --inventory reviewed-inventory.json --out plan.json + +sha256sum plan.json + +paper-atlas maintain edition-tracks \ + --pg-service foundry-research --pg-schema research \ + --plan plan.json --plan-sha256 <上一步的完整SHA256> --out applied.json +``` + +报告路径必须是已有目录中的新文件。预检只读。执行使用 PostgreSQL 单写入 advisory lock，或 SQLite BEGIN IMMEDIATE；整个分类修正、逐来源证据、标识、覆盖及验收在一个事务中完成。计划绑定数据库身份、全部来源文本/非修订元数据、标签、可选 annotation_events、投稿/决定/来源/标识内容及原覆盖快照；过期即停止。失败回滚数据，PostgreSQL 序列可能留下正常间隙，不能用 max(id) 当数量。 + +目标 edition 沿用原会议、年份、event_key，不删除或重命名原 edition。只更正既有 submissions.edition_id 和对应 submission_identifiers.edition_id。原文、来源检索时间、决定 ID/状态/证据及各表来源关系不变。 + +分类证据写入来源元数据的保留字段 `edition_scope_curation_v1`，包括原范围、目标 edition、官方分区、URL、哈希和计划摘要。普通来源导入不能伪造或替换此字段；后续 record 刷新保留修订。相同来源投稿通过旧范围调用 decision 时仍路由到修正后的 edition。旧整卷 main 文件通过 import records 刷新会报错并回滚，要求先按核实轨道重新整理，避免覆盖正确的统计。 + +覆盖保留原清单 URL、cursor、成功时间，按实际目标分配 expected_count/imported_count 并标 partial；本工具核实分类，未核实摘要或所有来源完整性。原覆盖整行保留在计划中，不能继续将整卷数当 main 论文数。应用后校验原 edition 身份、全部受保护内容、分类、标识、修订字段和覆盖；相同计划再次执行仅在这些验收条件仍成立时返回 already_applied。 + +生产使用前必须在已确认 runner 完成软件验证。正式数据修正后仍需独立验收，并在最终全库审计中复核；单元测试和应用回执不能代替生产验收。 diff --git a/paper_atlas/cli.py b/paper_atlas/cli.py index d68e147..fefb48c 100644 --- a/paper_atlas/cli.py +++ b/paper_atlas/cli.py @@ -27,6 +27,7 @@ COMMANDS = { ('catalogue','dblp'): ('paper_atlas.collectors.dblp', {}, '盘点 DBLP 历史会议目录'), ('import','arxiv'): ('paper_atlas.ingestion.arxiv', {}, '导入 arXiv 元数据 ZIP'), ('import','records'): ('paper_atlas.ingestion.records', {}, '导入核验过的会议文件'), + ('maintain','edition-tracks'): ('paper_atlas.ingestion.edition_tracks', {}, '按逐条来源证据修正既有投稿轨道并保护刷新'), ('maintain','joint-editions'): ('paper_atlas.ingestion.joint_editions', {}, '按已核验计划修正既有联合会议届次归属'), ('maintain','event-identity'): ('paper_atlas.ingestion.event_identity', {}, '按核验计划增加同年会议事件身份'), ('maintain','merge'): ('paper_atlas.ingestion.maintenance', {}, '协调单写入合并与恢复'), diff --git a/paper_atlas/db/edition_scopes.py b/paper_atlas/db/edition_scopes.py new file mode 100644 index 0000000..1781970 --- /dev/null +++ b/paper_atlas/db/edition_scopes.py @@ -0,0 +1,38 @@ +"""Retain reviewed event scopes when a source refresh repeats a whole-volume label.""" +import json + +KEY = 'edition_scope_curation_v1' + + +def preserve(metadata, previous): + old = json.loads(previous).get(KEY) if previous else None + if KEY in metadata and metadata[KEY] != old: + raise ValueError('Source import cannot supply or replace reviewed edition scope') + if old is not None: + metadata[KEY] = old + return metadata + + +def resolve(c, record, external, venue, year, track, event_key): + row = c.execute('SELECT metadata_json FROM source_records WHERE id=?', (record,)).fetchone() + entries = json.loads(row[0]).get(KEY, []) if row else [] + if not isinstance(entries, list): + raise ValueError('Invalid reviewed edition scope') + applicable = [] + for entry in entries: + if entry['external_id'] != external: + continue + old, target = entry['from'], entry['target'] + if int(year) != old['year'] or venue not in entry['venue_names']: + continue + if track not in (old['track'], target['track']): + continue + if event_key is not None and event_key != old['event_key']: + continue + current = c.execute('SELECT venue_id,year,track,event_key FROM editions WHERE id=?', (target['edition_id'],)).fetchone() + if not current or [current[i] for i in range(4)] != [target[k] for k in ('venue_id','year','track','event_key')]: + raise ValueError('Reviewed target edition has changed') + applicable.append(target['edition_id']) + if len(applicable) > 1: + raise ValueError('Ambiguous reviewed edition scope') + return applicable[0] if applicable else None diff --git a/paper_atlas/db/store.py b/paper_atlas/db/store.py index 17d187b..e625ca1 100644 --- a/paper_atlas/db/store.py +++ b/paper_atlas/db/store.py @@ -96,6 +96,8 @@ class Store: 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() + from paper_atlas.db.edition_scopes import preserve + metadata = preserve(dict(metadata or {}), old[4] if old else None) if old: matches.add(old[0]) key = normalized(title) @@ -175,7 +177,10 @@ class Store: for alias in aliases or []: c.execute('INSERT INTO venue_aliases VALUES(?,?) ON CONFLICT DO NOTHING',(venue_id,alias)) from paper_atlas.db.events import find_edition - edition = find_edition(self,venue_id,venue,int(year),track,record,external_id,event_key) + from paper_atlas.db.edition_scopes import resolve + edition = resolve(c,record,external_id,venue,year,track,event_key) + if edition is None: + edition = find_edition(self,venue_id,venue,int(year),track,record,external_id,event_key) # 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 []) diff --git a/paper_atlas/ingestion/edition_tracks.py b/paper_atlas/ingestion/edition_tracks.py new file mode 100644 index 0000000..ce39584 --- /dev/null +++ b/paper_atlas/ingestion/edition_tracks.py @@ -0,0 +1,243 @@ +"""Snapshot-bound reclassification of existing attempts, without reimport or merge.""" +import argparse +from collections import Counter +from contextlib import closing +import hashlib +import json +from pathlib import Path +import re + +from paper_atlas.db.backend import database_arguments, open_read_connection, open_store, inserted_id +from paper_atlas.db.edition_scopes import KEY +from paper_atlas.ingestion.event_identity import database_identity, digest, values + +MARKER = 'edition_tracks_v1:' + + +def validate(inventory): + if (not isinstance(inventory, dict) or inventory.get('format') != 1 + or not isinstance(inventory.get('records'), list) or not 0 < len(inventory['records']) <= 10000): + raise ValueError('Nonempty bounded reviewed inventory required') + ids = set(); records = set() + for row in inventory['records']: + required = {'submission_id','source_record_id','external_id','source','doi','from_edition_id', + 'year','venue','venue_names','track','section','evidence_url','evidence_sha256'} + if not isinstance(row, dict) or set(row) != required: + raise ValueError('Unexpected inventory record fields') + for field in ('submission_id','source_record_id','from_edition_id','year'): + if type(row[field]) is not int or row[field] <= 0: + raise ValueError('Positive integer inventory identity required') + if row['submission_id'] in ids or row['source_record_id'] in records: + raise ValueError('Repeated submission/source identity in inventory') + ids.add(row['submission_id']); records.add(row['source_record_id']) + for field in ('external_id','source','doi','venue','section','evidence_url'): + if not isinstance(row[field], str) or not row[field].strip(): + raise ValueError('Missing inventory evidence field') + if (not isinstance(row['venue_names'], list) or not row['venue_names'] + or not all(isinstance(v,str) and v for v in row['venue_names']) + or row['venue'] not in row['venue_names'] + or not re.fullmatch(r'[a-z][a-z0-9_]{0,63}', row['track']) + or not re.fullmatch(r'[0-9a-f]{64}', row['evidence_sha256']) + or not row['evidence_url'].startswith('https://')): + raise ValueError('Invalid scope or evidence binding') + return inventory + + +def bounded(c, sql, params): + rows = values(c.execute(sql+' LIMIT 100001', params)) + if len(rows) > 100000: + raise ValueError('Protected scope exceeds bounded reclassification limit') + return rows + + +def protect(c, ids): + marks = ','.join('?' for _ in ids) + subs = 'SELECT id FROM submissions WHERE id IN ('+marks+')' + papers = 'SELECT paper_id FROM submissions WHERE id IN ('+marks+')' + decisions = 'SELECT id FROM decisions WHERE submission_id IN ('+subs+')' + queries = { + 'submissions': ('SELECT id,paper_id,source_record_id,source_submission_id,submitted_at FROM submissions WHERE id IN ('+marks+') ORDER BY id'), + 'submission_identifiers': ('SELECT scheme,value,submission_id FROM submission_identifiers WHERE submission_id IN ('+subs+') ORDER BY submission_id,scheme,value'), + 'submission_sources': ('SELECT * FROM submission_sources WHERE submission_id IN ('+subs+') ORDER BY submission_id,source_record_id,source_submission_id'), + 'decisions': ('SELECT * FROM decisions WHERE submission_id IN ('+subs+') ORDER BY id'), + 'decision_sources': ('SELECT * FROM decision_sources WHERE decision_id IN ('+decisions+') ORDER BY decision_id,source_record_id,source_decision_id'), + 'papers': ('SELECT * FROM papers WHERE id IN ('+papers+') ORDER BY id'), + 'identifiers': ('SELECT * FROM identifiers WHERE paper_id IN ('+papers+') ORDER BY scheme,value'), + 'paper_tags': ('SELECT * FROM paper_tags WHERE paper_id IN ('+papers+') ORDER BY paper_id,tag_id,evidence_key'), + } + from paper_atlas.curation.annotations import available + if available(c): + queries['annotation_events'] = 'SELECT * FROM annotation_events WHERE paper_id IN ('+papers+') ORDER BY id' + result = {table: {'rows':len(rows), 'sha256':digest(rows)} for table, sql in queries.items() + for rows in [bounded(c,sql,ids)]} + rows = bounded(c, 'SELECT r.* FROM source_records r WHERE r.id IN (SELECT source_record_id FROM submission_sources WHERE submission_id IN ('+subs+')) ORDER BY r.id', ids) + # Only the reviewed curation field may change; every other metadata value + # and all raw text/IDs/retrieval times remain protected. + for row in rows: + metadata = json.loads(row[8]); metadata.pop(KEY,None); row[8] = metadata + result['source_records'] = {'rows':len(rows), 'sha256':digest(rows)} + return result + + +def inspect(c, inventory): + validate(inventory); changes=[]; ids=tuple(sorted(r['submission_id']for r in inventory['records'])) + original = sorted({r['from_edition_id'] for r in inventory['records']}) + actual_ids = {r[0] for r in c.execute('SELECT id FROM submissions WHERE edition_id IN ('+','.join('?'for _ in original)+')', original)} + if actual_ids != set(ids): + raise ValueError('Inventory must cover every attempt in each original edition') + for row in inventory['records']: + current = c.execute('''SELECT s.edition_id,s.source_record_id,s.source_submission_id,s.paper_id, + e.venue_id,e.year,e.track,e.event_key,v.acronym,r.external_id,r.metadata_json,src.name + FROM submissions s JOIN editions e ON e.id=s.edition_id JOIN venues v ON v.id=e.venue_id + JOIN source_records r ON r.id=s.source_record_id JOIN sources src ON src.id=r.source_id WHERE s.id=?''',(row['submission_id'],)).fetchone() + if not current or [current[i]for i in (0,1,2,5,8,9,11)] != [row[k]for k in ('from_edition_id','source_record_id','external_id','year','venue','external_id','source')]: + raise ValueError('Submission identity/scope differs from inventory') + aliases={r[0]for r in c.execute('SELECT alias FROM venue_aliases WHERE venue_id=?',(current[4],))}|{current[8]} + if set(row['venue_names'])!=aliases: + raise ValueError('Venue names differ from current aliases') + metadata=json.loads(current[10]) + if KEY in metadata: + raise ValueError('Existing curated scope requires explicit reconciliation') + if (metadata.get('record_identifiers') or {}).get('doi') != row['doi']: + raise ValueError('Stable DOI differs from reviewed source evidence') + links=values(c.execute('SELECT source_record_id,source_submission_id FROM submission_sources WHERE submission_id=?',(row['submission_id'],))) + if links != [[row['source_record_id'],row['external_id']]]: + raise ValueError('Shared-source attempt requires separate reconciliation') + decisions=values(c.execute('SELECT status FROM decisions WHERE submission_id=?',(row['submission_id'],))) + if not decisions or any(d[0]!='accepted'for d in decisions): + raise ValueError('Only publication observations can use this track correction') + target=c.execute('SELECT id FROM editions WHERE venue_id=? AND year=? AND track=? AND event_key=?',(current[4],current[5],row['track'],current[7])).fetchone() + if target and target[0] != current[0]: + if c.execute('SELECT 1 FROM submissions WHERE edition_id=? LIMIT 1',(target[0],)).fetchone(): + raise ValueError('Nonempty target edition requires separate reconciliation') + if c.execute('SELECT 1 FROM submission_identifiers WHERE edition_id=? LIMIT 1',(target[0],)).fetchone(): + raise ValueError('Target identifier scope already exists') + if c.execute('SELECT 1 FROM edition_sources WHERE edition_id=? LIMIT 1',(target[0],)).fetchone(): + raise ValueError('Target coverage already exists; reconcile first') + if c.execute('SELECT 1 FROM submission_identifiers WHERE submission_id=? AND edition_id<>? LIMIT 1',(row['submission_id'],current[0])).fetchone(): + raise ValueError('Existing identifier edition is inconsistent') + changes.append({**row,'from_track':current[6],'event_key':current[7], 'venue_id':current[4], + 'target_edition_id':target[0]if target else None}) + coverage=bounded(c,'SELECT * FROM edition_sources WHERE edition_id IN ('+','.join('?'for _ in original)+') ORDER BY edition_id,source_id,list_url',original) + if Counter(row[0]for row in coverage)!=Counter({edition:1 for edition in original}): + raise ValueError('Exactly one original coverage listing per edition required') + return {'records':changes, 'protected':protect(c,ids), 'coverage':coverage, + 'editions':bounded(c,'SELECT * FROM editions WHERE id IN ('+','.join('?'for _ in original)+') ORDER BY id',original)} + + +def proposal(c, inventory, identity): + return {'format':1,'database':identity,'inventory':inventory,'before':inspect(c,inventory)} + + +def verify(c, plan): + records=plan['before']['records']; ids=tuple(sorted(r['submission_id']for r in records)) + if protect(c,ids)!=plan['before']['protected']: + raise ValueError('Protected history/text/tags/identities changed; roll back') + for row in records: + event=c.execute('''SELECT e.id,e.venue_id,e.year,e.track,e.event_key FROM submissions s JOIN editions e ON e.id=s.edition_id WHERE s.id=?''',(row['submission_id'],)).fetchone() + if not event or [event[i]for i in range(1,5)] != [row['venue_id'],row['year'],row['track'],row['event_key']]: + raise ValueError('Target track assignment failed') + if c.execute('SELECT 1 FROM submission_identifiers WHERE submission_id=? AND edition_id<>? LIMIT 1',(row['submission_id'],event[0])).fetchone(): + raise ValueError('Target identifier assignment failed') + if row['track'] != row['from_track']: + metadata=json.loads(c.execute('SELECT metadata_json FROM source_records WHERE id=?',(row['source_record_id'],)).fetchone()[0]) + if metadata.get(KEY) != [curation(row,event[0],digest(plan))]: + raise ValueError('Reviewed refresh protection missing or changed') + original=sorted({r['from_edition_id']for r in records}) + if bounded(c,'SELECT * FROM editions WHERE id IN ('+','.join('?'for _ in original)+') ORDER BY id',original)!=plan['before']['editions']: + raise ValueError('Original edition identities changed') + for old in plan['before']['coverage']: + edition,source,url,_,_,_,cursor,success,_=old + expected=Counter() + for row in records: + if row['from_edition_id']==edition: + target=c.execute('SELECT edition_id FROM submissions WHERE id=?',(row['submission_id'],)).fetchone()[0] + expected[target]+=1 + for target,number in expected.items(): + actual=c.execute('SELECT status,expected_count,imported_count,cursor,last_success_at,last_error FROM edition_sources WHERE edition_id=? AND source_id=? AND list_url=?',(target,source,url)).fetchone() + note='Track scope corrected from reviewed per-record evidence; original whole-volume coverage retained in plan '+digest(plan) + if actual is None or [actual[i]for i in range(6)]!=['partial',number,number,cursor,success,note]: + raise ValueError('Corrected coverage differs from reviewed scope') + return {'records_verified':len(records),'protected':plan['before']['protected']} + + +def curation(row, target, signature): + return {'external_id':row['external_id'], 'venue_names':sorted(set(row['venue_names'])), + 'from':{'edition_id':row['from_edition_id'],'year':row['year'],'track':row['from_track'],'event_key':row['event_key']}, + 'target':{'edition_id':target,'venue_id':row['venue_id'],'year':row['year'],'track':row['track'],'event_key':row['event_key']}, + 'section':row['section'],'evidence_url':row['evidence_url'],'evidence_sha256':row['evidence_sha256'],'plan_digest':signature} + + +def apply_plan(c, plan): + if plan.get('format') != 1: + raise ValueError('Unsupported plan format') + signature=digest(plan);marker=MARKER+signature + if c.execute('SELECT 1 FROM schema_migrations WHERE name=?',(marker,)).fetchone(): + return {'status':'already_applied','events_moved':0,**verify(c,plan)} + if inspect(c,plan['inventory']) != plan['before']: + raise ValueError('Stale track plan; create and review a fresh plan') + targets={}; counts=Counter(); moved=0 + for row in plan['before']['records']: + key=(row['venue_id'],row['year'],row['track'],row['event_key']) + if key not in targets: + found=c.execute('SELECT id FROM editions WHERE venue_id=? AND year=? AND track=? AND event_key=?',key).fetchone() + targets[key]=found[0]if found else inserted_id(c,'INSERT INTO editions(venue_id,year,track,event_key) VALUES(?,?,?,?)',key) + target=targets[key];counts[(row['from_edition_id'],target)]+=1 + if target != row['from_edition_id']: + c.execute('UPDATE submission_identifiers SET edition_id=? WHERE submission_id=?',(target,row['submission_id'])) + if c.execute('UPDATE submissions SET edition_id=? WHERE id=? AND edition_id=?',(target,row['submission_id'],row['from_edition_id'])).rowcount!=1: + raise ValueError('Concurrent submission change') + metadata=json.loads(c.execute('SELECT metadata_json FROM source_records WHERE id=?',(row['source_record_id'],)).fetchone()[0]);metadata[KEY]=[curation(row,target,signature)] + c.execute('UPDATE source_records SET metadata_json=? WHERE id=?',(json.dumps(metadata,ensure_ascii=False),row['source_record_id']));moved+=1 + # Preserve each original source URL and timestamps, but do not retain its + # old whole-volume count as main-track coverage. Partial is deliberate: + # this operation verifies scope, not abstracts or all possible sources. + for old in plan['before']['coverage']: + edition,source,url,status,expected,imported,cursor,success,error=old + own={r['submission_id'] for r in plan['before']['records']if r['from_edition_id']==edition} + source_name=c.execute('SELECT name FROM sources WHERE id=?',(source,)).fetchone()[0] + if {r['source']for r in plan['before']['records']if r['submission_id']in own}!={source_name}: + raise ValueError('Coverage includes a source outside the reviewed inventory') + for (origin,target),number in counts.items(): + if origin!=edition:continue + note='Track scope corrected from reviewed per-record evidence; original whole-volume coverage retained in plan '+signature + c.execute('''INSERT INTO edition_sources(edition_id,source_id,list_url,status,expected_count,imported_count,cursor,last_success_at,last_error) + VALUES(?,?,?,?,?,?,?,?,?) ON CONFLICT(edition_id,source_id,list_url) DO UPDATE SET status=excluded.status, + expected_count=excluded.expected_count,imported_count=excluded.imported_count,last_error=excluded.last_error''', + (target,source,url,'partial',number,number,cursor,success,note)) + result=verify(c,plan) + c.execute('INSERT INTO schema_migrations(name) VALUES(?)',(marker,)) + return {'status':'applied','events_moved':moved,**result} + + +def run(db,out,*,pg_service=None,pg_schema=None,inventory=None,plan=None,plan_sha256=None): + out=Path(out) + if out.exists() or not out.parent.is_dir():raise ValueError('New report path in existing directory required') + expected=None + if plan: + raw=Path(plan).read_bytes() + if hashlib.sha256(raw).hexdigest()!=plan_sha256:raise ValueError('Plan hash differs') + expected=json.loads(raw);validate(expected['inventory']) + elif plan_sha256:raise ValueError('Plan path required') + if bool(expected)==bool(inventory):raise ValueError('Choose inventory preflight or apply plan') + if expected is None: + with closing(open_read_connection(db,pg_service=pg_service,pg_schema=pg_schema))as c: + result=proposal(c,validate(json.loads(Path(inventory).read_bytes())),database_identity(c,db)) + else: + store=open_store(db,pg_service=pg_service,pg_schema=pg_schema) + with closing(store.conn)as c: + try: + if expected['database']!=database_identity(c,db):raise ValueError('Plan belongs to another database') + if not getattr(c,'postgres',False):c.commit();c.execute('BEGIN IMMEDIATE') + result=apply_plan(c,expected);c.commit() + except BaseException:c.rollback();raise + out.write_text(json.dumps(result,ensure_ascii=False,indent=2)+'\n');return result + + +def main(argv=None,*,prog=None): + p=argparse.ArgumentParser(prog=prog);database_arguments(p) + p.add_argument('--inventory');p.add_argument('--plan');p.add_argument('--plan-sha256');p.add_argument('--out',required=True) + a=p.parse_args(argv) + if bool(a.plan)!=bool(a.plan_sha256):p.error('--plan and --plan-sha256 must be supplied together') + r=run(a.db,a.out,pg_service=a.pg_service,pg_schema=a.pg_schema,inventory=a.inventory,plan=a.plan,plan_sha256=a.plan_sha256) + print(json.dumps({'status':r.get('status','planned'),'events_moved':r.get('events_moved',0),'out':a.out}),flush=True) diff --git a/paper_atlas/ingestion/records.py b/paper_atlas/ingestion/records.py index 63898ca..5b03604 100644 --- a/paper_atlas/ingestion/records.py +++ b/paper_atlas/ingestion/records.py @@ -134,6 +134,12 @@ def ingest_file(db,path, *, pg_service=None, pg_schema=None): if events and status is None:raise ValueError('Explicit decision event requires a status') if status is not None and status not in ('accepted','rejected','desk_rejected','withdrawn','under_review','unknown'):raise ValueError('Unsupported decision event') edition=store.decision(paper,record,row['venue'],row['year'],row['track'],external,event.get('evidence_url',row['url']),status,event.get('raw_value',''),row.get('venue_aliases'),identities=identities,observed_at=event.get('observed_at') or row_observed,decision_date=event.get('decision_date'),source_decision_id=event.get('source_decision_id',''),submitted_at=row.get('submitted_at'),guard_status=source_name in (PUBLIC_SOURCE,ARCHIVE_SOURCE),event_key=row.get('event_key')) + from paper_atlas.db.edition_scopes import KEY + scope_metadata=json.loads(c.execute('SELECT metadata_json FROM source_records WHERE id=?',(record,)).fetchone()[0]) + if scope_metadata.get(KEY): + actual_track=c.execute('SELECT track FROM editions WHERE id=?',(edition,)).fetchone()[0] + if row['track']!=actual_track: + raise ValueError('Recollect this whole-volume source with reviewed per-record tracks before refresh') editions.add(edition) else: raise ValueError('This importer requires an explicit supported decision status') diff --git a/tests/test_edition_tracks.py b/tests/test_edition_tracks.py new file mode 100644 index 0000000..8d10273 --- /dev/null +++ b/tests/test_edition_tracks.py @@ -0,0 +1,135 @@ +"""Stable-attempt correction, rollback and stale whole-volume refresh on both engines.""" +from contextlib import closing, redirect_stdout +import copy +import hashlib +import io +import json +import os +from pathlib import Path +import tempfile +import unittest +import uuid + +from paper_atlas.db.backend import open_store +from paper_atlas.db.edition_scopes import KEY +from paper_atlas.db.store import Store +from paper_atlas.ingestion.edition_tracks import proposal,apply_plan,verify,validate,run +from paper_atlas.ingestion.event_identity import database_identity,values +from paper_atlas.ingestion.records import ingest_file +from paper_atlas.postgres.client import connect +from paper_atlas.postgres.migration import migrate +from paper_atlas.postgres.schema import quote + + +def seed(db): + s=Store(db); rows=[] + for i,track in enumerate(['main','survey','doctoral_consortium','proceedings_unspecified'],1): + external='entry'+str(i);doi='10.1234/entry'+str(i) + p,r=s.record('fixture',external,'https://example.org/'+external,'Publication '+external, + 'Abstract with NUL\x00 and SOH\x01',[],{'doi':doi}) + eid=s.decision(p,r,'JOINT',2022,'main',external,'https://example.org/'+external,'accepted','published',identities=[('doi',doi)],observed_at='2022-01-01T00:00:00Z') + sid=s.conn.execute('SELECT id FROM submissions WHERE source_record_id=?',(r,)).fetchone()[0] + s.record_tags(p,r,'fixture',[{'namespace':'fixture:subject','value':'Original'}]) + rows.append(dict(submission_id=sid,source_record_id=r,external_id=external,source='fixture',doi=doi, + from_edition_id=eid,year=2022,venue='JOINT',venue_names=['JOINT'],track=track, + section=track,evidence_url='https://example.org/proceedings/2022/',evidence_sha256='a'*64)) + source=s.conn.execute("SELECT id FROM sources WHERE name='fixture'").fetchone()[0] + s.conn.execute("INSERT INTO edition_sources VALUES(?,?,?,'complete',4,4,'checkpoint','2022-01-01T00:00:00Z',NULL)",(eid,source,'https://example.org/volume')) + s.conn.commit();s.conn.close();return {'format':1,'records':rows} + + +class TrackCases: + def store(self):return open_store(**self.target) + + def plan(self,c):return proposal(c,self.inventory,database_identity(c,self.target.get('db'))) + + def test_move_preserves_all_history_tags_text_and_stable_identities(self): + with closing(self.store().conn)as c: + plan=self.plan(c);before=plan['before']['protected'];r=apply_plan(c,plan);c.commit() + self.assertEqual((r['events_moved'],r['records_verified']),(3,4)) + self.assertEqual(r['protected'],before) + self.assertEqual(apply_plan(c,plan)['status'],'already_applied') + self.assertEqual(values(c.execute('SELECT status,expected_count,imported_count FROM edition_sources ORDER BY edition_id')),[['partial',1,1]]*4) + self.assertEqual(c.execute('SELECT count(*) FROM submissions').fetchone()[0],4) + self.assertEqual(c.execute('SELECT count(*) FROM decisions').fetchone()[0],4) + self.assertEqual(c.execute('SELECT count(*) FROM paper_merges').fetchone()[0],0) + + def test_source_record_refresh_keeps_scope_and_original_attempt(self): + with closing(self.store().conn)as c: + plan=self.plan(c);apply_plan(c,plan);c.commit() + with closing(self.store().conn)as c: + # Use Store methods with the same writer connection as production. + s=object.__new__(Store);s.conn=c;s.event_identity=True + r=self.inventory['records'][1];metadata=json.loads(c.execute('SELECT metadata_json FROM source_records WHERE id=?',(r['source_record_id'],)).fetchone()[0]);saved=metadata[KEY] + p,record=s.record('fixture',r['external_id'],'https://example.org/new','Refreshed title',None,[],{'doi':r['doi']},{'new_source_field':True}) + eid=s.decision(p,record,'JOINT',2022,'main',r['external_id'],'https://example.org/entry2','accepted','published',identities=[('doi',r['doi'])],observed_at='2022-01-01T00:00:00Z') + self.assertEqual(c.execute('SELECT track FROM editions WHERE id=?',(eid,)).fetchone()[0],'survey') + self.assertEqual(c.execute('SELECT count(*) FROM submissions').fetchone()[0],4) + self.assertEqual(json.loads(c.execute('SELECT metadata_json FROM source_records WHERE id=?',(record,)).fetchone()[0])[KEY],saved) + with self.assertRaisesRegex(ValueError,'cannot supply'): + s.record('fixture',r['external_id'],'https://example.org/new','Forged',None,[],metadata={KEY:[]}) + c.rollback() + + def test_stale_snapshot_and_partial_inventory_are_rejected(self): + with closing(self.store().conn)as c: + plan=self.plan(c) + c.execute("UPDATE papers SET title='Changed' WHERE id=(SELECT paper_id FROM submissions WHERE id=?)",(self.inventory['records'][0]['submission_id'],)) + with self.assertRaisesRegex(ValueError,'Stale'):apply_plan(c,plan) + c.rollback() + bad=copy.deepcopy(self.inventory);bad['records'].pop() + with self.assertRaisesRegex(ValueError,'every attempt'):proposal(c,bad,database_identity(c,self.target.get('db'))) + bad=copy.deepcopy(self.inventory);bad['records'][0]['doi']='10.1234/wrong' + with self.assertRaisesRegex(ValueError,'DOI'):proposal(c,bad,database_identity(c,self.target.get('db'))) + + def test_failure_after_updates_rolls_back_all_scope_and_metadata_changes(self): + with closing(self.store().conn)as c: + plan=self.plan(c) + # A late coverage failure must roll back earlier moved attempts. + source=c.execute("SELECT id FROM sources WHERE name='fixture'").fetchone()[0] + c.execute("INSERT INTO sources(name,homepage) VALUES('other','https://example.org')") + other=c.execute("SELECT id FROM sources WHERE name='other'").fetchone()[0] + c.execute('UPDATE edition_sources SET source_id=? WHERE source_id=?',(other,source));c.commit() + plan=self.plan(c) + try: + with self.assertRaisesRegex(ValueError,'outside'):apply_plan(c,plan) + finally:c.rollback() + self.assertEqual(self.plan(c)['before'],plan['before']) + self.assertEqual(c.execute('SELECT count(*) FROM editions').fetchone()[0],1) + + def test_import_old_whole_volume_fails_without_corrupting_coverage(self): + with closing(self.store().conn)as c: + plan=self.plan(c);apply_plan(c,plan);c.commit() + r=self.inventory['records'][1];path=self.root/'old.jsonl' + row=dict(source='fixture',external_id=r['external_id'],url='https://example.org/entry2',title='Old title',authors=[],identifiers={'doi':r['doi']},venue='JOINT',year=2022,track='main',status='accepted') + path.write_text(json.dumps(row)+'\n');path.with_suffix('.manifest.json').write_text(json.dumps(dict(source='fixture',venue='JOINT',year=2022,track='main',count=1,list_url='https://example.org/volume',status='complete'))) + with self.assertRaisesRegex(ValueError,'Recollect'):ingest_file(self.target.get('db'),path,**{k:v for k,v in self.target.items()if k!='db'}) + with closing(self.store().conn)as c:verify(c,plan) + + def test_cli_hash_binding_preflight_and_duplicate_inventory(self): + inventory=self.root/'inventory.json';inventory.write_text(json.dumps(self.inventory)) + planfile=self.root/'plan.json';p=run(self.target.get('db'),planfile,inventory=inventory,**{k:v for k,v in self.target.items()if k!='db'}) + with self.assertRaisesRegex(ValueError,'hash'):run(self.target.get('db'),self.root/'bad.json',plan=planfile,plan_sha256='0'*64,**{k:v for k,v in self.target.items()if k!='db'}) + report=run(self.target.get('db'),self.root/'applied.json',plan=planfile,plan_sha256=hashlib.sha256(planfile.read_bytes()).hexdigest(),**{k:v for k,v in self.target.items()if k!='db'}) + self.assertEqual(report['events_moved'],3) + bad=copy.deepcopy(self.inventory);bad['records'].append(bad['records'][0]) + with self.assertRaisesRegex(ValueError,'Repeated'):validate(bad) + + +class SQLiteTrackTests(TrackCases,unittest.TestCase): + def setUp(self): + t=tempfile.TemporaryDirectory();self.addCleanup(t.cleanup);self.root=Path(t.name);db=self.root/'source.sqlite' + self.inventory=seed(db);self.target={'db':db} + + +@unittest.skipUnless(os.environ.get('PAPER_ATLAS_TEST_PG_SERVICE'),'Requires approved remote PostgreSQL service') +class PostgreSQLTrackTests(TrackCases,unittest.TestCase): + def setUp(self): + t=tempfile.TemporaryDirectory();self.addCleanup(t.cleanup);self.root=Path(t.name);db=self.root/'source.sqlite';self.inventory=seed(db) + self.service=os.environ['PAPER_ATLAS_TEST_PG_SERVICE'];self.schema='pa_tracks_'+uuid.uuid4().hex[:12];self.addCleanup(self.cleanup) + with redirect_stdout(io.StringIO()):migrate(db,self.service,self.root/'migration.json',schema=self.schema,min_free_gib=0) + self.target={'pg_service':self.service,'pg_schema':self.schema} + + 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.schema,)): + c.execute('DROP SCHEMA '+quote(name)+' CASCADE') -- 2.52.0 From d2c016e3096f700b15484f8cb5bcaaa46e531730 Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Sun, 8 Nov 2026 23:48:40 +0800 Subject: [PATCH 11/11] Reject target conflicts and post-application scope evidence drift --- paper_atlas/ingestion/edition_tracks.py | 4 +++- tests/test_edition_tracks.py | 24 ++++++++++++++++++++++++ 2 files changed, 27 insertions(+), 1 deletion(-) diff --git a/paper_atlas/ingestion/edition_tracks.py b/paper_atlas/ingestion/edition_tracks.py index ce39584..f714f3b 100644 --- a/paper_atlas/ingestion/edition_tracks.py +++ b/paper_atlas/ingestion/edition_tracks.py @@ -139,10 +139,12 @@ def verify(c, plan): raise ValueError('Target track assignment failed') if c.execute('SELECT 1 FROM submission_identifiers WHERE submission_id=? AND edition_id<>? LIMIT 1',(row['submission_id'],event[0])).fetchone(): raise ValueError('Target identifier assignment failed') + metadata=json.loads(c.execute('SELECT metadata_json FROM source_records WHERE id=?',(row['source_record_id'],)).fetchone()[0]) if row['track'] != row['from_track']: - metadata=json.loads(c.execute('SELECT metadata_json FROM source_records WHERE id=?',(row['source_record_id'],)).fetchone()[0]) if metadata.get(KEY) != [curation(row,event[0],digest(plan))]: raise ValueError('Reviewed refresh protection missing or changed') + elif KEY in metadata: + raise ValueError('Unexpected scope revision on unchanged track') original=sorted({r['from_edition_id']for r in records}) if bounded(c,'SELECT * FROM editions WHERE id IN ('+','.join('?'for _ in original)+') ORDER BY id',original)!=plan['before']['editions']: raise ValueError('Original edition identities changed') diff --git a/tests/test_edition_tracks.py b/tests/test_edition_tracks.py index 8d10273..9638e1c 100644 --- a/tests/test_edition_tracks.py +++ b/tests/test_edition_tracks.py @@ -105,6 +105,30 @@ class TrackCases: with self.assertRaisesRegex(ValueError,'Recollect'):ingest_file(self.target.get('db'),path,**{k:v for k,v in self.target.items()if k!='db'}) with closing(self.store().conn)as c:verify(c,plan) + def test_nonempty_target_and_alias_drift_fail_before_moving(self): + with closing(self.store().conn)as c: + c.execute("INSERT INTO venue_aliases VALUES((SELECT id FROM venues WHERE acronym='JOINT'),'ADDED')") + with self.assertRaisesRegex(ValueError,'aliases'):self.plan(c) + c.rollback() + s=object.__new__(Store);s.conn=c;s.event_identity=True + p,r=s.record('fixture','collision','https://example.org/collision','Distinct contribution',None,[],{'doi':'10.1234/collision'}) + s.decision(p,r,'JOINT',2022,'survey','collision','https://example.org/collision','accepted','published') + with self.assertRaisesRegex(ValueError,'Nonempty target'):self.plan(c) + c.rollback() + self.assertEqual(c.execute('SELECT count(*) FROM submissions').fetchone()[0],4) + + def test_repeated_plan_detects_coverage_or_scope_evidence_drift(self): + with closing(self.store().conn)as c: + plan=self.plan(c);apply_plan(c,plan);c.commit() + c.execute("UPDATE edition_sources SET imported_count=99") + with self.assertRaisesRegex(ValueError,'coverage'):apply_plan(c,plan) + c.rollback() + record=self.inventory['records'][0]['source_record_id'] + m=json.loads(c.execute('SELECT metadata_json FROM source_records WHERE id=?',(record,)).fetchone()[0]);m[KEY]=[] + c.execute('UPDATE source_records SET metadata_json=? WHERE id=?',(json.dumps(m),record)) + with self.assertRaisesRegex(ValueError,'unchanged track'):apply_plan(c,plan) + c.rollback() + def test_cli_hash_binding_preflight_and_duplicate_inventory(self): inventory=self.root/'inventory.json';inventory.write_text(json.dumps(self.inventory)) planfile=self.root/'plan.json';p=run(self.target.get('db'),planfile,inventory=inventory,**{k:v for k,v in self.target.items()if k!='db'}) -- 2.52.0